Koopman MPC — Review-Only Convex Model Predictive Controller

actuation.koopman_mpc turns a fitted Koopman predictor into a convex model-predictive controller. Because the Koopman model (monitor.koopman_edmd) is linear, predictive control over it is a single convex quadratic programme — no nonlinear optimisation, no local minima. The controller is review-only: it returns a content-hashed proposal and never actuates; the first proposed input is the action a safety envelope (actuation.foundation_model_governor / actuation.control_barrier) admits, constrains, or rejects.

It is not a selectable live runtime.simulation.simulate() mode. The generic binding-spec simulator accepts only control_mode="supervisor_policy" and fails closed for koopman_mpc; the Koopman MPC remains in the offline dVOC damping and FMI co-simulation surfaces where the fitted predictor and plant boundary are explicit.

1. The condensed quadratic programme

Over a horizon H the lifted states are eliminated so the only decision variable is the input sequence U (Korda & Mezić 2018, eq. 24); the online cost is independent of the lift dimension N. The predicted outputs stack as

Y = Ψ ψ(x_k) + Θ U,        Ψ_i = C Aⁱ,        Θ_{i,j} = C A^{i-1-j} B (j < i),

and the controller minimises the tracking-and-effort cost

Σ_{i=1}^{H} (y_i − r)ᵀ Q (y_i − r) + u_{i-1}ᵀ R u_{i-1} + (y_H − r)ᵀ Q_f (y_H − r)

subject to actuator bounds u_min ≤ u_i ≤ u_max and optional move limits |u_i − u_{i-1}| ≤ Δ. Condensing gives min ½UᵀPU + qᵀU with P = 2(Θᵀ Q̄ Θ + R̄) and q = 2 Θᵀ Q̄ (Ψ ψ(x_k) − r̄).

The basic formulation penalises u (not u − u_eq), so it regulates to an equilibrium (oscillation damping) with no offset but tracks a non-equilibrium-input set-point with a small steady-state offset.

2. The QP layer

The quadratic programme is solved by actuation._qp, whose canonical path is a deterministic operator-splitting (ADMM) solver — the OSQP algorithm of Stellato et al. (2020), including the adaptive-ρ re-scaling that lets it converge on ill-conditioned predictive-control programmes. A review-only controller must produce a reproducible, content-hashable decision, so the deterministic floor is the default; the optional osqp C solver (the mpc extra) is held to the ADMM result by the parity gate (1e-5) and is never the silent default.

3. Python API

from scpn_phase_orchestrator.monitor.koopman_edmd import (
    KoopmanDictionary, fit_koopman_predictor,
)
from scpn_phase_orchestrator.actuation.koopman_mpc import (
    KoopmanMPCConfig, KoopmanMPCController,
)

predictor = fit_koopman_predictor(states, next_states, inputs, dictionary=...)
controller = KoopmanMPCController(
    predictor=predictor,
    config=KoopmanMPCConfig(horizon=20, input_lower=-1.0, input_upper=1.0),
)
decision = controller.solve(current_state)        # review-only proposal
action = decision.proposed_input                  # hand to the safety governor

KoopmanMPCController.solve returns a frozen KoopmanMPCDecision carrying the first proposed input, the full input plan, the predicted output trajectory, the objective, an OPTIMAL/MAX_ITER status, an active-bound flag, and the SHA-256 content_hash of the rounded payload.

4. Tested behaviour

• Oscillation damping — closed-loop regulation drives a lightly damped oscillatory plant from ‖x‖≈3.8 (uncontrolled) to ≈0.
• Set-point tracking — drives the state substantially toward a reachable equilibrium.
• Constraints — actuator bounds and move limits are satisfied; the QP reports OPTIMAL.
• Reproducibility — the same inputs yield the same content hash.
• QP parity — the ADMM floor matches osqp to 1e-5 on random programmes.
• Composition — the proposed input flows into the foundation-model governor.

5. Pipeline position

oscillation_modes / modal_participation (monitor) → koopman_edmd (model) → koopman_mpc (control, review-only) → foundation_model_governor / control_barrier (safety envelope) → prc_oscillation (assurance). The controller proposes; the envelope gates; nothing actuates without that review.

6. References

• Korda & Mezić 2018, Automatica 93, 149-160 (arXiv:1611.03537) — Koopman operator meets MPC.
• Stellato, Banjac, Goulart, Bemporad & Boyd 2020, Math. Program. Comput. 12, 637-672 (arXiv:1711.08013) — OSQP: an operator splitting solver for QPs.

7. API reference

koopman_mpc

A convex Koopman model-predictive controller for the dVOC oscillation pack.

A fitted Koopman predictor (monitor.koopman_edmd) supplies a linear model z_{k+1}=Az_k+Bu_k, y=Cz of an otherwise nonlinear system. Linear model predictive control over that lifted model is therefore a single convex quadratic programme, which this controller builds in condensed form (Korda & Mezić 2018, eq. 24): the lifted states are eliminated so the decision variable is the input sequence U alone, and the online cost is independent of the lift dimension N.

Over a horizon H the predicted outputs stack as Y = Ψ ψ(x_k) + Θ U with

Ψ_i = C Aⁱ,        Θ_{i,j} = C A^{i-1-j} B   (j < i),

and the controller minimises the tracking and effort cost

Σ_{i=1}^{H} (y_i − r)ᵀ Q (y_i − r) + u_{i-1}ᵀ R u_{i-1} + (y_H − r)ᵀ Q_f (y_H − r)

subject to actuator bounds u_min ≤ u_i ≤ u_max and optional move limits |u_i − u_{i-1}| ≤ Δ. The quadratic programme is solved by the deterministic ADMM floor of the QP layer so the decision is reproducible and content-hashable.

This controller is review-only: it returns a proposed input sequence sealed into a content-addressed :class:KoopmanMPCDecision; it never actuates. The first proposed input is the action a downstream safety envelope (actuation.foundation_model_governor / actuation.control_barrier) admits, constrains, or rejects before any hardware sees it.

References

• Korda & Mezić 2018, Automatica 93, 149-160 (arXiv:1611.03537) — linear predictors for nonlinear dynamical systems: Koopman operator meets MPC.

Classes

KoopmanMPCConfig dataclass

KoopmanMPCConfig(
    horizon: int,
    output_weight: float | FloatArray = 1.0,
    input_weight: float | FloatArray = 0.01,
    terminal_weight: float = 1.0,
    input_lower: float | FloatArray = -np.inf,
    input_upper: float | FloatArray = np.inf,
    move_limit: float | None = None,
)

Cost and constraint specification for the Koopman MPC.

Parameters

horizon : int The prediction horizon H (number of steps). output_weight : float | numpy.ndarray The output tracking weight Q (scalar or per-output diagonal). input_weight : float | numpy.ndarray The input effort weight R (scalar or per-input diagonal). terminal_weight : float A non-negative multiplier on Q for the terminal stage. input_lower, input_upper : float | numpy.ndarray The actuator bounds (scalar or per-input). move_limit : float | None An optional symmetric per-step move limit |u_i − u_{i-1}| ≤ Δ.

KoopmanMPCDecision dataclass

KoopmanMPCDecision(
    proposed_input: FloatArray,
    input_plan: FloatArray,
    predicted_outputs: FloatArray,
    objective: float,
    status: str,
    active_bounds: bool,
)

A review-only Koopman-MPC proposal, sealed by a content hash.

Parameters

proposed_input : numpy.ndarray The first input u_0 of the optimal sequence, shape (m,) — the action handed to the safety envelope. input_plan : numpy.ndarray The full optimal input sequence, shape (H, m). predicted_outputs : numpy.ndarray The predicted output trajectory y_1 … y_H, shape (H, n). objective : float The optimal quadratic-programme objective value. status : str "OPTIMAL" if the solver converged, otherwise "MAX_ITER". active_bounds : bool Whether any element of proposed_input sits on an actuator bound. content_hash : str The SHA-256 of the canonical, rounded decision payload.

KoopmanMPCController dataclass

KoopmanMPCController(
    predictor: KoopmanPredictor, config: KoopmanMPCConfig
)

A condensed convex Koopman model-predictive controller.

Parameters

predictor : KoopmanPredictor The fitted Koopman linear predictor supplying (A, B, C). config : KoopmanMPCConfig The cost and constraint specification.

Methods:

solve

solve(
    current_state: FloatArray,
    *,
    reference: FloatArray | None = None,
    previous_input: FloatArray | None = None,
) -> KoopmanMPCDecision

Solve the MPC programme and return a review-only proposal.

Parameters

current_state : numpy.ndarray The current physical state x_k of shape (n,). reference : numpy.ndarray | None The constant output set-point r of shape (n,); defaults to the origin (oscillation damping). previous_input : numpy.ndarray | None The previously applied input, required only when a move limit is configured.

Returns

KoopmanMPCDecision The sealed proposal.

Raises

ValueError If the shapes are inconsistent or a move limit is set without a previous input.

Source code in src/scpn_phase_orchestrator/actuation/koopman_mpc.py

def solve(
    self,
    current_state: FloatArray,
    *,
    reference: FloatArray | None = None,
    previous_input: FloatArray | None = None,
) -> KoopmanMPCDecision:
    """Solve the MPC programme and return a review-only proposal.

    Parameters
    ----------
    current_state : numpy.ndarray
        The current physical state ``x_k`` of shape ``(n,)``.
    reference : numpy.ndarray | None
        The constant output set-point ``r`` of shape ``(n,)``; defaults to
        the origin (oscillation damping).
    previous_input : numpy.ndarray | None
        The previously applied input, required only when a move limit is
        configured.

    Returns
    -------
    KoopmanMPCDecision
        The sealed proposal.

    Raises
    ------
    ValueError
        If the shapes are inconsistent or a move limit is set without a
        previous input.
    """
    horizon = self.config.horizon
    n_state = self.predictor.state_dim
    n_input = self.predictor.input_dim

    z0 = self.predictor.lift(current_state)
    target = (
        np.zeros(n_state, dtype=np.float64)
        if reference is None
        else _bound_vector(reference, name="reference", dim=n_state)
    )
    output_q = _diagonal_weight(
        self.config.output_weight, name="output_weight", dim=n_state
    )
    input_r = _diagonal_weight(
        self.config.input_weight, name="input_weight", dim=n_input
    )
    lower = _bound_vector(self.config.input_lower, name="input_lower", dim=n_input)
    upper = _bound_vector(self.config.input_upper, name="input_upper", dim=n_input)
    if np.any(upper < lower):
        raise ValueError("input_upper must not be below input_lower")

    psi, theta = _condensed_prediction(self.predictor, horizon)
    free_output = psi @ z0  # (H * n,)
    target_stack = np.tile(target, horizon)

    stage_q = np.tile(output_q, horizon)
    stage_q[(horizon - 1) * n_state :] *= self.config.terminal_weight
    stage_r = np.tile(input_r, horizon)

    # ½ Uᵀ P U + qᵀ U with P = 2(Θᵀ diag(Q) Θ + diag(R)).
    weighted_theta = theta * stage_q[:, None]
    hessian = 2.0 * (theta.T @ weighted_theta + np.diag(stage_r))
    hessian = 0.5 * (hessian + hessian.T)
    linear = 2.0 * theta.T @ (stage_q * (free_output - target_stack))

    constraint, con_lower, con_upper = _build_constraints(
        horizon, n_input, lower, upper, self.config.move_limit, previous_input
    )
    solution = solve_qp(hessian, linear, constraint, con_lower, con_upper)

    plan = solution.x.reshape(horizon, n_input)
    outputs = (free_output + theta @ solution.x).reshape(horizon, n_state)
    proposed = np.ascontiguousarray(plan[0], dtype=np.float64)
    on_bound = bool(
        np.any(np.isclose(proposed, lower) & np.isfinite(lower))
        or np.any(np.isclose(proposed, upper) & np.isfinite(upper))
    )
    return KoopmanMPCDecision(
        proposed_input=proposed,
        input_plan=np.ascontiguousarray(plan, dtype=np.float64),
        predicted_outputs=np.ascontiguousarray(outputs, dtype=np.float64),
        objective=solution.objective,
        status="OPTIMAL" if solution.converged else "MAX_ITER",
        active_bounds=on_bound,
    )

Functions:

8. Closed-loop oscillation-damping pipeline

runtime.dvoc_oscillation_damping closes the dVOC loop end to end: an underdamped oscillator rings down and the matrix-pencil estimator plus the NERC PRC screener flag its poorly-damped mode; an EDMD-with-control Koopman predictor is fitted and driven in closed loop by the Koopman MPC; the controlled ringdown is re-screened and the weakest mode is now better damped. The result carries both hash-sealed PRCOscillationEvidence records and exports one deterministic scpn_dvoc_oscillation_damping_audit_v1 record that binds the before/after evidence hashes, damping delta, terminal signal magnitudes, fitted-predictor residual, before/after mode_family_counts, and review_only_offline_no_live_actuation claim boundary. The spo koopman-mpc command runs this pipeline on a default grid oscillator and writes the same combined audit record when --output is supplied. The pipeline is review-only and offline — it performs no live actuation.

dvoc_oscillation_damping

Close the dVOC loop: detect a poorly-damped mode, damp it, prove it damped.

This integration wires the whole dVOC chain into one reviewable pipeline. An underdamped oscillator rings down; the matrix-pencil estimator (monitor.oscillation_modes) detects its electromechanical mode and the NERC PRC screener (assurance.prc_oscillation) flags it as poorly damped. An EDMD-with-control Koopman predictor (monitor.koopman_edmd) is then fitted from input-excited snapshots and driven in closed loop by the condensed Koopman MPC (actuation.koopman_mpc); the controlled ringdown is re-screened, and the weakest mode is now better damped. The result carries both hash-sealed PRC evidence records and their mode-family counts, so the damping improvement and inter-area/sub-synchronous review signal are auditable end to end.

The pipeline is review-only and offline: it operates on a caller-supplied discrete-time plant x_{k+1} = A x_k + B u_k and emits evidence; it performs no live actuation.

Classes

OscillationDampingResult dataclass

OscillationDampingResult(
    uncontrolled_signal: FloatArray,
    controlled_signal: FloatArray,
    uncontrolled_damping_ratio: float,
    controlled_damping_ratio: float,
    before_evidence: PRCOscillationEvidence,
    after_evidence: PRCOscillationEvidence,
    damping_improved: bool,
    fit_residual: float,
)

The before/after evidence of a closed-loop oscillation-damping run.

Parameters

uncontrolled_signal, controlled_signal : numpy.ndarray The observed ringdown coordinate without and with Koopman MPC. uncontrolled_damping_ratio, controlled_damping_ratio : float The weakest detected modal damping ratio before and after control. before_evidence, after_evidence : PRCOscillationEvidence The hash-sealed PRC screening records of the two ringdowns. damping_improved : bool Whether the controlled ringdown is better damped than the open-loop one. fit_residual : float Root-mean-square one-step residual of the fitted Koopman predictor.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return the hash-sealed before/after damping audit record.

Returns

dict[str, object] A JSON-safe deterministic record that binds the open-loop and closed-loop PRC evidence hashes, damping improvement, terminal signal magnitudes, fitted-predictor residual, and non-actuating claim boundary under a single content hash.

Source code in src/scpn_phase_orchestrator/runtime/dvoc_oscillation_damping.py

def to_audit_record(self) -> dict[str, object]:
    """Return the hash-sealed before/after damping audit record.

    Returns
    -------
    dict[str, object]
        A JSON-safe deterministic record that binds the open-loop and
        closed-loop PRC evidence hashes, damping improvement, terminal
        signal magnitudes, fitted-predictor residual, and non-actuating
        claim boundary under a single content hash.
    """
    payload = self._audit_payload()
    payload["content_hash"] = canonical_record_hash(payload)
    return payload

Functions:

underdamped_oscillator

underdamped_oscillator(
    *, frequency_hz: float, damping_ratio: float, dt: float
) -> tuple[FloatArray, FloatArray]

Build a discrete-time underdamped second-order oscillator with control.

The continuous plant is ẍ + 2ζω ẋ + ω² x = u with ω = 2π·f, written in state form [x, ẋ] and discretised by exact zero-order hold.

Parameters

frequency_hz : float Natural frequency f in hertz. damping_ratio : float Open-loop damping ratio ζ (a small value is poorly damped). dt : float Sampling interval.

Returns

tuple[numpy.ndarray, numpy.ndarray] The discrete state matrix A of shape (2, 2) and input matrix B of shape (2, 1).

Raises

ValueError If frequency_hz, dt are not positive or damping_ratio is negative.

Source code in src/scpn_phase_orchestrator/runtime/dvoc_oscillation_damping.py

def underdamped_oscillator(
    *, frequency_hz: float, damping_ratio: float, dt: float
) -> tuple[FloatArray, FloatArray]:
    """Build a discrete-time underdamped second-order oscillator with control.

    The continuous plant is ``ẍ + 2ζω ẋ + ω² x = u`` with ``ω = 2π·f``, written in
    state form ``[x, ẋ]`` and discretised by exact zero-order hold.

    Parameters
    ----------
    frequency_hz : float
        Natural frequency ``f`` in hertz.
    damping_ratio : float
        Open-loop damping ratio ``ζ`` (a small value is poorly damped).
    dt : float
        Sampling interval.

    Returns
    -------
    tuple[numpy.ndarray, numpy.ndarray]
        The discrete state matrix ``A`` of shape ``(2, 2)`` and input matrix
        ``B`` of shape ``(2, 1)``.

    Raises
    ------
    ValueError
        If ``frequency_hz``, ``dt`` are not positive or ``damping_ratio`` is
        negative.
    """
    if frequency_hz <= 0.0 or dt <= 0.0:
        raise ValueError("frequency_hz and dt must be positive")
    if damping_ratio < 0.0:
        raise ValueError("damping_ratio must be non-negative")
    omega = 2.0 * np.pi * frequency_hz
    cont_state = np.array([[0.0, 1.0], [-(omega**2), -2.0 * damping_ratio * omega]])
    cont_input = np.array([[0.0], [1.0]])
    # Exact zero-order-hold discretisation via the augmented matrix exponential.
    augmented = np.zeros((3, 3))
    augmented[:2, :2] = cont_state
    augmented[:2, 2:] = cont_input
    discrete = expm(augmented * dt)
    state_matrix = np.ascontiguousarray(discrete[:2, :2], dtype=np.float64)
    input_matrix = np.ascontiguousarray(discrete[:2, 2:], dtype=np.float64)
    return state_matrix, input_matrix

damp_oscillation

damp_oscillation(
    state_matrix: FloatArray,
    input_matrix: FloatArray,
    *,
    initial_state: FloatArray,
    horizon: int,
    fs: float,
    captured_at: str,
    config: KoopmanMPCConfig | None = None,
    event_prefix: str = "dvoc-damping",
    training_scale: float = 1.0,
    training_samples: int = 400,
    seed: int = 0,
) -> OscillationDampingResult

Damp a plant's oscillation with Koopman MPC and prove it with PRC evidence.

Parameters

state_matrix, input_matrix : numpy.ndarray The discrete-time plant A (n, n) and B (n, m). initial_state : numpy.ndarray The perturbed initial state x_0 of shape (n,). horizon : int Number of ringdown steps to simulate for each pass. fs : float The sampling rate in hertz used by the mode estimator and PRC screen. captured_at : str ISO-8601 capture timestamp stamped into the evidence records. config : KoopmanMPCConfig | None The MPC configuration; a damping-oriented default is used if omitted. event_prefix : str Prefix for the two PRC evidence event identifiers. training_scale : float Standard deviation of the random snapshots used to fit the predictor. training_samples : int Number of input-excited snapshots used to fit the predictor. seed : int Seed for the snapshot sampler.

Returns

OscillationDampingResult The before/after signals, weakest damping ratios, both PRC evidence records, the improvement flag, and the predictor fit residual.

Raises

ValueError If the plant, initial state, or horizon are inconsistent.

Source code in src/scpn_phase_orchestrator/runtime/dvoc_oscillation_damping.py

def damp_oscillation(
    state_matrix: FloatArray,
    input_matrix: FloatArray,
    *,
    initial_state: FloatArray,
    horizon: int,
    fs: float,
    captured_at: str,
    config: KoopmanMPCConfig | None = None,
    event_prefix: str = "dvoc-damping",
    training_scale: float = 1.0,
    training_samples: int = 400,
    seed: int = 0,
) -> OscillationDampingResult:
    """Damp a plant's oscillation with Koopman MPC and prove it with PRC evidence.

    Parameters
    ----------
    state_matrix, input_matrix : numpy.ndarray
        The discrete-time plant ``A`` ``(n, n)`` and ``B`` ``(n, m)``.
    initial_state : numpy.ndarray
        The perturbed initial state ``x_0`` of shape ``(n,)``.
    horizon : int
        Number of ringdown steps to simulate for each pass.
    fs : float
        The sampling rate in hertz used by the mode estimator and PRC screen.
    captured_at : str
        ISO-8601 capture timestamp stamped into the evidence records.
    config : KoopmanMPCConfig | None
        The MPC configuration; a damping-oriented default is used if omitted.
    event_prefix : str
        Prefix for the two PRC evidence event identifiers.
    training_scale : float
        Standard deviation of the random snapshots used to fit the predictor.
    training_samples : int
        Number of input-excited snapshots used to fit the predictor.
    seed : int
        Seed for the snapshot sampler.

    Returns
    -------
    OscillationDampingResult
        The before/after signals, weakest damping ratios, both PRC evidence
        records, the improvement flag, and the predictor fit residual.

    Raises
    ------
    ValueError
        If the plant, initial state, or horizon are inconsistent.
    """
    state = np.ascontiguousarray(np.asarray(initial_state, dtype=np.float64).ravel())
    if state.shape[0] != state_matrix.shape[0]:
        raise ValueError("initial_state length must match the state dimension")
    if horizon < 1:
        raise ValueError("horizon must be at least 1")

    uncontrolled = _open_loop_ringdown(state_matrix, state, horizon)
    before_modes = estimate_oscillation_modes(uncontrolled, fs)
    before_evidence = screen_oscillation_modes(
        before_modes,
        event_id=f"{event_prefix}-open-loop",
        captured_at=captured_at,
        signal_source="koopman-mpc/open-loop-ringdown",
        sampling_rate_hz=fs,
    )

    predictor = _fit_plant_koopman(
        state_matrix,
        input_matrix,
        scale=training_scale,
        samples=training_samples,
        seed=seed,
    )
    mpc_config = config if config is not None else _default_damping_config(horizon)
    controller = KoopmanMPCController(predictor, mpc_config)
    controlled = _closed_loop_ringdown(
        state_matrix, input_matrix, controller, state, horizon
    )
    after_modes = estimate_oscillation_modes(controlled, fs)
    after_evidence = screen_oscillation_modes(
        after_modes,
        event_id=f"{event_prefix}-closed-loop",
        captured_at=captured_at,
        signal_source="koopman-mpc/closed-loop-ringdown",
        sampling_rate_hz=fs,
    )

    before_damping = _weakest_damping(uncontrolled, fs)
    after_damping = _weakest_damping(controlled, fs)
    return OscillationDampingResult(
        uncontrolled_signal=uncontrolled,
        controlled_signal=controlled,
        uncontrolled_damping_ratio=before_damping,
        controlled_damping_ratio=after_damping,
        before_evidence=before_evidence,
        after_evidence=after_evidence,
        damping_improved=after_damping > before_damping,
        fit_residual=predictor.fit_residual,
    )

9. IEEE PMU concentrator adapter

runtime.pmu_ieee_adapter adapts the wide, multi-header CSV that phasor measurement concentrators and the oscillation-detection literature export into the two-column series runtime.pmu_ringdown consumes. read_ieee_pmu_recording locates the header block by its quantity-type row (T for time, F for frequency), enumerates the frequency channels with their exact-zero dropout and non-finite counts, and — when a unit row is present — confirms each frequency channel is reported in hertz. IEEEPMURecording.select_cleanest_channel returns the channel that is free of dropouts and within a plausible band of the nominal frequency, breaking ties toward the largest peak-to-peak swing, which carries the most oscillation content. write_ingester_csv writes that channel with Python's shortest round-tripping decimal and returns an AdaptedIngesterCSV provenance record whose SHA-256 digests link the derived CSV back to the source capture; adapt_ieee_pmu_csv runs the read, selection, and write in one call. The spo pmu-ieee-adapt command exposes this path, and the derived CSV feeds spo pmu-ringdown directly. This is a format-conversion path only; it never fits a plant model and never actuates.

pmu_ieee_adapter

Adapt an IEEE-format multi-header PMU concentrator CSV into ingester input.

Phasor-measurement concentrators and the oscillation-detection literature export captures in a wide, multi-header layout: a channel-label row, a quantity-type row (T for time, F for frequency, VM/VA/IM/IA for the phasor channels), a unit row, and a secondary-label row, followed by numeric samples with one time column and five channels per phasor-measurement unit. The ringdown screener consumes a two-column time_s,frequency_hz series instead. This module bridges the two: it parses the multi-header layout, enumerates the frequency channels with their dropout counts, selects the channel that is free of dropouts and sits within a plausible band of the nominal grid frequency (breaking ties toward the largest peak-to-peak swing, which carries the most oscillation content), and writes the selected channel as the screener's input with a hashed provenance record linking the derived CSV back to the source capture.

Classes

PMUFrequencyChannel dataclass

PMUFrequencyChannel(
    label: str,
    column_index: int,
    samples: FloatArray,
    zero_count: int,
    nonfinite_count: int,
    mean_hz: float,
    min_hz: float,
    max_hz: float,
)

One frequency channel extracted from an IEEE-format PMU capture.

Attributes

label : str Channel label from the header's label row (typically a substation and line identifier shared by the phasor unit's five channels). column_index : int Zero-based column index of the channel in the source CSV. samples : FloatArray Frequency samples in hertz, aligned with the recording's time vector. zero_count : int Number of exact-zero samples, the concentrator's dropout marker. nonfinite_count : int Number of non-finite samples (NaN or infinity). mean_hz : float Mean of the finite samples in hertz, or NaN if none are finite. min_hz : float Minimum finite sample in hertz, or NaN if none are finite. max_hz : float Maximum finite sample in hertz, or NaN if none are finite.

Attributes

peak_to_peak_hz property

peak_to_peak_hz: float

Return the finite peak-to-peak swing in hertz, or NaN if empty.

identifier property

identifier: str

Return a label and column identifier unique within the recording.

is_clean property

is_clean: bool

Return whether the channel is free of dropout and non-finite samples.

Methods:

is_within_band

is_within_band(
    nominal_frequency_hz: float, band_hz: float
) -> bool

Return whether the finite mean sits within band_hz of nominal.

Parameters

nominal_frequency_hz : float Nominal grid frequency the channel is expected to hover around. band_hz : float Half-width in hertz of the accepted band about the nominal frequency.

Returns

bool True when the finite mean is within the band, False when it is outside the band or no samples are finite.

Source code in src/scpn_phase_orchestrator/runtime/pmu_ieee_adapter.py

def is_within_band(self, nominal_frequency_hz: float, band_hz: float) -> bool:
    """Return whether the finite mean sits within ``band_hz`` of nominal.

    Parameters
    ----------
    nominal_frequency_hz : float
        Nominal grid frequency the channel is expected to hover around.
    band_hz : float
        Half-width in hertz of the accepted band about the nominal frequency.

    Returns
    -------
    bool
        ``True`` when the finite mean is within the band, ``False`` when it
        is outside the band or no samples are finite.
    """
    if not np.isfinite(self.mean_hz):
        return False
    return abs(self.mean_hz - nominal_frequency_hz) <= band_hz

IEEEPMURecording dataclass

IEEEPMURecording(
    source_name: str,
    source_sha256: str,
    times: FloatArray,
    channels: tuple[PMUFrequencyChannel, ...],
)

A parsed IEEE-format multi-header PMU capture.

Attributes

source_name : str Basename of the parsed source CSV. source_sha256 : str SHA-256 digest of the exact source CSV bytes. times : FloatArray Capture time vector in seconds shared by every channel. channels : tuple[PMUFrequencyChannel, ...] Frequency channels in source-column order.

Methods:

select_cleanest_channel

select_cleanest_channel(
    *,
    nominal_frequency_hz: float = 60.0,
    plausible_band_hz: float = 2.0,
) -> PMUFrequencyChannel

Return the dropout-free in-band channel with the largest swing.

A channel qualifies when it carries no dropout or non-finite samples and its mean sits within plausible_band_hz of the nominal frequency, which rejects dead channels reading zero and channels reported against a different nominal. Among the qualifying channels the one with the largest peak-to-peak swing is chosen, since it carries the most oscillation content for ringdown screening; ties break toward the lowest column index for determinism.

Parameters

nominal_frequency_hz : float Nominal grid frequency the channel is expected to hover around. plausible_band_hz : float Half-width in hertz of the band about the nominal frequency within which a channel mean is accepted.

Returns

PMUFrequencyChannel The selected frequency channel.

Raises

ValueError If the controls are invalid or no channel qualifies.

Source code in src/scpn_phase_orchestrator/runtime/pmu_ieee_adapter.py

def select_cleanest_channel(
    self,
    *,
    nominal_frequency_hz: float = 60.0,
    plausible_band_hz: float = 2.0,
) -> PMUFrequencyChannel:
    """Return the dropout-free in-band channel with the largest swing.

    A channel qualifies when it carries no dropout or non-finite samples and
    its mean sits within ``plausible_band_hz`` of the nominal frequency,
    which rejects dead channels reading zero and channels reported against a
    different nominal. Among the qualifying channels the one with the largest
    peak-to-peak swing is chosen, since it carries the most oscillation
    content for ringdown screening; ties break toward the lowest column index
    for determinism.

    Parameters
    ----------
    nominal_frequency_hz : float
        Nominal grid frequency the channel is expected to hover around.
    plausible_band_hz : float
        Half-width in hertz of the band about the nominal frequency within
        which a channel mean is accepted.

    Returns
    -------
    PMUFrequencyChannel
        The selected frequency channel.

    Raises
    ------
    ValueError
        If the controls are invalid or no channel qualifies.
    """
    nominal = _positive_float(nominal_frequency_hz, "nominal_frequency_hz")
    band = _positive_float(plausible_band_hz, "plausible_band_hz")
    qualifying = [
        channel
        for channel in self.channels
        if channel.is_clean and channel.is_within_band(nominal, band)
    ]
    if not qualifying:
        dropout = sum(1 for channel in self.channels if not channel.is_clean)
        raise ValueError(
            f"no frequency channel within {band} Hz of {nominal} Hz among "
            f"{len(self.channels)} channels ({dropout} carried dropouts)"
        )
    return max(
        qualifying,
        key=lambda channel: (channel.peak_to_peak_hz, -channel.column_index),
    )

AdaptedIngesterCSV dataclass

AdaptedIngesterCSV(
    source_name: str,
    source_sha256: str,
    output_name: str,
    output_sha256: str,
    channel_label: str,
    channel_column_index: int,
    time_column: str,
    frequency_column: str,
    row_count: int,
)

Provenance of a screener-ready CSV derived from an IEEE PMU capture.

Attributes

source_name : str Basename of the source IEEE PMU CSV. source_sha256 : str SHA-256 digest of the source CSV bytes. output_name : str Basename of the written ingester CSV. output_sha256 : str SHA-256 digest of the written ingester CSV bytes. channel_label : str Label of the selected frequency channel. channel_column_index : int Source-column index of the selected frequency channel. time_column : str Timestamp column name written to the ingester CSV. frequency_column : str Frequency column name written to the ingester CSV. row_count : int Number of sample rows written.

Functions:

read_ieee_pmu_recording

read_ieee_pmu_recording(
    path: str | Path,
) -> IEEEPMURecording

Parse an IEEE-format multi-header PMU CSV into a recording.

The header block is located by the quantity-type row — the first row whose first cell is the time token T — with the label row immediately above it and the data rows starting at the first row whose first cell parses as a number. When a unit row is present it is cross-checked so that every parsed frequency channel is reported in hertz.

Parameters

path : str | pathlib.Path Path to the IEEE-format PMU concentrator CSV.

Returns

IEEEPMURecording The time vector and the frequency channels with their dropout counts.

Raises

ValueError If the header block, time column, frequency channels, or numeric samples cannot be parsed.

Source code in src/scpn_phase_orchestrator/runtime/pmu_ieee_adapter.py

def read_ieee_pmu_recording(path: str | Path) -> IEEEPMURecording:
    """Parse an IEEE-format multi-header PMU CSV into a recording.

    The header block is located by the quantity-type row — the first row whose
    first cell is the time token ``T`` — with the label row immediately above it
    and the data rows starting at the first row whose first cell parses as a
    number. When a unit row is present it is cross-checked so that every parsed
    frequency channel is reported in hertz.

    Parameters
    ----------
    path : str | pathlib.Path
        Path to the IEEE-format PMU concentrator CSV.

    Returns
    -------
    IEEEPMURecording
        The time vector and the frequency channels with their dropout counts.

    Raises
    ------
    ValueError
        If the header block, time column, frequency channels, or numeric samples
        cannot be parsed.
    """
    csv_path = Path(path)
    source_bytes = csv_path.read_bytes()
    rows = list(csv.reader(io.StringIO(source_bytes.decode("utf-8"))))

    quantity_index = _quantity_row_index(rows)
    quantity = rows[quantity_index]
    labels = rows[quantity_index - 1]
    data_start = _data_start_index(rows, quantity_index)
    units = _unit_row(rows, quantity_index, data_start)

    frequency_columns = _frequency_columns(quantity, units)
    max_column = max(frequency_columns)
    times, channel_samples = _read_samples(
        rows, data_start, max_column, frequency_columns
    )
    channels = tuple(
        _build_channel(labels, column, channel_samples[column])
        for column in frequency_columns
    )
    return IEEEPMURecording(
        source_name=csv_path.name,
        source_sha256=hashlib.sha256(source_bytes).hexdigest(),
        times=times,
        channels=channels,
    )

write_ingester_csv

write_ingester_csv(
    recording: IEEEPMURecording,
    channel: PMUFrequencyChannel,
    dest: str | Path,
    *,
    time_column: str = "time_s",
    frequency_column: str = "frequency_hz",
) -> AdaptedIngesterCSV

Write one frequency channel as the ringdown screener's input CSV.

Samples are written with Python's shortest round-tripping decimal so the derived CSV is deterministic and reparses to the same values the screener would have read from the source.

Parameters

recording : IEEEPMURecording The parsed capture supplying the shared time vector. channel : PMUFrequencyChannel The frequency channel to write; its samples must align with the time vector. dest : str | pathlib.Path Destination path for the two-column ingester CSV. time_column : str Timestamp column name written to the CSV. frequency_column : str Frequency column name written to the CSV.

Returns

AdaptedIngesterCSV Provenance linking the written CSV back to the source capture.

Raises

ValueError If the column names are blank or the channel is not aligned with the recording's time vector.

Source code in src/scpn_phase_orchestrator/runtime/pmu_ieee_adapter.py

def write_ingester_csv(
    recording: IEEEPMURecording,
    channel: PMUFrequencyChannel,
    dest: str | Path,
    *,
    time_column: str = "time_s",
    frequency_column: str = "frequency_hz",
) -> AdaptedIngesterCSV:
    """Write one frequency channel as the ringdown screener's input CSV.

    Samples are written with Python's shortest round-tripping decimal so the
    derived CSV is deterministic and reparses to the same values the screener
    would have read from the source.

    Parameters
    ----------
    recording : IEEEPMURecording
        The parsed capture supplying the shared time vector.
    channel : PMUFrequencyChannel
        The frequency channel to write; its samples must align with the time
        vector.
    dest : str | pathlib.Path
        Destination path for the two-column ingester CSV.
    time_column : str
        Timestamp column name written to the CSV.
    frequency_column : str
        Frequency column name written to the CSV.

    Returns
    -------
    AdaptedIngesterCSV
        Provenance linking the written CSV back to the source capture.

    Raises
    ------
    ValueError
        If the column names are blank or the channel is not aligned with the
        recording's time vector.
    """
    time_field = _non_empty_str(time_column, "time_column")
    frequency_field = _non_empty_str(frequency_column, "frequency_column")
    if channel.samples.shape[0] != recording.times.shape[0]:
        raise ValueError(
            "channel samples are not aligned with the recording time vector"
        )
    buffer = io.StringIO(newline="")
    writer = csv.writer(buffer)
    writer.writerow([time_field, frequency_field])
    for time_value, frequency_value in zip(
        recording.times, channel.samples, strict=True
    ):
        writer.writerow([str(float(time_value)), str(float(frequency_value))])
    output_bytes = buffer.getvalue().encode("utf-8")
    dest_path = Path(dest)
    dest_path.write_bytes(output_bytes)
    return AdaptedIngesterCSV(
        source_name=recording.source_name,
        source_sha256=recording.source_sha256,
        output_name=dest_path.name,
        output_sha256=hashlib.sha256(output_bytes).hexdigest(),
        channel_label=channel.label,
        channel_column_index=channel.column_index,
        time_column=time_field,
        frequency_column=frequency_field,
        row_count=int(recording.times.shape[0]),
    )

adapt_ieee_pmu_csv

adapt_ieee_pmu_csv(
    source: str | Path,
    dest: str | Path,
    *,
    nominal_frequency_hz: float = 60.0,
    plausible_band_hz: float = 2.0,
    time_column: str = "time_s",
    frequency_column: str = "frequency_hz",
) -> AdaptedIngesterCSV

Adapt an IEEE PMU capture into the screener's input in one call.

Parameters

source : str | pathlib.Path Path to the IEEE-format PMU concentrator CSV. dest : str | pathlib.Path Destination path for the derived two-column ingester CSV. nominal_frequency_hz : float Nominal grid frequency used to reject out-of-band channels. plausible_band_hz : float Half-width in hertz of the accepted band about the nominal frequency. time_column : str Timestamp column name written to the derived CSV. frequency_column : str Frequency column name written to the derived CSV.

Returns

AdaptedIngesterCSV Provenance linking the derived CSV back to the source capture.

Raises

ValueError If the source cannot be parsed or no channel qualifies for selection.

Source code in src/scpn_phase_orchestrator/runtime/pmu_ieee_adapter.py

def adapt_ieee_pmu_csv(
    source: str | Path,
    dest: str | Path,
    *,
    nominal_frequency_hz: float = 60.0,
    plausible_band_hz: float = 2.0,
    time_column: str = "time_s",
    frequency_column: str = "frequency_hz",
) -> AdaptedIngesterCSV:
    """Adapt an IEEE PMU capture into the screener's input in one call.

    Parameters
    ----------
    source : str | pathlib.Path
        Path to the IEEE-format PMU concentrator CSV.
    dest : str | pathlib.Path
        Destination path for the derived two-column ingester CSV.
    nominal_frequency_hz : float
        Nominal grid frequency used to reject out-of-band channels.
    plausible_band_hz : float
        Half-width in hertz of the accepted band about the nominal frequency.
    time_column : str
        Timestamp column name written to the derived CSV.
    frequency_column : str
        Frequency column name written to the derived CSV.

    Returns
    -------
    AdaptedIngesterCSV
        Provenance linking the derived CSV back to the source capture.

    Raises
    ------
    ValueError
        If the source cannot be parsed or no channel qualifies for selection.
    """
    recording = read_ieee_pmu_recording(source)
    channel = recording.select_cleanest_channel(
        nominal_frequency_hz=nominal_frequency_hz,
        plausible_band_hz=plausible_band_hz,
    )
    return write_ingester_csv(
        recording,
        channel,
        dest,
        time_column=time_column,
        frequency_column=frequency_column,
    )

10. PMU ringdown evidence ingress

runtime.pmu_ringdown is the operator-data ingress for the same review-only PRC screening chain. screen_pmu_ringdown_csv reads a local PMU or historian CSV with time_s and frequency_hz columns, verifies finite data whose timestamps match a best-fit uniform grid (so decimal-rounded operator timestamps are accepted), converts measured frequency into nominal-frequency deviation, mean-detrends the deviation to remove the operating-point offset that would otherwise dominate the estimate, optionally block-mean decimates an over-sampled capture to a requested analysis rate, estimates oscillation modes under a bounded model order, and seals the resulting PRCOscillationEvidence with the source CSV SHA-256 digest. The record keeps both the raw capture rate and the post-decimation analysis rate. The spo pmu-ringdown command exposes the same path for reviewed operator captures and writes one deterministic scpn_pmu_ringdown_prc_audit_v1 record when --output is supplied. This is a data-screening path only; it never fits a plant model and never actuates.

pmu_ringdown

Review-only PRC oscillation screening for operator-provided PMU ringdowns.

The dVOC audit pack needs a real-data ingress surface before it can be validated against reviewed operator captures. This module provides that boundary without claiming live control: it reads a local CSV exported from a PMU or historian, validates finite uniformly sampled frequency measurements, converts them to a nominal-frequency deviation signal, runs the matrix-pencil oscillation estimator, and seals the resulting PRC evidence with a source-file digest.

Classes

PMURingdownEvidence dataclass

PMURingdownEvidence(
    schema: str,
    event_id: str,
    captured_at: str,
    signal_source: str,
    source_name: str,
    source_sha256: str,
    time_column: str,
    frequency_column: str,
    nominal_frequency_hz: float,
    sample_count: int,
    sampling_rate_hz: float,
    duration_s: float,
    detrend: str,
    analysis_rate_hz: float,
    analysis_sample_count: int,
    prc_evidence: PRCOscillationEvidence,
    claim_boundary: str = PMU_RINGDOWN_CLAIM_BOUNDARY,
    review_only: bool = True,
)

Hash-sealed PRC screening evidence for one PMU ringdown CSV.

Attributes

schema : str Audit schema identifier. event_id : str Caller-assigned event identifier. captured_at : str Capture timestamp supplied by the caller. signal_source : str Operator-facing source label for the PMU or historian signal. source_name : str Basename of the screened CSV path. source_sha256 : str SHA-256 digest of the exact source CSV bytes. time_column, frequency_column : str CSV columns consumed by the parser. nominal_frequency_hz : float Frequency subtracted from the measured PMU frequency before estimation. sample_count : int Number of accepted samples. sampling_rate_hz : float Uniform sampling rate inferred from the timestamp column (the raw capture rate, before any decimation). duration_s : float Capture duration from first to last sample. detrend : str Detrend mode applied to the deviation signal before estimation ("none" or "mean"). analysis_rate_hz : float Sampling rate of the signal actually fed to the estimator, after optional decimation. Equals sampling_rate_hz when no decimation was requested. analysis_sample_count : int Number of samples fed to the estimator after optional decimation. prc_evidence : PRCOscillationEvidence Hash-sealed PRC screening evidence for the frequency-deviation signal. claim_boundary : str Review-only claim boundary. review_only : bool Always True for this ingestion surface. content_hash : str SHA-256 of the canonical record excluding this field.

Methods:

__post_init__

__post_init__() -> None

Compute the content hash from the canonical evidence payload.

Source code in src/scpn_phase_orchestrator/runtime/pmu_ringdown.py

def __post_init__(self) -> None:
    """Compute the content hash from the canonical evidence payload."""
    object.__setattr__(
        self, "content_hash", canonical_record_hash(self._canonical_payload())
    )

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe mapping of the PMU ringdown evidence.

Returns

dict[str, object] The canonical payload plus the computed content_hash.

Source code in src/scpn_phase_orchestrator/runtime/pmu_ringdown.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the PMU ringdown evidence.

    Returns
    -------
    dict[str, object]
        The canonical payload plus the computed ``content_hash``.
    """
    record = self._canonical_payload()
    record["content_hash"] = self.content_hash
    return record

Functions:

screen_pmu_ringdown_csv

screen_pmu_ringdown_csv(
    path: str | Path,
    *,
    event_id: str,
    captured_at: str,
    signal_source: str,
    time_column: str = "time_s",
    frequency_column: str = "frequency_hz",
    nominal_frequency_hz: float = 60.0,
    detrend: str = "mean",
    analysis_rate_hz: float | None = None,
    min_samples: int = 8,
    max_analysis_samples: int = 1200,
    sampling_jitter_tolerance: float = 0.05,
    model_order: int | None = 8,
) -> PMURingdownEvidence

Screen a PMU frequency ringdown CSV into hash-sealed PRC evidence.

The defaults are tuned for real operator captures. A raw PMU frequency channel sits at a small offset from the nominal frequency (the operating point rarely equals exactly 60/50 Hz) and is reported on a decimal-rounded timestamp grid at the full reporting rate. Left untreated, the offset is fit as a dominant 0 Hz mode that buries the electromechanical oscillation, the rounded timestamps fail an over-tight uniformity check, and the full reporting rate makes a several-minute capture too long for the estimator. The defaults address all three: mean detrending removes the operating-point offset, analysis_rate_hz decimates an over-sampled capture, the uniformity tolerance accepts rounded timestamps, and a bounded model order keeps a noisy real signal from fragmenting into spurious modes.

Parameters

path : str | pathlib.Path CSV path with a timestamp column and a measured frequency column. event_id : str Caller-assigned event identifier for the PMU capture. captured_at : str Capture timestamp stamped into the PRC evidence. signal_source : str Operator-facing PMU or historian signal label. time_column : str Name of the timestamp column in seconds. frequency_column : str Name of the measured frequency column in hertz. nominal_frequency_hz : float Nominal grid frequency subtracted before mode estimation. detrend : str Deviation-signal detrend mode: "mean" (default) removes the operating-point offset, "none" disables detrending. A linear detrend is intentionally not offered — it distorts a decaying ringdown. analysis_rate_hz : float | None Target analysis rate in hertz. When set below the raw capture rate the deviation signal is anti-alias (block-mean) decimated to approximately this rate before estimation; None estimates at the raw rate. Use roughly ten times the highest mode frequency of interest. min_samples : int Minimum accepted raw sample count. Must be at least four. max_analysis_samples : int Upper bound on the post-decimation sample count fed to the estimator. A longer signal fails closed with guidance rather than making the matrix-pencil singular value decomposition intractable. sampling_jitter_tolerance : float Tolerance for timestamp uniformity, as a fraction of the sample interval, measured against the best-fit uniform grid (so decimal rounding does not accumulate). model_order : int | None Matrix-pencil model order forwarded to the estimator. The default of eight suits screening for a handful of dominant modes; None selects the order from the singular-value spectrum (only advisable for clean captures).

Returns

PMURingdownEvidence Deterministic, review-only PMU screening evidence.

Raises

ValueError If the CSV, timestamps, frequency samples, or scalar controls are invalid.

Source code in src/scpn_phase_orchestrator/runtime/pmu_ringdown.py

def screen_pmu_ringdown_csv(
    path: str | Path,
    *,
    event_id: str,
    captured_at: str,
    signal_source: str,
    time_column: str = "time_s",
    frequency_column: str = "frequency_hz",
    nominal_frequency_hz: float = 60.0,
    detrend: str = "mean",
    analysis_rate_hz: float | None = None,
    min_samples: int = 8,
    max_analysis_samples: int = 1200,
    sampling_jitter_tolerance: float = 5.0e-2,
    model_order: int | None = 8,
) -> PMURingdownEvidence:
    """Screen a PMU frequency ringdown CSV into hash-sealed PRC evidence.

    The defaults are tuned for real operator captures. A raw PMU frequency
    channel sits at a small offset from the nominal frequency (the operating
    point rarely equals exactly 60/50 Hz) and is reported on a decimal-rounded
    timestamp grid at the full reporting rate. Left untreated, the offset is
    fit as a dominant 0 Hz mode that buries the electromechanical oscillation,
    the rounded timestamps fail an over-tight uniformity check, and the full
    reporting rate makes a several-minute capture too long for the estimator.
    The defaults address all three: mean detrending removes the operating-point
    offset, ``analysis_rate_hz`` decimates an over-sampled capture, the
    uniformity tolerance accepts rounded timestamps, and a bounded model order
    keeps a noisy real signal from fragmenting into spurious modes.

    Parameters
    ----------
    path : str | pathlib.Path
        CSV path with a timestamp column and a measured frequency column.
    event_id : str
        Caller-assigned event identifier for the PMU capture.
    captured_at : str
        Capture timestamp stamped into the PRC evidence.
    signal_source : str
        Operator-facing PMU or historian signal label.
    time_column : str
        Name of the timestamp column in seconds.
    frequency_column : str
        Name of the measured frequency column in hertz.
    nominal_frequency_hz : float
        Nominal grid frequency subtracted before mode estimation.
    detrend : str
        Deviation-signal detrend mode: ``"mean"`` (default) removes the
        operating-point offset, ``"none"`` disables detrending. A linear
        detrend is intentionally not offered — it distorts a decaying ringdown.
    analysis_rate_hz : float | None
        Target analysis rate in hertz. When set below the raw capture rate the
        deviation signal is anti-alias (block-mean) decimated to approximately
        this rate before estimation; ``None`` estimates at the raw rate. Use
        roughly ten times the highest mode frequency of interest.
    min_samples : int
        Minimum accepted raw sample count. Must be at least four.
    max_analysis_samples : int
        Upper bound on the post-decimation sample count fed to the estimator.
        A longer signal fails closed with guidance rather than making the
        matrix-pencil singular value decomposition intractable.
    sampling_jitter_tolerance : float
        Tolerance for timestamp uniformity, as a fraction of the sample
        interval, measured against the best-fit uniform grid (so decimal
        rounding does not accumulate).
    model_order : int | None
        Matrix-pencil model order forwarded to the estimator. The default of
        eight suits screening for a handful of dominant modes; ``None`` selects
        the order from the singular-value spectrum (only advisable for clean
        captures).

    Returns
    -------
    PMURingdownEvidence
        Deterministic, review-only PMU screening evidence.

    Raises
    ------
    ValueError
        If the CSV, timestamps, frequency samples, or scalar controls are invalid.
    """
    csv_path = Path(path)
    event = _non_empty_str(event_id, "event_id")
    captured = _non_empty_str(captured_at, "captured_at")
    source = _non_empty_str(signal_source, "signal_source")
    time_field = _non_empty_str(time_column, "time_column")
    frequency_field = _non_empty_str(frequency_column, "frequency_column")
    nominal = _positive_real(nominal_frequency_hz, "nominal_frequency_hz")
    detrend_mode = _validated_detrend(detrend)
    sample_floor = _min_samples(min_samples)
    analysis_ceiling = _max_analysis_samples(max_analysis_samples)
    jitter = _non_negative_real(sampling_jitter_tolerance, "sampling_jitter_tolerance")
    target_rate = (
        None
        if analysis_rate_hz is None
        else _positive_real(analysis_rate_hz, "analysis_rate_hz")
    )

    source_bytes = csv_path.read_bytes()
    times, frequencies = _read_pmu_csv(csv_path, time_field, frequency_field)
    if times.shape[0] < sample_floor:
        raise ValueError(
            f"PMU ringdown CSV must contain at least {sample_floor} samples"
        )
    sample_interval = _uniform_sample_interval(times, time_field, jitter)
    sampling_rate = 1.0 / sample_interval
    deviation = np.ascontiguousarray(frequencies - nominal, dtype=np.float64)
    analysis_signal, analysis_rate = _decimate_signal(
        deviation, sampling_rate, target_rate
    )
    if analysis_signal.shape[0] > analysis_ceiling:
        raise ValueError(
            f"analysis signal has {analysis_signal.shape[0]} samples, above the "
            f"{analysis_ceiling} limit; set analysis_rate_hz to decimate the capture"
        )
    analysis_signal = _detrend_signal(analysis_signal, detrend_mode)
    modes = estimate_oscillation_modes(
        analysis_signal, analysis_rate, model_order=model_order
    )
    prc_evidence = screen_oscillation_modes(
        modes,
        event_id=event,
        captured_at=captured,
        signal_source=f"{source}/deviation",
        sampling_rate_hz=analysis_rate,
    )
    return PMURingdownEvidence(
        schema=PMU_RINGDOWN_AUDIT_SCHEMA,
        event_id=event,
        captured_at=captured,
        signal_source=source,
        source_name=csv_path.name,
        source_sha256=hashlib.sha256(source_bytes).hexdigest(),
        time_column=time_field,
        frequency_column=frequency_field,
        nominal_frequency_hz=nominal,
        sample_count=int(times.shape[0]),
        sampling_rate_hz=sampling_rate,
        duration_s=float(times[-1] - times[0]),
        detrend=detrend_mode,
        analysis_rate_hz=analysis_rate,
        analysis_sample_count=int(analysis_signal.shape[0]),
        prc_evidence=prc_evidence,
    )

11. IBR ride-through evidence ingress

runtime.ibr_ride_through is the PRC-029-ready operator-data ingress for voltage and frequency ride-through review. screen_ibr_ride_through_csv reads a local CSV with time_s, voltage_pu, and frequency_hz columns, rejects malformed or non-finite data, runs the PRC-029 ride-through screener, and seals the result with the source CSV SHA-256 digest. The spo ibr-ride-through command exposes the same path and writes one deterministic scpn_ibr_ride_through_prc029_audit_v1 record when --output is supplied. This is a data-screening path only; it never fits a plant model, never actuates, and never claims compliance.

ibr_ride_through

CSV ingress for review-only PRC-029 ride-through screening.

The C2 power-grid audit pack consumes local operator exports only. This module reads a timestamped voltage/frequency CSV, rejects malformed measurements before publication, invokes :mod:scpn_phase_orchestrator.assurance.prc_ride_through, and wraps the result with a source-file digest so the screened evidence can be reproduced byte-for-byte.

Classes

IBRRideThroughCsvEvidence dataclass

IBRRideThroughCsvEvidence(
    schema: str,
    event_id: str,
    captured_at: str,
    signal_source: str,
    source_name: str,
    source_sha256: str,
    time_column: str,
    voltage_column: str,
    frequency_column: str,
    ibr_category: str,
    sample_count: int,
    duration_s: float,
    prc029_evidence: PRCRideThroughEvidence,
    claim_boundary: str = IBR_RIDE_THROUGH_CLAIM_BOUNDARY,
    review_only: bool = True,
)

Hash-sealed PRC-029 screening evidence for one operator CSV.

Attributes

schema : str Audit schema identifier. event_id : str Caller-assigned event identifier. captured_at : str Measurement timestamp supplied by the caller. signal_source : str Operator-facing source label. source_name : str Basename of the screened CSV. source_sha256 : str SHA-256 digest of the exact source CSV bytes. time_column, voltage_column, frequency_column : str CSV columns consumed by the parser. ibr_category : str PRC-029 voltage-table category forwarded to the screener. sample_count : int Number of accepted samples. duration_s : float Elapsed time from first to last accepted sample. prc029_evidence : PRCRideThroughEvidence Hash-sealed PRC-029 screening evidence. claim_boundary : str Review-only claim boundary. review_only : bool Always True for this ingestion surface. content_hash : str SHA-256 of the canonical record excluding this field.

Methods:

__post_init__

__post_init__() -> None

Compute the content hash from the canonical evidence payload.

Source code in src/scpn_phase_orchestrator/runtime/ibr_ride_through.py

def __post_init__(self) -> None:
    """Compute the content hash from the canonical evidence payload."""
    object.__setattr__(
        self, "content_hash", canonical_record_hash(self._canonical_payload())
    )

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe mapping of the CSV evidence record.

Returns

dict[str, object] The canonical payload plus content_hash.

Source code in src/scpn_phase_orchestrator/runtime/ibr_ride_through.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the CSV evidence record.

    Returns
    -------
    dict[str, object]
        The canonical payload plus ``content_hash``.
    """
    record = self._canonical_payload()
    record["content_hash"] = self.content_hash
    return record

Functions:

screen_ibr_ride_through_csv

screen_ibr_ride_through_csv(
    path: str | Path,
    *,
    event_id: str,
    captured_at: str,
    signal_source: str,
    ibr_category: str = OTHER_IBR,
    time_column: str = "time_s",
    voltage_column: str = "voltage_pu",
    frequency_column: str = "frequency_hz",
) -> IBRRideThroughCsvEvidence

Screen an operator voltage/frequency CSV into PRC-029 evidence.

Parameters

path : str | pathlib.Path CSV path containing timestamp, voltage, and frequency columns. event_id : str Caller-assigned event identifier. captured_at : str Measurement timestamp stamped into the evidence. signal_source : str Operator-facing source label. ibr_category : str PRC-029 voltage-table selector. time_column : str Timestamp column in seconds. voltage_column : str Voltage column in per unit. frequency_column : str Frequency column in hertz.

Returns

IBRRideThroughCsvEvidence Deterministic, review-only CSV evidence package.

Raises

ValueError If the CSV, identifiers, category, or column values are invalid.

Source code in src/scpn_phase_orchestrator/runtime/ibr_ride_through.py

def screen_ibr_ride_through_csv(
    path: str | Path,
    *,
    event_id: str,
    captured_at: str,
    signal_source: str,
    ibr_category: str = OTHER_IBR,
    time_column: str = "time_s",
    voltage_column: str = "voltage_pu",
    frequency_column: str = "frequency_hz",
) -> IBRRideThroughCsvEvidence:
    """Screen an operator voltage/frequency CSV into PRC-029 evidence.

    Parameters
    ----------
    path : str | pathlib.Path
        CSV path containing timestamp, voltage, and frequency columns.
    event_id : str
        Caller-assigned event identifier.
    captured_at : str
        Measurement timestamp stamped into the evidence.
    signal_source : str
        Operator-facing source label.
    ibr_category : str
        PRC-029 voltage-table selector.
    time_column : str
        Timestamp column in seconds.
    voltage_column : str
        Voltage column in per unit.
    frequency_column : str
        Frequency column in hertz.

    Returns
    -------
    IBRRideThroughCsvEvidence
        Deterministic, review-only CSV evidence package.

    Raises
    ------
    ValueError
        If the CSV, identifiers, category, or column values are invalid.
    """
    csv_path = Path(path)
    event = _non_empty_str(event_id, "event_id")
    captured = _non_empty_str(captured_at, "captured_at")
    source = _non_empty_str(signal_source, "signal_source")
    time_field = _non_empty_str(time_column, "time_column")
    voltage_field = _non_empty_str(voltage_column, "voltage_column")
    frequency_field = _non_empty_str(frequency_column, "frequency_column")

    source_bytes = csv_path.read_bytes()
    times, voltage, frequency = _read_ibr_csv(
        csv_path, time_field, voltage_field, frequency_field
    )
    prc029_evidence = screen_ride_through_samples(
        times,
        voltage,
        frequency,
        event_id=event,
        captured_at=captured,
        signal_source=source,
        ibr_category=ibr_category,
    )
    return IBRRideThroughCsvEvidence(
        schema=IBR_RIDE_THROUGH_AUDIT_SCHEMA,
        event_id=event,
        captured_at=captured,
        signal_source=source,
        source_name=csv_path.name,
        source_sha256=hashlib.sha256(source_bytes).hexdigest(),
        time_column=time_field,
        voltage_column=voltage_field,
        frequency_column=frequency_field,
        ibr_category=prc029_evidence.ibr_category,
        sample_count=prc029_evidence.sample_count,
        duration_s=prc029_evidence.duration_s,
        prc029_evidence=prc029_evidence,
    )