Skip to content

gds.canonical

Bases: BaseModel

Canonical projection of a GDSSpec to formal GDS structure.

Pure derivation — always computable, never authoritative. GDSSpec remains ground truth.

Source code in packages/gds-framework/gds/canonical.py 
32
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
class CanonicalGDS(BaseModel):
    """Canonical projection of a GDSSpec to formal GDS structure.

    Pure derivation — always computable, never authoritative.
    GDSSpec remains ground truth.
    """

    model_config = ConfigDict(frozen=True)

    # State space X: entity.variable entries
    state_variables: tuple[tuple[str, str], ...] = ()

    # Parameter space Θ
    parameter_schema: ParameterSchema = Field(default_factory=ParameterSchema)

    # Input space U: (block_name, port_name) from BoundaryAction forward_out
    input_ports: tuple[tuple[str, str], ...] = ()

    # Decision space D: (block_name, port_name) from Policy forward_out
    decision_ports: tuple[tuple[str, str], ...] = ()

    # Structural decomposition: block names by role
    boundary_blocks: tuple[str, ...] = ()
    control_blocks: tuple[str, ...] = ()
    policy_blocks: tuple[str, ...] = ()
    mechanism_blocks: tuple[str, ...] = ()

    # Mechanism update targets: (entity, variable) per mechanism
    update_map: tuple[tuple[str, tuple[tuple[str, str], ...]], ...] = ()

    @property
    def has_parameters(self) -> bool:
        """True if the system has any parameters."""
        return len(self.parameter_schema) > 0

    def formula(self) -> str:
        """Render as mathematical formula string."""
        if self.has_parameters:
            return "h_θ : X → X  (h = f_θ ∘ g_θ, θ ∈ Θ)"
        return "h : X → X  (h = f ∘ g)"

has_parameters property

True if the system has any parameters.

formula()

Render as mathematical formula string.

Source code in packages/gds-framework/gds/canonical.py 
67
68
69
70
71
def formula(self) -> str:
    """Render as mathematical formula string."""
    if self.has_parameters:
        return "h_θ : X → X  (h = f_θ ∘ g_θ, θ ∈ Θ)"
    return "h : X → X  (h = f ∘ g)"

Pure function: GDSSpec → CanonicalGDS.

Deterministic, stateless. Never mutates the spec.

Source code in packages/gds-framework/gds/canonical.py 
 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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def project_canonical(spec: GDSSpec) -> CanonicalGDS:
    """Pure function: GDSSpec → CanonicalGDS.

    Deterministic, stateless. Never mutates the spec.
    """
    # 1. State space X: all entity variables
    state_variables: list[tuple[str, str]] = []
    for entity in spec.entities.values():
        for var_name in entity.variables:
            state_variables.append((entity.name, var_name))

    # 2. Parameter space Θ
    parameter_schema = spec.parameter_schema

    # 3. Classify blocks by role
    boundary_blocks: list[str] = []
    control_blocks: list[str] = []
    policy_blocks: list[str] = []
    mechanism_blocks: list[str] = []

    for bname, block in spec.blocks.items():
        if isinstance(block, BoundaryAction):
            boundary_blocks.append(bname)
        elif isinstance(block, ControlAction):
            control_blocks.append(bname)
        elif isinstance(block, Policy):
            policy_blocks.append(bname)
        elif isinstance(block, Mechanism):
            mechanism_blocks.append(bname)

    # 4. Input space U: BoundaryAction forward_out ports
    input_ports: list[tuple[str, str]] = []
    for bname in boundary_blocks:
        block = spec.blocks[bname]
        for p in block.interface.forward_out:
            input_ports.append((bname, p.name))

    # 5. Decision space D: Policy forward_out ports
    decision_ports: list[tuple[str, str]] = []
    for bname in policy_blocks:
        block = spec.blocks[bname]
        for p in block.interface.forward_out:
            decision_ports.append((bname, p.name))

    # 6. Mechanism update targets
    update_map: list[tuple[str, tuple[tuple[str, str], ...]]] = []
    for bname in mechanism_blocks:
        block = spec.blocks[bname]
        if isinstance(block, Mechanism):
            updates = tuple(tuple(pair) for pair in block.updates)
            update_map.append((bname, updates))  # type: ignore[arg-type]

    return CanonicalGDS(
        state_variables=tuple(state_variables),
        parameter_schema=parameter_schema,
        input_ports=tuple(input_ports),
        decision_ports=tuple(decision_ports),
        boundary_blocks=tuple(boundary_blocks),
        control_blocks=tuple(control_blocks),
        policy_blocks=tuple(policy_blocks),
        mechanism_blocks=tuple(mechanism_blocks),
        update_map=tuple(update_map),
    )