Skip to content

ogs.dsl.base

Bases: Block

Abstract base for all open games — both atomic components and composites.

Every open game has a name and a signature describing its boundary ports using the (X, Y, R, S) convention. The signature keyword is accepted at construction and stored in the GDS interface field.

Source code in packages/gds-games/ogs/dsl/base.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class OpenGame(Block):
    """Abstract base for all open games — both atomic components and composites.

    Every open game has a ``name`` and a ``signature`` describing its boundary
    ports using the (X, Y, R, S) convention. The ``signature`` keyword is
    accepted at construction and stored in the GDS ``interface`` field.
    """

    @model_validator(mode="before")
    @classmethod
    def _accept_signature_kwarg(cls, data: dict) -> dict:
        """Accept 'signature' as an alias for 'interface'."""
        if isinstance(data, dict) and "signature" in data:
            data["interface"] = data.pop("signature")
        return data

    @property
    def signature(self) -> Signature:
        """Game-theory alias for self.interface with x/y/r/s accessors."""
        iface = self.interface
        if isinstance(iface, Signature):
            return iface
        return Signature(
            forward_in=iface.forward_in,
            forward_out=iface.forward_out,
            backward_in=iface.backward_in,
            backward_out=iface.backward_out,
        )

    @abstractmethod
    def flatten(self) -> list[AtomicGame]:  # type: ignore[override]
        """Return all atomic games in evaluation order."""

    def __rshift__(self, other: OpenGame) -> SequentialComposition:  # type: ignore[override]
        """``g1 >> g2`` — sequential composition."""
        from ogs.dsl.composition import SequentialComposition

        return SequentialComposition(
            name=f"{self.name} >> {other.name}",
            first=self,
            second=other,
        )

    def __or__(self, other: OpenGame) -> ParallelComposition:  # type: ignore[override]
        """``g1 | g2`` — parallel composition."""
        from ogs.dsl.composition import ParallelComposition

        return ParallelComposition(
            name=f"{self.name} | {other.name}",
            left=self,
            right=other,
        )

    def feedback(self, wiring: list[Flow]) -> FeedbackLoop:  # type: ignore[override]
        """Wrap with contravariant S→R feedback within a single timestep."""
        from ogs.dsl.composition import FeedbackLoop

        return FeedbackLoop(
            name=f"{self.name} [feedback]",
            inner=self,
            feedback_wiring=wiring,
        )

    def corecursive(
        self, wiring: list[Flow], exit_condition: str = ""
    ) -> CorecursiveLoop:
        """Wrap with covariant Y→X temporal iteration across timesteps."""
        from ogs.dsl.composition import CorecursiveLoop

        return CorecursiveLoop(
            name=f"{self.name} [corecursive]",
            inner=self,
            corecursive_wiring=wiring,
            exit_condition=exit_condition,
        )

signature property

Game-theory alias for self.interface with x/y/r/s accessors.

flatten() abstractmethod

Return all atomic games in evaluation order.

Source code in packages/gds-games/ogs/dsl/base.py 
62
63
64
@abstractmethod
def flatten(self) -> list[AtomicGame]:  # type: ignore[override]
    """Return all atomic games in evaluation order."""

__rshift__(other)

g1 >> g2 — sequential composition.

Source code in packages/gds-games/ogs/dsl/base.py 
66
67
68
69
70
71
72
73
74
def __rshift__(self, other: OpenGame) -> SequentialComposition:  # type: ignore[override]
    """``g1 >> g2`` — sequential composition."""
    from ogs.dsl.composition import SequentialComposition

    return SequentialComposition(
        name=f"{self.name} >> {other.name}",
        first=self,
        second=other,
    )

__or__(other)

g1 | g2 — parallel composition.

Source code in packages/gds-games/ogs/dsl/base.py 
76
77
78
79
80
81
82
83
84
def __or__(self, other: OpenGame) -> ParallelComposition:  # type: ignore[override]
    """``g1 | g2`` — parallel composition."""
    from ogs.dsl.composition import ParallelComposition

    return ParallelComposition(
        name=f"{self.name} | {other.name}",
        left=self,
        right=other,
    )

feedback(wiring)

Wrap with contravariant S→R feedback within a single timestep.

Source code in packages/gds-games/ogs/dsl/base.py 
86
87
88
89
90
91
92
93
94
def feedback(self, wiring: list[Flow]) -> FeedbackLoop:  # type: ignore[override]
    """Wrap with contravariant S→R feedback within a single timestep."""
    from ogs.dsl.composition import FeedbackLoop

    return FeedbackLoop(
        name=f"{self.name} [feedback]",
        inner=self,
        feedback_wiring=wiring,
    )

corecursive(wiring, exit_condition='')

Wrap with covariant Y→X temporal iteration across timesteps.

Source code in packages/gds-games/ogs/dsl/base.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def corecursive(
    self, wiring: list[Flow], exit_condition: str = ""
) -> CorecursiveLoop:
    """Wrap with covariant Y→X temporal iteration across timesteps."""
    from ogs.dsl.composition import CorecursiveLoop

    return CorecursiveLoop(
        name=f"{self.name} [corecursive]",
        inner=self,
        corecursive_wiring=wiring,
        exit_condition=exit_condition,
    )