Observations

observations

Native named observations at stored output times, without dense interpolation.

CompiledObservableLike

Bases: Protocol

Read-only structural compiler surface; core imports no compiler or provider.

coordinates property

Output axes and their coordinate labels.

definition property

Scientific name, units and description.

eval_fn property

Native flat-state evaluator.

measure_requirements property

Physical measures already applied by the evaluator.

output_shape property

Output dimensions excluding time.

parameter_shapes property

Direct parameters and complete native buffer shapes.

state_dependencies property

Direct state base dependencies.

time_dependent property

Whether the expression explicitly references time.

ObservationDiagnostics

Bases: NamedTuple

Describe evaluation coverage without inventing a numerical error estimate.

Accuracy inherits both the supplied trajectory and evaluator. No method has a qualified continuous extension through this interface yet. Spatial evaluation is owned by the callback; the compiler adapter selects declared nodes and applies declared measures without interpolation.

ObservationEvaluator

Bases: Protocol

Evaluate one native output from an instantaneous state and parameters.

__call__(t, y, **params)

Return one native output with the declared shape and state dtype.

Source code in src/op_engine/observations.py

def __call__(self, t: object, y: object, **params: object) -> object:
    """Return one native output with the declared shape and state dtype."""
    ...

ObservationMeasure

Bases: NamedTuple

Describe weights already applied by an evaluator, without applying them again.

ObservationMetadata

Bases: NamedTuple

Static scientific output metadata, separate from solver state layout.

Attributes:

Name	Type	Description
output_shape	tuple[int, ...]	Shape after evaluator reductions, excluding time.
coordinates	tuple[tuple[str, tuple[str, ...]], ...]	Ordered output axis names and coordinate labels.
units	str | None	Declared output units; no dimensional verification is implied.
state_dependencies	tuple[str, ...]	Direct state base dependencies.
parameter_shapes	tuple[tuple[str, tuple[int, ...]], ...]	Direct parameters and complete input shapes.
measure_requirements	tuple[tuple[str, ObservationMeasure], ...]	Named measures already applied by the evaluator.
time_dependent	bool	Whether the expression explicitly references time.
description	str | None	Optional scientific interpretation.

ObservationResult

Bases: NamedTuple

Static times, read-only named series mapping, and evaluation diagnostics.

ObservationSeries

Bases: NamedTuple

Native (n_output_times, *output_shape) values with their specification.

ObservationSpec

Bases: NamedTuple

A named evaluator and its immutable output description.

Custom evaluators receive each state in the trajectory's natural shape. Use :func:observation_from_compiled to flatten each state for an op_system compiled observable, preserving that compiler's state ordering.

observation_from_compiled(observable)

Adapt a compiled observable structurally, without a runtime compiler dependency.

The input trajectory must retain the compiler's flattened state order. The adapter reshapes each state to a vector and delegates all spatial reductions and coordinate selection to the compiler, applying no additional weights.

Parameters:

Name	Type	Description	Default
observable	CompiledObservableLike	Artifact exposing op_system's compiled-observable surface.	required

Returns:

Type	Description
ObservationSpec	Native evaluator and detached scientific metadata.

Source code in src/op_engine/observations.py

def observation_from_compiled(observable: CompiledObservableLike) -> ObservationSpec:
    """Adapt a compiled observable structurally, without a runtime compiler dependency.

    The input trajectory must retain the compiler's flattened state order. The
    adapter reshapes each state to a vector and delegates all spatial reductions
    and coordinate selection to the compiler, applying no additional weights.

    Args:
        observable: Artifact exposing op_system's compiled-observable surface.

    Returns:
        Native evaluator and detached scientific metadata.
    """
    evaluate = observable.eval_fn

    def eval_fn(t: object, y: object, **params: object) -> object:
        xp = array_namespace(y)
        return evaluate(t, xp.reshape(y, (-1,)), **params)

    return ObservationSpec(
        observable.definition.name,
        eval_fn,
        ObservationMetadata(
            tuple(observable.output_shape),
            tuple((axis, tuple(labels)) for axis, labels in observable.coordinates),
            observable.definition.units,
            tuple(observable.state_dependencies),
            tuple((name, tuple(shape)) for name, shape in observable.parameter_shapes),
            tuple(
                (
                    axis,
                    ObservationMeasure(
                        measure.kind, tuple(measure.weights), measure.units
                    ),
                )
                for axis, measure in observable.measure_requirements
            ),
            observable.time_dependent,
            observable.definition.description,
        ),
    )

observe_model(core, observations, *, params=None)

Evaluate a completed ModelCore's stored history using its output grid.

Parameters:

Name	Type	Description	Default
core	ModelCore	Completed state manager configured to store its full trajectory.	required
observations	Sequence[ObservationSpec]	Named evaluators and their output metadata.	required
params	Mapping[str, object] | None	Optional observation parameters.	None

Returns:

Type	Description
ObservationResult	Native named series at the model's stored output times.

Raises:

Type	Description
ValueError	If history is unavailable or the model has not completed.

Source code in src/op_engine/observations.py

def observe_model(
    core: ModelCore,
    observations: Sequence[ObservationSpec],
    *,
    params: Mapping[str, object] | None = None,
) -> ObservationResult:
    """Evaluate a completed ModelCore's stored history using its output grid.

    Args:
        core: Completed state manager configured to store its full trajectory.
        observations: Named evaluators and their output metadata.
        params: Optional observation parameters.

    Returns:
        Native named series at the model's stored output times.

    Raises:
        ValueError: If history is unavailable or the model has not completed.
    """
    if not core.store_history or core.state_array is None:
        msg = "observations require stored history; streaming is not available"
        raise ValueError(msg)
    if core.current_step != core.n_timesteps - 1:
        msg = "observations require a completed model trajectory"
        raise ValueError(msg)
    return observe_solution(
        core.time_grid, core.state_array, observations, params=params
    )

observe_solution(times, trajectory, observations, *, params=None)

Evaluate native named outputs at every supplied stored output time.

Time is a static, finite, strictly increasing host grid. Each callback must return an array in the trajectory namespace and dtype with its declared shape. Native values are stacked without conversion or gradient detachment. Inputs and returned arrays should be treated as immutable.

Parameters:

Name	Type	Description	Default
times	Sequence[float] | ndarray	Stored output times, independent of internal solver step times.	required
trajectory	Array	Native state history with a leading output-time dimension.	required
observations	Sequence[ObservationSpec]	Unique named evaluators with static output metadata.	required
params	Mapping[str, object] | None	Parameters passed to each evaluator, including observation-only ones.	None

Returns:

Type	Description
ObservationResult	Native series and evaluation coverage. No interpolation/error estimate.

Raises:

Type	Description
ValueError	If grids, trajectory shape, metadata or output shapes disagree.
TypeError	If states/outputs violate the real floating native-array contract.

Source code in src/op_engine/observations.py

def observe_solution(
    times: Sequence[float] | np.ndarray,
    trajectory: Array,
    observations: Sequence[ObservationSpec],
    *,
    params: Mapping[str, object] | None = None,
) -> ObservationResult:
    """Evaluate native named outputs at every supplied stored output time.

    Time is a static, finite, strictly increasing host grid. Each callback must
    return an array in the trajectory namespace and dtype with its declared
    shape. Native values are stacked without conversion or gradient detachment.
    Inputs and returned arrays should be treated as immutable.

    Args:
        times: Stored output times, independent of internal solver step times.
        trajectory: Native state history with a leading output-time dimension.
        observations: Unique named evaluators with static output metadata.
        params: Parameters passed to each evaluator, including observation-only ones.

    Returns:
        Native series and evaluation coverage. No interpolation/error estimate.

    Raises:
        ValueError: If grids, trajectory shape, metadata or output shapes disagree.
        TypeError: If states/outputs violate the real floating native-array contract.
    """
    grid = np.asarray(times)
    if (
        grid.ndim != 1
        or grid.size == 0
        or grid.dtype.kind not in "fiu"
        or not np.all(np.isfinite(grid))
        or not np.all(grid[1:] > grid[:-1])
    ):
        msg = "observation times must be a nonempty finite strictly increasing 1D grid"
        raise ValueError(msg)
    if not trajectory.shape or trajectory.shape[0] != grid.size:
        msg = "trajectory must have one leading state row per observation time"
        raise ValueError(msg)
    xp = array_namespace(trajectory)
    if not xp.isdtype(trajectory.dtype, "real floating"):
        msg = "observation trajectory must have a real floating dtype"
        raise TypeError(msg)
    _validate_specs(observations)
    static_times = tuple(float(time) for time in grid)
    states = cast("_Indexable", trajectory)
    series = {}
    for spec in observations:
        values = []
        for index, time in enumerate(static_times):
            value = spec.eval_fn(
                time, states[index], **({} if params is None else params)
            )
            if (
                array_namespace(value) is not xp
                or getattr(value, "dtype", None) != trajectory.dtype
            ):
                msg = (
                    f"observation {spec.name!r} must preserve trajectory "
                    "namespace and dtype"
                )
                raise TypeError(msg)
            if getattr(value, "shape", None) != spec.metadata.output_shape:
                msg = (
                    f"observation {spec.name!r} must have shape "
                    f"{spec.metadata.output_shape}"
                )
                raise ValueError(msg)
            values.append(value)
        series[spec.name] = ObservationSeries(
            spec, cast("Array", xp.stack(values, axis=0))
        )
    return ObservationResult(
        static_times,
        MappingProxyType(series),
        ObservationDiagnostics(len(observations) * len(static_times)),
    )