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

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)),
    )