Skip to content

gds_sim.results

Columnar result storage and conversion helpers.

Columnar result storage with optional DataFrame conversion.

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