Skip to content

gds_sim.model

Model, Simulation, and Experiment configuration objects.

Model, Simulation, and Experiment configuration objects.

Model

Bases: BaseModel

A simulation model: initial state, update blocks, and parameter space.

Source code in packages/gds-sim/gds_sim/model.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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
72
73
74
75
76
77
78
79
80
class Model(BaseModel):
    """A simulation model: initial state, update blocks, and parameter space."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    initial_state: dict[str, Any]
    state_update_blocks: list[StateUpdateBlock]
    params: dict[str, list[Any]] = {}

    # Computed at validation time — not part of the public schema
    _param_subsets: list[Params]
    _state_keys: list[str]

    @model_validator(mode="before")
    @classmethod
    def _coerce_blocks(cls, data: Any) -> Any:
        """Allow passing plain dicts instead of StateUpdateBlock instances."""
        if isinstance(data, dict) and "state_update_blocks" in data:
            blocks = data["state_update_blocks"]
            data["state_update_blocks"] = [
                StateUpdateBlock(**b) if isinstance(b, dict) else b for b in blocks
            ]
        return data

    @model_validator(mode="after")
    def _validate_structure(self) -> Self:
        # 1. Cache state keys
        self._state_keys = list(self.initial_state.keys())
        state_key_set = set(self._state_keys)

        # 2. Verify all SUF keys exist in initial_state
        for i, block in enumerate(self.state_update_blocks):
            for var_key in block.variables:
                if var_key not in state_key_set:
                    msg = (
                        f"State update block {i} references variable "
                        f"'{var_key}' not found in initial_state. "
                        f"Available keys: {self._state_keys}"
                    )
                    raise ValueError(msg)

        # 3. Adapt cadCAD-style function signatures
        adapted_blocks: list[StateUpdateBlock] = []
        for block in self.state_update_blocks:
            new_policies = {k: adapt_policy(fn) for k, fn in block.policies.items()}
            new_variables = {k: adapt_suf(fn) for k, fn in block.variables.items()}
            adapted_blocks.append(
                StateUpdateBlock(policies=new_policies, variables=new_variables)
            )
        self.state_update_blocks = adapted_blocks

        # 4. Expand parameter sweep (cartesian product)
        if self.params:
            keys = list(self.params.keys())
            values = [self.params[k] for k in keys]
            self._param_subsets = [
                dict(zip(keys, combo, strict=True))
                for combo in itertools.product(*values)
            ]
        else:
            self._param_subsets = [{}]

        return self

Simulation

Bases: BaseModel

A runnable simulation: model + execution parameters.

Source code in packages/gds-sim/gds_sim/model.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class Simulation(BaseModel):
    """A runnable simulation: model + execution parameters."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    model: Model
    timesteps: int = 100
    runs: int = 1
    history: int | Literal["full"] | None = None
    hooks: Hooks = Hooks()

    def run(self) -> Any:
        """Execute this simulation and return Results."""
        from gds_sim.engine import execute_simulation

        return execute_simulation(self)

run()

Execute this simulation and return Results.

Source code in packages/gds-sim/gds_sim/model.py 
94
95
96
97
98
def run(self) -> Any:
    """Execute this simulation and return Results."""
    from gds_sim.engine import execute_simulation

    return execute_simulation(self)

Experiment

Bases: BaseModel

A collection of simulations, optionally run in parallel.

Source code in packages/gds-sim/gds_sim/model.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
class Experiment(BaseModel):
    """A collection of simulations, optionally run in parallel."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    simulations: list[Simulation]
    processes: int | None = None

    def run(self) -> Any:
        """Execute all simulations and return merged Results."""
        from gds_sim.parallel import execute_experiment

        return execute_experiment(self)

run()

Execute all simulations and return merged Results.

Source code in packages/gds-sim/gds_sim/model.py 
109
110
111
112
113
def run(self) -> Any:
    """Execute all simulations and return merged Results."""
    from gds_sim.parallel import execute_experiment

    return execute_experiment(self)