Skip to content

gds_sim

Public API -- top-level exports.

gds-sim: High-performance simulation engine for the GDS ecosystem.

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)

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)

Results

Columnar dict-of-lists result storage.

Pre-allocates capacity when the total row count is known, then fills via append(). Converts to pandas DataFrame or list-of-dicts on demand.

Source code in packages/gds-sim/gds_sim/results.py 
 14
 15
 16
 17
 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
 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
137
138
139
140
141
142
143
144
145
146
147
148
class Results:
    """Columnar dict-of-lists result storage.

    Pre-allocates capacity when the total row count is known,
    then fills via ``append()``. Converts to pandas DataFrame
    or list-of-dicts on demand.
    """

    __slots__ = ("_capacity", "_columns", "_size", "_state_keys")

    def __init__(self, state_keys: list[str], capacity: int = 0) -> None:
        self._state_keys = state_keys
        self._size = 0
        self._capacity = capacity

        # Build column storage: metadata + state variables
        self._columns: dict[str, list[Any]] = {}
        all_keys = list(_META_COLS) + state_keys
        if capacity > 0:
            for k in all_keys:
                self._columns[k] = [None] * capacity
        else:
            for k in all_keys:
                self._columns[k] = []

    # ------------------------------------------------------------------
    # Factory
    # ------------------------------------------------------------------

    @classmethod
    def preallocate(cls, sim: Simulation) -> Results:
        """Create a Results instance pre-allocated for the given simulation."""
        n_subsets = len(sim.model._param_subsets)
        n_blocks = len(sim.model.state_update_blocks)
        # Row 0 (initial state) + timesteps * substeps, per run per subset
        rows_per_run = 1 + sim.timesteps * max(n_blocks, 1)
        capacity = rows_per_run * sim.runs * n_subsets
        return cls(list(sim.model._state_keys), capacity)

    # ------------------------------------------------------------------
    # Append
    # ------------------------------------------------------------------

    def append(
        self,
        state: dict[str, Any],
        *,
        timestep: int,
        substep: int,
        run: int,
        subset: int,
    ) -> None:
        """Append a single row (state snapshot + metadata)."""
        cols = self._columns
        idx = self._size

        if self._capacity > 0 and idx < self._capacity:
            # Fast path: fill pre-allocated slots
            cols["timestep"][idx] = timestep
            cols["substep"][idx] = substep
            cols["run"][idx] = run
            cols["subset"][idx] = subset
            for k in self._state_keys:
                cols[k][idx] = state[k]
        else:
            # Fallback: dynamic append
            cols["timestep"].append(timestep)
            cols["substep"].append(substep)
            cols["run"].append(run)
            cols["subset"].append(subset)
            for k in self._state_keys:
                cols[k].append(state[k])

        self._size += 1

    # ------------------------------------------------------------------
    # Conversion
    # ------------------------------------------------------------------

    def to_dataframe(self) -> Any:
        """Convert to pandas DataFrame. Requires ``pandas`` installed."""
        try:
            import pandas as pd  # type: ignore[import-untyped]
        except ImportError as exc:  # pragma: no cover
            raise ImportError(
                "pandas is required for to_dataframe(). "
                "Install with: pip install gds-sim[pandas]"
            ) from exc

        data = self._trimmed_columns()
        return pd.DataFrame(data)

    def to_list(self) -> list[dict[str, Any]]:
        """Convert to list of row-dicts (cadCAD-compatible format)."""
        data = self._trimmed_columns()
        keys = list(data.keys())
        n = self._size
        return [{k: data[k][i] for k in keys} for i in range(n)]

    def _trimmed_columns(self) -> dict[str, list[Any]]:
        """Return columns trimmed to actual size (handles pre-allocation)."""
        if self._capacity > 0 and self._size < self._capacity:
            return {k: v[: self._size] for k, v in self._columns.items()}
        return self._columns

    # ------------------------------------------------------------------
    # Merge
    # ------------------------------------------------------------------

    @classmethod
    def merge(cls, results_list: list[Results]) -> Results:
        """Merge multiple Results into one."""
        if not results_list:
            return cls([])
        if len(results_list) == 1:
            return results_list[0]

        state_keys = results_list[0]._state_keys
        total = sum(r._size for r in results_list)
        merged = cls(state_keys, capacity=total)

        all_keys = list(_META_COLS) + state_keys
        offset = 0
        for r in results_list:
            trimmed = r._trimmed_columns()
            n = r._size
            for k in all_keys:
                merged._columns[k][offset : offset + n] = trimmed[k]
            offset += n

        merged._size = total
        return merged

    def __len__(self) -> int:
        return self._size

preallocate(sim) classmethod

Create a Results instance pre-allocated for the given simulation.

Source code in packages/gds-sim/gds_sim/results.py 
43
44
45
46
47
48
49
50
51
@classmethod
def preallocate(cls, sim: Simulation) -> Results:
    """Create a Results instance pre-allocated for the given simulation."""
    n_subsets = len(sim.model._param_subsets)
    n_blocks = len(sim.model.state_update_blocks)
    # Row 0 (initial state) + timesteps * substeps, per run per subset
    rows_per_run = 1 + sim.timesteps * max(n_blocks, 1)
    capacity = rows_per_run * sim.runs * n_subsets
    return cls(list(sim.model._state_keys), capacity)

append(state, *, timestep, substep, run, subset)

Append a single row (state snapshot + metadata).

Source code in packages/gds-sim/gds_sim/results.py 
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
def append(
    self,
    state: dict[str, Any],
    *,
    timestep: int,
    substep: int,
    run: int,
    subset: int,
) -> None:
    """Append a single row (state snapshot + metadata)."""
    cols = self._columns
    idx = self._size

    if self._capacity > 0 and idx < self._capacity:
        # Fast path: fill pre-allocated slots
        cols["timestep"][idx] = timestep
        cols["substep"][idx] = substep
        cols["run"][idx] = run
        cols["subset"][idx] = subset
        for k in self._state_keys:
            cols[k][idx] = state[k]
    else:
        # Fallback: dynamic append
        cols["timestep"].append(timestep)
        cols["substep"].append(substep)
        cols["run"].append(run)
        cols["subset"].append(subset)
        for k in self._state_keys:
            cols[k].append(state[k])

    self._size += 1

to_dataframe()

Convert to pandas DataFrame. Requires pandas installed.

Source code in packages/gds-sim/gds_sim/results.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def to_dataframe(self) -> Any:
    """Convert to pandas DataFrame. Requires ``pandas`` installed."""
    try:
        import pandas as pd  # type: ignore[import-untyped]
    except ImportError as exc:  # pragma: no cover
        raise ImportError(
            "pandas is required for to_dataframe(). "
            "Install with: pip install gds-sim[pandas]"
        ) from exc

    data = self._trimmed_columns()
    return pd.DataFrame(data)

to_list()

Convert to list of row-dicts (cadCAD-compatible format).

Source code in packages/gds-sim/gds_sim/results.py 
106
107
108
109
110
111
def to_list(self) -> list[dict[str, Any]]:
    """Convert to list of row-dicts (cadCAD-compatible format)."""
    data = self._trimmed_columns()
    keys = list(data.keys())
    n = self._size
    return [{k: data[k][i] for k in keys} for i in range(n)]

merge(results_list) classmethod

Merge multiple Results into one.

Source code in packages/gds-sim/gds_sim/results.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@classmethod
def merge(cls, results_list: list[Results]) -> Results:
    """Merge multiple Results into one."""
    if not results_list:
        return cls([])
    if len(results_list) == 1:
        return results_list[0]

    state_keys = results_list[0]._state_keys
    total = sum(r._size for r in results_list)
    merged = cls(state_keys, capacity=total)

    all_keys = list(_META_COLS) + state_keys
    offset = 0
    for r in results_list:
        trimmed = r._trimmed_columns()
        n = r._size
        for k in all_keys:
            merged._columns[k][offset : offset + n] = trimmed[k]
        offset += n

    merged._size = total
    return merged

Hooks

Bases: BaseModel

Lifecycle hooks for a simulation run.

Source code in packages/gds-sim/gds_sim/types.py 
56
57
58
59
60
61
62
63
class Hooks(BaseModel):
    """Lifecycle hooks for a simulation run."""

    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)

    before_run: BeforeRunHook | None = None
    after_run: AfterRunHook | None = None
    after_step: AfterStepHook | None = None

StateUpdateBlock

Bases: BaseModel

A partial state update block: policies produce signals, SUFs update state.

Source code in packages/gds-sim/gds_sim/types.py 
47
48
49
50
51
52
53
class StateUpdateBlock(BaseModel):
    """A partial state update block: policies produce signals, SUFs update state."""

    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)

    policies: dict[str, PolicyFn] = {}
    variables: dict[str, SUFn]