Skip to content

Observation Reductions

observation_reductions

Single-pass native endpoint and sampled-time observation summaries.

ObservationSummary

Bases: NamedTuple

A sampled window and bounded-size named reductions, without a trajectory.

The window is exactly the first/last supplied time. There is no off-grid clipping, extrapolation or qualified method-specific dense output. The diagnostics report callback count without an independent error estimate.

ObservationSummaryValue

Bases: NamedTuple

Native summary arrays with the original scientific output specification.

Attributes:

Name Type Description
spec ObservationSpec

Instantaneous output shape, coordinates, units and interpretation.

initial Array

Output at the first supplied sample.

final Array

Output at the last supplied sample.

change Array

Final minus initial; appropriate for a cumulative counter.

integral Array

Composite-trapezoid time integral over the supplied samples. Its units are output units multiplied by the caller's time unit.

reduce_observations(samples, observations, *, params=None)

Reduce a finite native sample stream in one pass without storing a trajectory.

Samples define the exact window endpoints. Native initial/final values, changes and composite-trapezoid integrals preserve evaluator shapes and gradients. A single sample has zero duration, change and integral. Callback outputs are snapshotted so a producer may reuse its numerical state buffer.

The reducer retains a constant number of output arrays and one current state reference. Producer storage and reverse-mode computation graphs are separate: no constant-memory gradient guarantee is implied. There is no persistent checkpoint or partial-result recovery; an interrupted call must restart from a producer-owned replay/checkpoint. Callbacks must be pure.

Parameters:

Name Type Description Default
samples Iterable[tuple[float, Array]]

Finite, nonempty iterable of static times and native states. Times must increase strictly; state shape, namespace and real floating dtype stay fixed. Only the supplied times define quadrature samples.

required
observations Sequence[ObservationSpec]

Unique named evaluators and scientific output metadata.

required
params Mapping[str, object] | None

Observation parameters passed to every callback.

None

Returns:

Type Description
ObservationSummary

Bounded-size native reductions with window, sample count and diagnostics.

ObservationSummary

Integrals inherit trajectory and sampled-quadrature error; no dense-output

ObservationSummary

order, event reconstruction or separate error estimate is asserted.

Raises:

Type Description
ValueError

If the stream is empty, intervals are invalid or shapes change.

TypeError

If states or outputs violate the native namespace/dtype contract.

Source code in src/op_engine/observation_reductions.py
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def reduce_observations(
    samples: Iterable[tuple[float, Array]],
    observations: Sequence[ObservationSpec],
    *,
    params: Mapping[str, object] | None = None,
) -> ObservationSummary:
    """Reduce a finite native sample stream in one pass without storing a trajectory.

    Samples define the exact window endpoints. Native initial/final values,
    changes and composite-trapezoid integrals preserve evaluator shapes and
    gradients. A single sample has zero duration, change and integral. Callback
    outputs are snapshotted so a producer may reuse its numerical state buffer.

    The reducer retains a constant number of output arrays and one current
    state reference. Producer storage and reverse-mode computation graphs are
    separate: no constant-memory gradient guarantee is implied. There is no
    persistent checkpoint or partial-result recovery; an interrupted call must
    restart from a producer-owned replay/checkpoint. Callbacks must be pure.

    Args:
        samples: Finite, nonempty iterable of static times and native states.
            Times must increase strictly; state shape, namespace and real floating
            dtype stay fixed. Only the supplied times define quadrature samples.
        observations: Unique named evaluators and scientific output metadata.
        params: Observation parameters passed to every callback.

    Returns:
        Bounded-size native reductions with window, sample count and diagnostics.
        Integrals inherit trajectory and sampled-quadrature error; no dense-output
        order, event reconstruction or separate error estimate is asserted.

    Raises:
        ValueError: If the stream is empty, intervals are invalid or shapes change.
        TypeError: If states or outputs violate the native namespace/dtype contract.
    """
    selection = tuple(observations)
    iterator = iter(samples)
    try:
        time, state = next(iterator)
    except StopIteration as error:
        msg = "observation summary requires at least one sample"
        raise ValueError(msg) from error
    first = _sample(_static_time(time), state, selection, params)
    layout = _StateLayout(state.shape, state.dtype, array_namespace(state))
    values = _initial_values(first)
    last_time = first.times[0]
    count = 1
    for sample_time, state in iterator:
        time = _static_time(sample_time)
        duration = time - last_time
        if not np.isfinite(duration) or duration <= 0:
            msg = "summary sample times must increase with finite intervals"
            raise ValueError(msg)
        if state.shape != layout.shape:
            msg = "summary samples must preserve state shape"
            raise ValueError(msg)
        if (
            state.dtype != layout.dtype
            or array_namespace(state) is not layout.namespace
        ):
            msg = "summary samples must preserve state namespace and dtype"
            raise TypeError(msg)
        current = _sample(time, state, selection, params)
        values = _advance_values(values, current, duration)
        last_time = time
        count += 1
    return ObservationSummary(
        (first.times[0], last_time),
        count,
        MappingProxyType(values),
        ObservationDiagnostics(count * len(selection)),
    )