Skip to content

Reconstruction API

SnapshotCollector

numen.reconstruction.collector.SnapshotCollector

Reconstructs GenericWorld snapshots from solver output at arbitrary times.

Parameters:

Name Type Description Default
world any

The original GenericWorld instance used for the solve.

 required
spec CompiledSpec

The CompiledSpec produced by compile_spec(world).

 required
result SolveResult

The SolveResult from a backend solve() call.

 required
Source code in src/numen/reconstruction/collector.py 
 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
class SnapshotCollector:
    """Reconstructs GenericWorld snapshots from solver output at arbitrary times.

    Args:
        world:  The original ``GenericWorld`` instance used for the solve.
        spec:   The ``CompiledSpec`` produced by ``compile_spec(world)``.
        result: The ``SolveResult`` from a backend ``solve()`` call.
    """

    def __init__(self, world: any, spec: CompiledSpec, result: SolveResult):
        self.world = world
        self.spec = spec
        self.result = result

    def at(self, t: float) -> any:
        """Reconstruct a full world snapshot at simulation time ``t``.

        Uses binary search on ``result.t`` to find the nearest saved time point,
        then rebuilds the world with all compiled state fields updated.

        Args:
            t: Simulation time in seconds.

        Returns:
            A deep copy of the original world with state fields updated to
            their values at time ``t``.
        """
        return reconstruct_snapshot(self.world, self.spec, self.result, t)

    def at_times(self, times: list[float]) -> list[tuple[float, any]]:
        """Reconstruct world snapshots at a list of times.

        Args:
            times: List of simulation times in seconds.

        Returns:
            List of ``(t, snapshot)`` tuples in the same order as ``times``.
        """
        return [(t, self.at(t)) for t in times]

    def uniform(self, n: int = 100) -> list[tuple[float, any]]:
        """Reconstruct world snapshots at ``n`` uniformly-spaced times.

        Args:
            n: Number of snapshots. Default 100.

        Returns:
            List of ``(t, snapshot)`` tuples spanning the full solve interval.
        """
        times = np.linspace(self.result.t[0], self.result.t[-1], n)
        return self.at_times(times.tolist())

    def field_series(
        self,
        entity_id: str,
        component_kind: str,
        field_name: str,
    ) -> tuple[np.ndarray, np.ndarray]:
        """Extract a single field's time series directly from the result arrays.

        Args:
            entity_id:      Entity key (e.g. "osc").
            component_kind: Component kind string (e.g. "nl_oscillator").
            field_name:     Field name on the component (e.g. "position").

        Returns:
            (t, values) — faster than reconstructing full snapshots.
        """
        key = f"{entity_id}.{component_kind}.{field_name}"
        if key in self.spec.state_index_map:
            start, end = self.spec.state_index_map[key]
            values = self.result.x[start:end, :]
        elif key in self.spec.param_index_map:
            start, end = self.spec.param_index_map[key]
            p = np.array(self.spec.p)
            values = np.tile(p[start:end], (len(self.result.t), 1)).T
        else:
            raise KeyError(f"Field '{key}' not found in compiled spec")

        if values.shape[0] == 1:
            return self.result.t, values[0]
        return self.result.t, values

at(t)

Reconstruct a full world snapshot at simulation time t.

Uses binary search on result.t to find the nearest saved time point, then rebuilds the world with all compiled state fields updated.

Parameters:

Name Type Description Default
t float

Simulation time in seconds.

 required

Returns:

Type Description
any

A deep copy of the original world with state fields updated to
any

their values at time t.
Source code in src/numen/reconstruction/collector.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def at(self, t: float) -> any:
    """Reconstruct a full world snapshot at simulation time ``t``.

    Uses binary search on ``result.t`` to find the nearest saved time point,
    then rebuilds the world with all compiled state fields updated.

    Args:
        t: Simulation time in seconds.

    Returns:
        A deep copy of the original world with state fields updated to
        their values at time ``t``.
    """
    return reconstruct_snapshot(self.world, self.spec, self.result, t)

at_times(times)

Reconstruct world snapshots at a list of times.

Parameters:

Name Type Description Default
times list[float]

List of simulation times in seconds.

 required

Returns:

Type Description
list[tuple[float, any]]

List of (t, snapshot) tuples in the same order as times.
Source code in src/numen/reconstruction/collector.py 
63
64
65
66
67
68
69
70
71
72
def at_times(self, times: list[float]) -> list[tuple[float, any]]:
    """Reconstruct world snapshots at a list of times.

    Args:
        times: List of simulation times in seconds.

    Returns:
        List of ``(t, snapshot)`` tuples in the same order as ``times``.
    """
    return [(t, self.at(t)) for t in times]

field_series(entity_id, component_kind, field_name)

Extract a single field's time series directly from the result arrays.

Parameters:

Name Type Description Default
entity_id str

Entity key (e.g. "osc").

 required
component_kind str

Component kind string (e.g. "nl_oscillator").

 required
field_name str

Field name on the component (e.g. "position").

 required

Returns:

Type Description
tuple[ndarray, ndarray]

(t, values) — faster than reconstructing full snapshots.
Source code in src/numen/reconstruction/collector.py 
 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
def field_series(
    self,
    entity_id: str,
    component_kind: str,
    field_name: str,
) -> tuple[np.ndarray, np.ndarray]:
    """Extract a single field's time series directly from the result arrays.

    Args:
        entity_id:      Entity key (e.g. "osc").
        component_kind: Component kind string (e.g. "nl_oscillator").
        field_name:     Field name on the component (e.g. "position").

    Returns:
        (t, values) — faster than reconstructing full snapshots.
    """
    key = f"{entity_id}.{component_kind}.{field_name}"
    if key in self.spec.state_index_map:
        start, end = self.spec.state_index_map[key]
        values = self.result.x[start:end, :]
    elif key in self.spec.param_index_map:
        start, end = self.spec.param_index_map[key]
        p = np.array(self.spec.p)
        values = np.tile(p[start:end], (len(self.result.t), 1)).T
    else:
        raise KeyError(f"Field '{key}' not found in compiled spec")

    if values.shape[0] == 1:
        return self.result.t, values[0]
    return self.result.t, values

uniform(n=100)

Reconstruct world snapshots at n uniformly-spaced times.

Parameters:

Name Type Description Default
n int

Number of snapshots. Default 100.

 100

Returns:

Type Description
list[tuple[float, any]]

List of (t, snapshot) tuples spanning the full solve interval.
Source code in src/numen/reconstruction/collector.py 
74
75
76
77
78
79
80
81
82
83
84
def uniform(self, n: int = 100) -> list[tuple[float, any]]:
    """Reconstruct world snapshots at ``n`` uniformly-spaced times.

    Args:
        n: Number of snapshots. Default 100.

    Returns:
        List of ``(t, snapshot)`` tuples spanning the full solve interval.
    """
    times = np.linspace(self.result.t[0], self.result.t[-1], n)
    return self.at_times(times.tolist())

reconstruct_snapshot

numen.reconstruction.snapshot.reconstruct_snapshot(world, spec, result, t)

Rebuild a GenericWorld snapshot from solver output at time t.

Returns a deep copy of world with all compiled fields updated to their values at t. State keys use the full path entity_id.component_kind.field_name.

Source code in src/numen/reconstruction/snapshot.py 
13
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
def reconstruct_snapshot(world: any, spec: CompiledSpec, result: SolveResult, t: float) -> any:
    """Rebuild a GenericWorld snapshot from solver output at time t.

    Returns a deep copy of world with all compiled fields updated to their values at t.
    State keys use the full path ``entity_id.component_kind.field_name``.
    """
    idx = int(np.searchsorted(result.t, t))
    idx = min(idx, result.x.shape[1] - 1)
    x = result.x[:, idx]

    snapshot = world.model_copy(deep=True)

    for entity_id, comps_by_kind in snapshot.components.items():
        for kind, component in list(comps_by_kind.items()):
            updates: dict[str, any] = {}
            prefix = f"{entity_id}.{kind}."
            for key, (start, end) in spec.state_index_map.items():
                if not key.startswith(prefix):
                    continue
                field_name = key[len(prefix):]
                value = float(x[start]) if end - start == 1 else x[start:end].tolist()
                updates[field_name] = value
            if updates:
                snapshot.components[entity_id][kind] = component.model_copy(update=updates)

    return snapshot