Adapters

Bridges between SPO and external systems in the SCPN ecosystem, observability platforms, and hardware controllers. Each adapter translates SPO's internal representations (phases, coupling matrices, regime states) into the wire format expected by the target system.

Pipeline position

UPDEEngine ──→ UPDEState ──→ Adapters (output)
                                │
                  ┌─────────────┼──────────────────┐
                  ↓             ↓                  ↓
          SCPN Ecosystem   Observability      Hardware
          │                │                  │
          ├─ scpn_control  ├─ OpenTelemetry   ├─ Modbus/TLS
          ├─ fusion_core   ├─ Prometheus      └─ gRPC
          ├─ neurocore     └─ Grafana
          ├─ plasma_control
          ├─ quantum_control
          └─ snn_bridge

External Systems ──→ Adapters (input) ──→ Oscillator Extractors

Adapters are bidirectional: output adapters export SPO state to external systems; input adapters import external signals for phase extraction.

SCPN Ecosystem Bridges

These adapters connect SPO to sibling packages in the SCPN ecosystem. They share the Kuramoto/UPDE phase representation but differ in scope:

Adapter	Target	Data flow
scpn_control_bridge	scpn-control (v0.18.0)	Bidirectional: phases, coupling, regime
fusion_core_bridge	SCPN-Fusion-Core (v3.9.3)	Export: sync metrics for fusion analysis
neurocore_bridge	sc-neurocore (v3.13.3)	Export: phase states for SNN processing
plasma_control_bridge	Plasma control systems	Import: magnetic diagnostics as P-channel
quantum_control_bridge	scpn-quantum-control (v0.9.1)	Export: coherence metrics for QPU scheduling
hybrid_cocompiler	Quantum + neuromorphic review package	Export: shared audit envelope for simulator handoff
snn_bridge	local SNN daemon	Export: local phase-dynamics signal packets

scpn-control Bridge

SCPNControlBridge(scpn_config: dict) — bidirectional adapter.

Method	Signature	Description
import_knm	(scpn_knm: NDArray) → CouplingState	Wrap external K_nm
import_omega	(scpn_omega: NDArray) → NDArray	Validate frequencies
export_state	(upde_state: UPDEState) → dict	Telemetry export

import_knm validates non-empty finite real-valued square matrices with a zero self-coupling diagonal. import_omega validates non-empty finite real-valued 1-D vectors with strictly positive natural frequencies. export_state produces a dict with regime, stability, layers (each with R, ψ, lock signatures).

scpn_control_bridge

SCPN-control bridge for validated coupling, frequency, and telemetry exchange.

The bridge accepts JSON-compatible configuration, imports finite dense coupling matrices and positive natural-frequency vectors, and exports reduced UPDE state telemetry with layer locks and cross-alignment. It is a data-shape adapter only; it does not invoke an external control engine or apply actions.

Classes

SCPNControlBridge

SCPNControlBridge(scpn_config: dict[str, Any])

Adapter between scpn-control telemetry and phase-orchestrator types.

Source code in src/scpn_phase_orchestrator/adapters/scpn_control_bridge.py

def __init__(self, scpn_config: dict[str, Any]):
    if not isinstance(scpn_config, dict):
        raise ValueError("scpn_config must be a dict")
    self._config = _validate_scpn_config(scpn_config)

Methods:

import_knm

import_knm(scpn_knm: FloatArray) -> CouplingState

Wrap an external Knm matrix into a CouplingState.

Parameters

scpn_knm : FloatArray An external coupling matrix, shape (N, N).

Returns

CouplingState The coupling state wrapping the external matrix.

Raises

ValueError If the coupling matrix is invalid.

Source code in src/scpn_phase_orchestrator/adapters/scpn_control_bridge.py

def import_knm(self, scpn_knm: FloatArray) -> CouplingState:
    """Wrap an external Knm matrix into a CouplingState.

    Parameters
    ----------
    scpn_knm : FloatArray
        An external coupling matrix, shape ``(N, N)``.

    Returns
    -------
    CouplingState
        The coupling state wrapping the external matrix.

    Raises
    ------
    ValueError
        If the coupling matrix is invalid.
    """
    knm = _as_real_numeric_array(scpn_knm, name="Knm")
    if knm.ndim != 2 or knm.shape[0] != knm.shape[1]:
        raise ValueError(f"Knm must be square, got shape {knm.shape}")
    if knm.size == 0:
        raise ValueError("Knm must be non-empty")
    if not np.all(np.isfinite(knm)):
        raise ValueError("Knm must contain only finite values")
    if np.any(np.diag(knm) != 0.0):
        raise ValueError("Knm self-coupling diagonal must be zero")
    n = knm.shape[0]
    return CouplingState(
        knm=knm,
        alpha=np.zeros((n, n), dtype=np.float64),
        active_template="scpn_import",
    )

import_omega

import_omega(scpn_omega: FloatArray) -> FloatArray

Validate and pass through natural frequencies.

Parameters

scpn_omega : FloatArray External natural frequencies, shape (N,).

Returns

FloatArray The validated natural frequencies.

Raises

ValueError If the natural frequencies are invalid.

Source code in src/scpn_phase_orchestrator/adapters/scpn_control_bridge.py

def import_omega(self, scpn_omega: FloatArray) -> FloatArray:
    """Validate and pass through natural frequencies.

    Parameters
    ----------
    scpn_omega : FloatArray
        External natural frequencies, shape ``(N,)``.

    Returns
    -------
    FloatArray
        The validated natural frequencies.

    Raises
    ------
    ValueError
        If the natural frequencies are invalid.
    """
    omega = _as_real_numeric_array(scpn_omega, name="omega")
    if omega.ndim != 1:
        raise ValueError(f"omega must be 1-D, got ndim={omega.ndim}")
    if omega.size == 0:
        raise ValueError("omega must be non-empty")
    if not np.all(np.isfinite(omega)):
        raise ValueError("omega must contain only finite values")
    if np.any(omega <= 0.0):
        raise ValueError("All natural frequencies must be positive")
    return omega

export_state

export_state(upde_state: UPDEState) -> dict[str, Any]

Convert UPDEState to scpn-control compatible telemetry dict.

Parameters

upde_state : UPDEState The UPDE state to export.

Returns

dict[str, Any] The scpn-control-compatible telemetry dict.

Source code in src/scpn_phase_orchestrator/adapters/scpn_control_bridge.py

def export_state(self, upde_state: UPDEState) -> dict[str, Any]:
    """Convert UPDEState to scpn-control compatible telemetry dict.

    Parameters
    ----------
    upde_state : UPDEState
        The UPDE state to export.

    Returns
    -------
    dict[str, Any]
        The scpn-control-compatible telemetry dict.
    """
    return {
        "regime": upde_state.regime_id,
        "stability": upde_state.stability_proxy,
        "layers": [
            {
                "R": ls.R,
                "psi": ls.psi,
                "locks": {
                    k: {"plv": v.plv, "lag": v.mean_lag}
                    for k, v in ls.lock_signatures.items()
                },
            }
            for ls in upde_state.layers
        ],
        "cross_alignment": upde_state.cross_layer_alignment.tolist(),
    }

Fusion Core Bridge

FusionCoreBridge is a non-executing review bridge for scpn-fusion-core equilibrium summaries. It maps positive q-profile bounds, non-negative normalised beta, confinement time, sawtooth/ELM event counts, and non-negative MHD amplitude into bounded phase channels. The feedback path rejects empty phase vectors before computing the complex order parameter, so exported R_global, mean phase, and mean frequency records stay finite. Stability checks also reject negative beta and confinement-ratio payloads instead of silently converting them into ordinary soft violations.

fusion_core_bridge

Fusion-Core bridge for phase encoding and stability-review diagnostics.

The bridge maps fusion equilibrium observables into bounded phase vectors, returns aggregate phase feedback summaries, normalises q-profile/equilibrium payloads, and checks local fusion stability invariants. It is pure NumPy/dict code and does not require or call a live fusion solver; outputs are review signals and feedback dictionaries for explicit downstream handoff.

Classes

FusionCoreBridge

FusionCoreBridge(n_layers: int = 6)

Adapter between scpn-fusion-core equilibrium data and phase-orchestrator.

All methods work without scpn-fusion-core (pure numpy + dict).

Source code in src/scpn_phase_orchestrator/adapters/fusion_core_bridge.py

def __init__(self, n_layers: int = 6):
    if (
        isinstance(n_layers, bool)
        or not isinstance(n_layers, Integral)
        or not 1 <= n_layers <= len(_OBS_NAMES)
    ):
        raise ValueError(
            f"n_layers must be an integer in [1, {len(_OBS_NAMES)}], "
            f"got {n_layers!r}"
        )
    self._n_layers = int(n_layers)

Methods:

observables_to_phases

observables_to_phases(
    snapshot: dict[str, Any],
) -> FloatArray

Map 6 fusion observables to [0, 2*pi) phases.

Observable → Phase formula: q_profile → 2pi(q - q_min)/(q_max - q_min) beta_n → 2pibeta_n/beta_limit tau_e → 2pitau_e/tau_ref sawtooth_count → countpi mod 2pi elm_count → countpi mod 2pi mhd_amplitude → 2piamplitude/threshold

Parameters

snapshot : dict[str, Any] Fusion observable values keyed by name.

Returns

FloatArray The oscillator phases in [0, 2π), shape (N,).

Raises

ValueError If the snapshot is missing required observables.

Source code in src/scpn_phase_orchestrator/adapters/fusion_core_bridge.py

def observables_to_phases(self, snapshot: dict[str, Any]) -> FloatArray:
    """Map 6 fusion observables to [0, 2*pi) phases.

    Observable → Phase formula:
      q_profile       → 2*pi*(q - q_min)/(q_max - q_min)
      beta_n          → 2*pi*beta_n/beta_limit
      tau_e           → 2*pi*tau_e/tau_ref
      sawtooth_count  → count*pi mod 2*pi
      elm_count       → count*pi mod 2*pi
      mhd_amplitude   → 2*pi*amplitude/threshold

    Parameters
    ----------
    snapshot : dict[str, Any]
        Fusion observable values keyed by name.

    Returns
    -------
    FloatArray
        The oscillator phases in ``[0, 2π)``, shape ``(N,)``.

    Raises
    ------
    ValueError
        If the snapshot is missing required observables.
    """
    if not isinstance(snapshot, dict):
        raise ValueError("snapshot must be a dict")
    q = _finite_positive_real(snapshot.get("q_profile", 1.5), name="q_profile")
    q_min = _finite_positive_real(snapshot.get("q_min", 1.0), name="q_min")
    q_max = _finite_positive_real(snapshot.get("q_max", 5.0), name="q_max")
    _validate_q_bounds(q_min, q_max)
    beta_n = _finite_non_negative_real(snapshot.get("beta_n", 1.0), name="beta_n")
    tau_e = _finite_non_negative_real(snapshot.get("tau_e", 1.0), name="tau_e")
    saw_count = non_negative_int(
        snapshot.get("sawtooth_count", 0),
        name="sawtooth_count",
    )
    elm_count = non_negative_int(snapshot.get("elm_count", 0), name="elm_count")
    mhd_amp = _finite_non_negative_real(
        snapshot.get("mhd_amplitude", 0.0),
        name="mhd_amplitude",
    )

    denom_q = q_max - q_min if q_max != q_min else 1.0
    phases = np.array(
        [
            TWO_PI * np.clip((q - q_min) / denom_q, 0.0, 1.0),
            TWO_PI * np.clip(beta_n / BETA_N_LIMIT, 0.0, 1.0),
            TWO_PI * np.clip(tau_e / TAU_E_REF_S, 0.0, 1.0),
            (saw_count * np.pi) % TWO_PI,
            (elm_count * np.pi) % TWO_PI,
            TWO_PI * np.clip(mhd_amp / MHD_THRESHOLD, 0.0, 1.0),
        ],
        dtype=np.float64,
    )

    return phases[: self._n_layers]

phases_to_feedback

phases_to_feedback(
    phases: FloatArray, omegas: FloatArray
) -> dict[str, Any]

Convert phase state back to feedback signals for the equilibrium solver.

Parameters

phases : FloatArray Oscillator phases in radians, shape (N,). omegas : FloatArray Natural frequencies in rad/s, shape (N,).

Returns

dict[str, Any] The feedback signals for the equilibrium solver.

Raises

ValueError If the phases or omegas are invalid.

Source code in src/scpn_phase_orchestrator/adapters/fusion_core_bridge.py

def phases_to_feedback(
    self,
    phases: FloatArray,
    omegas: FloatArray,
) -> dict[str, Any]:
    """Convert phase state back to feedback signals for the equilibrium solver.

    Parameters
    ----------
    phases : FloatArray
        Oscillator phases in radians, shape ``(N,)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.

    Returns
    -------
    dict[str, Any]
        The feedback signals for the equilibrium solver.

    Raises
    ------
    ValueError
        If the phases or omegas are invalid.
    """
    phases = _finite_vector(phases, name="phases")
    omegas = _finite_vector(omegas, name="omegas")
    if omegas.size < min(phases.size, self._n_layers):
        raise ValueError("omegas length must cover feedback oscillator count")
    n = min(len(phases), self._n_layers)
    z = np.exp(1j * phases[:n])
    order = z.mean()
    r = float(np.abs(order))
    psi = float(np.angle(order) % TWO_PI)
    return {
        "R_global": r,
        "mean_phase": psi,
        "mean_omega": float(np.mean(omegas[:n])),
        "n_oscillators": n,
    }

import_q_profile

import_q_profile(
    q_profile_or_dict: object,
) -> dict[str, Any]

Parse a q-profile from dict or scpn-fusion-core object.

Returns normalised dict with keys: q_min, q_max, q_axis, q_edge.

Parameters

q_profile_or_dict : object A q-profile as an scpn-fusion-core object or a dict.

Returns

dict[str, Any] The parsed q-profile as a dict.

Raises

ValueError If the q-profile cannot be parsed.

Source code in src/scpn_phase_orchestrator/adapters/fusion_core_bridge.py

def import_q_profile(self, q_profile_or_dict: object) -> dict[str, Any]:
    """Parse a q-profile from dict or scpn-fusion-core object.

    Returns normalised dict with keys: q_min, q_max, q_axis, q_edge.

    Parameters
    ----------
    q_profile_or_dict : object
        A q-profile as an scpn-fusion-core object or a dict.

    Returns
    -------
    dict[str, Any]
        The parsed q-profile as a dict.

    Raises
    ------
    ValueError
        If the q-profile cannot be parsed.
    """
    if isinstance(q_profile_or_dict, dict):
        q_min = _finite_positive_real(
            q_profile_or_dict.get("q_min", 1.0),
            name="q_min",
        )
        q_max = _finite_positive_real(
            q_profile_or_dict.get("q_max", 5.0),
            name="q_max",
        )
        q_axis = _finite_real(
            q_profile_or_dict.get("q_axis", q_min),
            name="q_axis",
        )
        q_edge = _finite_real(
            q_profile_or_dict.get("q_edge", q_max),
            name="q_edge",
        )
    else:
        q_min = _finite_positive_real(
            getattr(q_profile_or_dict, "q_min", 1.0),
            name="q_min",
        )
        q_max = _finite_positive_real(
            getattr(q_profile_or_dict, "q_max", 5.0),
            name="q_max",
        )
        q_axis = _finite_real(
            getattr(q_profile_or_dict, "q_axis", q_min),
            name="q_axis",
        )
        q_edge = _finite_real(
            getattr(q_profile_or_dict, "q_edge", q_max),
            name="q_edge",
        )
    _validate_q_bounds(q_min, q_max)
    if not q_min <= q_axis <= q_max:
        raise ValueError("q_axis must be within q_min and q_max")
    if not q_min <= q_edge <= q_max:
        raise ValueError("q_edge must be within q_min and q_max")
    return {"q_min": q_min, "q_max": q_max, "q_axis": q_axis, "q_edge": q_edge}

import_equilibrium

import_equilibrium(
    kernel_result: dict[str, Any],
) -> dict[str, Any]

Extract equilibrium observables from a fusion kernel result dict.

Parameters

kernel_result : dict[str, Any] An scpn-fusion-core kernel result dict.

Returns

dict[str, Any] The equilibrium observables extracted from the kernel result.

Raises

ValueError If the kernel result is malformed.

Source code in src/scpn_phase_orchestrator/adapters/fusion_core_bridge.py

def import_equilibrium(self, kernel_result: dict[str, Any]) -> dict[str, Any]:
    """Extract equilibrium observables from a fusion kernel result dict.

    Parameters
    ----------
    kernel_result : dict[str, Any]
        An scpn-fusion-core kernel result dict.

    Returns
    -------
    dict[str, Any]
        The equilibrium observables extracted from the kernel result.

    Raises
    ------
    ValueError
        If the kernel result is malformed.
    """
    if not isinstance(kernel_result, dict):
        raise ValueError("kernel_result must be a dict")
    return {
        "q_profile": _finite_positive_real(
            kernel_result.get("q_profile", 1.5),
            name="q_profile",
        ),
        "beta_n": _finite_non_negative_real(
            kernel_result.get("beta_n", 1.0),
            name="beta_n",
        ),
        "tau_e": _finite_non_negative_real(
            kernel_result.get("tau_e", 1.0),
            name="tau_e",
        ),
        "sawtooth_count": non_negative_int(
            kernel_result.get("sawtooth_count", 0),
            name="sawtooth_count",
        ),
        "elm_count": non_negative_int(
            kernel_result.get("elm_count", 0),
            name="elm_count",
        ),
        "mhd_amplitude": _finite_non_negative_real(
            kernel_result.get("mhd_amplitude", 0.0),
            name="mhd_amplitude",
        ),
    }

check_stability

check_stability(
    observables: dict[str, Any],
) -> list[dict[str, Any]]

Check fusion stability invariants.

Returns a list of violation dicts (empty if all invariants hold).

Parameters

observables : dict[str, Any] Fusion observable values keyed by name.

Returns

list[dict[str, Any]] The list of stability-invariant violations.

Raises

ValueError If the observables are invalid.

Source code in src/scpn_phase_orchestrator/adapters/fusion_core_bridge.py

def check_stability(self, observables: dict[str, Any]) -> list[dict[str, Any]]:
    """Check fusion stability invariants.

    Returns a list of violation dicts (empty if all invariants hold).

    Parameters
    ----------
    observables : dict[str, Any]
        Fusion observable values keyed by name.

    Returns
    -------
    list[dict[str, Any]]
        The list of stability-invariant violations.

    Raises
    ------
    ValueError
        If the observables are invalid.
    """
    if not isinstance(observables, dict):
        raise ValueError("stability observables must be a dict")
    violations: list[dict[str, Any]] = []
    q_min = observables.get("q_min")
    if q_min is None:
        q_min = observables.get("q_profile")
    if q_min is not None:
        q_min = _finite_real(q_min, name="stability q_min")
    if q_min is not None and q_min < Q_MIN_STABLE:
        violations.append(
            {
                "variable": "q_min",
                "value": q_min,
                "threshold": Q_MIN_STABLE,
                "severity": "hard",
                "message": f"q_min={q_min:.3f} < {Q_MIN_STABLE}",
            }
        )
    beta_n = observables.get("beta_n")
    if beta_n is not None:
        beta_n = _finite_non_negative_real(beta_n, name="stability beta_n")
    if beta_n is not None and beta_n > BETA_N_LIMIT:
        violations.append(
            {
                "variable": "beta_n",
                "value": beta_n,
                "threshold": BETA_N_LIMIT,
                "severity": "hard",
                "message": f"beta_n={beta_n:.3f} > {BETA_N_LIMIT} (Troyon no-wall)",
            }
        )
    tau_ratio = observables.get("tau_e_ratio")
    if tau_ratio is not None:
        tau_ratio = _finite_non_negative_real(
            tau_ratio,
            name="stability tau_e_ratio",
        )
    if tau_ratio is not None and tau_ratio < 0.5:
        violations.append(
            {
                "variable": "tau_e_ratio",
                "value": tau_ratio,
                "threshold": 0.5,
                "severity": "soft",
                "message": f"tau_e_ratio={tau_ratio:.3f} < 0.5",
            }
        )
    return violations

Functions:

Plasma Control Bridge

PlasmaControlBridge is the non-actuating plasma telemetry boundary. It accepts finite layer-coupling matrices, phase snapshots, Lyapunov review scores, and invariant payloads only after rejecting boolean numeric aliases, non-zero layer self-coupling, empty phase snapshots, negative beta and Greenwald ratios, and non-positive safety-factor minima. This keeps the Kronecker-expanded K_nm graph off-diagonal and prevents placeholder phase-state exports from entering downstream review paths.

plasma_control_bridge

Plasma-control bridge for telemetry import, coupling expansion, and review.

The bridge converts plasma-layer coupling matrices, phase snapshots, natural frequencies, Lyapunov verdicts, and proposed control action dictionaries into SPO-compatible data structures under strict finite-shape validation. It also checks local plasma invariants for review. The adapter performs no live plasma actuation and does not require scpn-control to be installed.

Classes

PlasmaControlBridge

PlasmaControlBridge(n_layers: int = 8)

Adapter between scpn-control plasma telemetry and phase-orchestrator types.

All methods work without scpn-control installed (pure numpy + dict).

Source code in src/scpn_phase_orchestrator/adapters/plasma_control_bridge.py

def __init__(self, n_layers: int = 8):
    n_layers = _validate_positive_int(n_layers, name="n_layers")
    if n_layers > _PLASMA_LAYER_COUNT:
        raise ValueError(
            f"n_layers must be <= {_PLASMA_LAYER_COUNT}, got {n_layers!r}"
        )
    self._n_layers = n_layers

Methods:

import_knm_spec

import_knm_spec(knm_spec_or_dict: object) -> CouplingState

Expand an (L, L) layer-coupling matrix to (N, N) via Kronecker replication.

Accepts a dict with 'matrix' key (list of lists or ndarray) and optional 'n_osc_per_layer' (int, default 2), or a raw ndarray of shape (L, L).

Parameters

knm_spec_or_dict : object A layer-coupling matrix as an object or dict.

Returns

CouplingState The expanded (N, N) coupling state.

Raises

ValueError If the coupling spec is malformed.

Source code in src/scpn_phase_orchestrator/adapters/plasma_control_bridge.py

def import_knm_spec(self, knm_spec_or_dict: object) -> CouplingState:
    """Expand an (L, L) layer-coupling matrix to (N, N) via Kronecker replication.

    Accepts a dict with 'matrix' key (list of lists or ndarray) and optional
    'n_osc_per_layer' (int, default 2), or a raw ndarray of shape (L, L).

    Parameters
    ----------
    knm_spec_or_dict : object
        A layer-coupling matrix as an object or dict.

    Returns
    -------
    CouplingState
        The expanded ``(N, N)`` coupling state.

    Raises
    ------
    ValueError
        If the coupling spec is malformed.
    """
    if isinstance(knm_spec_or_dict, dict):
        layer_knm = _finite_array(knm_spec_or_dict["matrix"], name="Layer Knm")
        n_per = _validate_positive_int(
            knm_spec_or_dict.get("n_osc_per_layer", 2),
            name="n_osc_per_layer",
        )
    else:
        layer_knm = _finite_array(knm_spec_or_dict, name="Layer Knm")
        n_per = 2

    if layer_knm.ndim != 2 or layer_knm.shape[0] != layer_knm.shape[1]:
        raise ValueError(f"Layer Knm must be square, got shape {layer_knm.shape}")
    if layer_knm.shape != (self._n_layers, self._n_layers):
        raise ValueError(
            f"Layer Knm shape {layer_knm.shape} must match "
            f"n_layers={self._n_layers}"
        )
    if not np.allclose(np.diag(layer_knm), 0.0, rtol=0.0, atol=1e-12):
        raise ValueError("Layer Knm self-coupling diagonal must be zero")

    # Kronecker expansion: each layer block shares the inter-layer coupling
    n_total = layer_knm.shape[0] * n_per
    knm = np.kron(layer_knm, np.ones((n_per, n_per), dtype=np.float64)).astype(
        np.float64
    )
    np.fill_diagonal(knm, 0.0)

    return CouplingState(
        knm=knm,
        alpha=np.zeros((n_total, n_total), dtype=np.float64),
        active_template="plasma_import",
    )

import_plasma_omega

import_plasma_omega(n_osc_per_layer: int = 1) -> FloatArray

Generate natural frequencies spanning plasma timescales.

Returns frequencies ordered: micro_turbulence(fast) → plasma_wall(slow).

Parameters

n_osc_per_layer : int Number of oscillators per layer.

Returns

FloatArray The natural frequencies spanning plasma timescales.

Source code in src/scpn_phase_orchestrator/adapters/plasma_control_bridge.py

def import_plasma_omega(self, n_osc_per_layer: int = 1) -> FloatArray:
    """Generate natural frequencies spanning plasma timescales.

    Returns frequencies ordered: micro_turbulence(fast) → plasma_wall(slow).

    Parameters
    ----------
    n_osc_per_layer : int
        Number of oscillators per layer.

    Returns
    -------
    FloatArray
        The natural frequencies spanning plasma timescales.
    """
    n_osc_per_layer = _validate_positive_int(
        n_osc_per_layer,
        name="n_osc_per_layer",
    )
    # 8 characteristic plasma timescales (normalised, rad/s)
    layer_omegas = np.array(
        [
            10.0,  # micro_turbulence
            8.0,  # zonal_flow
            3.0,  # mhd_tearing
            5.0,  # sawtooth_elm
            0.5,  # transport_barrier
            0.3,  # current_profile
            0.1,  # global_equilibrium
            1.0,  # plasma_wall
        ],
        dtype=np.float64,
    )[: self._n_layers]

    if n_osc_per_layer == 1:
        return layer_omegas
    return np.repeat(layer_omegas, n_osc_per_layer)

import_snapshot

import_snapshot(tick_result: dict[str, Any]) -> UPDEState

Convert an scpn-control tick result dict to UPDEState.

Expected keys: 'phases' (1-D array), optional 'regime', 'layer_sizes'.

Parameters

tick_result : dict[str, Any] An scpn-control tick result dict.

Returns

UPDEState The UPDE state for the tick result.

Raises

ValueError If the tick result is malformed.

Source code in src/scpn_phase_orchestrator/adapters/plasma_control_bridge.py

def import_snapshot(self, tick_result: dict[str, Any]) -> UPDEState:
    """Convert an scpn-control tick result dict to UPDEState.

    Expected keys: 'phases' (1-D array), optional 'regime', 'layer_sizes'.

    Parameters
    ----------
    tick_result : dict[str, Any]
        An scpn-control tick result dict.

    Returns
    -------
    UPDEState
        The UPDE state for the tick result.

    Raises
    ------
    ValueError
        If the tick result is malformed.
    """
    if not isinstance(tick_result, dict):
        raise ValueError("tick_result must be a dict")
    phases = _finite_array(tick_result["phases"], name="phases")
    if phases.ndim != 1:
        raise ValueError("phases must be a 1-D array")
    if phases.size == 0:
        raise ValueError("phases must not be empty")
    phases = phases % TWO_PI
    regime = _label(tick_result.get("regime", "NOMINAL"), name="regime")
    layer_sizes = tick_result.get("layer_sizes")

    if layer_sizes is None:
        n_per = max(1, len(phases) // self._n_layers)
        layer_sizes = [n_per] * self._n_layers
    else:
        layer_sizes = _validate_layer_sizes(
            layer_sizes,
            n_phases=len(phases),
            n_layers=self._n_layers,
        )

    layers: list[LayerState] = []
    offset = 0
    for size in layer_sizes:
        group = phases[offset : offset + size]
        offset += size
        if len(group) == 0:
            layers.append(LayerState(R=0.0, psi=0.0))
            continue
        z = np.exp(1j * group)
        order = z.mean()
        r_val = float(np.abs(order))
        psi_val = float(np.angle(order) % TWO_PI)
        layers.append(LayerState(R=r_val, psi=psi_val))

    n_l = len(layers)
    cross = np.eye(n_l, dtype=np.float64)
    stability = _unit_interval(tick_result.get("stability", 0.5), name="stability")

    return UPDEState(
        layers=layers,
        cross_layer_alignment=cross,
        stability_proxy=stability,
        regime_id=regime,
    )

import_lyapunov_verdict

import_lyapunov_verdict(
    verdict_or_dict: object,
) -> dict[str, Any]

Map a Lyapunov verdict to a boundary-compatible signal dict.

Accepts dict with 'score' (float in [0,1]).

Parameters

verdict_or_dict : object A Lyapunov verdict as an object or dict.

Returns

dict[str, Any] The boundary-compatible signal dict for the verdict.

Source code in src/scpn_phase_orchestrator/adapters/plasma_control_bridge.py

def import_lyapunov_verdict(self, verdict_or_dict: object) -> dict[str, Any]:
    """Map a Lyapunov verdict to a boundary-compatible signal dict.

    Accepts dict with 'score' (float in [0,1]).

    Parameters
    ----------
    verdict_or_dict : object
        A Lyapunov verdict as an object or dict.

    Returns
    -------
    dict[str, Any]
        The boundary-compatible signal dict for the verdict.
    """
    if isinstance(verdict_or_dict, dict):
        score = _unit_interval(verdict_or_dict.get("score", 0.0), name="score")
    else:
        score = _unit_interval(getattr(verdict_or_dict, "score", 0.0), name="score")
    return {
        "lyapunov_score": score,
        "stable": score > 0.3,
    }

export_control_actions

export_control_actions(
    actions: list[Any],
) -> dict[str, Any]

Package a list of control action dicts for scpn-control consumption.

Parameters

actions : list[Any] Control action dicts to package.

Returns

dict[str, Any] The control actions packaged for scpn-control.

Raises

ValueError If an action dict is invalid.

Source code in src/scpn_phase_orchestrator/adapters/plasma_control_bridge.py

def export_control_actions(self, actions: list[Any]) -> dict[str, Any]:
    """Package a list of control action dicts for scpn-control consumption.

    Parameters
    ----------
    actions : list[Any]
        Control action dicts to package.

    Returns
    -------
    dict[str, Any]
        The control actions packaged for scpn-control.

    Raises
    ------
    ValueError
        If an action dict is invalid.
    """
    if not isinstance(actions, list):
        raise ValueError("actions must be a list of dicts")
    exported: list[dict[str, object]] = []
    for action in actions:
        if not isinstance(action, dict):
            raise ValueError("actions must contain dict entries")
        knob = _label(action.get("knob", "K"), name="knob")
        scope = _label(action.get("scope", "global"), name="scope")
        value = _finite_real(action.get("value", 0.0), name="value")
        exported.append({"knob": knob, "scope": scope, "value": value})
    return {
        "actions": exported,
    }

check_physics_invariants

check_physics_invariants(
    values: dict[str, Any],
) -> list[dict[str, Any]]

Check plasma physics invariants against local thresholds.

Returns a list of violation dicts (empty if all invariants hold).

Parameters

values : dict[str, Any] Plasma observable values keyed by name.

Returns

list[dict[str, Any]] The list of physics-invariant violations.

Raises

ValueError If the observable values are invalid.

Source code in src/scpn_phase_orchestrator/adapters/plasma_control_bridge.py

def check_physics_invariants(self, values: dict[str, Any]) -> list[dict[str, Any]]:
    """Check plasma physics invariants against local thresholds.

    Returns a list of violation dicts (empty if all invariants hold).

    Parameters
    ----------
    values : dict[str, Any]
        Plasma observable values keyed by name.

    Returns
    -------
    list[dict[str, Any]]
        The list of physics-invariant violations.

    Raises
    ------
    ValueError
        If the observable values are invalid.
    """
    if not isinstance(values, dict):
        raise ValueError("physics invariant values must be a dict")
    violations: list[dict[str, Any]] = []
    q_min = values.get("q_min")
    if q_min is not None:
        q_min = _finite_positive_real(q_min, name="physics invariant q_min")
    if q_min is not None and q_min < Q_MIN_STABLE:
        violations.append(
            {
                "variable": "q_min",
                "value": q_min,
                "threshold": Q_MIN_STABLE,
                "severity": "hard",
                "message": f"q_min={q_min:.3f} < {Q_MIN_STABLE}",
            }
        )
    beta_n = values.get("beta_n")
    if beta_n is not None:
        beta_n = _finite_non_negative_real(
            beta_n,
            name="physics invariant beta_n",
        )
    if beta_n is not None and beta_n > BETA_N_LIMIT:
        violations.append(
            {
                "variable": "beta_n",
                "value": beta_n,
                "threshold": BETA_N_LIMIT,
                "severity": "hard",
                "message": f"beta_n={beta_n:.3f} > {BETA_N_LIMIT} (Troyon no-wall)",
            }
        )
    greenwald = values.get("greenwald")
    if greenwald is not None:
        greenwald = _finite_non_negative_real(
            greenwald,
            name="physics invariant greenwald",
        )
    if greenwald is not None and greenwald > GREENWALD_LIMIT:
        violations.append(
            {
                "variable": "greenwald",
                "value": greenwald,
                "threshold": GREENWALD_LIMIT,
                "severity": "hard",
                "message": f"greenwald={greenwald:.3f} > {GREENWALD_LIMIT}",
            }
        )
    return violations

Quantum Control Bridge

quantum_control_bridge

Quantum-control bridge for reviewable Hamiltonian and phase handoffs.

The bridge imports quantum phase artifacts into UPDE diagnostics, exports UPDE state summaries, validates coupling/frequency arrays, and can build deterministic OpenQASM manifest handoffs with parity hashes and actuation disabled. Live Hamiltonian or Q-UPDE execution is delegated only when external quantum-control packages are explicitly imported by the called method.

Classes

QuantumControlBridge

QuantumControlBridge(
    n_oscillators: int, trotter_order: int = 1
)

Adapter between scpn-quantum-control artifacts and phase-orchestrator types.

The QuantumControlBridge enables the mapping of classical Kuramoto phase dynamics onto Quantum Hardware (isomorphic XY spin Hamiltonian). It supports Hamiltonian construction, Trotterized time evolution (Q-UPDE), and variational synchronization minimization.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def __init__(self, n_oscillators: int, trotter_order: int = 1):
    n_oscillators = _require_positive_integer(n_oscillators, name="n_oscillators")
    if isinstance(trotter_order, bool) or not isinstance(trotter_order, Integral):
        raise ValueError("trotter_order must be an integer >= 1")
    if trotter_order < 1:
        raise ValueError("trotter_order must be an integer >= 1")
    self._n: int = int(n_oscillators)
    self._trotter_order: int = int(trotter_order)

Methods:

import_artifact

import_artifact(artifact_dict: dict[str, Any]) -> UPDEState

Convert a scpn-quantum-control result dict into UPDEState.

Parameters

artifact_dict : dict[str, Any] An scpn-quantum-control result dict.

Returns

UPDEState The UPDE state for the quantum result.

Raises

ValueError If the artifact dict is malformed.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def import_artifact(self, artifact_dict: dict[str, Any]) -> UPDEState:
    """Convert a scpn-quantum-control result dict into UPDEState.

    Parameters
    ----------
    artifact_dict : dict[str, Any]
        An scpn-quantum-control result dict.

    Returns
    -------
    UPDEState
        The UPDE state for the quantum result.

    Raises
    ------
    ValueError
        If the artifact dict is malformed.
    """
    artifact = _require_mapping(artifact_dict, name="artifact_dict")
    if "phases" not in artifact:
        raise ValueError("artifact_dict must include 'phases'")
    phases = _finite_array(artifact["phases"], name="phases")
    if phases.shape != (self._n,):
        raise ValueError(
            f"phases shape {phases.shape} does not match n_oscillators={self._n}"
        )
    phases = phases % TWO_PI
    fidelity = _require_fidelity(artifact.get("fidelity", 0.0), name="fidelity")

    layer_assignments = artifact.get("layer_assignments")
    if layer_assignments is None:
        mid = len(phases) // 2
        layer_assignments = [list(range(mid)), list(range(mid, len(phases)))]
    layer_assignments = _validate_layer_assignments(
        layer_assignments,
        n_phases=len(phases),
    )

    layers: list[LayerState] = []
    for group in layer_assignments:
        if len(group) == 0:
            layers.append(LayerState(R=0.0, psi=0.0))
            continue
        z = np.exp(1j * phases[group])
        order = z.mean()
        r_val = float(np.abs(order))
        psi_val = float(np.angle(order) % TWO_PI)
        layers.append(LayerState(R=r_val, psi=psi_val))

    n_layers = len(layers)
    cross = np.eye(n_layers, dtype=np.float64)
    regime = str(artifact.get("regime", "NOMINAL"))

    return UPDEState(
        layers=layers,
        cross_layer_alignment=cross,
        stability_proxy=fidelity,
        regime_id=regime,
    )

export_artifact

export_artifact(state: UPDEState) -> dict[str, Any]

Convert UPDEState back to a dict compatible with scpn-quantum-control.

Parameters

state : UPDEState The current UPDE state.

Returns

dict[str, Any] The scpn-quantum-control-compatible state dict.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def export_artifact(self, state: UPDEState) -> dict[str, Any]:
    """Convert UPDEState back to a dict compatible with scpn-quantum-control.

    Parameters
    ----------
    state : UPDEState
        The current UPDE state.

    Returns
    -------
    dict[str, Any]
        The scpn-quantum-control-compatible state dict.
    """
    layers, cross_layer_alignment = _validate_upde_state(state)
    fidelity = _require_fidelity(
        state.stability_proxy,
        name="state.stability_proxy",
    )
    return {
        "regime": state.regime_id,
        "fidelity": fidelity,
        "layers": [{"R": ls.R, "psi": ls.psi} for ls in layers],
        "cross_alignment": cross_layer_alignment.tolist(),
    }

import_knm

import_knm(knm_array: FloatArray) -> CouplingState

Wrap a coupling matrix from quantum calibration into CouplingState.

Parameters

knm_array : FloatArray A coupling matrix from quantum calibration, shape (N, N).

Returns

CouplingState The coupling state wrapping the calibration matrix.

Raises

ValueError If the coupling matrix is invalid.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def import_knm(self, knm_array: FloatArray) -> CouplingState:
    """Wrap a coupling matrix from quantum calibration into CouplingState.

    Parameters
    ----------
    knm_array : FloatArray
        A coupling matrix from quantum calibration, shape ``(N, N)``.

    Returns
    -------
    CouplingState
        The coupling state wrapping the calibration matrix.

    Raises
    ------
    ValueError
        If the coupling matrix is invalid.
    """
    knm = _finite_array(knm_array, name="Knm")
    if knm.ndim != 2 or knm.shape[0] != knm.shape[1]:
        raise ValueError(f"Knm must be square, got shape {knm.shape}")
    if knm.shape != (self._n, self._n):
        raise ValueError(
            f"Knm shape {knm.shape} does not match n_oscillators={self._n}"
        )
    n = knm.shape[0]
    return CouplingState(
        knm=knm.copy(),
        alpha=np.zeros((n, n), dtype=np.float64),
        active_template="quantum_import",
    )

import_scpn_upde_edge

import_scpn_upde_edge(
    edge_payload: dict[str, object],
) -> dict[str, object]

Import a QUANTUM knm.scpn-upde edge under a bounded scope.

The accepted edge carries K_nm and omega arrays plus Trotter metadata. SPO recomputes the payload digests and its own deterministic compiler manifest. It does not permit QPU execution or actuation.

Parameters

edge_payload : dict[str, object] A payload emitted by scpn_quantum_control.bridge.scpn_upde_edge.

Returns

dict[str, object] Import evidence containing the coupling state and compiler manifest.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def import_scpn_upde_edge(
    self, edge_payload: dict[str, object]
) -> dict[str, object]:
    """Import a QUANTUM ``knm.scpn-upde`` edge under a bounded scope.

    The accepted edge carries ``K_nm`` and ``omega`` arrays plus Trotter
    metadata. SPO recomputes the payload digests and its own deterministic
    compiler manifest. It does not permit QPU execution or actuation.

    Parameters
    ----------
    edge_payload : dict[str, object]
        A payload emitted by
        ``scpn_quantum_control.bridge.scpn_upde_edge``.

    Returns
    -------
    dict[str, object]
        Import evidence containing the coupling state and compiler manifest.
    """
    edge = _require_mapping(edge_payload, name="edge_payload")
    knm, omegas, dt = self._validate_scpn_upde_edge(edge)
    coupling = self.import_knm(knm)
    manifest = self.build_quantum_compiler_manifest(knm, omegas, dt=dt)
    record: dict[str, object] = {
        "schema": "spo.quantum-control.scpn-upde-import.v1",
        "status": "accepted_computational_agreement",
        "accepted_schema": SCPN_UPDE_EDGE_SCHEMA,
        "scope_envelope": SCPN_UPDE_SCOPE_ENVELOPE,
        "edge_sha256": edge["edge_sha256"],
        "n_oscillators": self._n,
        "coupling_state": coupling,
        "compiler_manifest": manifest,
        "qpu_execution_permitted": False,
        "actuation_permitted": False,
    }
    canonical_record = {
        key: value
        for key, value in record.items()
        if key not in {"coupling_state", "compiler_manifest"}
    }
    record["import_sha256"] = _digest_mapping(canonical_record)
    return record

build_quantum_compiler_manifest

build_quantum_compiler_manifest(
    knm: FloatArray, omegas: FloatArray, *, dt: float
) -> dict[str, object]

Return a deterministic OpenQASM handoff with parity evidence.

The manifest is dependency-free review output for Qiskit/PennyLane simulator handoff. It does not execute on a QPU and does not permit live actuation.

Parameters

knm : FloatArray Coupling matrix K_nm, shape (N, N). omegas : FloatArray Natural frequencies in rad/s, shape (N,). dt : float Integration step size.

Returns

dict[str, object] The deterministic OpenQASM handoff with parity evidence.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def build_quantum_compiler_manifest(
    self,
    knm: FloatArray,
    omegas: FloatArray,
    *,
    dt: float,
) -> dict[str, object]:
    """Return a deterministic OpenQASM handoff with parity evidence.

    The manifest is dependency-free review output for Qiskit/PennyLane
    simulator handoff. It does not execute on a QPU and does not permit
    live actuation.

    Parameters
    ----------
    knm : FloatArray
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.
    dt : float
        Integration step size.

    Returns
    -------
    dict[str, object]
        The deterministic OpenQASM handoff with parity evidence.
    """
    knm_array, omega_array = self._validate_compiler_inputs(knm, omegas, dt=dt)
    frequency_terms: list[dict[str, object]] = [
        {
            "qubit": idx,
            "omega": float(omega_array[idx]),
            "angle": float(omega_array[idx] * dt),
        }
        for idx in range(self._n)
    ]
    coupling_terms = self._quantum_coupling_terms(knm_array, dt=dt)
    openqasm = self._render_openqasm(frequency_terms, coupling_terms)
    qasm_hash = sha256(openqasm.encode("utf-8")).hexdigest()
    parity = self._quantum_compiler_parity(
        omega_array,
        frequency_terms,
        knm_array,
        coupling_terms,
    )
    conformance = check_openqasm3(openqasm)
    parity["qasm_parse_ok"] = conformance.conformant
    manifest: dict[str, object] = {
        "manifest_kind": "quantum_compiler_manifest",
        "schema_version": 1,
        "status": (
            "co_simulation_parity_passed"
            if parity["max_abs_frequency_error"] == 0.0
            and parity["max_abs_coupling_error"] == 0.0
            else "co_simulation_parity_failed"
        ),
        "target_backends": ["qiskit_openqasm3", "pennylane_qasm"],
        "n_qubits": self._n,
        "trotter_order": self._trotter_order,
        "dt": float(dt),
        "qpu_execution_permitted": False,
        "actuation_permitted": False,
        "frequency_terms": frequency_terms,
        "coupling_terms": coupling_terms,
        "openqasm": openqasm,
        "qasm_sha256": qasm_hash,
        "openqasm_conformance": conformance.to_audit_record(),
        "co_simulation_parity": parity,
        "operator_commands": [
            "review quantum_compiler_manifest.json",
            "run Qiskit or PennyLane simulator parity before QPU handoff",
        ],
    }
    canonical = json.dumps(manifest, sort_keys=True, separators=(",", ":"))
    manifest["manifest_sha256"] = sha256(canonical.encode("utf-8")).hexdigest()
    return manifest

audit_qpu_target_readiness

audit_qpu_target_readiness(
    manifest: dict[str, object],
    *,
    target_backend: str,
    provider: str,
    credentials_configured: bool = False,
    operator_approved: bool = False,
) -> dict[str, object]

Return non-executing QPU target-readiness evidence.

The audit validates that a quantum compiler manifest is suitable for a named target backend and records whether operator preconditions are in place. It never runs a simulator, submits a QPU job, or flips the manifest execution/actuation permissions.

Parameters

manifest : dict[str, object] The compiler manifest to audit. target_backend : str Name of the target backend. provider : str Name of the hardware provider. credentials_configured : bool Whether provider credentials are configured. operator_approved : bool Whether a human operator approved the target.

Returns

dict[str, object] The non-executing QPU target-readiness evidence.

Raises

ValueError If the manifest or target details are invalid.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def audit_qpu_target_readiness(
    self,
    manifest: dict[str, object],
    *,
    target_backend: str,
    provider: str,
    credentials_configured: bool = False,
    operator_approved: bool = False,
) -> dict[str, object]:
    """Return non-executing QPU target-readiness evidence.

    The audit validates that a quantum compiler manifest is suitable for a
    named target backend and records whether operator preconditions are in
    place. It never runs a simulator, submits a QPU job, or flips the
    manifest execution/actuation permissions.

    Parameters
    ----------
    manifest : dict[str, object]
        The compiler manifest to audit.
    target_backend : str
        Name of the target backend.
    provider : str
        Name of the hardware provider.
    credentials_configured : bool
        Whether provider credentials are configured.
    operator_approved : bool
        Whether a human operator approved the target.

    Returns
    -------
    dict[str, object]
        The non-executing QPU target-readiness evidence.

    Raises
    ------
    ValueError
        If the manifest or target details are invalid.
    """
    manifest_record = _require_mapping(manifest, name="manifest")
    target_backend = _require_non_empty_text(
        target_backend,
        name="target_backend",
    )
    provider = _require_non_empty_text(provider, name="provider")
    if not isinstance(credentials_configured, bool):
        raise ValueError("credentials_configured must be a boolean")
    if not isinstance(operator_approved, bool):
        raise ValueError("operator_approved must be a boolean")
    if manifest_record.get("manifest_kind") != "quantum_compiler_manifest":
        raise ValueError("manifest must be a quantum_compiler_manifest")

    target_backends = manifest_record.get("target_backends")
    if not isinstance(target_backends, list) or not all(
        isinstance(item, str) for item in target_backends
    ):
        raise ValueError("manifest target_backends must be a list of strings")
    if target_backend not in target_backends:
        raise ValueError("target_backend is not declared by manifest")

    blocked_reasons: list[str] = []
    if manifest_record.get("status") != "co_simulation_parity_passed":
        blocked_reasons.append("co_simulation_parity_not_passed")
    if manifest_record.get("qpu_execution_permitted") is not False:
        blocked_reasons.append("qpu_execution_permission_must_remain_false")
    if manifest_record.get("actuation_permitted") is not False:
        blocked_reasons.append("actuation_permission_must_remain_false")
    if not credentials_configured:
        blocked_reasons.append("credentials_not_configured")
    if not operator_approved:
        blocked_reasons.append("operator_approval_missing")

    manifest_sha = str(manifest_record.get("manifest_sha256", ""))
    record: dict[str, object] = {
        "schema": "scpn_quantum_target_readiness_v1",
        "provider": provider,
        "target_backend": target_backend,
        "manifest_sha256": manifest_sha,
        "status": "blocked" if blocked_reasons else "ready_not_executed",
        "blocked_reasons": blocked_reasons,
        "credentials_configured": credentials_configured,
        "operator_approved": operator_approved,
        "qpu_execution_permitted": False,
        "actuation_permitted": False,
        "operator_commands": [
            "review quantum_compiler_manifest.json",
            "run simulator parity outside SPO before target handoff",
            "submit QPU job only from an approved external operator workflow",
        ],
    }
    canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
    record["readiness_sha256"] = sha256(canonical.encode("utf-8")).hexdigest()
    return record

build_hamiltonian

build_hamiltonian(
    knm: FloatArray, omegas: FloatArray
) -> object

Build Kuramoto XY Hamiltonian as SparsePauliOp.

Requires scpn-quantum-control.

Parameters

knm : FloatArray Coupling matrix K_nm, shape (N, N). omegas : FloatArray Natural frequencies in rad/s, shape (N,).

Returns

object The Kuramoto XY Hamiltonian as a SparsePauliOp.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def build_hamiltonian(self, knm: FloatArray, omegas: FloatArray) -> object:
    """Build Kuramoto XY Hamiltonian as SparsePauliOp.

    Requires scpn-quantum-control.

    Parameters
    ----------
    knm : FloatArray
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.

    Returns
    -------
    object
        The Kuramoto XY Hamiltonian as a SparsePauliOp.
    """
    knm, omegas = self._validate_knm_omegas(knm, omegas)
    from scpn_quantum_control.bridge.knm_hamiltonian import knm_to_hamiltonian

    return knm_to_hamiltonian(knm, omegas)

solve_q_upde

solve_q_upde(
    knm: FloatArray,
    omegas: FloatArray,
    t_max: float = 1.0,
    dt: float = 0.1,
    trotter_per_step: int = 5,
) -> dict[str, Any]

Execute Trotterized quantum simulation of the phase network (Q-UPDE).

This method maps the classical sin(delta theta) interaction to the XY spin exchange interaction (XX + YY) and natural frequencies to Z-axis magnetic fields.

Requires scpn-quantum-control.

Parameters

knm : FloatArray Coupling matrix K_nm, shape (N, N). omegas : FloatArray Natural frequencies in rad/s, shape (N,). t_max : float Total simulation time. dt : float Integration step size. trotter_per_step : int Number of Trotter steps per outer step.

Returns

dict[str, Any] The Trotterised Q-UPDE simulation result.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def solve_q_upde(
    self,
    knm: FloatArray,
    omegas: FloatArray,
    t_max: float = 1.0,
    dt: float = 0.1,
    trotter_per_step: int = 5,
) -> dict[str, Any]:
    """Execute Trotterized quantum simulation of the phase network (Q-UPDE).

    This method maps the classical sin(delta theta) interaction to
    the XY spin exchange interaction (XX + YY) and natural frequencies
    to Z-axis magnetic fields.

    Requires scpn-quantum-control.

    Parameters
    ----------
    knm : FloatArray
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.
    t_max : float
        Total simulation time.
    dt : float
        Integration step size.
    trotter_per_step : int
        Number of Trotter steps per outer step.

    Returns
    -------
    dict[str, Any]
        The Trotterised Q-UPDE simulation result.
    """
    t_max = _require_positive_real(t_max, name="t_max")
    dt = _require_positive_real(dt, name="dt")
    trotter_per_step = _require_positive_integer(
        trotter_per_step,
        name="trotter_per_step",
    )
    knm, omegas = self._validate_knm_omegas(knm, omegas)
    from scpn_quantum_control.phase.xy_kuramoto import QuantumKuramotoSolver

    solver = QuantumKuramotoSolver(
        n_oscillators=len(omegas),
        K_coupling=knm,
        omega_natural=omegas,
        trotter_order=self._trotter_order,
    )
    return cast(
        "dict[str, object]",
        solver.run(t_max=t_max, dt=dt, trotter_per_step=trotter_per_step),
    )

orchestrator_to_quantum

orchestrator_to_quantum(state: UPDEState) -> FloatArray

Convert orchestrator UPDEState to quantum phase array.

Parameters

state : UPDEState The current UPDE state.

Returns

FloatArray The quantum phase array for the UPDE state.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def orchestrator_to_quantum(
    self,
    state: UPDEState,
) -> FloatArray:
    """Convert orchestrator UPDEState to quantum phase array.

    Parameters
    ----------
    state : UPDEState
        The current UPDE state.

    Returns
    -------
    FloatArray
        The quantum phase array for the UPDE state.
    """
    payload = self.export_artifact(state)
    from scpn_quantum_control import (  # noqa: PLC0415
        orchestrator_to_quantum_phases,
    )

    layer_phases = {
        f"layer_{i}": ls["psi"] for i, ls in enumerate(payload["layers"])
    }
    return cast("FloatArray", orchestrator_to_quantum_phases(layer_phases))

quantum_to_orchestrator

quantum_to_orchestrator(
    quantum_theta: FloatArray,
) -> dict[str, Any]

Convert quantum phase array back to orchestrator-compatible dict.

Parameters

quantum_theta : FloatArray Quantum phase array, shape (N,).

Returns

dict[str, Any] The orchestrator-compatible dict for the quantum phases.

Raises

ValueError If the quantum phase array is invalid.

Source code in src/scpn_phase_orchestrator/adapters/quantum_control_bridge.py

def quantum_to_orchestrator(
    self,
    quantum_theta: FloatArray,
) -> dict[str, Any]:
    """Convert quantum phase array back to orchestrator-compatible dict.

    Parameters
    ----------
    quantum_theta : FloatArray
        Quantum phase array, shape ``(N,)``.

    Returns
    -------
    dict[str, Any]
        The orchestrator-compatible dict for the quantum phases.

    Raises
    ------
    ValueError
        If the quantum phase array is invalid.
    """
    theta = _finite_array(quantum_theta, name="quantum_theta")
    if theta.shape != (self._n,):
        raise ValueError(
            f"quantum_theta shape {theta.shape} does not match "
            f"n_oscillators={self._n}"
        )
    from scpn_quantum_control import (  # noqa: PLC0415
        quantum_to_orchestrator_phases,
    )

    return cast("dict[str, Any]", quantum_to_orchestrator_phases(theta.copy()))

Functions:

OpenQASM 3 Conformance

check_openqasm3 statically validates the structural conformance of the OpenQASM 3 text the quantum-control bridge emits: version header, includes, qubit-register declarations, custom gate declarations, and every gate application's parameter/qubit arity and register bounds. Its gate registry is split into honestly-labelled tiers — the gates stdgates.inc actually defines versus the two-qubit Pauli-rotation extensions (rxx/ryy/rzz/rzx) that Qiskit and PennyLane provide as builtins — so a program using the extensions is recorded as portable-to-those-backends rather than mislabelled as pure-standard conformant. The bridge embeds the report under the manifest's openqasm_conformance key and surfaces qasm_parse_ok in the co-simulation parity evidence.

openqasm_conformance

Deterministic structural conformance checker for OpenQASM 3 programs.

This module validates the structural conformance of an OpenQASM 3 program: version header, includes, quantum-register declarations, custom gate declarations, and gate applications (measure / reset / barrier are recognised as non-gate operations). For every gate application it resolves the gate name against a known registry and checks that the classical-parameter count and qubit-operand count match the gate's arity, and that every indexed qubit operand refers to a declared register within bounds.

Scope and honesty boundary

The checker is a static structural validator, not a full OpenQASM 3 parser, type checker, or simulator: it does not evaluate parameter expressions, classical control flow (if / for / while), subroutines (def), timing, or pulse-level constructs. Statements outside the checked subset are surfaced in :attr:OpenQasm3ConformanceReport.unchecked_statements and never silently pass as "conformant".

The known-gate registry is split into two honestly-labelled tiers:

• :data:STANDARD_LIBRARY_GATES — the gates the OpenQASM 3 standard library header stdgates.inc actually defines. This table is transcribed from the reference definitions at https://github.com/openqasm/openqasm/blob/main/examples/stdgates.inc and the standard-library documentation, verified at source on 2026-07-21. It notably does not contain the two-qubit Pauli-rotation gates rxx / ryy / rzz / rzx.
• :data:BACKEND_EXTENSION_GATES — the two-qubit Pauli-rotation gates that common OpenQASM 3 target backends (Qiskit's qiskit.qasm3 importer and PennyLane's QASM loader) provide as builtins even though they are absent from stdgates.inc. A program that uses these is portable to those backends but is not pure-stdgates.inc conformant, so the report records their use explicitly via :attr:OpenQasm3ConformanceReport.extension_gates_used rather than blurring the distinction.

Classes

OpenQasm3ConformanceReport dataclass

OpenQasm3ConformanceReport(
    conformant: bool,
    qasm_version: str | None,
    includes: tuple[str, ...],
    qubit_registers: tuple[tuple[str, int], ...],
    gate_call_count: int,
    stdgates_used: tuple[str, ...],
    extension_gates_used: tuple[str, ...],
    custom_gates_declared: tuple[str, ...],
    issues: tuple[str, ...] = tuple(),
    unchecked_statements: tuple[str, ...] = tuple(),
)

Structured result of an OpenQASM 3 structural conformance check.

Attributes

uses_non_stdgates_extensions property

uses_non_stdgates_extensions: bool

Return whether the program uses gates absent from stdgates.inc.

Returns

bool True when at least one applied gate is a backend extension (:data:BACKEND_EXTENSION_GATES) rather than a standard-library or in-program gate.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a deterministic JSON-safe audit mapping of this report.

Returns

dict[str, object] A sorted, JSON-serialisable mapping suitable for embedding in a review manifest.

Source code in src/scpn_phase_orchestrator/adapters/openqasm_conformance.py

def to_audit_record(self) -> dict[str, object]:
    """Return a deterministic JSON-safe audit mapping of this report.

    Returns
    -------
    dict[str, object]
        A sorted, JSON-serialisable mapping suitable for embedding in a
        review manifest.
    """
    return {
        "conformant": self.conformant,
        "qasm_version": self.qasm_version,
        "includes": list(self.includes),
        "qubit_registers": [
            {"name": name, "size": size} for name, size in self.qubit_registers
        ],
        "gate_call_count": self.gate_call_count,
        "stdgates_used": list(self.stdgates_used),
        "extension_gates_used": list(self.extension_gates_used),
        "custom_gates_declared": list(self.custom_gates_declared),
        "uses_non_stdgates_extensions": self.uses_non_stdgates_extensions,
        "issue_count": len(self.issues),
        "issues": list(self.issues),
        "unchecked_statement_count": len(self.unchecked_statements),
    }

Functions:

check_openqasm3

check_openqasm3(program: str) -> OpenQasm3ConformanceReport

Check the structural conformance of an OpenQASM 3 program.

The function never raises: any structural violation is recorded as an issue and reflected in :attr:OpenQasm3ConformanceReport.conformant.

Parameters

program : str The OpenQASM 3 source text to validate.

Returns

OpenQasm3ConformanceReport The structural conformance report. conformant is True only when a version header is present and no structural issue was found.

Raises

TypeError If program is not a string.

Source code in src/scpn_phase_orchestrator/adapters/openqasm_conformance.py

def check_openqasm3(program: str) -> OpenQasm3ConformanceReport:
    """Check the structural conformance of an OpenQASM 3 program.

    The function never raises: any structural violation is recorded as an issue
    and reflected in :attr:`OpenQasm3ConformanceReport.conformant`.

    Parameters
    ----------
    program : str
        The OpenQASM 3 source text to validate.

    Returns
    -------
    OpenQasm3ConformanceReport
        The structural conformance report. ``conformant`` is ``True`` only when
        a version header is present and no structural issue was found.

    Raises
    ------
    TypeError
        If *program* is not a string.
    """
    if not isinstance(program, str):
        raise TypeError("program must be a string")
    state = _ConformanceState()
    statements = _split_statements(_strip_comments(program))
    for index, statement in enumerate(statements):
        _process_statement(state, statement, index)
    if state.version is None:
        state.issues.append("missing OPENQASM version header")
    # Only ``stdgates.inc`` gates require the include; the ``U``/``gphase``
    # builtins are available without any include.
    uses_stdgates = any(gate in STANDARD_LIBRARY_GATES for gate in state.stdgates_used)
    if uses_stdgates and "stdgates.inc" not in state.includes:
        state.issues.append(
            'standard-library gates used without include "stdgates.inc"'
        )
    return OpenQasm3ConformanceReport(
        conformant=not state.issues,
        qasm_version=state.version,
        includes=tuple(state.includes),
        qubit_registers=tuple(sorted(state.qubit_registers.items())),
        gate_call_count=state.gate_calls,
        stdgates_used=tuple(sorted(state.stdgates_used)),
        extension_gates_used=tuple(sorted(state.extension_gates_used)),
        custom_gates_declared=tuple(sorted(state.custom_gates)),
        issues=tuple(state.issues),
        unchecked_statements=tuple(state.unchecked),
    )

Hybrid Co-Compiler

hybrid_cocompiler

Deterministic hybrid co-compiler review manifests.

Functions:

build_hybrid_cocompiler_manifest

build_hybrid_cocompiler_manifest(
    quantum_manifest: Mapping[str, object],
    neuromorphic_manifest: Mapping[str, object],
    *,
    n_channel_semantics: Sequence[str] = (
        "Q_control",
        "S_spike",
        "audit",
    ),
) -> dict[str, object]

Combine quantum and spiking manifests under one audit envelope.

Parameters

quantum_manifest : Mapping[str, object] The quantum compiler manifest. neuromorphic_manifest : Mapping[str, object] The neuromorphic schedule manifest. n_channel_semantics : Sequence[str] Per-channel semantic labels.

Returns

dict[str, object] The combined quantum/neuromorphic hybrid manifest.

Source code in src/scpn_phase_orchestrator/adapters/hybrid_cocompiler.py

def build_hybrid_cocompiler_manifest(
    quantum_manifest: Mapping[str, object],
    neuromorphic_manifest: Mapping[str, object],
    *,
    n_channel_semantics: Sequence[str] = ("Q_control", "S_spike", "audit"),
) -> dict[str, object]:
    """Combine quantum and spiking manifests under one audit envelope.

    Parameters
    ----------
    quantum_manifest : Mapping[str, object]
        The quantum compiler manifest.
    neuromorphic_manifest : Mapping[str, object]
        The neuromorphic schedule manifest.
    n_channel_semantics : Sequence[str]
        Per-channel semantic labels.

    Returns
    -------
    dict[str, object]
        The combined quantum/neuromorphic hybrid manifest.
    """
    quantum_manifest = _validate_manifest_mapping(
        quantum_manifest,
        label="quantum_manifest",
    )
    neuromorphic_manifest = _validate_manifest_mapping(
        neuromorphic_manifest,
        label="neuromorphic_manifest",
    )
    _validate_manifest_kind(
        quantum_manifest,
        expected="quantum_compiler_manifest",
        label="quantum manifest kind",
    )
    _validate_manifest_kind(
        neuromorphic_manifest,
        expected="neuromorphic_schedule_manifest",
        label="neuromorphic manifest kind",
    )
    _validate_permission_fields(
        quantum_manifest,
        label="quantum",
        fields=("qpu_execution_permitted", "actuation_permitted"),
    )
    _validate_permission_fields(
        neuromorphic_manifest,
        label="neuromorphic",
        fields=("hardware_write_permitted", "actuation_permitted"),
    )
    semantics = _normalise_semantics(n_channel_semantics)
    blocked_reasons = _blocked_reasons(quantum_manifest, neuromorphic_manifest)
    target_backends = _target_backends(quantum_manifest, neuromorphic_manifest)
    component_hashes = _component_hashes(quantum_manifest, neuromorphic_manifest)
    parity = {
        "engine": "hybrid_manifest_status_reconstruction",
        "quantum_status": quantum_manifest.get("status"),
        "neuromorphic_status": neuromorphic_manifest.get("status"),
        "quantum_term_count": _term_count(quantum_manifest.get("co_simulation_parity")),
        "neuromorphic_sample_count": _sample_count(
            neuromorphic_manifest.get("simulator_parity")
        ),
    }
    manifest: dict[str, object] = {
        "manifest_kind": "hybrid_neuromorphic_quantum_cocompiler",
        "schema_version": 1,
        "status": "blocked" if blocked_reasons else "co_simulation_parity_passed",
        "target_backends": target_backends,
        "n_channel_semantics": semantics,
        "component_hashes": component_hashes,
        "co_simulation_parity": parity,
        "blocked_reasons": blocked_reasons,
        "qpu_execution_permitted": False,
        "hardware_write_permitted": False,
        "actuation_permitted": False,
        "operator_commands": [
            "review hybrid_neuromorphic_quantum_cocompiler.json",
            "run quantum and neuromorphic simulators under the shared audit envelope",
        ],
    }
    canonical = json.dumps(manifest, sort_keys=True, separators=(",", ":"))
    manifest["hybrid_manifest_sha256"] = sha256(canonical.encode("utf-8")).hexdigest()
    return manifest

audit_hybrid_target_readiness

audit_hybrid_target_readiness(
    hybrid_manifest: Mapping[str, object],
    quantum_readiness: Mapping[str, object],
    neuromorphic_readiness: Mapping[str, object],
    *,
    hybrid_operator_approved: bool = False,
) -> dict[str, object]

Return non-executing hybrid target-readiness evidence.

The audit links the already review-only hybrid manifest to the independent quantum and neuromorphic target-readiness records. It never submits work to a QPU, simulator, neuromorphic backend, or actuator.

Parameters

hybrid_manifest : Mapping[str, object] The combined hybrid co-compiler manifest. quantum_readiness : Mapping[str, object] Quantum target-readiness evidence. neuromorphic_readiness : Mapping[str, object] Neuromorphic target-readiness evidence. hybrid_operator_approved : bool Whether a human operator approved the hybrid target.

Returns

dict[str, object] The non-executing hybrid target-readiness evidence.

Raises

ValueError If the manifests or readiness evidence are invalid.

Source code in src/scpn_phase_orchestrator/adapters/hybrid_cocompiler.py

def audit_hybrid_target_readiness(
    hybrid_manifest: Mapping[str, object],
    quantum_readiness: Mapping[str, object],
    neuromorphic_readiness: Mapping[str, object],
    *,
    hybrid_operator_approved: bool = False,
) -> dict[str, object]:
    """Return non-executing hybrid target-readiness evidence.

    The audit links the already review-only hybrid manifest to the independent
    quantum and neuromorphic target-readiness records. It never submits work to
    a QPU, simulator, neuromorphic backend, or actuator.

    Parameters
    ----------
    hybrid_manifest : Mapping[str, object]
        The combined hybrid co-compiler manifest.
    quantum_readiness : Mapping[str, object]
        Quantum target-readiness evidence.
    neuromorphic_readiness : Mapping[str, object]
        Neuromorphic target-readiness evidence.
    hybrid_operator_approved : bool
        Whether a human operator approved the hybrid target.

    Returns
    -------
    dict[str, object]
        The non-executing hybrid target-readiness evidence.

    Raises
    ------
    ValueError
        If the manifests or readiness evidence are invalid.
    """
    hybrid_manifest = _validate_manifest_mapping(
        hybrid_manifest,
        label="hybrid_manifest",
    )
    quantum_readiness = _validate_manifest_mapping(
        quantum_readiness,
        label="quantum_readiness",
    )
    neuromorphic_readiness = _validate_manifest_mapping(
        neuromorphic_readiness,
        label="neuromorphic_readiness",
    )
    if not isinstance(hybrid_operator_approved, bool):
        raise ValueError("hybrid_operator_approved must be a bool")
    _validate_manifest_kind(
        hybrid_manifest,
        expected="hybrid_neuromorphic_quantum_cocompiler",
        label="hybrid manifest kind",
    )
    _validate_manifest_kind(
        quantum_readiness,
        expected="scpn_quantum_target_readiness_v1",
        label="quantum_readiness schema",
        key="schema",
    )
    _validate_manifest_kind(
        neuromorphic_readiness,
        expected="scpn_neuromorphic_target_readiness_v1",
        label="neuromorphic_readiness schema",
        key="schema",
    )
    _validate_permission_fields(
        hybrid_manifest,
        label="hybrid",
        fields=(
            "qpu_execution_permitted",
            "hardware_write_permitted",
            "actuation_permitted",
        ),
    )
    _validate_permission_fields(
        quantum_readiness,
        label="quantum_readiness",
        fields=("qpu_execution_permitted", "actuation_permitted"),
    )
    _validate_permission_fields(
        neuromorphic_readiness,
        label="neuromorphic_readiness",
        fields=("hardware_write_permitted", "actuation_permitted"),
    )
    component_hashes = _validate_component_hash_mapping(
        hybrid_manifest.get("component_hashes")
    )
    hybrid_sha = _hash_text(hybrid_manifest, "hybrid_manifest_sha256")
    quantum_readiness_sha = _hash_text(quantum_readiness, "readiness_sha256")
    neuromorphic_readiness_sha = _hash_text(
        neuromorphic_readiness,
        "readiness_sha256",
    )
    blocked_reasons = _hybrid_readiness_blocked_reasons(
        hybrid_manifest,
        quantum_readiness,
        neuromorphic_readiness,
        component_hashes,
        hybrid_operator_approved=hybrid_operator_approved,
    )
    record: dict[str, object] = {
        "schema": "scpn_hybrid_target_readiness_v1",
        "status": "blocked" if blocked_reasons else "ready_not_executed",
        "blocked_reasons": blocked_reasons,
        "hybrid_manifest_sha256": hybrid_sha,
        "quantum_readiness_sha256": quantum_readiness_sha,
        "neuromorphic_readiness_sha256": neuromorphic_readiness_sha,
        "component_manifest_hashes": component_hashes,
        "component_statuses": {
            "hybrid": hybrid_manifest.get("status"),
            "quantum": quantum_readiness.get("status"),
            "neuromorphic": neuromorphic_readiness.get("status"),
        },
        "hybrid_operator_approved": hybrid_operator_approved,
        "qpu_execution_permitted": False,
        "hardware_write_permitted": False,
        "actuation_permitted": False,
        "operator_commands": [
            "review hybrid_neuromorphic_quantum_cocompiler.json",
            "verify quantum and neuromorphic readiness hashes before handoff",
            (
                "submit hybrid execution only from an approved external "
                "operator workflow"
            ),
        ],
    }
    canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
    record["readiness_sha256"] = sha256(canonical.encode("utf-8")).hexdigest()
    return record

build_hybrid_operator_handoff_package

build_hybrid_operator_handoff_package(
    hybrid_manifest: Mapping[str, object],
    hybrid_readiness: Mapping[str, object],
) -> dict[str, object]

Build a deterministic non-executing package for external operators.

Parameters

hybrid_manifest : Mapping[str, object] The combined hybrid co-compiler manifest. hybrid_readiness : Mapping[str, object] Hybrid target-readiness evidence.

Returns

dict[str, object] The deterministic non-executing operator handoff package.

Raises

ValueError If the manifest or readiness evidence is invalid.

Source code in src/scpn_phase_orchestrator/adapters/hybrid_cocompiler.py

def build_hybrid_operator_handoff_package(
    hybrid_manifest: Mapping[str, object],
    hybrid_readiness: Mapping[str, object],
) -> dict[str, object]:
    """Build a deterministic non-executing package for external operators.

    Parameters
    ----------
    hybrid_manifest : Mapping[str, object]
        The combined hybrid co-compiler manifest.
    hybrid_readiness : Mapping[str, object]
        Hybrid target-readiness evidence.

    Returns
    -------
    dict[str, object]
        The deterministic non-executing operator handoff package.

    Raises
    ------
    ValueError
        If the manifest or readiness evidence is invalid.
    """
    hybrid_manifest = _validate_manifest_mapping(
        hybrid_manifest,
        label="hybrid_manifest",
    )
    hybrid_readiness = _validate_manifest_mapping(
        hybrid_readiness,
        label="hybrid_readiness",
    )
    _validate_manifest_kind(
        hybrid_manifest,
        expected="hybrid_neuromorphic_quantum_cocompiler",
        label="hybrid manifest kind",
    )
    _validate_manifest_kind(
        hybrid_readiness,
        expected="scpn_hybrid_target_readiness_v1",
        label="hybrid_readiness schema",
        key="schema",
    )
    _validate_permission_fields(
        hybrid_manifest,
        label="hybrid",
        fields=(
            "qpu_execution_permitted",
            "hardware_write_permitted",
            "actuation_permitted",
        ),
    )
    _validate_permission_fields(
        hybrid_readiness,
        label="hybrid_readiness",
        fields=(
            "qpu_execution_permitted",
            "hardware_write_permitted",
            "actuation_permitted",
        ),
    )
    hybrid_manifest_sha = _hash_text(hybrid_manifest, "hybrid_manifest_sha256")
    hybrid_readiness_sha = _hash_text(hybrid_readiness, "readiness_sha256")
    if hybrid_readiness.get("hybrid_manifest_sha256") != hybrid_manifest_sha:
        raise ValueError("hybrid readiness manifest hash must match hybrid manifest")

    blocked_reasons = _string_list(
        hybrid_readiness.get("blocked_reasons"),
        "hybrid_readiness blocked_reasons",
    )
    package: dict[str, object] = {
        "schema": "scpn_hybrid_operator_handoff_package_v1",
        "status": hybrid_readiness.get("status"),
        "blocked_reasons": blocked_reasons,
        "hybrid_manifest_sha256": hybrid_manifest_sha,
        "hybrid_readiness_sha256": hybrid_readiness_sha,
        "component_manifest_hashes": hybrid_manifest.get("component_hashes"),
        "component_statuses": hybrid_readiness.get("component_statuses"),
        "target_backends": _target_backends_from_hybrid_manifest(hybrid_manifest),
        "execution_permitted": False,
        "qpu_execution_permitted": False,
        "hardware_write_permitted": False,
        "actuation_permitted": False,
        "operator_commands": [
            "review hybrid_neuromorphic_quantum_cocompiler.json",
            "review scpn_hybrid_target_readiness_v1.json",
            "verify package_sha256 before external operator handoff",
            "execute only outside SPO from an approved operator workflow",
        ],
    }
    canonical = json.dumps(package, sort_keys=True, separators=(",", ":"))
    package["package_sha256"] = sha256(canonical.encode("utf-8")).hexdigest()
    return package

SNN Bridge

SNNControllerBridge maps finite UPDE layer order-parameter magnitudes in [0, 1] to LIF input currents, validates non-negative real-valued spike rates before action projection, and rejects boolean or complex array aliases before schedule-manifest generation.

snn_bridge

Spiking-neural-network bridge for NumPy LIF estimates and review manifests.

SNNControllerBridge maps UPDE layer coherence to input currents, estimates steady-state LIF rates, proposes control actions from rate thresholds, and builds deterministic Lava/PyNN schedule manifests with hardware writes disabled. Optional Lava process construction is isolated to its explicit method. The core bridge remains pure NumPy and does not perform neuromorphic actuation.

Classes

SNNControllerBridge

SNNControllerBridge(
    n_neurons: int = 100,
    tau_rc: float = TAU_RC,
    tau_ref: float = TAU_REF,
)

Bridge between UPDE state and spiking neural network controllers.

All methods are pure-numpy — no external SNN libraries required.

Source code in src/scpn_phase_orchestrator/adapters/snn_bridge.py

def __init__(
    self,
    n_neurons: int = 100,
    tau_rc: float = TAU_RC,
    tau_ref: float = TAU_REF,
) -> None:
    self.n_neurons = _require_positive_int(n_neurons, field="n_neurons")
    self.tau_rc = _require_positive_real(tau_rc, field="tau_rc")
    self.tau_ref = _require_positive_real(tau_ref, field="tau_ref")

Methods:

upde_state_to_input_current

upde_state_to_input_current(
    state: UPDEState, i_scale: float = 1.0
) -> FloatArray

Map R values from each layer to LIF input currents.

Parameters

state : UPDEState The current UPDE state. i_scale : float Scaling factor from order parameter to input current.

Returns

FloatArray The LIF input currents for each layer.

Source code in src/scpn_phase_orchestrator/adapters/snn_bridge.py

def upde_state_to_input_current(
    self, state: UPDEState, i_scale: float = 1.0
) -> FloatArray:
    """Map R values from each layer to LIF input currents.

    Parameters
    ----------
    state : UPDEState
        The current UPDE state.
    i_scale : float
        Scaling factor from order parameter to input current.

    Returns
    -------
    FloatArray
        The LIF input currents for each layer.
    """
    i_scale = _require_positive_real(i_scale, field="i_scale")
    r_values: FloatArray = np.array(
        [
            _require_order_parameter(layer.R, field=f"layer {idx} R")
            for idx, layer in enumerate(state.layers)
        ],
        dtype=np.float64,
    )
    result: FloatArray = r_values * i_scale
    return result

spike_rates_to_actions

spike_rates_to_actions(
    rates: FloatArray,
    layer_assignments: list[int],
    threshold_hz: float = 50.0,
) -> list[ControlAction]

Convert spike rates to control actions.

rates: 1-D array of mean firing rates (Hz) per neuron group. layer_assignments: maps each rate index to a layer. threshold_hz: rates above this trigger coupling boost.

Parameters

rates : FloatArray Per-layer spike rates. layer_assignments : list[int] Per-neuron layer assignments. threshold_hz : float Firing-rate threshold in Hz for emitting actions.

Returns

list[ControlAction] The control actions for the spike rates.

Raises

ValueError If the rates or assignments are invalid.

Source code in src/scpn_phase_orchestrator/adapters/snn_bridge.py

def spike_rates_to_actions(
    self,
    rates: FloatArray,
    layer_assignments: list[int],
    threshold_hz: float = 50.0,
) -> list[ControlAction]:
    """Convert spike rates to control actions.

    *rates*: 1-D array of mean firing rates (Hz) per neuron group.
    *layer_assignments*: maps each rate index to a layer.
    *threshold_hz*: rates above this trigger coupling boost.

    Parameters
    ----------
    rates : FloatArray
        Per-layer spike rates.
    layer_assignments : list[int]
        Per-neuron layer assignments.
    threshold_hz : float
        Firing-rate threshold in Hz for emitting actions.

    Returns
    -------
    list[ControlAction]
        The control actions for the spike rates.

    Raises
    ------
    ValueError
        If the rates or assignments are invalid.
    """
    rates = _require_finite_array(rates, field="rates")
    if rates.ndim != 1:
        raise ValueError("rates must be 1-D")
    if np.any(rates < 0.0):
        raise ValueError("rates must be non-negative")
    layer_assignments = _validated_layer_assignments(layer_assignments)
    if len(layer_assignments) != rates.size:
        raise ValueError("layer_assignments length must match rates length")
    threshold_hz = _require_positive_real(threshold_hz, field="threshold_hz")
    actions: list[ControlAction] = []
    for idx, (rate, layer) in enumerate(
        zip(rates, layer_assignments, strict=False)
    ):
        if rate > threshold_hz:
            excess = (rate - threshold_hz) / threshold_hz
            actions.append(
                ControlAction(
                    knob="K",
                    scope=f"layer_{layer}",
                    value=0.05 * excess,
                    ttl_s=5.0,
                    justification=f"SNN group {idx}: {rate:.1f} Hz",
                )
            )
    return actions

lif_rate_estimate

lif_rate_estimate(currents: FloatArray) -> FloatArray

Analytic LIF steady-state firing rate (Abbott 1999, Eq. 1).

rate = 1 / (tau_ref - tau_rc * ln(1 - 1/J)) for J > 1

Parameters

currents : FloatArray Per-neuron input currents.

Returns

FloatArray The analytic LIF steady-state firing rates.

Raises

ValueError If the input currents are invalid.

Source code in src/scpn_phase_orchestrator/adapters/snn_bridge.py

def lif_rate_estimate(self, currents: FloatArray) -> FloatArray:
    """Analytic LIF steady-state firing rate (Abbott 1999, Eq. 1).

    rate = 1 / (tau_ref - tau_rc * ln(1 - 1/J))  for J > 1

    Parameters
    ----------
    currents : FloatArray
        Per-neuron input currents.

    Returns
    -------
    FloatArray
        The analytic LIF steady-state firing rates.

    Raises
    ------
    ValueError
        If the input currents are invalid.
    """
    currents = _require_finite_array(currents, field="currents")
    if currents.ndim != 1:
        raise ValueError("currents must be 1-D")
    rates: FloatArray = np.zeros_like(currents, dtype=np.float64)
    above = currents > 1.0
    if above.any():
        j = currents[above]
        rates[above] = 1.0 / (self.tau_ref - self.tau_rc * np.log(1.0 - 1.0 / j))
    return rates

build_numpy_network

build_numpy_network(
    n_layers: int, seed: int = 0, synapse: float = 0.01
) -> SimpleNamespace

Build a pure-numpy LIF network for UPDE-SNN coupling.

Returns a SimpleNamespace with input_node, ensemble, output_node attributes and a step() method.

Parameters

n_layers : int Number of SCPN layers. seed : int Seed for the deterministic RNG. synapse : float Synaptic weight scale.

Returns

SimpleNamespace The pure-numpy LIF network.

Source code in src/scpn_phase_orchestrator/adapters/snn_bridge.py

def build_numpy_network(
    self, n_layers: int, seed: int = 0, synapse: float = 0.01
) -> SimpleNamespace:
    """Build a pure-numpy LIF network for UPDE-SNN coupling.

    Returns a SimpleNamespace with input_node, ensemble, output_node
    attributes and a step() method.

    Parameters
    ----------
    n_layers : int
        Number of SCPN layers.
    seed : int
        Seed for the deterministic RNG.
    synapse : float
        Synaptic weight scale.

    Returns
    -------
    SimpleNamespace
        The pure-numpy LIF network.
    """
    n_layers = _require_positive_int(n_layers, field="n_layers")
    seed = _require_nonnegative_int(seed, field="seed")
    synapse = _require_positive_real(synapse, field="synapse")
    rng = np.random.default_rng(seed)
    n = self.n_neurons
    encoders = rng.choice([-1.0, 1.0], (n, n_layers))
    max_rates = rng.uniform(100, 200, n)
    intercepts = rng.uniform(-0.5, 0.5, n)

    J_max = 1.0 / (1.0 - np.exp((self.tau_ref - 1.0 / max_rates) / self.tau_rc))
    alpha = (J_max - 1.0) / (1.0 - intercepts)
    J_bias = 1.0 - alpha * intercepts

    return SimpleNamespace(
        input_node=np.zeros(n_layers),
        ensemble=SimpleNamespace(
            n_neurons=n,
            encoders=encoders,
            alpha=alpha,
            J_bias=J_bias,
        ),
        output_node=np.zeros(n_layers),
        synapse=synapse,
        n_layers=n_layers,
    )

build_lava_process

build_lava_process(n_layers: int) -> object

Build a Lava LIF process for UPDE-SNN coupling.

Raises ImportError if lava-nc is not installed.

Parameters

n_layers : int Number of SCPN layers.

Returns

object The Lava LIF process.

Source code in src/scpn_phase_orchestrator/adapters/snn_bridge.py

def build_lava_process(self, n_layers: int) -> object:
    """Build a Lava LIF process for UPDE-SNN coupling.

    Raises ImportError if lava-nc is not installed.

    Parameters
    ----------
    n_layers : int
        Number of SCPN layers.

    Returns
    -------
    object
        The Lava LIF process.
    """
    _require_positive_int(n_layers, field="n_layers")
    # type ignore: lava-nc is an optional dependency without bundled stubs.
    from lava.proc.lif.process import LIF  # type: ignore[import-not-found]

    return LIF(
        shape=(self.n_neurons,),
        du=1.0 / self.tau_rc,
        dv=1.0 / self.tau_ref,
        vth=1.0,
    )

build_neuromorphic_schedule_manifest

build_neuromorphic_schedule_manifest(
    state: UPDEState,
    *,
    i_scale: float = 1.0,
    threshold_hz: float = 50.0,
    projection_delay_ms: float = 1.0,
) -> dict[str, object]

Compile a reviewable Lava/PyNN schedule from a UPDE state.

The manifest is deterministic and contains simulator-parity evidence from the pure-numpy LIF rate path. It opens no hardware handles and does not permit actuation.

Parameters

state : UPDEState The current UPDE state. i_scale : float Scaling factor from order parameter to input current. threshold_hz : float Firing-rate threshold in Hz for emitting actions. projection_delay_ms : float Projection delay in milliseconds.

Returns

dict[str, object] The reviewable Lava/PyNN schedule manifest.

Source code in src/scpn_phase_orchestrator/adapters/snn_bridge.py

def build_neuromorphic_schedule_manifest(
    self,
    state: UPDEState,
    *,
    i_scale: float = 1.0,
    threshold_hz: float = 50.0,
    projection_delay_ms: float = 1.0,
) -> dict[str, object]:
    """Compile a reviewable Lava/PyNN schedule from a UPDE state.

    The manifest is deterministic and contains simulator-parity evidence
    from the pure-numpy LIF rate path. It opens no hardware handles and
    does not permit actuation.

    Parameters
    ----------
    state : UPDEState
        The current UPDE state.
    i_scale : float
        Scaling factor from order parameter to input current.
    threshold_hz : float
        Firing-rate threshold in Hz for emitting actions.
    projection_delay_ms : float
        Projection delay in milliseconds.

    Returns
    -------
    dict[str, object]
        The reviewable Lava/PyNN schedule manifest.
    """
    self._validate_schedule_inputs(
        state,
        i_scale=i_scale,
        threshold_hz=threshold_hz,
        projection_delay_ms=projection_delay_ms,
    )
    currents = self.upde_state_to_input_current(state, i_scale=i_scale)
    rates = self.lif_rate_estimate(currents)
    parity_rates = self.lif_rate_estimate(currents)
    rate_error = float(np.max(np.abs(rates - parity_rates))) if rates.size else 0.0

    populations = [
        self._population_record(
            layer_index=idx,
            layer=layer,
            input_current=float(currents[idx]),
            estimated_rate_hz=float(rates[idx]),
        )
        for idx, layer in enumerate(state.layers)
    ]
    projections = self._projection_records(
        state.cross_layer_alignment,
        delay_ms=projection_delay_ms,
    )
    nir_graph = to_nir_graph(
        populations,
        projections,
        tau_membrane_ms=self.tau_rc * 1000.0,
        tau_refractory_ms=self.tau_ref * 1000.0,
    )
    manifest: dict[str, object] = {
        "manifest_kind": "neuromorphic_schedule_manifest",
        "schema_version": 1,
        "status": (
            "simulator_parity_passed"
            if rate_error == 0.0
            else "simulator_parity_failed"
        ),
        "target_backends": ["lava", "pynn"],
        "n_layers": len(state.layers),
        "n_neurons_per_population": self.n_neurons,
        "tau_rc_s": self.tau_rc,
        "tau_ref_s": self.tau_ref,
        "input_scale": float(i_scale),
        "threshold_hz": float(threshold_hz),
        "actuation_permitted": False,
        "hardware_write_permitted": False,
        "populations": populations,
        "projections": projections,
        "control_actions": [
            {
                "knob": action.knob,
                "scope": action.scope,
                "value": action.value,
                "ttl_s": action.ttl_s,
                "justification": action.justification,
            }
            for action in self.spike_rates_to_actions(
                rates,
                layer_assignments=list(range(len(state.layers))),
                threshold_hz=threshold_hz,
            )
        ],
        "simulator_parity": {
            "engine": "numpy_lif_rate_estimate",
            "max_abs_rate_error_hz": rate_error,
            "sample_count": len(state.layers),
        },
        "operator_commands": [
            "review neuromorphic_schedule_manifest.json",
            "run Lava or PyNN simulator parity before hardware handoff",
        ],
        "neuromorphic_ir": nir_graph.to_record(),
        "nir_sha256": nir_graph.sha256,
    }
    canonical = json.dumps(manifest, sort_keys=True, separators=(",", ":"))
    manifest["schedule_sha256"] = sha256(canonical.encode("utf-8")).hexdigest()
    return manifest

audit_hardware_target_readiness

audit_hardware_target_readiness(
    manifest: dict[str, object],
    *,
    target_backend: str,
    hardware_site: str,
    credentials_configured: bool = False,
    operator_approved: bool = False,
    external_simulator_parity_verified: bool = False,
) -> dict[str, object]

Return non-executing neuromorphic hardware readiness evidence.

The audit validates a schedule manifest against a declared target and records whether external operator preconditions are present. It never opens a backend connection, submits a hardware job, or enables actuation/hardware-write permissions.

Parameters

manifest : dict[str, object] The compiler manifest to audit. target_backend : str Name of the target backend. hardware_site : str Name of the deployment hardware site. credentials_configured : bool Whether provider credentials are configured. operator_approved : bool Whether a human operator approved the target. external_simulator_parity_verified : bool Whether external-simulator parity has been verified.

Returns

dict[str, object] The non-executing neuromorphic hardware readiness evidence.

Raises

ValueError If the manifest or target details are invalid.

Source code in src/scpn_phase_orchestrator/adapters/snn_bridge.py

def audit_hardware_target_readiness(
    self,
    manifest: dict[str, object],
    *,
    target_backend: str,
    hardware_site: str,
    credentials_configured: bool = False,
    operator_approved: bool = False,
    external_simulator_parity_verified: bool = False,
) -> dict[str, object]:
    """Return non-executing neuromorphic hardware readiness evidence.

    The audit validates a schedule manifest against a declared target and
    records whether external operator preconditions are present. It never
    opens a backend connection, submits a hardware job, or enables
    actuation/hardware-write permissions.

    Parameters
    ----------
    manifest : dict[str, object]
        The compiler manifest to audit.
    target_backend : str
        Name of the target backend.
    hardware_site : str
        Name of the deployment hardware site.
    credentials_configured : bool
        Whether provider credentials are configured.
    operator_approved : bool
        Whether a human operator approved the target.
    external_simulator_parity_verified : bool
        Whether external-simulator parity has been verified.

    Returns
    -------
    dict[str, object]
        The non-executing neuromorphic hardware readiness evidence.

    Raises
    ------
    ValueError
        If the manifest or target details are invalid.
    """
    manifest_record = _require_mapping(manifest, field="manifest")
    target_backend = _require_non_empty_text(
        target_backend,
        field="target_backend",
    )
    hardware_site = _require_non_empty_text(hardware_site, field="hardware_site")
    for field, value in (
        ("credentials_configured", credentials_configured),
        ("operator_approved", operator_approved),
        ("external_simulator_parity_verified", external_simulator_parity_verified),
    ):
        if not isinstance(value, bool):
            raise ValueError(f"{field} must be a boolean")
    if manifest_record.get("manifest_kind") != "neuromorphic_schedule_manifest":
        raise ValueError("manifest must be a neuromorphic_schedule_manifest")

    target_backends = manifest_record.get("target_backends")
    if not isinstance(target_backends, list) or not all(
        isinstance(item, str) for item in target_backends
    ):
        raise ValueError("manifest target_backends must be a list of strings")
    if target_backend not in target_backends:
        raise ValueError("target_backend is not declared by manifest")

    blocked_reasons: list[str] = []
    if manifest_record.get("status") != "simulator_parity_passed":
        blocked_reasons.append("simulator_parity_not_passed")
    if manifest_record.get("hardware_write_permitted") is not False:
        blocked_reasons.append("hardware_write_permission_must_remain_false")
    if manifest_record.get("actuation_permitted") is not False:
        blocked_reasons.append("actuation_permission_must_remain_false")
    if not credentials_configured:
        blocked_reasons.append("credentials_not_configured")
    if not operator_approved:
        blocked_reasons.append("operator_approval_missing")
    if not external_simulator_parity_verified:
        blocked_reasons.append("external_simulator_parity_not_verified")

    manifest_sha = str(manifest_record.get("schedule_sha256", ""))
    record: dict[str, object] = {
        "schema": "scpn_neuromorphic_target_readiness_v1",
        "target_backend": target_backend,
        "hardware_site": hardware_site,
        "manifest_sha256": manifest_sha,
        "status": "blocked" if blocked_reasons else "ready_not_executed",
        "blocked_reasons": blocked_reasons,
        "credentials_configured": credentials_configured,
        "operator_approved": operator_approved,
        "external_simulator_parity_verified": external_simulator_parity_verified,
        "hardware_write_permitted": False,
        "actuation_permitted": False,
        "operator_commands": [
            "review neuromorphic_schedule_manifest.json",
            "run target simulator parity outside SPO before hardware handoff",
            (
                "submit neuromorphic hardware job only from an approved "
                "operator workflow"
            ),
        ],
    }
    canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
    record["readiness_sha256"] = sha256(canonical.encode("utf-8")).hexdigest()
    return record

Functions:

Neuromorphic IR Export

to_nir_graph serialises a schedule's LIF populations and inter-population projections into a deterministic, SHA-256-hashed graph in the shape of the Neuromorphic Intermediate Representation (neuromorphs/NIR): nodes are neuron populations, edges are weighted (source, target) connections. The export is an honestly-labelled structural subset (conformance = "structural_subset"), not a spec-validated NIR export: the SCPN Abbott-rate LIF defines only the membrane/refractory time constants and a normalised firing threshold, so the NIR physical parameters it does not model (R, v_leak, v_reset) are listed in unmodelled_nir_lif_parameters rather than fabricated. The SNN bridge embeds the graph under the schedule manifest's neuromorphic_ir key with a nir_sha256 digest.

neuromorphic_ir_export

Deterministic NIR-structural graph export for SNN schedule manifests.

This module serialises the LIF populations and inter-population projections of a reviewable neuromorphic schedule into a portable, deterministic, SHA-256-hashed graph in the shape of the Neuromorphic Intermediate Representation (neuromorphs/NIR <https://github.com/neuromorphs/NIR>_): a directed graph whose nodes are neuron populations and whose edges are (source, target) connection tuples with weights.

Honesty boundary

The output is a structural subset, not a spec-validated NIR export, and it says so in its metadata (conformance = "structural_subset"). The reference NIR LIF primitive is parametrised by tau [ms], R [Ω], v_leak [mV], v_reset [mV], and v_threshold [mV] (verified at source, 2026-07-21). The SCPN SNN bridge models an Abbott-1999 analytic-rate LIF that genuinely defines only the membrane and refractory time constants and a normalised firing threshold; it does not define the NIR physical parameters R, v_leak, or v_reset. Rather than fabricate those values, the export emits only the parameters the model actually holds, marks the firing threshold as normalised, and lists the unmodelled NIR parameters explicitly in unmodelled_nir_lif_parameters. It adds no dependency and touches no hardware.

Classes

NeuromorphicIRGraph dataclass

NeuromorphicIRGraph(
    nodes: tuple[dict[str, object], ...],
    edges: tuple[dict[str, object], ...],
    metadata: dict[str, object],
)

A deterministic NIR-structural graph of LIF nodes and weighted edges.

Attributes

sha256 property

sha256: str

Return the SHA-256 hex digest of the canonical JSON encoding.

Returns

str The 64-character lowercase hexadecimal SHA-256 digest.

Methods:

to_record

to_record() -> dict[str, object]

Return a deterministic, JSON-safe mapping of the graph.

Returns

dict[str, object] The metadata / nodes / edges mapping, with the node and edge tuples materialised as lists in insertion order.

Source code in src/scpn_phase_orchestrator/adapters/neuromorphic_ir_export.py

def to_record(self) -> dict[str, object]:
    """Return a deterministic, JSON-safe mapping of the graph.

    Returns
    -------
    dict[str, object]
        The ``metadata`` / ``nodes`` / ``edges`` mapping, with the node and
        edge tuples materialised as lists in insertion order.
    """
    return {
        "metadata": dict(self.metadata),
        "nodes": [dict(node) for node in self.nodes],
        "edges": [dict(edge) for edge in self.edges],
    }

canonical_json

canonical_json() -> str

Return the canonical (sorted, compact) JSON encoding of the graph.

Returns

str A deterministic JSON string with sorted keys and no whitespace.

Source code in src/scpn_phase_orchestrator/adapters/neuromorphic_ir_export.py

def canonical_json(self) -> str:
    """Return the canonical (sorted, compact) JSON encoding of the graph.

    Returns
    -------
    str
        A deterministic JSON string with sorted keys and no whitespace.
    """
    return json.dumps(self.to_record(), sort_keys=True, separators=(",", ":"))

Functions:

to_nir_graph

to_nir_graph(
    populations: list[dict[str, object]],
    projections: list[dict[str, object]],
    *,
    tau_membrane_ms: float,
    tau_refractory_ms: float,
    v_threshold_normalised: float = 1.0,
) -> NeuromorphicIRGraph

Compile schedule populations and projections into a NIR-structural graph.

Parameters

populations : list[dict[str, object]] Per-population records from a neuromorphic schedule manifest; each must carry a non-empty name and a non-negative estimated_rate_hz. projections : list[dict[str, object]] Inter-population projection records; each must carry non-empty source / target node names and a non-negative weight. tau_membrane_ms : float LIF membrane time constant in milliseconds. tau_refractory_ms : float LIF refractory period in milliseconds. v_threshold_normalised : float, optional The normalised firing threshold of the Abbott-rate LIF (dimensionless; not the NIR v_threshold in mV, which the model does not define).

Returns

NeuromorphicIRGraph The deterministic NIR-structural graph. Every edge's source and target is guaranteed to reference a declared node id.

Raises

ValueError If a record is malformed, or an edge references an undeclared node.

Source code in src/scpn_phase_orchestrator/adapters/neuromorphic_ir_export.py

def to_nir_graph(
    populations: list[dict[str, object]],
    projections: list[dict[str, object]],
    *,
    tau_membrane_ms: float,
    tau_refractory_ms: float,
    v_threshold_normalised: float = 1.0,
) -> NeuromorphicIRGraph:
    """Compile schedule populations and projections into a NIR-structural graph.

    Parameters
    ----------
    populations : list[dict[str, object]]
        Per-population records from a neuromorphic schedule manifest; each must
        carry a non-empty ``name`` and a non-negative ``estimated_rate_hz``.
    projections : list[dict[str, object]]
        Inter-population projection records; each must carry non-empty
        ``source`` / ``target`` node names and a non-negative ``weight``.
    tau_membrane_ms : float
        LIF membrane time constant in milliseconds.
    tau_refractory_ms : float
        LIF refractory period in milliseconds.
    v_threshold_normalised : float, optional
        The normalised firing threshold of the Abbott-rate LIF (dimensionless;
        not the NIR ``v_threshold`` in mV, which the model does not define).

    Returns
    -------
    NeuromorphicIRGraph
        The deterministic NIR-structural graph. Every edge's ``source`` and
        ``target`` is guaranteed to reference a declared node id.

    Raises
    ------
    ValueError
        If a record is malformed, or an edge references an undeclared node.
    """
    tau_membrane_ms = _require_finite_non_negative(
        tau_membrane_ms, field="tau_membrane_ms"
    )
    tau_refractory_ms = _require_finite_non_negative(
        tau_refractory_ms, field="tau_refractory_ms"
    )
    v_threshold_normalised = _require_finite_non_negative(
        v_threshold_normalised, field="v_threshold_normalised"
    )
    nodes = tuple(
        _node_from_population(
            population,
            tau_membrane_ms=tau_membrane_ms,
            tau_refractory_ms=tau_refractory_ms,
            v_threshold_normalised=v_threshold_normalised,
        )
        for population in populations
    )
    node_ids = {node["id"] for node in nodes}
    edges = tuple(_edge_from_projection(projection) for projection in projections)
    for edge in edges:
        for endpoint in ("source", "target"):
            if edge[endpoint] not in node_ids:
                raise ValueError(
                    f"projection {endpoint} {edge[endpoint]!r} references an "
                    "undeclared node"
                )
    metadata: dict[str, object] = {
        "format": NIR_STRUCTURAL_FORMAT,
        "format_version": NIR_STRUCTURAL_FORMAT_VERSION,
        "reference_spec": "neuromorphs/NIR",
        "conformance": "structural_subset",
        "node_type": "LIF",
        "edge_type": "Linear",
        "v_threshold_unit": "normalised",
        "unmodelled_nir_lif_parameters": list(UNMODELLED_NIR_LIF_PARAMETERS),
        "node_count": len(nodes),
        "edge_count": len(edges),
    }
    return NeuromorphicIRGraph(nodes=nodes, edges=edges, metadata=metadata)

Neurocore Bridge

NeurocoreBridge maps bounded UPDE layer coherence to stochastic LIF input currents, accepts only non-negative deterministic seeds, and validates real-valued non-negative rate vectors from action inputs or Rust backend output before producing coupling actions.

neurocore_bridge

Bridge between sc-neurocore stochastic neurons and phase-orchestrator.

sc-neurocore provides StochasticLIFNeuron, SCIzhikevichNeuron, and HomeostaticLIFNeuron with get_state()/step(current)/reset_state() API.

This bridge: 1. Maps UPDE layer R values to neuron input currents 2. Runs a LIF ensemble matching sc-neurocore dynamics 3. Converts spike rates back to orchestrator ControlActions

Backend priority

1. Rust (spo_kernel.PyLIFEnsemble) — ~1000x faster than scalar Python
2. NumPy vectorised — ~50-100x faster than scalar Python
3. sc-neurocore scalar — per-neuron Python objects (validation only)

LIF parameters match sc-neurocore v3.13.3 defaults (Gerstner & Kistler 2002): v_rest=0, v_threshold=1, tau_mem=20ms, R=1, dt=1ms.

Install sc-neurocore: pip install sc-neurocore

Classes

NeurocoreBridge

NeurocoreBridge(
    n_layers: int,
    neurons_per_layer: int = 8,
    current_scale: float = 2.0,
    spike_threshold_hz: float = 40.0,
    noise_std: float = 0.0,
    backend: str = "auto",
    seed: int | None = None,
)

Live integration with sc-neurocore StochasticLIFNeuron ensemble.

Each layer in the UPDE state maps to a group of stochastic LIF neurons. Layer coherence R drives input current; spike rates above threshold generate coupling boost actions.

Backend selection (automatic): - "rust" — Rust LIF via spo_kernel.PyLIFEnsemble (fastest) - "numpy" — vectorised numpy LIF integration - "scalar" — per-neuron sc-neurocore objects (requires sc-neurocore)

Pass backend="numpy" or backend="scalar" to force a specific backend. Default: best available.

Source code in src/scpn_phase_orchestrator/adapters/neurocore_bridge.py

def __init__(
    self,
    n_layers: int,
    neurons_per_layer: int = 8,
    current_scale: float = 2.0,
    spike_threshold_hz: float = 40.0,
    noise_std: float = 0.0,
    backend: str = "auto",
    seed: int | None = None,
) -> None:
    n_layers = _require_positive_int(n_layers, field="n_layers")
    neurons_per_layer = _require_positive_int(
        neurons_per_layer,
        field="neurons_per_layer",
    )
    current_scale = _require_finite_real(
        current_scale,
        field="current_scale",
        positive=True,
    )
    spike_threshold_hz = _require_finite_real(
        spike_threshold_hz,
        field="spike_threshold_hz",
        positive=True,
    )
    noise_std = _require_finite_real(
        noise_std,
        field="noise_std",
        positive=False,
    )
    if not isinstance(backend, str):
        raise ValueError("backend must be a string")
    seed = _require_seed(seed)

    self._n_layers = n_layers
    self._n_per = neurons_per_layer
    self._n_total = n_layers * neurons_per_layer
    self._scale = current_scale
    self._threshold_hz = spike_threshold_hz
    self._dt = 0.001  # 1ms step (for rate Hz conversion)

    # Resolve backend
    if backend == "auto":
        backend = "rust" if _HAS_RUST else "numpy"
    self._backend = backend

    if backend == "rust":
        from spo_kernel import PyLIFEnsemble

        self._rust_ensemble = PyLIFEnsemble(n_layers, neurons_per_layer, noise_std)
    elif backend == "numpy":
        self._v = np.full(self._n_total, _V_REST)
        self._refractory = np.zeros(self._n_total, dtype=np.int32)
        self._noise_std = noise_std
        self._rng = np.random.default_rng(seed)
    elif backend == "scalar":
        if not HAS_NEUROCORE:  # pragma: no cover
            msg = "sc-neurocore not installed. pip install sc-neurocore"
            raise ImportError(msg)
        self._neurons: list[Any] = []
        for _ in range(self._n_total):
            self._neurons.append(StochasticLIFNeuron())
    else:
        msg = (
            f"Unknown backend {backend!r}, "
            "expected 'auto', 'rust', 'numpy', or 'scalar'"
        )
        raise ValueError(msg)

    self._spike_counts = np.zeros(self._n_total, dtype=np.int64)
    self._step_count = 0

Attributes

backend property

backend: str

Active backend: 'rust', 'numpy', or 'scalar'.

Returns

str Active backend: 'rust', 'numpy', or 'scalar'.

Methods:

step

step(state: UPDEState, n_substeps: int = 10) -> FloatArray

Run neuron ensemble for n_substeps, return per-layer spike rates.

Parameters

state : UPDEState The current UPDE state. n_substeps : int Number of inner substeps to run.

Returns

FloatArray The per-layer spike rates after n_substeps.

Raises

ValueError If the state or substep count is invalid.

Source code in src/scpn_phase_orchestrator/adapters/neurocore_bridge.py

def step(self, state: UPDEState, n_substeps: int = 10) -> FloatArray:
    """Run neuron ensemble for n_substeps, return per-layer spike rates.

    Parameters
    ----------
    state : UPDEState
        The current UPDE state.
    n_substeps : int
        Number of inner substeps to run.

    Returns
    -------
    FloatArray
        The per-layer spike rates after ``n_substeps``.

    Raises
    ------
    ValueError
        If the state or substep count is invalid.
    """
    n_substeps = _require_positive_int(n_substeps, field="n_substeps")
    if len(state.layers) < self._n_layers:
        raise ValueError("state.layers must cover configured n_layers")
    validated_r_values = [
        _require_unit_interval(layer.R, field=f"layer {idx} R")
        for idx, layer in enumerate(state.layers[: self._n_layers])
    ]
    r_values = np.array(
        validated_r_values,
        dtype=np.float64,
    )
    layer_currents = r_values * self._scale

    if self._backend == "rust":
        return self._step_rust(layer_currents, n_substeps)
    if self._backend == "numpy":
        currents = np.repeat(layer_currents, self._n_per)
        self._step_numpy(currents, n_substeps)
    else:
        currents = np.repeat(layer_currents, self._n_per)
        self._step_scalar(currents, n_substeps)

    duration_s = self._step_count * self._dt
    if duration_s == 0:  # pragma: no cover
        return np.zeros(self._n_layers)

    spikes_2d = self._spike_counts.reshape(self._n_layers, self._n_per)
    layer_spikes: FloatArray = spikes_2d.sum(axis=1)
    return layer_spikes / (self._n_per * duration_s)

rates_to_actions

rates_to_actions(rates: FloatArray) -> list[ControlAction]

Convert per-layer spike rates to coupling boost actions.

Parameters

rates : FloatArray Per-layer spike rates.

Returns

list[ControlAction] The coupling-boost control actions for the spike rates.

Source code in src/scpn_phase_orchestrator/adapters/neurocore_bridge.py

def rates_to_actions(self, rates: FloatArray) -> list[ControlAction]:
    """Convert per-layer spike rates to coupling boost actions.

    Parameters
    ----------
    rates : FloatArray
        Per-layer spike rates.

    Returns
    -------
    list[ControlAction]
        The coupling-boost control actions for the spike rates.
    """
    rates = _require_rate_vector(rates, n_layers=self._n_layers)
    actions: list[ControlAction] = []
    for layer_idx, rate in enumerate(rates):
        if rate > self._threshold_hz:
            excess = (rate - self._threshold_hz) / self._threshold_hz
            actions.append(
                ControlAction(
                    knob="K",
                    scope=f"layer_{layer_idx}",
                    value=0.05 * min(excess, 2.0),
                    ttl_s=5.0,
                    justification=(f"neurocore layer {layer_idx}: {rate:.1f} Hz"),
                )
            )
    return actions

step_and_act

step_and_act(
    state: UPDEState, n_substeps: int = 10
) -> list[ControlAction]

Step the ensemble and return control actions.

Parameters

state : UPDEState The current UPDE state. n_substeps : int Number of inner substeps to run.

Returns

list[ControlAction] The control actions after stepping the ensemble.

Source code in src/scpn_phase_orchestrator/adapters/neurocore_bridge.py

def step_and_act(
    self,
    state: UPDEState,
    n_substeps: int = 10,
) -> list[ControlAction]:
    """Step the ensemble and return control actions.

    Parameters
    ----------
    state : UPDEState
        The current UPDE state.
    n_substeps : int
        Number of inner substeps to run.

    Returns
    -------
    list[ControlAction]
        The control actions after stepping the ensemble.
    """
    rates = self.step(state, n_substeps)
    return self.rates_to_actions(rates)

get_neuron_states

get_neuron_states() -> list[dict[str, Any]]

Return voltage/refractory state for all neurons.

Returns

list[dict[str, Any]] Return voltage/refractory state for all neurons.

Source code in src/scpn_phase_orchestrator/adapters/neurocore_bridge.py

def get_neuron_states(self) -> list[dict[str, Any]]:
    """Return voltage/refractory state for all neurons.

    Returns
    -------
    list[dict[str, Any]]
        Return voltage/refractory state for all neurons.
    """
    if self._backend == "rust":
        states: list[dict[str, Any]] = self._rust_ensemble.get_neuron_states()
        return states
    if self._backend == "numpy":
        return [
            {"v": float(self._v[i]), "refractory": int(self._refractory[i])}
            for i in range(self._n_total)
        ]
    return [n.get_state() for n in self._neurons]

reset

reset() -> None

Reset all neurons and counters.

Source code in src/scpn_phase_orchestrator/adapters/neurocore_bridge.py

def reset(self) -> None:
    """Reset all neurons and counters."""
    if self._backend == "rust":
        self._rust_ensemble.reset()
    elif self._backend == "numpy":
        self._v[:] = _V_REST
        self._refractory[:] = 0
    else:
        for n in self._neurons:
            n.reset_state()
    self._spike_counts[:] = 0
    self._step_count = 0

Observability

Adapters for production monitoring and tracing.

OpenTelemetry

Exports SPO metrics and traces to any OTLP-compatible backend (Jaeger, Zipkin, Grafana Tempo). Requires opentelemetry-api.

OTelExporter API:

Method	Signature	Description
record_step	(upde_state, step_idx)	Record metrics for one engine step
record_regime_change	(old, new)	Record regime transition event

Metrics exported:

Metric	Type	Description
spo.order_parameter	Gauge	Current R value
spo.regime	Gauge	Current regime (0-3)
spo.step_latency_ms	Histogram	Engine step duration
spo.coupling_mean	Gauge	Mean K_nm value

opentelemetry

Compatibility alias for scpn_phase_orchestrator.runtime.observability.

Prometheus

Fetches Prometheus instant and range metrics as a validated telemetry input boundary. The adapter rejects malformed decoded JSON, malformed result/sample structures, non-finite JSON constants, boolean/negative/non-real sample timestamps, and non-finite sample values before returning arrays or scalars.

PrometheusAdapter API:

PrometheusAdapter(endpoint: str, timeout: float = 5.0)

Method	Signature	Description
fetch_metric	(query, start, end, step) -> NDArray[np.float64]	Fetch a range-vector metric as finite values
fetch_instant	(query) -> float	Fetch one instant-vector scalar

prometheus

Prometheus HTTP adapter for validated instant and range metric queries.

PrometheusAdapter validates endpoint URLs, timeouts, query text, range bounds, and step size before issuing standard Prometheus API requests. Network failures are reported as ConnectionError and malformed responses as ValueError. The adapter fetches metric values only; it does not run a server or mutate orchestration state.

Classes

PrometheusAdapter

PrometheusAdapter(endpoint: str, timeout: float = 10.0)

Fetch time-series metrics from a Prometheus endpoint.

Source code in src/scpn_phase_orchestrator/adapters/prometheus.py

def __init__(self, endpoint: str, timeout: float = 10.0):
    if not isinstance(endpoint, str) or not endpoint:
        raise ValueError("Prometheus endpoint must be a non-empty http(s) URL")
    parsed = urlparse(endpoint)
    if parsed.scheme not in ("http", "https") or not parsed.netloc:
        raise ValueError("Prometheus endpoint must be a non-empty http(s) URL")
    if isinstance(timeout, bool):
        raise ValueError("Prometheus timeout must be finite and positive")
    try:
        parsed_timeout = float(timeout)
    except (TypeError, ValueError) as exc:
        raise ValueError("Prometheus timeout must be finite and positive") from exc
    if not isfinite(parsed_timeout) or parsed_timeout <= 0.0:
        raise ValueError("Prometheus timeout must be finite and positive")
    self._endpoint = endpoint.rstrip("/")
    self._timeout = parsed_timeout

Methods:

fetch_metric

fetch_metric(
    query: str, start: float, end: float, step: float
) -> FloatArray

Query Prometheus range API, return values as 1-D float array.

Raises ConnectionError on network failure, ValueError on bad response.

Parameters

query : str PromQL query string. start : float Range start time as a UNIX timestamp. end : float Range end time as a UNIX timestamp. step : float Sampling step in seconds.

Returns

FloatArray The range query values as a 1-D float array.

Raises

ConnectionError If the Prometheus server is unreachable. ValueError If the query or response is invalid.

Source code in src/scpn_phase_orchestrator/adapters/prometheus.py

def fetch_metric(
    self, query: str, start: float, end: float, step: float
) -> FloatArray:
    """Query Prometheus range API, return values as 1-D float array.

    Raises ConnectionError on network failure, ValueError on bad response.

    Parameters
    ----------
    query : str
        PromQL query string.
    start : float
        Range start time as a UNIX timestamp.
    end : float
        Range end time as a UNIX timestamp.
    step : float
        Sampling step in seconds.

    Returns
    -------
    FloatArray
        The range query values as a 1-D float array.

    Raises
    ------
    ConnectionError
        If the Prometheus server is unreachable.
    ValueError
        If the query or response is invalid.
    """
    query_text = _require_query_text(query)
    start_f = _require_finite_float(start, "start")
    end_f = _require_finite_float(end, "end")
    step_f = _require_finite_float(step, "step")
    if end_f < start_f:
        raise ValueError("Prometheus end must be >= start")
    if step_f <= 0.0:
        raise ValueError("Prometheus step must be positive")
    params = urlencode(
        {"query": query_text, "start": start_f, "end": end_f, "step": step_f}
    )
    url = f"{self._endpoint}/api/v1/query_range?{params}"
    req = Request(url, headers={"Accept": "application/json"})
    try:
        # Endpoint scheme is constrained to http/https in __init__, so
        # B310's file://-style scheme concern does not apply here.
        with urlopen(req, timeout=self._timeout) as resp:  # nosec B310
            body = _load_response_body(resp.read())
    except (URLError, OSError):
        raise ConnectionError("Prometheus query failed") from None

    if body.get("status") != "success":
        raise ValueError(f"Prometheus returned status={body.get('status')}")

    results = _response_results(body)
    if not results:
        return np.array([], dtype=np.float64)

    values = _range_values(results[0])
    result: FloatArray = np.array(values, dtype=np.float64)
    return result

fetch_instant

fetch_instant(query: str) -> float

Query Prometheus instant API, return scalar value.

Parameters

query : str PromQL query string.

Returns

float The instant query scalar value.

Raises

ConnectionError If the Prometheus server is unreachable. ValueError If the query or response is invalid.

Source code in src/scpn_phase_orchestrator/adapters/prometheus.py

def fetch_instant(self, query: str) -> float:
    """Query Prometheus instant API, return scalar value.

    Parameters
    ----------
    query : str
        PromQL query string.

    Returns
    -------
    float
        The instant query scalar value.

    Raises
    ------
    ConnectionError
        If the Prometheus server is unreachable.
    ValueError
        If the query or response is invalid.
    """
    query_text = _require_query_text(query)
    params = urlencode({"query": query_text})
    url = f"{self._endpoint}/api/v1/query?{params}"
    req = Request(url, headers={"Accept": "application/json"})
    try:
        # Endpoint scheme is constrained to http/https in __init__, so
        # B310's file://-style scheme concern does not apply here.
        with urlopen(req, timeout=self._timeout) as resp:  # nosec B310
            body = _load_response_body(resp.read())
    except (URLError, OSError):
        raise ConnectionError("Prometheus query failed") from None

    if body.get("status") != "success":
        raise ValueError(f"Prometheus returned status={body.get('status')}")

    results = _response_results(body)
    if not results:
        raise ValueError("Prometheus returned empty result set")

    return _instant_value(results[0])

Metrics Exporter

Lightweight metrics export helpers used by services that do not need the full OpenTelemetry adapter.

metrics_exporter

Compatibility alias for scpn_phase_orchestrator.runtime.observability.

Redis Store

Optional Redis-backed state exchange for deployments that need shared runtime state outside the local process.

redis_store

Redis-backed JSON state persistence adapter with explicit dependency checks.

RedisStateStore validates host, port, database, and key parameters before using an injected client or constructing a Redis client when the optional package is installed. Stored payloads must be JSON objects, and missing keys return None. The adapter persists caller-provided state only; it does not manage simulation lifecycle or background synchronization.

Classes

RedisStateStore

RedisStateStore(
    host: str = "localhost",
    port: int = 6379,
    db: int = 0,
    key: str = "spo:sim_state",
    client: Any = None,
    password: str | None = None,
    ssl: bool = True,
    ssl_ca_certs: str | Path | None = None,
    ssl_certfile: str | Path | None = None,
    ssl_keyfile: str | Path | None = None,
)

Persist simulation state in Redis for survival across restarts.

When redis is not installed, all operations raise RuntimeError.

Source code in src/scpn_phase_orchestrator/adapters/redis_store.py

def __init__(
    self,
    host: str = "localhost",
    port: int = 6379,
    db: int = 0,
    key: str = "spo:sim_state",
    client: Any = None,
    password: str | None = None,
    ssl: bool = True,
    ssl_ca_certs: str | Path | None = None,
    ssl_certfile: str | Path | None = None,
    ssl_keyfile: str | Path | None = None,
) -> None:
    self._host = require_non_empty_str(host, field="Redis host")
    self._port = require_tcp_port(port, field="Redis port")
    self._db = require_non_negative_int(db, field="Redis db")
    self._key = require_non_empty_str(key, field="Redis key")
    if not isinstance(ssl, bool):
        raise ValueError("Redis ssl must be a bool")
    if password is not None:
        password = require_non_empty_str(password, field="Redis password")
    self._ssl = ssl
    self._password = password
    self._ssl_ca_certs = _optional_path(ssl_ca_certs, "Redis TLS CA bundle")
    self._ssl_certfile = _optional_path(ssl_certfile, "Redis TLS certificate")
    self._ssl_keyfile = _optional_path(ssl_keyfile, "Redis TLS key")
    if (self._ssl_certfile is None) != (self._ssl_keyfile is None):
        raise ValueError("Redis TLS certificate and key must be provided together")
    if not self._ssl and self._host not in _LOOPBACK_HOSTS:
        raise ValueError(
            "plaintext Redis connections are allowed only for loopback hosts"
        )
    if self._host not in _LOOPBACK_HOSTS and self._password is None:
        raise ValueError("remote Redis connections require password authentication")
    if (
        self._host not in _LOOPBACK_HOSTS
        and self._ssl
        and self._ssl_ca_certs is None
    ):
        raise ValueError("remote Redis TLS connections require a CA bundle")
    if client is not None:
        self._client = client
    elif not _HAS_REDIS:
        raise RuntimeError("redis package not installed — pip install redis")
    else:
        client_kwargs: dict[str, object] = {
            "host": self._host,
            "port": self._port,
            "db": self._db,
            "ssl": self._ssl,
        }
        if self._password is not None:
            client_kwargs["password"] = self._password
        if self._ssl:
            client_kwargs["ssl_cert_reqs"] = "required"
            if self._ssl_ca_certs is not None:
                client_kwargs["ssl_ca_certs"] = self._ssl_ca_certs
            if self._ssl_certfile is not None:
                client_kwargs["ssl_certfile"] = self._ssl_certfile
            if self._ssl_keyfile is not None:
                client_kwargs["ssl_keyfile"] = self._ssl_keyfile
        self._client = _redis_mod.Redis(
            **client_kwargs,
        )

Attributes

key property

key: str

Redis key used for state storage.

Returns

str Redis key used for state storage.

Methods:

save_state

save_state(sim_state: dict[str, Any]) -> None

Serialise state dict to JSON and store in Redis.

Parameters

sim_state : dict[str, Any] The simulation state dict to store.

Raises

ValueError If the state dict is not JSON-serialisable.

Source code in src/scpn_phase_orchestrator/adapters/redis_store.py

def save_state(self, sim_state: dict[str, Any]) -> None:
    """Serialise state dict to JSON and store in Redis.

    Parameters
    ----------
    sim_state : dict[str, Any]
        The simulation state dict to store.

    Raises
    ------
    ValueError
        If the state dict is not JSON-serialisable.
    """
    if not isinstance(sim_state, dict):
        raise ValueError("sim_state must be a JSON-serializable dict")
    try:
        payload = json.dumps(sim_state, allow_nan=False)
    except (TypeError, ValueError) as exc:
        raise ValueError("sim_state must be JSON serializable") from exc
    self._client.set(self._key, payload)

load_state

load_state() -> dict[str, Any] | None

Load state from Redis. Returns None if key does not exist.

Returns

dict[str, Any] | None Load state from Redis. Returns None if key does not exist.

Raises

ValueError If the stored payload is malformed.

Source code in src/scpn_phase_orchestrator/adapters/redis_store.py

def load_state(self) -> dict[str, Any] | None:
    """Load state from Redis. Returns None if key does not exist.

    Returns
    -------
    dict[str, Any] | None
        Load state from Redis. Returns None if key does not exist.

    Raises
    ------
    ValueError
        If the stored payload is malformed.
    """
    raw = self._client.get(self._key)
    if raw is None:
        return None
    try:
        result = json.loads(raw, parse_constant=_reject_json_constant)
    except (TypeError, ValueError, json.JSONDecodeError) as exc:
        raise ValueError("Redis payload must be a JSON object") from exc
    if not isinstance(result, dict):
        raise ValueError("Redis payload must be a JSON object")
    _require_finite_json_numbers(result)
    return result

delete_state

delete_state() -> None

Remove the stored state key.

Source code in src/scpn_phase_orchestrator/adapters/redis_store.py

def delete_state(self) -> None:
    """Remove the stored state key."""
    self._client.delete(self._key)

Functions:

Hardware Adapters

Modbus/TLS

Industrial control interface for power grids, HVAC, and manufacturing. Translates ControlAction to Modbus register writes over TLS.

modbus_tls

Secure Modbus TCP adapter with mutual TLS.

Wraps pymodbus with an ssl.SSLContext for certificate-authenticated connections to SCADA endpoints per IEC 62443 zone/conduit requirements. Server certificate verification is always enabled: pass ca_cert_path for a deployment CA bundle or rely on the operating-system trust store.

Classes

SecureModbusAdapter

SecureModbusAdapter(
    host: str,
    port: int,
    tls_cert_path: str | Path,
    tls_key_path: str | Path,
    ca_cert_path: str | Path | None = None,
)

Modbus TCP client with TLS mutual authentication.

Parameters

host : str Target Modbus device hostname or IP. port : int TCP port (default Modbus/TLS: 802). tls_cert_path : str | Path Path to client certificate (PEM). tls_key_path : str | Path Path to client private key (PEM). ca_cert_path : str | Path | None Optional CA bundle (PEM) for server verification. When omitted, the operating-system trust store is used. Server verification is never disabled by this adapter.

Source code in src/scpn_phase_orchestrator/adapters/modbus_tls.py

def __init__(
    self,
    host: str,
    port: int,
    tls_cert_path: str | Path,
    tls_key_path: str | Path,
    ca_cert_path: str | Path | None = None,
) -> None:
    self._host = require_non_empty_str(host, field="Modbus host")
    self._port = require_tcp_port(port, field="Modbus port")
    self._cert = Path(tls_cert_path)
    self._key = Path(tls_key_path)
    self._ca = Path(ca_cert_path) if ca_cert_path is not None else None
    self._ctx = self._build_tls_context()
    self._client = self._connect()

Methods:

read_register

read_register(address: int) -> int

Read a single holding register.

Raises ConnectionError if the read fails or returns an error frame.

Parameters

address : int Modbus register address.

Returns

int The holding-register value.

Raises

ConnectionError If the read fails or the device returns a Modbus error frame.

Source code in src/scpn_phase_orchestrator/adapters/modbus_tls.py

def read_register(self, address: int) -> int:
    """Read a single holding register.

    Raises ConnectionError if the read fails or returns an error frame.

    Parameters
    ----------
    address : int
        Modbus register address.

    Returns
    -------
    int
        The holding-register value.

    Raises
    ------
    ConnectionError
        If the read fails or the device returns a Modbus error frame.
    """
    address = non_negative_int(address, name="address")
    # type ignore: optional pymodbus client is stored as object after runtime guard.
    result = self._client.read_holding_registers(  # type: ignore[attr-defined]
        address, count=1
    )
    if result.isError():
        raise ConnectionError(f"Modbus read error at address {address}: {result}")
    return int(result.registers[0])

write_register

write_register(address: int, value: int) -> None

Write a single holding register.

Raises ConnectionError if the write fails.

Parameters

address : int Modbus register address. value : int Register value to write.

Raises

ConnectionError If the write fails or the device returns a Modbus error frame.

Source code in src/scpn_phase_orchestrator/adapters/modbus_tls.py

def write_register(self, address: int, value: int) -> None:
    """Write a single holding register.

    Raises ConnectionError if the write fails.

    Parameters
    ----------
    address : int
        Modbus register address.
    value : int
        Register value to write.

    Raises
    ------
    ConnectionError
        If the write fails or the device returns a Modbus error frame.
    """
    address = non_negative_int(address, name="address")
    value = _int_value(value, field="value")
    # type ignore: optional pymodbus client is stored as object after runtime guard.
    result = self._client.write_register(address, value)  # type: ignore[attr-defined]
    if result.isError():
        raise ConnectionError(f"Modbus write error at address {address}: {result}")

validate_connection

validate_connection() -> bool

Return True if the TLS-wrapped Modbus connection is active.

Returns

bool Return True if the TLS-wrapped Modbus connection is active.

Source code in src/scpn_phase_orchestrator/adapters/modbus_tls.py

def validate_connection(self) -> bool:
    """Return True if the TLS-wrapped Modbus connection is active.

    Returns
    -------
    bool
        Return True if the TLS-wrapped Modbus connection is active.
    """
    try:
        # type ignore: optional pymodbus client is stored as object.
        return bool(self._client.connected)  # type: ignore[attr-defined]
    except (AttributeError, OSError, RuntimeError):
        return False

close

close() -> None

Close the Modbus/TLS client when the pymodbus client exposes close().

Source code in src/scpn_phase_orchestrator/adapters/modbus_tls.py

def close(self) -> None:
    """Close the Modbus/TLS client when the pymodbus client exposes close()."""
    close = getattr(self._client, "close", None)
    if callable(close):
        close()

__enter__

__enter__() -> SecureModbusAdapter

Return self for context-manager use.

Source code in src/scpn_phase_orchestrator/adapters/modbus_tls.py

def __enter__(self) -> SecureModbusAdapter:
    """Return self for context-manager use."""
    return self

__exit__

__exit__(exc_type: object, exc: object, tb: object) -> None

Close the client on context-manager exit.

Source code in src/scpn_phase_orchestrator/adapters/modbus_tls.py

def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
    """Close the client on context-manager exit."""
    self.close()

Functions:

OPC-UA SCADA Bridge

Read-only ingestion from OPC-UA servers for Industry 4.0 SCADA/DCS systems. OpcUaTag and OpcUaBridgeConfig validate the endpoint and tag mapping; OpcUaPhaseBridge.extract_phases turns decoded process-tag sample series (temperatures, pressures, flow rates) into physical-channel phase states with the tag's declared waveform extractor (hilbert/physical, wavelet, or zero_crossing), with no network dependency. Live reads use the optional asyncua dependency (opcua extra): collect_live connects, reads samples_per_tag values per tag, and disconnects; read_live reads from an already-connected client. The bridge never writes to the server.

from scpn_phase_orchestrator.adapters import OpcUaTag, OpcUaPhaseBridge

bridge = OpcUaPhaseBridge.from_tags(
    "opc.tcp://plc.local:4840/scada",
    [
        OpcUaTag(
            node_id="ns=2;s=Reactor.Temp",
            name="reactor_temp",
            sample_rate_hz=10.0,
            extractor_type="wavelet",
        )
    ],
)
samples = await bridge.collect_live(samples_per_tag=128)
phases = bridge.extract_phases(samples)

opcua_bridge

OPC-UA bridge for industrial SCADA/DCS phase extraction.

Reads oscillator-relevant process tags (temperatures, pressures, flow rates) from an OPC-UA server and maps each tag's sampled waveform to a physical-channel phase state via the tag's declared waveform extractor. Tags default to Hilbert extraction and can select the wavelet-ridge or zero-crossing extractor where those algorithms match the measured signal.

The bridge separates three concerns so the bulk is testable without a server:

• Configuration — :class:OpcUaTag and :class:OpcUaBridgeConfig validate the endpoint URL, tag declarations, and security posture eagerly.
• Phase extraction — :meth:OpcUaPhaseBridge.extract_phases turns decoded tag sample series into per-tag :class:PhaseState objects with no network or asyncua dependency, and :meth:OpcUaPhaseBridge.collect_samples polls an injected synchronous reader callable.
• Live read — :meth:OpcUaPhaseBridge.read_live and :meth:OpcUaPhaseBridge.collect_live use asyncua (optional dependency, opcua extra) to read node values from a connected client.

The bridge never writes to the OPC-UA server; it is a read-only ingestion path.

Classes

OpcUaTag dataclass

OpcUaTag(
    node_id: str,
    name: str,
    channel: str = "P",
    scale: float = 1.0,
    offset: float = 0.0,
    sample_rate_hz: float = 1.0,
    extractor_type: str = "hilbert",
)

Declares how one OPC-UA node maps to a physical oscillator.

Attributes

node_id : str OPC-UA node identifier (e.g. "ns=2;i=4" or "ns=2;s=Reactor.Temp"). name : str Oscillator name the extracted phase state is bound to. channel : str SPO channel label; one of "P", "R", "E", "S" (default "P" for the physical channel). scale, offset : float Affine calibration applied to each raw sample as scale * x + offset. sample_rate_hz : float Sampling rate of the tag waveform in hertz, used by waveform phase extraction. extractor_type : str Waveform extractor type or channel alias. "physical" resolves to "hilbert"; "wavelet" and "zero_crossing" select the corresponding physical-channel algorithms.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the tag.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the tag fields.

Source code in src/scpn_phase_orchestrator/adapters/opcua_bridge.py

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

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the tag fields.
    """
    return {
        "node_id": self.node_id,
        "name": self.name,
        "channel": self.channel,
        "scale": self.scale,
        "offset": self.offset,
        "sample_rate_hz": self.sample_rate_hz,
        "extractor_type": self.extractor_type,
    }

OpcUaBridgeConfig dataclass

OpcUaBridgeConfig(
    endpoint_url: str,
    tags: tuple[OpcUaTag, ...],
    security_policy: str = "None",
    security_mode: str = "None",
    request_timeout_s: float = 4.0,
)

Validated OPC-UA connection and tag-mapping configuration.

Attributes

endpoint_url : str OPC-UA endpoint, must use the opc.tcp:// scheme. tags : tuple[OpcUaTag, ...] The tags to read; node identifiers and oscillator names must be unique. security_policy : str OPC-UA security policy (default "None"). security_mode : str Message security mode (default "None"). request_timeout_s : float Per-request timeout in seconds for the live client.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the configuration.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the configuration fields.

Source code in src/scpn_phase_orchestrator/adapters/opcua_bridge.py

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

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the configuration fields.
    """
    return {
        "endpoint_url": self.endpoint_url,
        "tags": [tag.to_audit_record() for tag in self.tags],
        "security_policy": self.security_policy,
        "security_mode": self.security_mode,
        "request_timeout_s": self.request_timeout_s,
    }

OpcUaPhaseBridge dataclass

OpcUaPhaseBridge(config: OpcUaBridgeConfig)

Read-only OPC-UA tag ingestion mapped to physical phase states.

Attributes

config : OpcUaBridgeConfig The validated bridge configuration.

Methods:

from_tags classmethod

from_tags(
    endpoint_url: str,
    tags: Sequence[OpcUaTag],
    **config_kwargs: object,
) -> OpcUaPhaseBridge

Build a bridge from an endpoint and a tag sequence.

Parameters

endpoint_url : str OPC-UA endpoint (opc.tcp:// scheme). tags : Sequence[OpcUaTag] The tags to read. **config_kwargs : object Forwarded to :class:OpcUaBridgeConfig (security and timeout).

Returns

OpcUaPhaseBridge A configured bridge.

Source code in src/scpn_phase_orchestrator/adapters/opcua_bridge.py

@classmethod
def from_tags(
    cls,
    endpoint_url: str,
    tags: Sequence[OpcUaTag],
    **config_kwargs: object,
) -> OpcUaPhaseBridge:
    """Build a bridge from an endpoint and a tag sequence.

    Parameters
    ----------
    endpoint_url : str
        OPC-UA endpoint (``opc.tcp://`` scheme).
    tags : Sequence[OpcUaTag]
        The tags to read.
    **config_kwargs : object
        Forwarded to :class:`OpcUaBridgeConfig` (security and timeout).

    Returns
    -------
    OpcUaPhaseBridge
        A configured bridge.
    """
    config = OpcUaBridgeConfig(
        endpoint_url=endpoint_url,
        tags=tuple(tags),
        # type ignore: forwarded **kwargs are validated by
        # OpcUaBridgeConfig.__post_init__ at construction time.
        **config_kwargs,  # type: ignore[arg-type]
    )
    return cls(config=config)

extract_phases

extract_phases(
    tag_samples: Mapping[str, Sequence[float]],
) -> dict[str, PhaseState]

Map decoded tag sample series to per-tag physical phase states.

Parameters

tag_samples : Mapping[str, Sequence[float]] Sample series keyed by tag name; every declared tag must be present with at least one finite sample.

Returns

dict[str, PhaseState] The latest instantaneous phase state per tag, keyed by tag name.

Raises

ValueError If a declared tag is missing, a series is empty, or a sample is not a finite real value.

Source code in src/scpn_phase_orchestrator/adapters/opcua_bridge.py

def extract_phases(
    self,
    tag_samples: Mapping[str, Sequence[float]],
) -> dict[str, PhaseState]:
    """Map decoded tag sample series to per-tag physical phase states.

    Parameters
    ----------
    tag_samples : Mapping[str, Sequence[float]]
        Sample series keyed by tag name; every declared tag must be present
        with at least one finite sample.

    Returns
    -------
    dict[str, PhaseState]
        The latest instantaneous phase state per tag, keyed by tag name.

    Raises
    ------
    ValueError
        If a declared tag is missing, a series is empty, or a sample is not a
        finite real value.
    """
    phases: dict[str, PhaseState] = {}
    for tag in self.config.tags:
        if tag.name not in tag_samples:
            raise ValueError(f"missing samples for tag {tag.name!r}")
        raw = tag_samples[tag.name]
        if len(raw) == 0:
            raise ValueError(f"tag {tag.name!r} has no samples")
        calibrated = np.asarray(
            [_finite_real(value, field_name=f"{tag.name} sample") for value in raw],
            dtype=np.float64,
        )
        calibrated = tag.scale * calibrated + tag.offset
        states = self._extractors[tag.name].extract(calibrated, tag.sample_rate_hz)
        phases[tag.name] = states[0]
    return phases

collect_samples

collect_samples(
    reader: Callable[[str], float], *, samples_per_tag: int
) -> dict[str, list[float]]

Poll a synchronous reader callable into per-tag sample series.

Parameters

reader : Callable[[str], float] Returns the current value for a node identifier; called samples_per_tag times per tag in declaration order. samples_per_tag : int Number of samples to collect per tag.

Returns

dict[str, list[float]] Sample series keyed by tag name.

Raises

ValueError If samples_per_tag is not a positive integer or a read value is not a finite real number.

Source code in src/scpn_phase_orchestrator/adapters/opcua_bridge.py

def collect_samples(
    self,
    reader: Callable[[str], float],
    *,
    samples_per_tag: int,
) -> dict[str, list[float]]:
    """Poll a synchronous reader callable into per-tag sample series.

    Parameters
    ----------
    reader : Callable[[str], float]
        Returns the current value for a node identifier; called
        ``samples_per_tag`` times per tag in declaration order.
    samples_per_tag : int
        Number of samples to collect per tag.

    Returns
    -------
    dict[str, list[float]]
        Sample series keyed by tag name.

    Raises
    ------
    ValueError
        If ``samples_per_tag`` is not a positive integer or a read value is
        not a finite real number.
    """
    count = _positive_int(samples_per_tag, field_name="samples_per_tag")
    samples: dict[str, list[float]] = {tag.name: [] for tag in self.config.tags}
    for _ in range(count):
        for tag in self.config.tags:
            value = _finite_real(
                reader(tag.node_id), field_name=f"{tag.name} reading"
            )
            samples[tag.name].append(value)
    return samples

connect

connect() -> Client

Create a (not yet connected) asyncua client for the endpoint.

Returns

asyncua.Client A client configured for the endpoint and request timeout. Use it as an async context manager to open and close the connection.

Raises

RuntimeError If the optional asyncua dependency is not installed.

Source code in src/scpn_phase_orchestrator/adapters/opcua_bridge.py

def connect(self) -> Client:
    """Create a (not yet connected) ``asyncua`` client for the endpoint.

    Returns
    -------
    asyncua.Client
        A client configured for the endpoint and request timeout. Use it as
        an async context manager to open and close the connection.

    Raises
    ------
    RuntimeError
        If the optional ``asyncua`` dependency is not installed.
    """
    if not HAS_ASYNCUA:
        raise RuntimeError(
            "asyncua is not installed; install the 'opcua' extra to read live"
        )
    from asyncua import Client as AsyncuaClient

    return AsyncuaClient(
        url=self.config.endpoint_url,
        timeout=self.config.request_timeout_s,
    )

read_live async

read_live(
    client: Any,
    *,
    samples_per_tag: int,
    interval_s: float = 0.0,
) -> dict[str, list[float]]

Read node values from a connected asyncua client.

Parameters

client : asyncua.Client A connected client (or any object exposing get_node(node_id) with an awaitable read_value). samples_per_tag : int Number of samples to read per tag. interval_s : float, optional Delay in seconds between sampling rounds (default 0).

Returns

dict[str, list[float]] Sample series keyed by tag name.

Raises

ValueError If samples_per_tag is not positive, interval_s is negative, or a read value is not a finite real number.

Source code in src/scpn_phase_orchestrator/adapters/opcua_bridge.py

async def read_live(
    self,
    client: Any,
    *,
    samples_per_tag: int,
    interval_s: float = 0.0,
) -> dict[str, list[float]]:
    """Read node values from a connected ``asyncua`` client.

    Parameters
    ----------
    client : asyncua.Client
        A connected client (or any object exposing ``get_node(node_id)``
        with an awaitable ``read_value``).
    samples_per_tag : int
        Number of samples to read per tag.
    interval_s : float, optional
        Delay in seconds between sampling rounds (default ``0``).

    Returns
    -------
    dict[str, list[float]]
        Sample series keyed by tag name.

    Raises
    ------
    ValueError
        If ``samples_per_tag`` is not positive, ``interval_s`` is negative, or
        a read value is not a finite real number.
    """
    count = _positive_int(samples_per_tag, field_name="samples_per_tag")
    interval = _finite_real(interval_s, field_name="interval_s")
    if interval < 0.0:
        raise ValueError("interval_s must be >= 0")
    nodes = {tag.name: client.get_node(tag.node_id) for tag in self.config.tags}
    samples: dict[str, list[float]] = {tag.name: [] for tag in self.config.tags}
    for index in range(count):
        if index > 0 and interval > 0.0:
            await asyncio.sleep(interval)
        for tag in self.config.tags:
            value = await nodes[tag.name].read_value()
            samples[tag.name].append(
                _finite_real(value, field_name=f"{tag.name} reading")
            )
    return samples

collect_live async

collect_live(
    *, samples_per_tag: int, interval_s: float = 0.0
) -> dict[str, list[float]]

Connect, read samples_per_tag samples per tag, and disconnect.

Parameters

samples_per_tag : int Number of samples to read per tag. interval_s : float, optional Delay in seconds between sampling rounds (default 0).

Returns

dict[str, list[float]] Sample series keyed by tag name.

Raises

RuntimeError If the optional asyncua dependency is not installed.

Source code in src/scpn_phase_orchestrator/adapters/opcua_bridge.py

async def collect_live(
    self,
    *,
    samples_per_tag: int,
    interval_s: float = 0.0,
) -> dict[str, list[float]]:
    """Connect, read ``samples_per_tag`` samples per tag, and disconnect.

    Parameters
    ----------
    samples_per_tag : int
        Number of samples to read per tag.
    interval_s : float, optional
        Delay in seconds between sampling rounds (default ``0``).

    Returns
    -------
    dict[str, list[float]]
        Sample series keyed by tag name.

    Raises
    ------
    RuntimeError
        If the optional ``asyncua`` dependency is not installed.
    """
    client = self.connect()
    async with client:
        return await self.read_live(
            client,
            samples_per_tag=samples_per_tag,
            interval_s=interval_s,
        )

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the bridge configuration.

Returns

dict[str, object] Deterministic, JSON-safe mapping with the configuration and the asyncua availability flag.

Source code in src/scpn_phase_orchestrator/adapters/opcua_bridge.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the bridge configuration.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping with the configuration and the
        ``asyncua`` availability flag.
    """
    return {
        "config": self.config.to_audit_record(),
        "asyncua_available": HAS_ASYNCUA,
    }

Functions:

MQTT Edge Bridge

Read-only ingestion from MQTT brokers for edge/IoT sensor fleets. MqttTag and MqttBridgeConfig validate the broker endpoint and topic mapping; decode_payload parses raw or JSON payloads, ingest_messages folds a batch of received (topic, payload) messages into per-tag sample series, and extract_phases turns those series into physical-channel phase states with the tag's declared waveform extractor (hilbert/physical, wavelet, or zero_crossing) — all with no network dependency. collect_live subscribes via the optional paho-mqtt dependency (mqtt extra) and accumulates messages; it accepts an injected client for testing. The bridge never publishes to the broker.

from scpn_phase_orchestrator.adapters import MqttTag, MqttPhaseBridge

bridge = MqttPhaseBridge.from_tags(
    "broker.local",
    [
        MqttTag(
            topic="plant/reactor/temp",
            name="reactor_temp",
            sample_rate_hz=10.0,
            extractor_type="zero_crossing",
        )
    ],
)
samples = bridge.collect_live(samples_per_tag=128)
phases = bridge.extract_phases(samples)

mqtt_bridge

MQTT bridge for edge/IoT sensor phase extraction.

Subscribes to MQTT topics carrying oscillator-relevant process measurements and maps each topic's sampled waveform to a physical-channel phase state via the tag's declared waveform extractor. Tags default to Hilbert extraction and can select the wavelet-ridge or zero-crossing extractor where those algorithms match the measured signal.

Like the OPC-UA bridge, the bulk is testable without a broker:

• Configuration — :class:MqttTag and :class:MqttBridgeConfig validate the broker endpoint and topic mapping eagerly.
• Decoding and ingestion — :meth:MqttPhaseBridge.decode_payload parses raw or JSON payloads, :meth:MqttPhaseBridge.ingest_messages folds a batch of (topic, payload) messages into per-tag sample series, and :meth:MqttPhaseBridge.extract_phases turns those series into :class:PhaseState objects — all with no network or paho-mqtt dependency.
• Live subscribe — :meth:MqttPhaseBridge.collect_live uses paho-mqtt (optional dependency, mqtt extra) to subscribe and accumulate messages.

The bridge is read-only: it never publishes to the broker.

Classes

MqttTag dataclass

MqttTag(
    topic: str,
    name: str,
    channel: str = "P",
    scale: float = 1.0,
    offset: float = 0.0,
    sample_rate_hz: float = 1.0,
    extractor_type: str = "hilbert",
    payload_format: str = "raw",
)

Declares how one MQTT topic maps to a physical oscillator.

Attributes

topic : str MQTT topic the sensor publishes to. name : str Oscillator name the extracted phase state is bound to. channel : str SPO channel label; one of "P", "R", "E", "S". scale, offset : float Affine calibration applied to each decoded value as scale * x + offset. sample_rate_hz : float Publish rate of the topic in hertz, used by waveform phase extraction. extractor_type : str Waveform extractor type or channel alias. "physical" resolves to "hilbert"; "wavelet" and "zero_crossing" select the corresponding physical-channel algorithms. payload_format : str "raw" (a decimal number as text) or "json" (a JSON number or a JSON object with a "value" field).

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the tag.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the tag fields.

Source code in src/scpn_phase_orchestrator/adapters/mqtt_bridge.py

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

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the tag fields.
    """
    return {
        "topic": self.topic,
        "name": self.name,
        "channel": self.channel,
        "scale": self.scale,
        "offset": self.offset,
        "sample_rate_hz": self.sample_rate_hz,
        "extractor_type": self.extractor_type,
        "payload_format": self.payload_format,
    }

MqttBridgeConfig dataclass

MqttBridgeConfig(
    broker_host: str,
    tags: tuple[MqttTag, ...],
    broker_port: int = 1883,
    keepalive_s: int = 60,
    client_id: str = "spo-mqtt-bridge",
    use_tls: bool = False,
)

Validated MQTT connection and topic-mapping configuration.

Attributes

broker_host : str MQTT broker hostname or address. tags : tuple[MqttTag, ...] Topics to subscribe to; topics and oscillator names must be unique. broker_port : int Broker TCP port (default 1883). keepalive_s : int Keep-alive interval in seconds for the live client. client_id : str MQTT client identifier. use_tls : bool Whether the live client should negotiate TLS.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the configuration.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the configuration fields.

Source code in src/scpn_phase_orchestrator/adapters/mqtt_bridge.py

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

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the configuration fields.
    """
    return {
        "broker_host": self.broker_host,
        "broker_port": self.broker_port,
        "keepalive_s": self.keepalive_s,
        "client_id": self.client_id,
        "use_tls": self.use_tls,
        "tags": [tag.to_audit_record() for tag in self.tags],
    }

MqttPhaseBridge dataclass

MqttPhaseBridge(config: MqttBridgeConfig)

Read-only MQTT topic ingestion mapped to physical phase states.

Attributes

config : MqttBridgeConfig The validated bridge configuration.

Methods:

from_tags classmethod

from_tags(
    broker_host: str,
    tags: Sequence[MqttTag],
    **config_kwargs: object,
) -> MqttPhaseBridge

Build a bridge from a broker host and a tag sequence.

Parameters

broker_host : str MQTT broker hostname or address. tags : Sequence[MqttTag] The topics to subscribe to. **config_kwargs : object Forwarded to :class:MqttBridgeConfig.

Returns

MqttPhaseBridge A configured bridge.

Source code in src/scpn_phase_orchestrator/adapters/mqtt_bridge.py

@classmethod
def from_tags(
    cls,
    broker_host: str,
    tags: Sequence[MqttTag],
    **config_kwargs: object,
) -> MqttPhaseBridge:
    """Build a bridge from a broker host and a tag sequence.

    Parameters
    ----------
    broker_host : str
        MQTT broker hostname or address.
    tags : Sequence[MqttTag]
        The topics to subscribe to.
    **config_kwargs : object
        Forwarded to :class:`MqttBridgeConfig`.

    Returns
    -------
    MqttPhaseBridge
        A configured bridge.
    """
    config = MqttBridgeConfig(
        broker_host=broker_host,
        tags=tuple(tags),
        # type ignore: forwarded **kwargs are validated by
        # MqttBridgeConfig.__post_init__ at construction time.
        **config_kwargs,  # type: ignore[arg-type]
    )
    return cls(config=config)

decode_payload

decode_payload(tag: MqttTag, payload: bytes | str) -> float

Decode one MQTT payload into a finite calibrated sample value.

Parameters

tag : MqttTag The tag whose payload_format and calibration apply. payload : bytes or str The raw message payload.

Returns

float The decoded scale * value + offset sample.

Raises

ValueError If the payload cannot be decoded to a finite real value.

Source code in src/scpn_phase_orchestrator/adapters/mqtt_bridge.py

def decode_payload(self, tag: MqttTag, payload: bytes | str) -> float:
    """Decode one MQTT payload into a finite calibrated sample value.

    Parameters
    ----------
    tag : MqttTag
        The tag whose ``payload_format`` and calibration apply.
    payload : bytes or str
        The raw message payload.

    Returns
    -------
    float
        The decoded ``scale * value + offset`` sample.

    Raises
    ------
    ValueError
        If the payload cannot be decoded to a finite real value.
    """
    text = payload.decode("utf-8") if isinstance(payload, bytes) else payload
    if tag.payload_format == "json":
        try:
            decoded = json.loads(text)
        except json.JSONDecodeError as exc:
            raise ValueError(f"tag {tag.name!r} payload is not valid JSON") from exc
        if isinstance(decoded, Mapping):
            decoded = decoded.get("value")
        raw = _finite_real(decoded, field_name=f"{tag.name} value")
    else:
        try:
            raw = float(text)
        except (TypeError, ValueError) as exc:
            raise ValueError(
                f"tag {tag.name!r} payload is not a number: {text!r}"
            ) from exc
        if not np.isfinite(raw):
            raise ValueError(f"tag {tag.name!r} payload must be finite")
    return tag.scale * raw + tag.offset

ingest_messages

ingest_messages(
    messages: Sequence[tuple[str, bytes | str]],
) -> dict[str, list[float]]

Fold a batch of (topic, payload) messages into per-tag series.

Messages on unknown topics are ignored. Decoded values are appended in arrival order to the series of the tag matching their topic.

Parameters

messages : Sequence[tuple[str, bytes | str]] The received messages.

Returns

dict[str, list[float]] Calibrated sample series keyed by tag name (empty list per tag with no matching messages).

Source code in src/scpn_phase_orchestrator/adapters/mqtt_bridge.py

def ingest_messages(
    self,
    messages: Sequence[tuple[str, bytes | str]],
) -> dict[str, list[float]]:
    """Fold a batch of ``(topic, payload)`` messages into per-tag series.

    Messages on unknown topics are ignored. Decoded values are appended in
    arrival order to the series of the tag matching their topic.

    Parameters
    ----------
    messages : Sequence[tuple[str, bytes | str]]
        The received messages.

    Returns
    -------
    dict[str, list[float]]
        Calibrated sample series keyed by tag name (empty list per tag with
        no matching messages).
    """
    samples: dict[str, list[float]] = {tag.name: [] for tag in self.config.tags}
    for topic, payload in messages:
        tag = self._by_topic.get(topic)
        if tag is None:
            continue
        samples[tag.name].append(self.decode_payload(tag, payload))
    return samples

extract_phases

extract_phases(
    topic_samples: Mapping[str, Sequence[float]],
) -> dict[str, PhaseState]

Map per-tag sample series to physical phase states.

Parameters

topic_samples : Mapping[str, Sequence[float]] Calibrated sample series keyed by tag name; every declared tag must be present with at least one finite sample.

Returns

dict[str, PhaseState] The latest instantaneous phase state per tag, keyed by tag name.

Raises

ValueError If a declared tag is missing, a series is empty, or a sample is not a finite real value.

Source code in src/scpn_phase_orchestrator/adapters/mqtt_bridge.py

def extract_phases(
    self,
    topic_samples: Mapping[str, Sequence[float]],
) -> dict[str, PhaseState]:
    """Map per-tag sample series to physical phase states.

    Parameters
    ----------
    topic_samples : Mapping[str, Sequence[float]]
        Calibrated sample series keyed by tag name; every declared tag must be
        present with at least one finite sample.

    Returns
    -------
    dict[str, PhaseState]
        The latest instantaneous phase state per tag, keyed by tag name.

    Raises
    ------
    ValueError
        If a declared tag is missing, a series is empty, or a sample is not a
        finite real value.
    """
    phases: dict[str, PhaseState] = {}
    for tag in self.config.tags:
        if tag.name not in topic_samples:
            raise ValueError(f"missing samples for tag {tag.name!r}")
        raw = topic_samples[tag.name]
        if len(raw) == 0:
            raise ValueError(f"tag {tag.name!r} has no samples")
        series = np.asarray(
            [_finite_real(value, field_name=f"{tag.name} sample") for value in raw],
            dtype=np.float64,
        )
        states = self._extractors[tag.name].extract(series, tag.sample_rate_hz)
        phases[tag.name] = states[0]
    return phases

connect

connect() -> Client

Create a configured (not yet connected) paho-mqtt client.

Returns

paho.mqtt.client.Client A client bound to the configured client id and TLS posture.

Raises

RuntimeError If the optional paho-mqtt dependency is not installed.

Source code in src/scpn_phase_orchestrator/adapters/mqtt_bridge.py

def connect(self) -> Client:
    """Create a configured (not yet connected) ``paho-mqtt`` client.

    Returns
    -------
    paho.mqtt.client.Client
        A client bound to the configured client id and TLS posture.

    Raises
    ------
    RuntimeError
        If the optional ``paho-mqtt`` dependency is not installed.
    """
    if not HAS_PAHO_MQTT:
        raise RuntimeError(
            "paho-mqtt is not installed; install the 'mqtt' extra to read live"
        )
    from paho.mqtt.client import CallbackAPIVersion, Client

    client = Client(CallbackAPIVersion.VERSION2, client_id=self.config.client_id)
    if self.config.use_tls:
        client.tls_set()
    return client

collect_live

collect_live(
    *,
    samples_per_tag: int,
    timeout_s: float = 10.0,
    client: Any = None,
) -> dict[str, list[float]]

Subscribe to the configured topics and accumulate samples.

Parameters

samples_per_tag : int Target number of samples to collect for every tag before returning. timeout_s : float, optional Maximum seconds to wait for the target to be reached (default 10). client : paho.mqtt.client.Client, optional An existing client; a new one is created via :meth:connect when omitted.

Returns

dict[str, list[float]] Sample series keyed by tag name (each truncated to at most samples_per_tag).

Raises

ValueError If samples_per_tag is not positive or timeout_s is not finite and positive.

Source code in src/scpn_phase_orchestrator/adapters/mqtt_bridge.py

def collect_live(
    self,
    *,
    samples_per_tag: int,
    timeout_s: float = 10.0,
    client: Any = None,
) -> dict[str, list[float]]:
    """Subscribe to the configured topics and accumulate samples.

    Parameters
    ----------
    samples_per_tag : int
        Target number of samples to collect for every tag before returning.
    timeout_s : float, optional
        Maximum seconds to wait for the target to be reached (default ``10``).
    client : paho.mqtt.client.Client, optional
        An existing client; a new one is created via :meth:`connect` when
        omitted.

    Returns
    -------
    dict[str, list[float]]
        Sample series keyed by tag name (each truncated to at most
        ``samples_per_tag``).

    Raises
    ------
    ValueError
        If ``samples_per_tag`` is not positive or ``timeout_s`` is not finite
        and positive.
    """
    count = _positive_int(samples_per_tag, field_name="samples_per_tag")
    deadline = _positive_real(timeout_s, field_name="timeout_s")
    active = client if client is not None else self.connect()
    samples: dict[str, list[float]] = {tag.name: [] for tag in self.config.tags}

    def on_message(_client: Any, _userdata: Any, message: Any) -> None:
        """Handle an incoming MQTT message (paho client callback)."""
        tag = self._by_topic.get(message.topic)
        if tag is None:
            return
        series = samples[tag.name]
        if len(series) < count:
            series.append(self.decode_payload(tag, message.payload))

    active.on_message = on_message
    active.connect(
        self.config.broker_host, self.config.broker_port, self.config.keepalive_s
    )
    for tag in self.config.tags:
        active.subscribe(tag.topic)
    active.loop_start()
    try:
        start = time.monotonic()
        while time.monotonic() - start < deadline:
            if all(len(series) >= count for series in samples.values()):
                break
            time.sleep(0.01)
    finally:
        active.loop_stop()
        active.disconnect()
    return samples

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the bridge configuration.

Returns

dict[str, object] Deterministic, JSON-safe mapping with the configuration and the paho-mqtt availability flag.

Source code in src/scpn_phase_orchestrator/adapters/mqtt_bridge.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the bridge configuration.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping with the configuration and the
        ``paho-mqtt`` availability flag.
    """
    return {
        "config": self.config.to_audit_record(),
        "paho_mqtt_available": HAS_PAHO_MQTT,
    }

Functions:

IEEE C37.118.2 Synchrophasor Codec

Dependency-free decoder for IEEE C37.118.2-2011 synchrophasor CONFIG-2 and DATA frames from raw bytes (no network I/O). SynchrophasorFrameCodec.decode_config2 recovers each PMU's measurement layout (FORMAT flags, phasor/analog/digital counts, nominal frequency); decode_data then decodes the phasor, frequency, and analog/digital measurements, interpreting FREQ as a deviation from nominal (millihertz when integer, hertz when float). Every frame is CRC-CCITT validated before its body is read, and malformed input raises a typed SynchrophasorFrameError subclass rather than returning partial data. The byte layout and CRC parameters were cross-checked against two independent open-source implementations (iicsys/pypmu and marsolla/Open-C37.118). data_frames_to_frequency_series assembles a (time_s, frequency_hz) series in the exact layout the PMU ringdown screener consumes, so a decoded stream feeds directly into hash-sealed ringdown evidence. The live-socket ingestion path is C37118SessionClient (pure-standard-library asyncio, no optional extra required); this codec handles only bytes already read.

from scpn_phase_orchestrator.adapters import (
    SynchrophasorFrameCodec,
    data_frames_to_frequency_series,
)

codec = SynchrophasorFrameCodec()
config = codec.decode_config2(config2_bytes)
frames = tuple(codec.decode_data(data_bytes, config) for data_bytes in stream)
time_s, frequency_hz = data_frames_to_frequency_series(config, frames, pmu_index=0)

synchrophasor_c37118

Pure decoder for IEEE C37.118.2-2011 synchrophasor CONFIG-2 and DATA frames.

This module decodes the binary framing of the IEEE synchrophasor data-transfer protocol without any network I/O: given the raw bytes of a CONFIG-2 frame it recovers the per-PMU measurement layout, and given the bytes of a DATA frame plus that configuration it recovers the phasor, frequency, and analog/digital measurements. Every frame is checksum-validated (CRC-CCITT) before its body is read, and malformed input raises a typed :class:SynchrophasorFrameError subclass rather than returning partial data.

The byte layout, CRC parameters, and field semantics were cross-checked against two independent open-source implementations of the standard: the pypmu Python library (iicsys/pypmu, synchrophasor/frame.py) and the C++ Open-C37.118 library (marsolla/Open-C37.118, src/c37118*.{h,cpp}). Both agree on the 14-byte common header (SYNC FRAMESIZE IDCODE SOC FRACSEC, big-endian), the FORMAT-word field sizes, and the CRC-CCITT checksum (polynomial 0x1021, initial value 0xFFFF, no final mask, computed over every byte except the trailing two). The FREQ field is a deviation from the PMU nominal frequency: a signed 16-bit integer in millihertz when the FORMAT freq bit is clear, or a 32-bit float in hertz when it is set. The live-socket ingestion path is :class:~scpn_phase_orchestrator.adapters.synchrophasor_client.C37118SessionClient (pure-standard-library asyncio, no optional extra required); this module deliberately handles only bytes already read.

Classes

SynchrophasorFrameError

Bases: ValueError

Base class for all synchrophasor frame decoding failures.

FrameTruncationError

Bases: SynchrophasorFrameError

Raised when a frame is shorter than its declared or required length.

FrameChecksumError

Bases: SynchrophasorFrameError

Raised when the trailing CRC-CCITT checksum does not match the body.

UnsupportedFrameError

Bases: SynchrophasorFrameError

Raised for a frame whose SYNC/type is not the expected decodable kind.

SynchrophasorHeader dataclass

SynchrophasorHeader(
    frame_type: int,
    version: int,
    framesize: int,
    id_code: int,
    soc: int,
    fracsec_raw: int,
)

Decoded 14-byte common header shared by every synchrophasor frame.

Attributes

frame_type : int Frame-type code from SYNC byte 2 (bits 6-4); e.g. :data:FRAME_TYPE_DATA or :data:FRAME_TYPE_CONFIG2. version : int Protocol version number from SYNC byte 2 (bits 3-0). framesize : int Declared total frame size in bytes, including SYNC and CRC. id_code : int Data-stream / PMU identification code. soc : int Second-of-century timestamp (UNIX seconds). fracsec_raw : int Raw 32-bit FRACSEC word (time-quality byte plus fraction count).

Attributes

fraction_count property

fraction_count: int

Return the raw fraction-of-second count (lower 24 bits of FRACSEC).

message_time_quality property

message_time_quality: int

Return the 4-bit message time-quality code from the FRACSEC top byte.

leap_second_pending property

leap_second_pending: bool

Return whether the leap-second-pending flag is set.

leap_second_occurred property

leap_second_occurred: bool

Return whether the leap-second-occurred flag is set.

leap_second_direction property

leap_second_direction: str

Return the leap-second direction (- if flagged, else +).

Methods:

seconds_of_second

seconds_of_second(time_base: int) -> float

Return the fractional-second offset as a float given time_base.

Parameters

time_base : int The CONFIG-2 TIME_BASE resolution of the fractional timestamp.

Returns

float The fraction of a second, fraction_count / time_base.

Raises

SynchrophasorFrameError If time_base is not a positive integer.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

def seconds_of_second(self, time_base: int) -> float:
    """Return the fractional-second offset as a float given ``time_base``.

    Parameters
    ----------
    time_base : int
        The CONFIG-2 ``TIME_BASE`` resolution of the fractional timestamp.

    Returns
    -------
    float
        The fraction of a second, ``fraction_count / time_base``.

    Raises
    ------
    SynchrophasorFrameError
        If ``time_base`` is not a positive integer.
    """
    if time_base <= 0:
        raise SynchrophasorFrameError("time_base must be a positive integer")
    return self.fraction_count / time_base

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the header fields.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the header fields.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the header fields.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the header fields.
    """
    return {
        "frame_type": self.frame_type,
        "version": self.version,
        "framesize": self.framesize,
        "id_code": self.id_code,
        "soc": self.soc,
        "fraction_count": self.fraction_count,
        "message_time_quality": self.message_time_quality,
        "leap_second_pending": self.leap_second_pending,
        "leap_second_occurred": self.leap_second_occurred,
        "leap_second_direction": self.leap_second_direction,
    }

PhasorUnit dataclass

PhasorUnit(is_current: bool, scale: int)

Conversion factor for one phasor channel (a decoded PHUNIT word).

Attributes

is_current : bool True if the channel is a current phasor, False for voltage (PHUNIT most-significant byte). scale : int Unsigned 24-bit scale factor in 10**-5 volts or amperes per bit, used to convert 16-bit integer phasor components to engineering units. Ignored for floating-point phasors, which are already in engineering units.

Attributes

volts_or_amperes_per_bit property

volts_or_amperes_per_bit: float

Return the engineering-unit scale per integer bit (scale * 1e-5).

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the phasor unit.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the phasor conversion factor.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the phasor unit.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the phasor conversion factor.
    """
    return {"is_current": self.is_current, "scale": self.scale}

PmuConfiguration dataclass

PmuConfiguration(
    station_name: str,
    id_code: int,
    phasor_polar: bool,
    phasor_float: bool,
    analog_float: bool,
    freq_float: bool,
    phasor_count: int,
    analog_count: int,
    digital_word_count: int,
    channel_names: tuple[str, ...],
    nominal_frequency_hz: float,
    phasor_units: tuple[PhasorUnit, ...] = (),
)

Per-PMU measurement layout decoded from a CONFIG-2 frame.

Attributes

station_name : str Human-readable station name (trimmed of NUL/space padding). id_code : int PMU identification code. phasor_polar : bool True if phasors are polar (magnitude, angle); False if rectangular. phasor_float : bool True if phasors use 32-bit floats; False if 16-bit integers. analog_float : bool True if analog values use 32-bit floats; False if 16-bit integers. freq_float : bool True if FREQ/DFREQ use 32-bit floats (hertz); False if 16-bit integers (millihertz deviation). phasor_count, analog_count, digital_word_count : int PHNMR, ANNMR, and DGNMR counts respectively. channel_names : tuple[str, ...] Phasor, analog, and digital channel labels in declared order. nominal_frequency_hz : float Nominal line frequency (50.0 or 60.0 Hz) from the FNOM word. phasor_units : tuple[PhasorUnit, ...] Per-phasor conversion factors (PHUNIT), one per phasor channel.

Attributes

phasor_size property

phasor_size: int

Return the byte size of one phasor (8 if float, else 4).

freq_size property

freq_size: int

Return the byte size of the FREQ/DFREQ field (4 if float, else 2).

analog_size property

analog_size: int

Return the byte size of one analog value (4 if float, else 2).

data_block_size property

data_block_size: int

Return the byte size of this PMU's block within a DATA frame.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the PMU configuration.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the PMU configuration fields.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

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

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the PMU configuration fields.
    """
    return {
        "station_name": self.station_name,
        "id_code": self.id_code,
        "phasor_polar": self.phasor_polar,
        "phasor_float": self.phasor_float,
        "analog_float": self.analog_float,
        "freq_float": self.freq_float,
        "phasor_count": self.phasor_count,
        "analog_count": self.analog_count,
        "digital_word_count": self.digital_word_count,
        "channel_names": list(self.channel_names),
        "nominal_frequency_hz": self.nominal_frequency_hz,
        "phasor_units": [unit.to_audit_record() for unit in self.phasor_units],
    }

ConfigurationFrame2 dataclass

ConfigurationFrame2(
    header: SynchrophasorHeader,
    time_base: int,
    pmus: tuple[PmuConfiguration, ...],
    data_rate: int,
)

Decoded CONFIG-2 frame describing every PMU in the data stream.

Attributes

header : SynchrophasorHeader The decoded common header. time_base : int Resolution of the fractional-second timestamp (TIME_BASE). pmus : tuple[PmuConfiguration, ...] Per-PMU measurement layouts in declared order. data_rate : int Reporting rate: frames per second if positive, seconds per frame if negative.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the configuration frame.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the configuration frame.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the configuration frame.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the configuration frame.
    """
    return {
        "header": self.header.to_audit_record(),
        "time_base": self.time_base,
        "pmu_count": len(self.pmus),
        "pmus": [pmu.to_audit_record() for pmu in self.pmus],
        "data_rate": self.data_rate,
    }

PmuMeasurement dataclass

PmuMeasurement(
    stat: int,
    phasors: tuple[tuple[float, float], ...],
    frequency_hz: float,
    frequency_deviation: float,
    df_dt: float,
    analogs: tuple[float, ...],
    digitals: tuple[int, ...],
)

One PMU's measurements decoded from a DATA frame block.

Attributes

stat : int 16-bit STAT flag word. phasors : tuple[tuple[float, float], ...] Phasor components in the frame's native representation: rectangular (real, imag) or polar (magnitude, angle) per the PMU's FORMAT. frequency_hz : float Absolute frequency in hertz (nominal plus the decoded deviation). frequency_deviation : float Raw FREQ deviation from nominal (millihertz if integer, hertz if float). df_dt : float Rate-of-change of frequency (DFREQ) in the frame's native units. analogs : tuple[float, ...] Analog channel values. digitals : tuple[int, ...] Digital status words.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the PMU measurement.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the PMU measurement.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

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

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the PMU measurement.
    """
    return {
        "stat": self.stat,
        "phasors": [list(component) for component in self.phasors],
        "frequency_hz": self.frequency_hz,
        "frequency_deviation": self.frequency_deviation,
        "df_dt": self.df_dt,
        "analogs": list(self.analogs),
        "digitals": list(self.digitals),
    }

DataFrame dataclass

DataFrame(
    header: SynchrophasorHeader,
    measurements: tuple[PmuMeasurement, ...],
)

Decoded DATA frame carrying one measurement per configured PMU.

Attributes

header : SynchrophasorHeader The decoded common header. measurements : tuple[PmuMeasurement, ...] Per-PMU measurements aligned with the configuration's PMU order.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the data frame.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the data frame.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the data frame.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the data frame.
    """
    return {
        "header": self.header.to_audit_record(),
        "measurements": [m.to_audit_record() for m in self.measurements],
    }

SynchrophasorFrameCodec

Stateless decoder for IEEE C37.118.2-2011 CONFIG-2 and DATA frames.

The codec performs no network I/O: each method accepts the raw bytes of a single frame, validates its SYNC word, declared size, and CRC-CCITT checksum, and returns a fully decoded, immutable frame object. Any structural fault raises a :class:SynchrophasorFrameError subclass; the codec never returns partially decoded data.

Methods:

decode_config2

decode_config2(frame: bytes) -> ConfigurationFrame2

Decode a CONFIG-2 frame into its per-PMU measurement layout.

Parameters

frame : bytes The complete CONFIG-2 frame, including SYNC and trailing CRC.

Returns

ConfigurationFrame2 The decoded configuration.

Raises

SynchrophasorFrameError If the frame is truncated, has the wrong SYNC/type, or fails CRC.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

def decode_config2(self, frame: bytes) -> ConfigurationFrame2:
    """Decode a CONFIG-2 frame into its per-PMU measurement layout.

    Parameters
    ----------
    frame : bytes
        The complete CONFIG-2 frame, including SYNC and trailing CRC.

    Returns
    -------
    ConfigurationFrame2
        The decoded configuration.

    Raises
    ------
    SynchrophasorFrameError
        If the frame is truncated, has the wrong SYNC/type, or fails CRC.
    """
    header = self._validate_common(frame, expected_type=FRAME_TYPE_CONFIG2)
    reader = _FrameReader(frame, start=_HEADER_SIZE, end=len(frame) - _CRC_SIZE)
    time_base = reader.u32() & 0x00FFFFFF
    pmu_count = reader.u16()
    pmus = tuple(self._decode_pmu_config(reader) for _ in range(pmu_count))
    data_rate = reader.i16()
    return ConfigurationFrame2(
        header=header,
        time_base=time_base,
        pmus=pmus,
        data_rate=data_rate,
    )

decode_data

decode_data(
    frame: bytes, config: ConfigurationFrame2
) -> DataFrame

Decode a DATA frame using a previously decoded CONFIG-2 layout.

Parameters

frame : bytes The complete DATA frame, including SYNC and trailing CRC. config : ConfigurationFrame2 The configuration describing each PMU's measurement layout.

Returns

DataFrame The decoded measurements, one block per configured PMU.

Raises

SynchrophasorFrameError If the frame is truncated, has the wrong SYNC/type, or fails CRC.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

def decode_data(self, frame: bytes, config: ConfigurationFrame2) -> DataFrame:
    """Decode a DATA frame using a previously decoded CONFIG-2 layout.

    Parameters
    ----------
    frame : bytes
        The complete DATA frame, including SYNC and trailing CRC.
    config : ConfigurationFrame2
        The configuration describing each PMU's measurement layout.

    Returns
    -------
    DataFrame
        The decoded measurements, one block per configured PMU.

    Raises
    ------
    SynchrophasorFrameError
        If the frame is truncated, has the wrong SYNC/type, or fails CRC.
    """
    header = self._validate_common(frame, expected_type=FRAME_TYPE_DATA)
    reader = _FrameReader(frame, start=_HEADER_SIZE, end=len(frame) - _CRC_SIZE)
    measurements = tuple(
        self._decode_pmu_measurement(reader, pmu) for pmu in config.pmus
    )
    return DataFrame(header=header, measurements=measurements)

Functions:

compute_crc_ccitt

compute_crc_ccitt(data: bytes) -> int

Compute the IEEE C37.118.2 CRC-CCITT checksum of a byte string.

The checksum uses the generating polynomial 0x1021 (X^16 + X^12 + X^5 + 1), an initial register value of 0xFFFF, and no final mask, processing each byte most-significant-bit first. This matches the checksum both reference implementations apply over every frame byte except the trailing two CRC bytes.

Parameters

data : bytes The bytes to checksum (a full frame excluding its trailing CRC field).

Returns

int The 16-bit CRC-CCITT value.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

def compute_crc_ccitt(data: bytes) -> int:
    """Compute the IEEE C37.118.2 CRC-CCITT checksum of a byte string.

    The checksum uses the generating polynomial ``0x1021``
    (``X^16 + X^12 + X^5 + 1``), an initial register value of ``0xFFFF``, and no
    final mask, processing each byte most-significant-bit first. This matches the
    checksum both reference implementations apply over every frame byte except
    the trailing two CRC bytes.

    Parameters
    ----------
    data : bytes
        The bytes to checksum (a full frame excluding its trailing CRC field).

    Returns
    -------
    int
        The 16-bit CRC-CCITT value.
    """
    crc = _CRC_INIT
    for byte in data:
        crc ^= byte << 8
        for _ in range(8):
            crc = ((crc << 1) ^ _CRC_POLY if crc & 0x8000 else crc << 1) & 0xFFFF
    return crc

data_frames_to_frequency_series

data_frames_to_frequency_series(
    config: ConfigurationFrame2,
    frames: tuple[DataFrame, ...],
    *,
    pmu_index: int = 0,
) -> tuple[tuple[float, ...], tuple[float, ...]]

Assemble a (time_s, frequency_hz) series for one PMU across frames.

The time vector is relative to the first frame, combining the second-of- century count and the fractional-second offset resolved against the configuration's TIME_BASE; the frequency vector reports each frame's absolute frequency for the selected PMU. The result mirrors the two-column time_s,frequency_hz layout consumed by the PMU ringdown screener, so a decoded synchrophasor stream feeds directly into ringdown evidence.

Parameters

config : ConfigurationFrame2 The configuration whose TIME_BASE and PMU order the frames follow. frames : tuple[DataFrame, ...] The DATA frames in acquisition order. pmu_index : int, optional Index of the PMU whose frequency series is extracted (default 0).

Returns

tuple[tuple[float, ...], tuple[float, ...]] The relative-time vector in seconds and the frequency vector in hertz.

Raises

SynchrophasorFrameError If frames is empty or pmu_index is out of range for a frame.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_c37118.py

def data_frames_to_frequency_series(
    config: ConfigurationFrame2,
    frames: tuple[DataFrame, ...],
    *,
    pmu_index: int = 0,
) -> tuple[tuple[float, ...], tuple[float, ...]]:
    """Assemble a ``(time_s, frequency_hz)`` series for one PMU across frames.

    The time vector is relative to the first frame, combining the second-of-
    century count and the fractional-second offset resolved against the
    configuration's ``TIME_BASE``; the frequency vector reports each frame's
    absolute frequency for the selected PMU. The result mirrors the two-column
    ``time_s,frequency_hz`` layout consumed by the PMU ringdown screener, so a
    decoded synchrophasor stream feeds directly into ringdown evidence.

    Parameters
    ----------
    config : ConfigurationFrame2
        The configuration whose ``TIME_BASE`` and PMU order the frames follow.
    frames : tuple[DataFrame, ...]
        The DATA frames in acquisition order.
    pmu_index : int, optional
        Index of the PMU whose frequency series is extracted (default ``0``).

    Returns
    -------
    tuple[tuple[float, ...], tuple[float, ...]]
        The relative-time vector in seconds and the frequency vector in hertz.

    Raises
    ------
    SynchrophasorFrameError
        If ``frames`` is empty or ``pmu_index`` is out of range for a frame.
    """
    if not frames:
        raise SynchrophasorFrameError("at least one DATA frame is required")
    if not 0 <= pmu_index < len(config.pmus):
        raise SynchrophasorFrameError(
            f"pmu_index {pmu_index} out of range for {len(config.pmus)} PMUs"
        )
    times: list[float] = []
    frequencies: list[float] = []
    first = frames[0].header
    base_seconds = first.soc + first.seconds_of_second(config.time_base)
    for frame in frames:
        if pmu_index >= len(frame.measurements):
            raise SynchrophasorFrameError(
                f"pmu_index {pmu_index} out of range for a frame with "
                f"{len(frame.measurements)} measurements"
            )
        absolute = frame.header.soc + frame.header.seconds_of_second(config.time_base)
        times.append(absolute - base_seconds)
        frequencies.append(frame.measurements[pmu_index].frequency_hz)
    return tuple(times), tuple(frequencies)

IEEE C37.118.2 Phase Bridge

Review-only bridge mapping decoded PMU phasors to oscillator PhaseStates. A PMU phasor is already phase-resolved, so — unlike the OPC-UA/MQTT waveform bridges — C37118PhaseBridge reads the phase directly instead of running a waveform extractor: theta is the phasor angle (rectangular atan2(imag, real), which is scale-independent, or a floating-point polar angle in radians), omega is 2*pi times the frame's measured frequency, amplitude is the phasor magnitude in engineering units (integer components scaled by the PHUNIT 10**-5 V/A factor), and quality derives from the STAT data-error and time-sync bits. Integer polar phasors raise rather than emit a fabricated angle (the standard and the reference implementations disagree on the integer polar angle scale). The bridge never actuates (non_actuating / execution_disabled).

from scpn_phase_orchestrator.adapters import (
    C37118PhaseBridge,
    PhasorBinding,
    SynchrophasorFrameCodec,
)

codec = SynchrophasorFrameCodec()
config = codec.decode_config2(config2_bytes)
frames = [codec.decode_data(data_bytes, config) for data_bytes in stream]
bridge = C37118PhaseBridge.from_bindings([PhasorBinding("bus1_va", phasor_index=0)])
phases = bridge.extract_phases(config, frames)

synchrophasor_phase_bridge

Map decoded IEEE C37.118.2 PMU phasor measurements to oscillator phase states.

A phasor measurement unit already reports a phase-resolved quantity: each voltage or current phasor carries a magnitude and an angle, and the frame carries the measured line frequency. Unlike the scalar SCADA tags of the OPC-UA bridge — a raw waveform from which a phase must be extracted — a PMU phasor's angle is the instantaneous phase, so this bridge reads it directly rather than running a Hilbert or zero-crossing extractor:

• theta is the phasor angle, canonicalised to [0, 2*pi). For a rectangular phasor it is atan2(imag, real) (scale-independent, so the PHUNIT conversion factor never enters the angle); for a floating-point polar phasor it is the reported angle in radians.
• omega is the instantaneous angular frequency, 2*pi times the frame's absolute measured frequency in hertz.
• amplitude is the phasor magnitude in engineering units — integer components are scaled by the PHUNIT 10**-5 V/A-per-bit factor; float components are already in engineering units.
• quality is derived only from the STAT word's verified data-error field (bits 15-14) and time-sync bit (bit 13).

Integer polar phasors are an honest boundary: the standard scales an integer polar angle differently from the magnitude, and the open-source references disagree on that scaling, so rather than guess an angle unit this bridge raises for integer polar phasors instead of emitting a fabricated angle. The bridge is review-only: it produces phase states for observation and never actuates.

Classes

PhasorBinding dataclass

PhasorBinding(
    oscillator: str,
    pmu_index: int = 0,
    phasor_index: int = 0,
)

Bind one PMU phasor channel to an SPO oscillator.

Attributes

oscillator : str Oscillator name the phasor's phase state is bound to. pmu_index : int Index of the PMU within the configuration/data frame (default 0). phasor_index : int Index of the phasor channel within the PMU block (default 0).

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the binding.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the binding fields.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_phase_bridge.py

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

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the binding fields.
    """
    return {
        "oscillator": self.oscillator,
        "pmu_index": self.pmu_index,
        "phasor_index": self.phasor_index,
    }

C37118PhaseBridge dataclass

C37118PhaseBridge(bindings: tuple[PhasorBinding, ...])

Review-only bridge from decoded PMU phasors to oscillator phase states.

Attributes

bindings : tuple[PhasorBinding, ...] The phasor-to-oscillator bindings; oscillator names must be unique. non_actuating : bool Always True — the bridge observes and never drives hardware. execution_disabled : bool Always True — no control action is emitted from this bridge.

Methods:

from_bindings classmethod

from_bindings(
    bindings: Sequence[PhasorBinding],
) -> C37118PhaseBridge

Build a bridge from a sequence of phasor bindings.

Parameters

bindings : Sequence[PhasorBinding] The phasor-to-oscillator bindings.

Returns

C37118PhaseBridge A configured, review-only bridge.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_phase_bridge.py

@classmethod
def from_bindings(cls, bindings: Sequence[PhasorBinding]) -> C37118PhaseBridge:
    """Build a bridge from a sequence of phasor bindings.

    Parameters
    ----------
    bindings : Sequence[PhasorBinding]
        The phasor-to-oscillator bindings.

    Returns
    -------
    C37118PhaseBridge
        A configured, review-only bridge.
    """
    return cls(bindings=tuple(bindings))

extract_phases

extract_phases(
    config: ConfigurationFrame2, frames: Sequence[DataFrame]
) -> dict[str, PhaseState]

Map the most recent frame's phasors to per-oscillator phase states.

Parameters

config : ConfigurationFrame2 The configuration describing each PMU's measurement layout. frames : Sequence[DataFrame] The decoded DATA frames; the last frame provides the current state.

Returns

dict[str, PhaseState] The latest instantaneous phase state per bound oscillator.

Raises

ValueError If frames is empty, a binding's PMU index is out of range, or a bound phasor cannot be interpreted (integer polar angle).

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_phase_bridge.py

def extract_phases(
    self,
    config: ConfigurationFrame2,
    frames: Sequence[DataFrame],
) -> dict[str, PhaseState]:
    """Map the most recent frame's phasors to per-oscillator phase states.

    Parameters
    ----------
    config : ConfigurationFrame2
        The configuration describing each PMU's measurement layout.
    frames : Sequence[DataFrame]
        The decoded DATA frames; the last frame provides the current state.

    Returns
    -------
    dict[str, PhaseState]
        The latest instantaneous phase state per bound oscillator.

    Raises
    ------
    ValueError
        If ``frames`` is empty, a binding's PMU index is out of range, or a
        bound phasor cannot be interpreted (integer polar angle).
    """
    if not frames:
        raise ValueError("at least one DATA frame is required")
    latest = frames[-1]
    phases: dict[str, PhaseState] = {}
    for binding in self.bindings:
        if not 0 <= binding.pmu_index < len(config.pmus):
            raise ValueError(
                f"pmu_index {binding.pmu_index} out of range for "
                f"{len(config.pmus)} PMUs"
            )
        if binding.pmu_index >= len(latest.measurements):
            raise ValueError(
                f"pmu_index {binding.pmu_index} out of range for a frame "
                f"with {len(latest.measurements)} measurements"
            )
        pmu = config.pmus[binding.pmu_index]
        measurement = latest.measurements[binding.pmu_index]
        theta, amplitude = _phasor_theta_amplitude(
            pmu, measurement, binding.phasor_index
        )
        phases[binding.oscillator] = PhaseState(
            theta=theta,
            omega=_TWO_PI * measurement.frequency_hz,
            amplitude=amplitude,
            quality=_stat_quality(measurement.stat),
            channel="P",
            node_id=binding.oscillator,
        )
    return phases

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the bridge.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the bridge configuration and its review-only posture.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_phase_bridge.py

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

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the bridge configuration and
        its review-only posture.
    """
    return {
        "bindings": [binding.to_audit_record() for binding in self.bindings],
        "non_actuating": self.non_actuating,
        "execution_disabled": self.execution_disabled,
    }

Functions:

IEEE C37.118.2 Live Session Client

C37118SessionClient reads synchrophasor frames from a PDC/PMU over a TCP stream using only the standard library's asyncio (no third-party dependency). It issues the standard C37.118.2 command frames — request CONFIG-2, turn data on, turn data off — which are a benign protocol handshake that controls only the measurement data stream; the client writes no device setpoints and cannot actuate grid equipment (non_actuating). build_command_frame constructs a CRC-sealed COMMAND frame and read_frame reassembles one frame from the stream via its SYNC/FRAMESIZE prefix. The command-word values were verified at source against the pypmu CommandFrame table and the Wireshark synchrophasor dissector (which cites the standard's Table 15). Decoding is delegated to SynchrophasorFrameCodec.

from scpn_phase_orchestrator.adapters import C37118SessionClient

client = C37118SessionClient(id_code=7)
reader, writer = await client.open_connection("pdc.local", 4712)
try:
    config = await client.request_configuration(reader, writer)
    frames = await client.collect_data_frames(reader, writer, config, count=30)
finally:
    writer.close()

synchrophasor_client

Live asynchronous IEEE C37.118.2 synchrophasor session client.

Reads synchrophasor frames from a phasor data concentrator (PDC) or PMU over a TCP stream using only the standard library's :mod:asyncio — no third-party dependency. The client issues the standard C37.118.2 command frames to drive the data stream: it requests the CONFIG-2 frame, turns data transmission on, reads the requested number of DATA frames, and turns transmission off again. These command frames are a benign protocol handshake that controls only the measurement data stream; the client never writes device setpoints and cannot actuate grid equipment (non_actuating).

The command-word values were verified at source against two independent references: the iicsys/pypmu CommandFrame table and the Wireshark synchrophasor dissector (epan/dissectors/packet-synphasor.c, which cites the standard's Table 15): 0x0001 data-off, 0x0002 data-on, 0x0003 send HDR, 0x0004 send CONFIG-1, 0x0005 send CONFIG-2, 0x0006 send CONFIG-3. A command frame is the 14-byte common header plus a 2-byte command word plus the CRC (18 bytes total). Frames are reassembled from the stream using the SYNC/FRAMESIZE prefix, and decoding is delegated to :class:~scpn_phase_orchestrator.adapters.synchrophasor_c37118.SynchrophasorFrameCodec.

Classes

C37118SessionClient dataclass

C37118SessionClient(id_code: int)

Review-only async client that drives a C37.118.2 measurement stream.

Attributes

id_code : int The destination data-stream identification code (0..65535). non_actuating : bool Always True — the client issues only stream-control command frames and never writes device setpoints.

Methods:

request_configuration async

request_configuration(
    reader: StreamReader, writer: StreamWriter
) -> ConfigurationFrame2

Request and decode the CONFIG-2 frame from the stream.

Parameters

reader : asyncio.StreamReader The stream to read frames from. writer : asyncio.StreamWriter The stream to write the command frame to.

Returns

ConfigurationFrame2 The decoded configuration.

Raises

FrameTruncationError If the stream ends before a CONFIG-2 frame arrives.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_client.py

async def request_configuration(
    self,
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
) -> ConfigurationFrame2:
    """Request and decode the CONFIG-2 frame from the stream.

    Parameters
    ----------
    reader : asyncio.StreamReader
        The stream to read frames from.
    writer : asyncio.StreamWriter
        The stream to write the command frame to.

    Returns
    -------
    ConfigurationFrame2
        The decoded configuration.

    Raises
    ------
    FrameTruncationError
        If the stream ends before a CONFIG-2 frame arrives.
    """
    await self._send(writer, COMMAND_SEND_CONFIG2)
    while True:
        frame = await read_frame(reader)
        if _frame_type(frame) == FRAME_TYPE_CONFIG2:
            return self._codec.decode_config2(frame)

collect_data_frames async

collect_data_frames(
    reader: StreamReader,
    writer: StreamWriter,
    config: ConfigurationFrame2,
    *,
    count: int,
) -> list[DataFrame]

Turn data on, collect count DATA frames, then turn data off.

Parameters

reader : asyncio.StreamReader The stream to read frames from. writer : asyncio.StreamWriter The stream to write command frames to. config : ConfigurationFrame2 The configuration used to decode DATA frames. count : int Number of DATA frames to collect; must be a positive integer.

Returns

list[DataFrame] The decoded DATA frames in arrival order.

Raises

ValueError If count is not a positive integer. FrameTruncationError If the stream ends before count DATA frames arrive.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_client.py

async def collect_data_frames(
    self,
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
    config: ConfigurationFrame2,
    *,
    count: int,
) -> list[DataFrame]:
    """Turn data on, collect ``count`` DATA frames, then turn data off.

    Parameters
    ----------
    reader : asyncio.StreamReader
        The stream to read frames from.
    writer : asyncio.StreamWriter
        The stream to write command frames to.
    config : ConfigurationFrame2
        The configuration used to decode DATA frames.
    count : int
        Number of DATA frames to collect; must be a positive integer.

    Returns
    -------
    list[DataFrame]
        The decoded DATA frames in arrival order.

    Raises
    ------
    ValueError
        If ``count`` is not a positive integer.
    FrameTruncationError
        If the stream ends before ``count`` DATA frames arrive.
    """
    if isinstance(count, bool) or not isinstance(count, int) or count <= 0:
        raise ValueError("count must be a positive integer")
    await self._send(writer, COMMAND_DATA_ON)
    frames: list[DataFrame] = []
    try:
        while len(frames) < count:
            frame = await read_frame(reader)
            if _frame_type(frame) == FRAME_TYPE_DATA:
                frames.append(self._codec.decode_data(frame, config))
    finally:
        await self._send(writer, COMMAND_DATA_OFF)
    return frames

open_connection async

open_connection(
    host: str, port: int
) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]

Open a TCP connection to a PDC/PMU endpoint.

Parameters

host : str Hostname or address of the concentrator/PMU. port : int TCP port of the C37.118.2 data stream.

Returns

tuple[asyncio.StreamReader, asyncio.StreamWriter] The connected stream reader and writer.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_client.py

async def open_connection(
    self, host: str, port: int
) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
    """Open a TCP connection to a PDC/PMU endpoint.

    Parameters
    ----------
    host : str
        Hostname or address of the concentrator/PMU.
    port : int
        TCP port of the C37.118.2 data stream.

    Returns
    -------
    tuple[asyncio.StreamReader, asyncio.StreamWriter]
        The connected stream reader and writer.
    """
    return await asyncio.open_connection(host, port)

Functions:

build_command_frame

build_command_frame(
    id_code: int,
    command: int,
    *,
    soc: int = 0,
    fracsec: int = 0,
    version: int = 1,
) -> bytes

Build a CRC-sealed IEEE C37.118.2 COMMAND frame.

Parameters

id_code : int Destination data-stream identification code (0..65535). command : int Command word; one of the COMMAND_* constants. soc : int Second-of-century timestamp for the command (default 0). fracsec : int Fraction-of-second word for the command (default 0). version : int Protocol version number placed in the SYNC word (default 1).

Returns

bytes The complete COMMAND frame including SYNC and trailing CRC.

Raises

ValueError If id_code is out of range or command is not a known command.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_client.py

def build_command_frame(
    id_code: int,
    command: int,
    *,
    soc: int = 0,
    fracsec: int = 0,
    version: int = 1,
) -> bytes:
    """Build a CRC-sealed IEEE C37.118.2 COMMAND frame.

    Parameters
    ----------
    id_code : int
        Destination data-stream identification code (0..65535).
    command : int
        Command word; one of the ``COMMAND_*`` constants.
    soc : int
        Second-of-century timestamp for the command (default ``0``).
    fracsec : int
        Fraction-of-second word for the command (default ``0``).
    version : int
        Protocol version number placed in the SYNC word (default ``1``).

    Returns
    -------
    bytes
        The complete COMMAND frame including SYNC and trailing CRC.

    Raises
    ------
    ValueError
        If ``id_code`` is out of range or ``command`` is not a known command.
    """
    if not 0 <= id_code <= _UINT16_MAX:
        raise ValueError("id_code must be in the range 0..65535")
    if command not in _VALID_COMMANDS:
        raise ValueError(f"unknown command word 0x{command:04X}")
    framesize = _HEADER_SIZE + 2 + _CRC_SIZE
    header = (
        bytes([_SYNC_LEAD, (_FRAME_TYPE_COMMAND << 4) | (version & 0x0F)])
        + struct.pack(">H", framesize)
        + struct.pack(">H", id_code)
        + struct.pack(">I", soc)
        + struct.pack(">I", fracsec)
    )
    frame = header + struct.pack(">H", command)
    return frame + struct.pack(">H", compute_crc_ccitt(frame))

read_frame async

read_frame(reader: StreamReader) -> bytes

Read one complete synchrophasor frame from an async stream.

The frame is reassembled by reading the 4-byte SYNC/FRAMESIZE prefix, validating the SYNC lead, and reading exactly FRAMESIZE bytes in total.

Parameters

reader : asyncio.StreamReader The stream to read from.

Returns

bytes The complete frame, including SYNC and trailing CRC.

Raises

FrameTruncationError If the stream ends before a full frame, or the declared frame size is smaller than the minimum header-plus-CRC length. UnsupportedFrameError If the frame does not begin with the SYNC lead byte 0xAA.

Source code in src/scpn_phase_orchestrator/adapters/synchrophasor_client.py

async def read_frame(reader: asyncio.StreamReader) -> bytes:
    """Read one complete synchrophasor frame from an async stream.

    The frame is reassembled by reading the 4-byte SYNC/FRAMESIZE prefix,
    validating the SYNC lead, and reading exactly ``FRAMESIZE`` bytes in total.

    Parameters
    ----------
    reader : asyncio.StreamReader
        The stream to read from.

    Returns
    -------
    bytes
        The complete frame, including SYNC and trailing CRC.

    Raises
    ------
    FrameTruncationError
        If the stream ends before a full frame, or the declared frame size is
        smaller than the minimum header-plus-CRC length.
    UnsupportedFrameError
        If the frame does not begin with the SYNC lead byte ``0xAA``.
    """
    try:
        prefix = await reader.readexactly(_PREFIX_SIZE)
    except asyncio.IncompleteReadError as exc:
        raise FrameTruncationError(
            f"stream ended after {len(exc.partial)} bytes before a frame prefix"
        ) from exc
    if prefix[0] != _SYNC_LEAD:
        raise UnsupportedFrameError(
            f"frame does not begin with SYNC lead 0xAA (got 0x{prefix[0]:02X})"
        )
    framesize = struct.unpack(">H", prefix[2:4])[0]
    if framesize < _HEADER_SIZE + _CRC_SIZE:
        raise FrameTruncationError(
            f"declared framesize {framesize} is below the minimum "
            f"{_HEADER_SIZE + _CRC_SIZE}"
        )
    try:
        rest = await reader.readexactly(framesize - _PREFIX_SIZE)
    except asyncio.IncompleteReadError as exc:
        raise FrameTruncationError(
            f"stream ended after {len(exc.partial)} of {framesize - _PREFIX_SIZE} "
            "body bytes"
        ) from exc
    return prefix + rest

Hardware I/O

Generic hardware I/O abstraction for digital/analogue outputs.

Hardware I/O sample buffers and simulated-board frequency configuration accept only finite real sensor amplitudes and finite positive real frequencies; boolean and complex aliases are rejected before buffering or synthetic EEG generation so flags and phasors cannot enter real sensor channels.

hardware_io

Real-time hardware I/O via BrainFlow (EEG, PPG, EMG) and SCADA/Modbus.

BrainFlow supports: OpenBCI, Muse, Emotiv, NeuroSky, BrainBit, Enobio, and simulated boards for development. Install: pip install brainflow

SCADA: Modbus TCP via pymodbus. Install: pip install pymodbus

Classes

SampleBuffer dataclass

SampleBuffer(capacity: int, n_channels: int)

Ring buffer for streaming sensor data.

Methods:

push

push(samples: FloatArray) -> None

Push (n_channels, n_samples) into the ring buffer.

Parameters

samples : FloatArray Sample block, shape (n_channels, n_samples).

Raises

ValueError If the sample block shape is invalid.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def push(self, samples: FloatArray) -> None:
    """Push (n_channels, n_samples) into the ring buffer.

    Parameters
    ----------
    samples : FloatArray
        Sample block, shape ``(n_channels, n_samples)``.

    Raises
    ------
    ValueError
        If the sample block shape is invalid.
    """
    if samples.ndim != 2:
        raise ValueError(
            "samples must be a 2D array shaped (n_channels, n_samples)"
        )
    if samples.shape[0] != self.n_channels:
        raise ValueError(
            f"samples first dimension must match n_channels={self.n_channels}"
        )
    if _has_non_real_numeric_alias(samples):
        raise ValueError("samples must contain real numeric values")
    if not np.issubdtype(samples.dtype, np.number):
        raise ValueError("samples must be numeric")
    if not np.all(np.isfinite(samples)):
        raise ValueError("samples must contain only finite values")
    n_samples = samples.shape[1]
    for i in range(n_samples):
        self.buffer[:, self.write_idx % self.capacity] = samples[:, i]
        self.write_idx += 1
    self.count = min(self.count + n_samples, self.capacity)

get_recent

get_recent(n: int) -> FloatArray

Get the last n samples as (n_channels, n).

Parameters

n : int Number of most-recent samples to return.

Returns

FloatArray The most recent n samples, shape (n_channels, n).

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def get_recent(self, n: int) -> FloatArray:
    """Get the last n samples as (n_channels, n).

    Parameters
    ----------
    n : int
        Number of most-recent samples to return.

    Returns
    -------
    FloatArray
        The most recent ``n`` samples, shape ``(n_channels, n)``.
    """
    n = _positive_int(n, field="n")
    n = min(n, self.count)
    if n == 0:
        return np.zeros((self.n_channels, 0))
    end = self.write_idx % self.capacity
    if end >= n:
        return self.buffer[:, end - n : end]
    return np.concatenate(
        [self.buffer[:, self.capacity - (n - end) :], self.buffer[:, :end]],
        axis=1,
    )

BrainFlowAdapter

BrainFlowAdapter(
    board_id: int = 0,
    serial_port: str = "",
    buffer_size: int = 4096,
)

Streams EEG/PPG/EMG data from BrainFlow-supported devices.

Usage

adapter = BrainFlowAdapter(board_id=BoardIds.SYNTHETIC_BOARD) adapter.start() signal = adapter.get_channel_data(0, n_samples=256) adapter.stop()

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def __init__(
    self,
    board_id: int = 0,
    serial_port: str = "",
    buffer_size: int = 4096,
) -> None:
    if not HAS_BRAINFLOW:
        msg = "brainflow not installed. pip install brainflow"
        raise ImportError(msg)
    params = BrainFlowInputParams()
    if serial_port:
        params.serial_port = serial_port
    self._board = BoardShim(board_id, params)
    self._board_id = board_id
    self._eeg_channels = BoardShim.get_eeg_channels(board_id)
    self._sample_rate = BoardShim.get_sampling_rate(board_id)
    self._buffer_size = buffer_size
    self._running = False

Attributes

sample_rate property

sample_rate: int

Board sampling rate in Hz.

Returns

int Board sampling rate in Hz.

eeg_channels property

eeg_channels: list[int]

BrainFlow EEG channel indices for this board.

Returns

list[int] BrainFlow EEG channel indices for this board.

n_channels property

n_channels: int

Number of EEG channels.

Returns

int Number of EEG channels.

Methods:

start

start() -> None

Prepare and start the BrainFlow data stream.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def start(self) -> None:
    """Prepare and start the BrainFlow data stream."""
    self._board.prepare_session()
    self._board.start_stream(self._buffer_size)
    self._running = True

stop

stop() -> None

Stop the stream and release the board session.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def stop(self) -> None:
    """Stop the stream and release the board session."""
    if self._running:
        self._board.stop_stream()
        self._board.release_session()
        self._running = False

get_channel_data

get_channel_data(
    channel_idx: int, n_samples: int = 256
) -> FloatArray

Get recent samples from one EEG channel.

Parameters

channel_idx : int Index of the channel to read. n_samples : int Number of samples to return.

Returns

FloatArray Recent samples from the channel, shape (n_samples,).

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def get_channel_data(self, channel_idx: int, n_samples: int = 256) -> FloatArray:
    """Get recent samples from one EEG channel.

    Parameters
    ----------
    channel_idx : int
        Index of the channel to read.
    n_samples : int
        Number of samples to return.

    Returns
    -------
    FloatArray
        Recent samples from the channel, shape ``(n_samples,)``.
    """
    data = self._board.get_current_board_data(n_samples)
    ch = self._eeg_channels[channel_idx]
    return np.asarray(data[ch])

get_all_eeg

get_all_eeg(n_samples: int = 256) -> FloatArray

Get (n_eeg_channels, n_samples) of recent EEG data.

Parameters

n_samples : int Number of samples to return.

Returns

FloatArray Recent EEG data, shape (n_eeg_channels, n_samples).

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def get_all_eeg(self, n_samples: int = 256) -> FloatArray:
    """Get (n_eeg_channels, n_samples) of recent EEG data.

    Parameters
    ----------
    n_samples : int
        Number of samples to return.

    Returns
    -------
    FloatArray
        Recent EEG data, shape ``(n_eeg_channels, n_samples)``.
    """
    data = self._board.get_current_board_data(n_samples)
    return np.asarray(data[self._eeg_channels])

SimulatedBoardAdapter

SimulatedBoardAdapter(
    n_channels: int = 8,
    sample_rate: int = 256,
    frequencies: FloatArray | None = None,
)

Generates synthetic sinusoidal signals for development without hardware.

Matches the BrainFlowAdapter interface.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def __init__(
    self,
    n_channels: int = 8,
    sample_rate: int = 256,
    frequencies: FloatArray | None = None,
) -> None:
    self._n_channels = _positive_int(n_channels, field="n_channels")
    self._sample_rate = _positive_int(sample_rate, field="sample_rate")
    self._freqs = _validated_frequencies(frequencies, n_channels=self._n_channels)
    self._t = 0.0
    self._running = False

Attributes

sample_rate property

sample_rate: int

Simulated sampling rate in Hz.

Returns

int Simulated sampling rate in Hz.

n_channels property

n_channels: int

Number of simulated channels.

Returns

int Number of simulated channels.

Methods:

start

start() -> None

Reset time counter and begin generating data.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def start(self) -> None:
    """Reset time counter and begin generating data."""
    self._t = 0.0
    self._running = True

stop

stop() -> None

Mark the simulated board as stopped.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def stop(self) -> None:
    """Mark the simulated board as stopped."""
    self._running = False

get_channel_data

get_channel_data(
    channel_idx: int, n_samples: int = 256
) -> FloatArray

Return synthetic sinusoidal samples for one channel.

Parameters

channel_idx : int Index of the channel to read. n_samples : int Number of samples to return.

Returns

FloatArray Synthetic samples for the channel, shape (n_samples,).

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def get_channel_data(self, channel_idx: int, n_samples: int = 256) -> FloatArray:
    """Return synthetic sinusoidal samples for one channel.

    Parameters
    ----------
    channel_idx : int
        Index of the channel to read.
    n_samples : int
        Number of samples to return.

    Returns
    -------
    FloatArray
        Synthetic samples for the channel, shape ``(n_samples,)``.
    """
    channel_idx = _channel_index(channel_idx, n_channels=self._n_channels)
    n_samples = _positive_int(n_samples, field="n_samples")
    sr = self._sample_rate
    t = np.arange(n_samples) / sr + self._t
    self._t += n_samples / sr
    return np.asarray(np.sin(2.0 * np.pi * self._freqs[channel_idx] * t))

get_all_eeg

get_all_eeg(n_samples: int = 256) -> FloatArray

Return synthetic (n_channels, n_samples) sinusoidal data.

Parameters

n_samples : int Number of samples to return.

Returns

FloatArray Synthetic EEG data, shape (n_channels, n_samples).

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def get_all_eeg(self, n_samples: int = 256) -> FloatArray:
    """Return synthetic (n_channels, n_samples) sinusoidal data.

    Parameters
    ----------
    n_samples : int
        Number of samples to return.

    Returns
    -------
    FloatArray
        Synthetic EEG data, shape ``(n_channels, n_samples)``.
    """
    n_samples = _positive_int(n_samples, field="n_samples")
    sr = self._sample_rate
    t = np.arange(n_samples) / sr + self._t
    self._t += n_samples / sr
    return np.array([np.sin(2.0 * np.pi * f * t) for f in self._freqs])

ModbusAdapter

ModbusAdapter(host: str, port: int = 502)

Reads SCADA/PLC registers via Modbus TCP for industrial control.

Usage

adapter = ModbusAdapter("192.168.1.100", port=502) adapter.connect() values = adapter.read_holding_registers(0, count=10) adapter.disconnect()

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def __init__(self, host: str, port: int = 502) -> None:
    if not HAS_MODBUS:
        msg = "pymodbus not installed. pip install pymodbus"
        raise ImportError(msg)
    host_text = require_non_empty_str(host, field="Modbus host")
    self._host = host_text
    self._port = require_tcp_port(port, field="Modbus port")
    self._client = ModbusTcpClient(host_text, port=self._port)
    self._connected = False

Methods:

connect

connect() -> None

Open Modbus TCP connection.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def connect(self) -> None:
    """Open Modbus TCP connection."""
    self._client.connect()
    self._connected = True

disconnect

disconnect() -> None

Close Modbus TCP connection if open.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def disconnect(self) -> None:
    """Close Modbus TCP connection if open."""
    if self._connected:
        self._client.close()
        self._connected = False

read_holding_registers

read_holding_registers(
    address: int, count: int = 1
) -> FloatArray

Read holding registers, return as float64 array.

Parameters

address : int Modbus register address. count : int Number of registers to read.

Returns

FloatArray The register values as a float64 array.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def read_holding_registers(self, address: int, count: int = 1) -> FloatArray:
    """Read holding registers, return as float64 array.

    Parameters
    ----------
    address : int
        Modbus register address.
    count : int
        Number of registers to read.

    Returns
    -------
    FloatArray
        The register values as a float64 array.
    """
    address = _non_negative_int(address, field="address")
    count = _positive_int(count, field="count")
    result = self._client.read_holding_registers(address, count=count)
    if result.isError():
        return np.zeros(count)
    return np.array(result.registers, dtype=np.float64)

write_register

write_register(address: int, value: int) -> bool

Write a single holding register.

Parameters

address : int Modbus register address. value : int Register value to write.

Returns

bool True when the write succeeds.

Raises

ValueError If the value is out of range.

Source code in src/scpn_phase_orchestrator/adapters/hardware_io.py

def write_register(self, address: int, value: int) -> bool:
    """Write a single holding register.

    Parameters
    ----------
    address : int
        Modbus register address.
    value : int
        Register value to write.

    Returns
    -------
    bool
        ``True`` when the write succeeds.

    Raises
    ------
    ValueError
        If the value is out of range.
    """
    address = _non_negative_int(address, field="address")
    if isinstance(value, bool) or not isinstance(value, int):
        raise ValueError("value must be an integer")
    result = self._client.write_register(address, value)
    return not result.isError()

Functions:

Gaian Mesh Bridge

The implements Layer 12: Distributed Mesh of the SCPN architecture. It provides decentralized inter-node synchronization via stateless UDP heartbeats.

Multiple independent instances of SPO running across different machines can "couple" together. Instead of exchanging raw (N)$ phases, nodes exchange their macroscopic Order Parameters ({global}, \Psi_{global}\(). The bridge integrates the peer fields and translates them into external forcing parameters (\)\zeta$, \(\Psi\)) for the local .

Features

• Stateless UDP Broadcasting: Designed for high-frequency, loss-tolerant mesh topologies.
• Topological Consensus: Enables thousands of independent agents (drones, servers) to synchronize without a central command node.
• Timeout-Aware: Automatically drops stale peers from the mean-field calculation to prevent phantom drag.

Peer and local psi values are finite real phases on the circle; negative finite phases are canonicalised modulo 2*pi before mesh-drive computation.

gaian_mesh_bridge

UDP Gaian mesh bridge for exchanging reduced macroscopic order parameters.

GaianMeshNode validates peer addresses and local order-parameter updates, then uses background UDP loops to broadcast and receive reduced R/psi heartbeats. Mesh drive computation filters stale or malformed peer state and returns an external drive proposal only. The bridge exchanges no raw phases, coupling matrices, credentials, or actuation commands.

Classes

PeerState dataclass

PeerState(
    node_id: str, R: float, psi: float, timestamp: float
)

State received from a peer node in the mesh.

GaianMeshNode

GaianMeshNode(
    node_id: str,
    host: str = "127.0.0.1",
    port: int = 12000,
    peer_addresses: list[tuple[str, int]] | None = None,
    mesh_coupling_strength: float = 1.0,
    heartbeat_interval_s: float = 0.05,
    peer_timeout_s: float = 1.0,
)

Distributed Gaian Mesh (Layer 12) Coupling Bridge.

Allows multiple independent instances of scpn-phase-orchestrator running on different servers/machines to 'couple' together over UDP. They exchange aggregate Order Parameters (R, Psi) acting as a massive, decentralized super-oscillator.

The peers' macroscopic fields are combined into a resultant vector, which is then applied to the local integration engine via the external driver parameters zeta and psi.

Source code in src/scpn_phase_orchestrator/adapters/gaian_mesh_bridge.py

def __init__(
    self,
    node_id: str,
    host: str = "127.0.0.1",
    port: int = 12000,
    peer_addresses: list[tuple[str, int]] | None = None,
    mesh_coupling_strength: float = 1.0,
    heartbeat_interval_s: float = 0.05,
    peer_timeout_s: float = 1.0,
):
    self.node_id = require_non_empty_str(node_id, field="node_id")
    self.host = require_non_empty_str(host, field="host")
    self.port = require_tcp_port(port, field="port")
    self.peer_addresses = _validated_peer_addresses(peer_addresses)
    self.mesh_coupling_strength = _require_finite_real(
        mesh_coupling_strength,
        field="mesh_coupling_strength",
        positive=False,
    )
    self.heartbeat_interval_s = _require_finite_real(
        heartbeat_interval_s,
        field="heartbeat_interval_s",
        positive=True,
    )
    self.peer_timeout_s = _require_finite_real(
        peer_timeout_s,
        field="peer_timeout_s",
        positive=True,
    )

    self._peers: dict[str, PeerState] = {}
    self._running = False
    self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    self._sock.bind((self.host, self.port))

    self._local_R = 0.0
    self._local_psi = 0.0

    self._listen_thread: threading.Thread | None = None
    self._broadcast_thread: threading.Thread | None = None

Methods:

start

start() -> None

Start the mesh networking threads.

Raises

RuntimeError If the mesh networking threads cannot start.

Source code in src/scpn_phase_orchestrator/adapters/gaian_mesh_bridge.py

def start(self) -> None:
    """Start the mesh networking threads.

    Raises
    ------
    RuntimeError
        If the mesh networking threads cannot start.
    """
    if self._running:
        return

    self._running = True
    if self._listen_thread is None:
        self._listen_thread = threading.Thread(
            target=self._listen_loop,
            daemon=True,
        )
        self._broadcast_thread = threading.Thread(
            target=self._broadcast_loop,
            daemon=True,
        )

    listen_thread = self._listen_thread
    broadcast_thread = self._broadcast_thread
    if listen_thread is None or broadcast_thread is None:
        raise RuntimeError("mesh threads were not initialised")

    if listen_thread.is_alive() or broadcast_thread.is_alive():
        return

    listen_thread.start()
    broadcast_thread.start()

stop

stop() -> None

Stop the mesh networking threads.

Source code in src/scpn_phase_orchestrator/adapters/gaian_mesh_bridge.py

def stop(self) -> None:
    """Stop the mesh networking threads."""
    self._running = False
    if self._sock.fileno() != -1:
        with contextlib.suppress(OSError):
            self._sock.close()

    if self._listen_thread is not None and self._listen_thread.is_alive():
        self._listen_thread.join(timeout=1.0)
    if self._broadcast_thread is not None and self._broadcast_thread.is_alive():
        self._broadcast_thread.join(timeout=1.0)

__enter__

__enter__() -> GaianMeshNode

Start networking threads on with GaianMeshNode(...) as node:.

Source code in src/scpn_phase_orchestrator/adapters/gaian_mesh_bridge.py

def __enter__(self) -> GaianMeshNode:
    """Start networking threads on ``with GaianMeshNode(...) as node:``."""
    self.start()
    return self

__exit__

__exit__(exc_type: object, exc: object, tb: object) -> None

Stop networking threads and release the UDP socket on context exit.

Source code in src/scpn_phase_orchestrator/adapters/gaian_mesh_bridge.py

def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
    """Stop networking threads and release the UDP socket on context exit."""
    self.stop()

update_local_state

update_local_state(R: float, psi: float) -> None

Update the local macro state to be broadcasted to peers.

Parameters

R : float Kuramoto order parameter. psi : float Mean phase in radians.

Source code in src/scpn_phase_orchestrator/adapters/gaian_mesh_bridge.py

def update_local_state(self, R: float, psi: float) -> None:
    """Update the local macro state to be broadcasted to peers.

    Parameters
    ----------
    R : float
        Kuramoto order parameter.
    psi : float
        Mean phase in radians.
    """
    self._local_R = _require_unit_interval(R, field="R")
    self._local_psi = _require_phase(psi, field="psi")

compute_mesh_drive

compute_mesh_drive() -> tuple[float, float]

Compute the effective external drive (zeta, psi) from the mesh.

Returns

zeta: The magnitude of the mesh mean field.
psi_target: The phase angle of the mesh mean field.

Source code in src/scpn_phase_orchestrator/adapters/gaian_mesh_bridge.py

def compute_mesh_drive(self) -> tuple[float, float]:
    """Compute the effective external drive (zeta, psi) from the mesh.

    Returns
    -------
        zeta: The magnitude of the mesh mean field.
        psi_target: The phase angle of the mesh mean field.
    """
    now = time.time()

    # Filter out stale peers
    active_peers = [
        p
        for p in self._peers.values()
        if _valid_peer_state(p, now=now, timeout_s=self.peer_timeout_s)
    ]

    if not active_peers:
        return 0.0, 0.0

    # Combine peer order parameters into a resultant complex vector
    z_mesh = 0j
    for p in active_peers:
        z_mesh += p.R * np.exp(1j * p.psi)

    z_mesh /= len(active_peers)

    # Multiply by coupling strength
    zeta = self.mesh_coupling_strength * float(np.abs(z_mesh))
    psi_target = float(np.angle(z_mesh))
    if psi_target < 0:
        psi_target += 2 * np.pi

    return zeta, psi_target

Functions:

LSL BCI Entrainment Bridge

The LSLBCIBridge implements Phase 9: Biological Integration of the SCPN augmentation roadmap. It establishes a real-time feedback loop between human neural oscillations and the phase orchestrator.

By utilizing the Lab Streaming Layer (LSL) protocol, the bridge can ingest live EEG data from a wide range of hardware (OpenBCI, Muse, Neuralink, etc.). It extracts the instantaneous phase of target brainwaves (e.g., Alpha or Gamma rhythms) and provides them as input to the ActiveInferenceAgent for predictive entrainment.

Captured samples must be finite real EEG amplitudes, not boolean aliases, and LSL timestamps must be finite non-negative values before samples enter the Hilbert phase buffer.

Features

• Real-Time Phase Extraction: Uses Hilbert transforms on sliding windows to track neural phase state.
• Hardware Agnostic: Supports any EEG device with an LSL outlet.
• Review-only stimulation targets (research scaffold): can compute proposed auditory/visual stimulation targets from the measured phase, for offline research use only. This is an unvalidated experimental adapter — it makes no clinical claim and must not be used to drive stimulation of a person.

lsl_bci_bridge

Lab Streaming Layer BCI bridge for buffered phase extraction.

The bridge optionally connects to a configured LSL stream, captures one target channel into a bounded background buffer, and extracts the current Hilbert phase from recent samples. Configuration rejects invalid stream names, channels, and buffer durations. When LSL is unavailable or disconnected, it fails without leaking stream identifiers in shared error messages.

Classes

LSLBCIBridge

LSLBCIBridge(
    stream_name: str = "EEG",
    target_channel: int = 0,
    buffer_size_s: float = 2.0,
)

Real-time BCI Entrainment Bridge via Lab Streaming Layer (LSL).

This bridge enables direct human-machine synchronization. It streams raw EEG data from LSL (e.g., from OpenBCI or Muse), extracts the instantaneous phase of target neural oscillations, and provides them to the SCPN orchestrator for closed-loop entrainment.

Attributes

stream_name: Name of the LSL stream to listen to.
target_channel: Index of the EEG channel to use.
sampling_rate: Sampling rate of the EEG stream (Hz).

Source code in src/scpn_phase_orchestrator/adapters/lsl_bci_bridge.py

def __init__(
    self,
    stream_name: str = "EEG",
    target_channel: int = 0,
    buffer_size_s: float = 2.0,
):
    self.stream_name = _validate_stream_name(stream_name)
    self.target_channel = _validate_target_channel(target_channel)
    self.buffer_size_s = _validate_buffer_size_s(buffer_size_s)

    self._running = False
    self._thread_lock = threading.Lock()
    self._inlet: Any = None
    self._data_buffer: list[float] = []
    self._lock = threading.Lock()
    self._thread: threading.Thread | None = None
    self._buffer_len: int = 0

Methods:

connect

connect(timeout: float = 5.0) -> bool

Resolve and connect to the LSL stream.

Parameters

timeout : float Connection timeout in seconds.

Returns

bool True when the LSL stream is resolved and connected.

Source code in src/scpn_phase_orchestrator/adapters/lsl_bci_bridge.py

def connect(self, timeout: float = 5.0) -> bool:
    """Resolve and connect to the LSL stream.

    Parameters
    ----------
    timeout : float
        Connection timeout in seconds.

    Returns
    -------
    bool
        ``True`` when the LSL stream is resolved and connected.
    """
    timeout = _validate_connect_timeout(timeout)

    if not HAS_LSL or pylsl is None:
        return False

    streams = pylsl.resolve_byprop("name", self.stream_name, timeout=timeout)
    if not streams:
        return False

    inlet = pylsl.StreamInlet(streams[0])
    info = inlet.info()
    try:
        nominal_srate = _validate_nominal_srate(info.nominal_srate())
    except ValueError:
        return False
    self._inlet = inlet
    self.sampling_rate = nominal_srate
    self._buffer_len = max(1, int(ceil(self.buffer_size_s * nominal_srate)))

    return True

start

start() -> None

Start the background capture thread.

Raises

RuntimeError If the capture thread cannot start.

Source code in src/scpn_phase_orchestrator/adapters/lsl_bci_bridge.py

def start(self) -> None:
    """Start the background capture thread.

    Raises
    ------
    RuntimeError
        If the capture thread cannot start.
    """
    with self._thread_lock:
        if self._running:
            return

        thread = self._thread
        is_alive = getattr(thread, "is_alive", None)
        if thread is not None and callable(is_alive) and is_alive():
            # Defensively avoid duplicate capture loops.
            self._running = True
            return

        if self._inlet is None:
            try:
                connected = self.connect()
            except ValueError:
                connected = False
        else:
            connected = True

        if not connected:
            # Do not echo the configured stream_name — keep the error
            # message independent of the user-configured identifier so
            # shared logs never broadcast deployment topology.
            raise RuntimeError(
                "Could not connect to configured LSL stream "
                "(check stream configuration)"
            )

        self._running = True
        if self._thread is None:
            self._thread = threading.Thread(target=self._capture_loop, daemon=True)
        self._thread.start()

stop

stop() -> None

Stop capture and disconnect.

Source code in src/scpn_phase_orchestrator/adapters/lsl_bci_bridge.py

def stop(self) -> None:
    """Stop capture and disconnect."""
    with self._thread_lock:
        self._running = False
        thread = self._thread

    if thread and thread.is_alive():
        thread.join(timeout=1.0)

    with self._thread_lock:
        self._inlet = None
        if thread and thread.is_alive():
            return
        self._thread = None

get_instantaneous_phase

get_instantaneous_phase() -> float

Extract the current phase from the buffered signal.

Uses Hilbert transform on the recent buffer window. Returns phase in [0, 2*pi).

Returns

float Extract the current phase from the buffered signal.

Source code in src/scpn_phase_orchestrator/adapters/lsl_bci_bridge.py

def get_instantaneous_phase(self) -> float:
    """Extract the current phase from the buffered signal.

    Uses Hilbert transform on the recent buffer window.
    Returns phase in [0, 2*pi).

    Returns
    -------
    float
        Extract the current phase from the buffered signal.
    """
    with self._lock:
        signal = np.array(
            [sample for sample in self._data_buffer if _is_valid_sample(sample)],
            dtype=float,
        )
        if len(signal) < 32:  # Hilbert needs >=32 samples for stability
            return 0.0

    # Analytic signal via Hilbert transform
    analytic_signal = hilbert(signal)
    # Instantaneous phase is the angle of the complex signal
    # We take the last value as the 'current' phase
    phase = np.angle(analytic_signal[-1])

    return float(phase % (2 * np.pi))

Remanentia Bridge

remanentia_bridge

Bidirectional bridge between SPO coherence monitoring and Remanentia memory.

Direction 1 (SPO -> Remanentia): Agent coherence metrics feed consolidation decisions. High R = agents aligned = consolidate their outputs together. Low R = agents diverged = index separately, flag conflicts.

Direction 2 (Remanentia -> SPO): Memory recall novelty feeds coupling adaptation. Novel recall = agents exploring new ground = boost K. Stale recall = repetitive work = decay K.

Requires: Remanentia API running at http://localhost:8001

Classes

CoherenceMemorySnapshot dataclass

CoherenceMemorySnapshot(
    R_global: float,
    regime: str,
    n_entities: int,
    n_memories: int,
    novelty_score: float,
    consolidation_suggested: bool,
)

Combined coherence + memory state.

RemanentiaBridge

RemanentiaBridge(
    remanentia_url: str = "http://localhost:8002",
    timeout: float = 5.0,
)

Bidirectional SPO <-> Remanentia bridge.

Usage::

bridge = RemanentiaBridge(remanentia_url="http://localhost:8001")

# After each SPO step:
bridge.report_coherence(R=0.85, regime="nominal", agent_phases={...})

# Get memory-informed coupling adjustment:
novelty = bridge.get_novelty_score("What coupling topology works?")
K_boost = novelty * 0.5  # novel = explore more = stronger coupling

# Trigger consolidation when agents are aligned:
if R > 0.8:
    bridge.trigger_consolidation()

Source code in src/scpn_phase_orchestrator/adapters/remanentia_bridge.py

def __init__(
    self,
    remanentia_url: str = "http://localhost:8002",
    timeout: float = 5.0,
):
    self._url = _validated_remanentia_url(remanentia_url)
    self._timeout = _validated_timeout(timeout)
    self._last_R = 0.0
    self._last_regime = "unknown"
    self._last_novelty_score = 0.5
    self._last_entities = 0
    self._last_memories = 0

Methods:

health_check

health_check() -> bool

Check if Remanentia is running.

Returns

bool Check if Remanentia is running.

Raises

Exception Re-raises any non-transport, non-decode error; transport and decode failures are caught and False is returned.

Source code in src/scpn_phase_orchestrator/adapters/remanentia_bridge.py

def health_check(self) -> bool:
    """Check if Remanentia is running.

    Returns
    -------
    bool
        Check if Remanentia is running.

    Raises
    ------
    Exception
        Re-raises any non-transport, non-decode error; transport and decode failures
        are caught and ``False`` is returned.
    """
    try:
        resp = _validated_response_payload(self._get("/health"), name="health")
        return resp.get("status") == "ok"
    except BaseException as exc:
        if not self._is_transport_or_decode_error(exc):
            raise
        logger.warning("remanentia.health_check_failed: %s", type(exc).__name__)
        return False

report_coherence

report_coherence(
    R: float,
    regime: str,
    agent_phases: dict[str, float] | None = None,
) -> None

Report current SPO coherence state to Remanentia.

Remanentia can use this to decide when to consolidate: high R = aligned agents = good time to merge their traces.

Parameters

R : float Kuramoto order parameter. regime : str The current control regime label. agent_phases : dict[str, float] | None Per-agent phase values, or None.

Raises

ValueError If R, regime, or agent_phases is invalid. Exception Re-raises any non-transport, non-decode error; transport and decode failures are caught and logged.

Source code in src/scpn_phase_orchestrator/adapters/remanentia_bridge.py

def report_coherence(
    self,
    R: float,
    regime: str,
    agent_phases: dict[str, float] | None = None,
) -> None:
    """Report current SPO coherence state to Remanentia.

    Remanentia can use this to decide when to consolidate:
    high R = aligned agents = good time to merge their traces.

    Parameters
    ----------
    R : float
        Kuramoto order parameter.
    regime : str
        The current control regime label.
    agent_phases : dict[str, float] | None
        Per-agent phase values, or ``None``.

    Raises
    ------
    ValueError
        If ``R``, ``regime``, or ``agent_phases`` is invalid.
    Exception
        Re-raises any non-transport, non-decode error; transport and decode failures
        are caught and logged.
    """
    R = _validated_unit_interval(R, name="R")
    regime = _validated_label(regime, name="regime")
    _validated_agent_phases(agent_phases)
    self._last_R = R
    self._last_regime = regime
    # Store as a reasoning trace that Remanentia can index
    try:
        self._post(
            "/recall",
            {
                "query": f"SPO coherence report: R={R:.3f} regime={regime}",
                "top_k": 0,
            },
        )
    except BaseException as exc:
        if not self._is_transport_or_decode_error(exc):
            raise
        logger.warning(
            "remanentia.report_coherence_trace_failed: %s",
            type(exc).__name__,
        )

get_novelty_score

get_novelty_score(query: str) -> float

Query Remanentia and estimate novelty from recall results.

If recall returns many relevant memories -> low novelty (known ground). If recall returns few/none -> high novelty (unexplored territory). Novelty feeds SPO coupling: novel = boost K (explore together).

Parameters

query : str PromQL query string.

Returns

float The estimated novelty score for the query.

Raises

ValueError If query is invalid. Exception Re-raises any non-transport, non-decode error; transport and decode failures fall back to the last novelty score.

Source code in src/scpn_phase_orchestrator/adapters/remanentia_bridge.py

def get_novelty_score(self, query: str) -> float:
    """Query Remanentia and estimate novelty from recall results.

    If recall returns many relevant memories -> low novelty (known ground).
    If recall returns few/none -> high novelty (unexplored territory).
    Novelty feeds SPO coupling: novel = boost K (explore together).

    Parameters
    ----------
    query : str
        PromQL query string.

    Returns
    -------
    float
        The estimated novelty score for the query.

    Raises
    ------
    ValueError
        If ``query`` is invalid.
    Exception
        Re-raises any non-transport, non-decode error; transport and decode failures
        fall back to the last novelty score.
    """
    query = _validated_query(query)
    try:
        resp = self._post("/recall", {"query": query, "top_k": 5})
        scores = _validated_recall_scores(resp)
        if not scores:
            self._last_novelty_score = 1.0
            return 1.0  # fully novel — no relevant memories
        # Novelty = 1 - mean relevance score, clamped to [0, 1]. Recall
        # scores may be cosine similarities in [-1, 1]; a negative mean
        # would push 1 - mean above 1 and later violate the unit-interval
        # contract enforced by CoherenceMemorySnapshot.
        novelty = float(min(1.0, max(0.0, 1.0 - float(np.mean(scores)))))
        self._last_novelty_score = novelty
        return novelty
    except BaseException as exc:
        if not self._is_transport_or_decode_error(exc):
            raise
        logger.warning(
            "remanentia.get_novelty_score_failed: %s; using_last=%.6f",
            type(exc).__name__,
            self._last_novelty_score,
        )
        return self._last_novelty_score

get_entity_count

get_entity_count() -> int

Get number of entities in Remanentia's knowledge graph.

Returns

int Get number of entities in Remanentia's knowledge graph.

Raises

Exception Re-raises any non-transport, non-decode error; transport and decode failures fall back to the last entity count.

Source code in src/scpn_phase_orchestrator/adapters/remanentia_bridge.py

def get_entity_count(self) -> int:
    """Get number of entities in Remanentia's knowledge graph.

    Returns
    -------
    int
        Get number of entities in Remanentia's knowledge graph.

    Raises
    ------
    Exception
        Re-raises any non-transport, non-decode error; transport and decode failures
        fall back to the last entity count.
    """
    try:
        resp = self._get("/status")
        entities, memories = _validated_status_payload(resp)
        self._last_entities = entities
        self._last_memories = memories
        return entities
    except BaseException as exc:
        if not self._is_transport_or_decode_error(exc):
            raise
        logger.warning(
            "remanentia.get_entity_count_failed: %s; using_last=%d",
            type(exc).__name__,
            self._last_entities,
        )
        return self._last_entities

trigger_consolidation

trigger_consolidation(force: bool = False) -> bool

Trigger memory consolidation in Remanentia.

Best called when R is high (agents aligned, traces coherent).

Parameters

force : bool Whether to force consolidation regardless of thresholds.

Returns

bool True when consolidation was triggered.

Raises

ValueError If the consolidation request is rejected.

Source code in src/scpn_phase_orchestrator/adapters/remanentia_bridge.py

def trigger_consolidation(self, force: bool = False) -> bool:
    """Trigger memory consolidation in Remanentia.

    Best called when R is high (agents aligned, traces coherent).

    Parameters
    ----------
    force : bool
        Whether to force consolidation regardless of thresholds.

    Returns
    -------
    bool
        ``True`` when consolidation was triggered.

    Raises
    ------
    ValueError
        If the consolidation request is rejected.
    """
    if not isinstance(force, bool):
        raise ValueError("force must be a bool")
    try:
        resp = _validated_response_payload(
            self._post("/consolidate", {"force": force}),
            name="consolidate",
        )
        return resp.get("status") == "ok"
    except BaseException as exc:
        if not self._is_transport_or_decode_error(exc):
            raise
        logger.warning(
            "remanentia.trigger_consolidation_failed: %s",
            type(exc).__name__,
        )
        return False

novelty_to_coupling_delta

novelty_to_coupling_delta(
    queries: list[str], scale: float = 0.5
) -> FloatArray

Convert per-agent novelty scores to coupling adjustment.

Each agent's recent work is queried against Remanentia. Novel agents get coupling boosted (explore together). Redundant agents get coupling decayed (avoid repetition).

Returns (N,) array of per-agent K multipliers.

Parameters

queries : list[str] Per-agent novelty queries. scale : float Scaling factor applied to the coupling adjustment.

Returns

FloatArray The per-agent coupling adjustments.

Raises

ValueError If the queries or scale are invalid.

Source code in src/scpn_phase_orchestrator/adapters/remanentia_bridge.py

def novelty_to_coupling_delta(
    self,
    queries: list[str],
    scale: float = 0.5,
) -> FloatArray:
    """Convert per-agent novelty scores to coupling adjustment.

    Each agent's recent work is queried against Remanentia.
    Novel agents get coupling boosted (explore together).
    Redundant agents get coupling decayed (avoid repetition).

    Returns (N,) array of per-agent K multipliers.

    Parameters
    ----------
    queries : list[str]
        Per-agent novelty queries.
    scale : float
        Scaling factor applied to the coupling adjustment.

    Returns
    -------
    FloatArray
        The per-agent coupling adjustments.

    Raises
    ------
    ValueError
        If the queries or scale are invalid.
    """
    if not isinstance(queries, list):
        raise ValueError("queries must be a list of non-empty strings")
    scale = _validated_positive_real(scale, name="scale")
    deltas = []
    for q in queries:
        novelty = self.get_novelty_score(_validated_query(q))
        deltas.append(1.0 + novelty * scale)
    return np.array(deltas, dtype=np.float64)

snapshot

snapshot() -> CoherenceMemorySnapshot

Return the combined coherence and memory state.

Returns

CoherenceMemorySnapshot Return the combined coherence and memory state.

Raises

Exception Re-raises any non-transport, non-decode error; transport and decode failures fall back to cached memory counts.

Source code in src/scpn_phase_orchestrator/adapters/remanentia_bridge.py

def snapshot(self) -> CoherenceMemorySnapshot:
    """Return the combined coherence and memory state.

    Returns
    -------
    CoherenceMemorySnapshot
        Return the combined coherence and memory state.

    Raises
    ------
    Exception
        Re-raises any non-transport, non-decode error; transport and decode failures
        fall back to cached memory counts.
    """
    n_ent = self.get_entity_count()
    try:
        status = self._get("/status")
        entities, n_memories = _validated_status_payload(status)
        self._last_entities = entities
        self._last_memories = n_memories
    except BaseException as exc:
        if not self._is_transport_or_decode_error(exc):
            raise
        logger.warning(
            "remanentia.snapshot_status_failed: %s; using_last=%d",
            type(exc).__name__,
            self._last_memories,
        )
        n_memories = self._last_memories

    return CoherenceMemorySnapshot(
        R_global=self._last_R,
        regime=self._last_regime,
        n_entities=n_ent,
        n_memories=n_memories,
        novelty_score=self._last_novelty_score,
        consolidation_suggested=self._last_R > 0.8,
    )

Synapse Bridges

The synapse bridges translate phase-channel and coupling data into sibling service contracts while keeping the normal audit path unchanged. The channel bridge treats hub WebSocket frames as untrusted JSON: decoded messages must use finite JSON values and unique object keys before sender, type, or payload fields can affect phase-channel state.

synapse_channel_bridge

Live bridge from SYNAPSE_CHANNEL hub events to SPO phase dynamics.

Connects to the SYNAPSE_CHANNEL WebSocket hub and maps agent activity into oscillator phases for real-time coherence monitoring.

Mapping: - Heartbeat interval → P-channel frequency (regular = coherent) - Task claim/release rate → I-channel frequency (balanced = coherent) - Chat message similarity → S-channel coupling (same topic = coupled)

Usage::

bridge = SynapseChannelBridge(
    hub_uri="ws://localhost:8876",
    agents=["Agent-A", "Agent-B", "Agent-C", "Human"],
)
await bridge.connect()

# In your SPO loop:
phases = bridge.get_phases()
knm = bridge.get_coupling()

Classes

AgentState dataclass

AgentState(
    last_heartbeat: float = 0.0,
    heartbeat_intervals: list[float] = list(),
    task_events: list[float] = list(),
    message_count: int = 0,
    current_task: str | None = None,
    phase_p: float = 0.0,
    phase_i: float = 0.0,
    phase_s: float = 0.0,
)

Tracked state for one agent.

SynapseChannelBridge

SynapseChannelBridge(
    hub_uri: str = "ws://localhost:8876",
    agents: list[str] | None = None,
)

Live bridge from SYNAPSE_CHANNEL to SPO oscillator phases.

Source code in src/scpn_phase_orchestrator/adapters/synapse_channel_bridge.py

def __init__(
    self,
    hub_uri: str = "ws://localhost:8876",
    agents: list[str] | None = None,
):
    self._uri = _validate_hub_uri(hub_uri)
    self._agents = _validate_agents(agents)
    self._agent_idx: dict[str, int] = {
        name: i for i, name in enumerate(self._agents)
    }
    self._states: dict[str, AgentState] = {
        name: AgentState() for name in self._agents
    }
    self._ws: Any = None
    self._running = False
    self._n = len(self._agents)

Attributes

n_oscillators property

n_oscillators: int

Return the number of configured agent oscillators.

Returns

int Return the number of configured agent oscillators.

Methods:

connect async

connect() -> None

Connect to SYNAPSE_CHANNEL hub and start listening.

Raises

ImportError If the SYNAPSE channel client is not installed.

Source code in src/scpn_phase_orchestrator/adapters/synapse_channel_bridge.py

async def connect(self) -> None:
    """Connect to SYNAPSE_CHANNEL hub and start listening.

    Raises
    ------
    ImportError
        If the SYNAPSE channel client is not installed.
    """
    try:
        import websockets
    except ImportError:
        raise ImportError("websockets required: pip install websockets") from None

    self._ws = await websockets.connect(self._uri)
    # Register as observer
    await self._ws.send(
        json.dumps(
            {
                "type": "chat",
                "sender": "SPO-Bridge",
                "target": "all",
                "payload": "SPO coherence bridge connected",
            }
        )
    )
    self._running = True

listen_once async

listen_once() -> None

Process one message from the hub.

Source code in src/scpn_phase_orchestrator/adapters/synapse_channel_bridge.py

async def listen_once(self) -> None:
    """Process one message from the hub."""
    if self._ws is None:
        return
    try:
        raw = await asyncio.wait_for(self._ws.recv(), timeout=1.0)
        msg = _loads_hub_json(raw)
        self._process_message(msg)
    except TimeoutError:
        return  # no message within timeout — normal
    except (json.JSONDecodeError, ValueError):
        logger.warning("synapse.listen_once_invalid_json")
        return
    except (ConnectionError, OSError, RuntimeError) as exc:
        logger.warning(
            "synapse.listen_once_transport_error: %s", type(exc).__name__
        )
        return  # connection error — caller should retry

get_phases

get_phases() -> FloatArray

Extract current oscillator phases from agent heartbeat activity.

The returned phase for each agent is the P-channel phase, which advances once per call at the agent's recent mean heartbeat frequency (a regular heartbeat yields a steady phase advance). The per-agent phase_i (task cadence) and phase_s (topic alignment) channels are tracked on :class:AgentState but are not folded into this one-dimensional phase vector.

Returns

FloatArray The P-channel phase per agent, shape (n_oscillators,).

Source code in src/scpn_phase_orchestrator/adapters/synapse_channel_bridge.py

def get_phases(self) -> FloatArray:
    """Extract current oscillator phases from agent heartbeat activity.

    The returned phase for each agent is the P-channel phase, which
    advances once per call at the agent's recent mean heartbeat frequency
    (a regular heartbeat yields a steady phase advance). The per-agent
    ``phase_i`` (task cadence) and ``phase_s`` (topic alignment) channels
    are tracked on :class:`AgentState` but are not folded into this
    one-dimensional phase vector.

    Returns
    -------
    FloatArray
        The P-channel phase per agent, shape ``(n_oscillators,)``.
    """
    phases = np.zeros(self._n)

    for name, state in self._states.items():
        idx = self._agent_idx[name]

        # P: heartbeat regularity → phase
        if state.heartbeat_intervals:
            mean_interval = float(np.mean(state.heartbeat_intervals[-5:]))
            freq = 1.0 / max(mean_interval, 0.1)
            state.phase_p = (state.phase_p + TWO_PI * freq * 1.0) % TWO_PI
        phases[idx] = state.phase_p

    return phases

get_coupling

get_coupling() -> FloatArray

Compute coupling from shared task context.

Agents working on related tasks couple strongly. Agents with no task decouple.

Returns

FloatArray Compute coupling from shared task context.

Source code in src/scpn_phase_orchestrator/adapters/synapse_channel_bridge.py

def get_coupling(self) -> FloatArray:
    """Compute coupling from shared task context.

    Agents working on related tasks couple strongly.
    Agents with no task decouple.

    Returns
    -------
    FloatArray
        Compute coupling from shared task context.
    """
    knm = np.zeros((self._n, self._n))
    for name_i, state_i in self._states.items():
        for name_j, state_j in self._states.items():
            if name_i == name_j:
                continue
            i = self._agent_idx[name_i]
            j = self._agent_idx[name_j]

            # Both active → couple
            if state_i.current_task and state_j.current_task:
                knm[i, j] = 1.0
            # One idle → weak coupling
            elif state_i.current_task or state_j.current_task:
                knm[i, j] = 0.3

    return knm

get_agent_summary

get_agent_summary() -> dict[str, dict[str, Any]]

Return per-agent summary for display.

Returns

dict[str, dict[str, Any]] Return per-agent summary for display.

Source code in src/scpn_phase_orchestrator/adapters/synapse_channel_bridge.py

def get_agent_summary(self) -> dict[str, dict[str, Any]]:
    """Return per-agent summary for display.

    Returns
    -------
    dict[str, dict[str, Any]]
        Return per-agent summary for display.
    """
    summary = {}
    for name, state in self._states.items():
        summary[name] = {
            "phase": state.phase_p,
            "task": state.current_task,
            "messages": state.message_count,
            "heartbeats": len(state.heartbeat_intervals),
        }
    return summary

close async

close() -> None

Stop the bridge and close the active hub WebSocket connection.

Source code in src/scpn_phase_orchestrator/adapters/synapse_channel_bridge.py

async def close(self) -> None:
    """Stop the bridge and close the active hub WebSocket connection."""
    self._running = False
    if self._ws:
        await self._ws.close()

synapse_coupling_bridge

Bridge sc-neurocore synapse dynamics into the SPO coupling matrix K_nm.

Maps three synapse types to SPO coupling parameters:

1. STDP weight changes → K_nm deltas: spike-timing dependent plasticity modifies pairwise coupling strengths. Potentiation (dW > 0) strengthens K_ij; depression (dW < 0) weakens it.

2. Gap junction conductance → phase coupling: electrical synapses provide direct bidirectional coupling. g_c maps linearly to K_ij (symmetric).

3. Tripartite astrocyte Ca²⁺ → imprint modulation: astrocyte oscillations modulate the imprint memory vector m_k. High Ca²⁺ enhances imprint accumulation; low Ca²⁺ accelerates decay.

Requires: pip install sc-neurocore>=3.13.0

Classes

SynapseSnapshot dataclass

SynapseSnapshot(
    knm_delta: FloatArray,
    gap_coupling: FloatArray,
    astrocyte_modulation: FloatArray,
    mean_weight_change: float,
    mean_conductance: float,
    mean_ca: float,
)

Snapshot of synapse state mapped to SPO parameters.

SynapseCouplingBridge

SynapseCouplingBridge(
    n_oscillators: int,
    stdp_scale: float = 1.0,
    gap_scale: float = 1.0,
    ca_scale: float = 1.0,
)

Map sc-neurocore synapse dynamics to SPO coupling parameters.

Usage::

from sc_neurocore.synapses.triplet_stdp import TripletSTDP
from sc_neurocore.synapses.gap_junction import GapJunction

bridge = SynapseCouplingBridge(n_oscillators=8)

# After each SNN step, feed weight changes
bridge.update_stdp_weights(weight_matrix)
bridge.update_gap_conductances(conductance_matrix)
bridge.update_astrocyte_ca(ca_levels)

# Get SPO coupling delta
snap = bridge.snapshot()
knm_new = knm_base + snap.knm_delta

Source code in src/scpn_phase_orchestrator/adapters/synapse_coupling_bridge.py

def __init__(
    self,
    n_oscillators: int,
    stdp_scale: float = 1.0,
    gap_scale: float = 1.0,
    ca_scale: float = 1.0,
):
    self._n = _validate_n_oscillators(n_oscillators)
    self._stdp_scale = _validate_positive_scale("stdp_scale", stdp_scale)
    self._gap_scale = _validate_positive_scale("gap_scale", gap_scale)
    self._ca_scale = _validate_positive_scale("ca_scale", ca_scale)

    self._stdp_weights: FloatArray = np.zeros(
        (self._n, self._n),
        dtype=np.float64,
    )
    self._prev_weights: FloatArray = np.zeros(
        (self._n, self._n),
        dtype=np.float64,
    )
    self._gap_conductances: FloatArray = np.zeros(
        (self._n, self._n),
        dtype=np.float64,
    )
    self._ca_levels: FloatArray = np.zeros(self._n, dtype=np.float64)

Methods:

update_stdp_weights

update_stdp_weights(weights: FloatArray) -> None

Feed current STDP weight matrix from sc-neurocore.

The bridge computes dW = weights - prev_weights and maps to K_nm deltas.

Parameters

weights : FloatArray STDP weight matrix, shape (N, N).

Source code in src/scpn_phase_orchestrator/adapters/synapse_coupling_bridge.py

def update_stdp_weights(self, weights: FloatArray) -> None:
    """Feed current STDP weight matrix from sc-neurocore.

    The bridge computes dW = weights - prev_weights and maps
    to K_nm deltas.

    Parameters
    ----------
    weights : FloatArray
        STDP weight matrix, shape ``(N, N)``.
    """
    # Validate the incoming weights BEFORE advancing the previous-weight
    # baseline, so a rejected update leaves the dW reference intact rather
    # than silently zeroing the next delta.
    validated = _validate_square_matrix(weights, "weights", self._n)
    self._prev_weights = self._stdp_weights.copy()
    self._stdp_weights = validated

update_gap_conductances

update_gap_conductances(conductances: FloatArray) -> None

Feed gap junction conductance matrix.

Symmetric: g_c(i,j) = g_c(j,i). Maps directly to K_ij.

Parameters

conductances : FloatArray Gap-junction conductance matrix, shape (N, N).

Source code in src/scpn_phase_orchestrator/adapters/synapse_coupling_bridge.py

def update_gap_conductances(self, conductances: FloatArray) -> None:
    """Feed gap junction conductance matrix.

    Symmetric: g_c(i,j) = g_c(j,i). Maps directly to K_ij.

    Parameters
    ----------
    conductances : FloatArray
        Gap-junction conductance matrix, shape ``(N, N)``.
    """
    g = _validate_nonnegative_square_matrix(
        conductances,
        "conductances",
        self._n,
    )
    self._gap_conductances = 0.5 * (g + g.T)
    np.fill_diagonal(self._gap_conductances, 0.0)

update_astrocyte_ca

update_astrocyte_ca(ca_levels: FloatArray) -> None

Feed astrocyte Ca²⁺ concentration per oscillator.

High Ca²⁺ → strong imprint modulation (facilitates learning).

Parameters

ca_levels : FloatArray Per-oscillator astrocyte Ca²⁺ concentrations.

Source code in src/scpn_phase_orchestrator/adapters/synapse_coupling_bridge.py

def update_astrocyte_ca(self, ca_levels: FloatArray) -> None:
    """Feed astrocyte Ca²⁺ concentration per oscillator.

    High Ca²⁺ → strong imprint modulation (facilitates learning).

    Parameters
    ----------
    ca_levels : FloatArray
        Per-oscillator astrocyte Ca²⁺ concentrations.
    """
    self._ca_levels = _validate_nonnegative_vector(
        ca_levels,
        "ca_levels",
        self._n,
    )

snapshot

snapshot() -> SynapseSnapshot

Compute SPO coupling parameters from current synapse state.

Returns

SynapseSnapshot Compute SPO coupling parameters from current synapse state.

Source code in src/scpn_phase_orchestrator/adapters/synapse_coupling_bridge.py

def snapshot(self) -> SynapseSnapshot:
    """Compute SPO coupling parameters from current synapse state.

    Returns
    -------
    SynapseSnapshot
        Compute SPO coupling parameters from current synapse state.
    """
    # STDP → K_nm delta
    dw = self._stdp_weights - self._prev_weights
    knm_delta = dw * self._stdp_scale
    np.fill_diagonal(knm_delta, 0.0)

    # Gap junction → symmetric coupling
    gap_coupling = self._gap_conductances * self._gap_scale

    # Astrocyte Ca²⁺ → imprint modulation vector
    ca_norm = self._ca_levels / max(self._ca_levels.max(), 1e-10)
    astro_mod = ca_norm * self._ca_scale

    return SynapseSnapshot(
        knm_delta=knm_delta,
        gap_coupling=gap_coupling,
        astrocyte_modulation=astro_mod,
        mean_weight_change=float(np.mean(np.abs(dw))),
        mean_conductance=float(np.mean(self._gap_conductances)),
        mean_ca=float(np.mean(self._ca_levels)),
    )

apply_to_knm

apply_to_knm(knm_base: FloatArray) -> FloatArray

Apply all synapse-derived modifications to a base K_nm.

Parameters

knm_base : FloatArray Base coupling matrix to modify, shape (N, N).

Returns

FloatArray The base coupling matrix with synapse-derived modifications.

Source code in src/scpn_phase_orchestrator/adapters/synapse_coupling_bridge.py

def apply_to_knm(self, knm_base: FloatArray) -> FloatArray:
    """Apply all synapse-derived modifications to a base K_nm.

    Parameters
    ----------
    knm_base : FloatArray
        Base coupling matrix to modify, shape ``(N, N)``.

    Returns
    -------
    FloatArray
        The base coupling matrix with synapse-derived modifications.
    """
    validated_knm_base = _validate_square_matrix(knm_base, "knm_base", self._n)
    snap = self.snapshot()
    knm = validated_knm_base + snap.knm_delta + snap.gap_coupling
    result: FloatArray = np.maximum(knm, 0.0)
    np.fill_diagonal(result, 0.0)
    return result

apply_to_imprint

apply_to_imprint(m_k: FloatArray) -> FloatArray

Modulate imprint vector by astrocyte Ca²⁺ levels.

Parameters

m_k : FloatArray Imprint vector, shape (N,).

Returns

FloatArray The imprint vector modulated by astrocyte Ca²⁺.

Source code in src/scpn_phase_orchestrator/adapters/synapse_coupling_bridge.py

def apply_to_imprint(self, m_k: FloatArray) -> FloatArray:
    """Modulate imprint vector by astrocyte Ca²⁺ levels.

    Parameters
    ----------
    m_k : FloatArray
        Imprint vector, shape ``(N,)``.

    Returns
    -------
    FloatArray
        The imprint vector modulated by astrocyte Ca²⁺.
    """
    validated_m_k = _validate_vector(m_k, "m_k", self._n)
    snap = self.snapshot()
    result: FloatArray = validated_m_k * (1.0 + snap.astrocyte_modulation)
    return result

FMI 3.0 Co-Simulation Export

adapters.fmi_cosimulation wraps the Koopman MPC controller as an FMI 3.0 co-simulation slave so a simulation master (Dymola, OpenModelica, FMPy) can drive the SPO controller as a block: set the measured state and set point, call do_step, read back the proposed control. The slave, the modelDescription.xml generator and the .fmu packager are pure NumPy and produce a conformant FMI 3.0 model interface; loading the package inside a third-party FMI tool additionally needs the C-ABI binary shim, which is an optional, separately-installed build step (e.g. the unifmu toolchain) outside this module. The reverse import direction is cosimulate, a co-simulation master that drives the controller slave against a plant supplied as a step callable — an external plant FMU plugs in by wrapping its FMI runtime (e.g. fmpy) as that callable.

fmi_cosimulation

Export the Koopman MPC controller as an FMI 3.0 co-simulation slave.

The Functional Mock-up Interface (FMI 3.0, modelica.org) is the industrial standard for coupling simulation tools. This adapter wraps the condensed Koopman MPC (actuation.koopman_mpc) as an FMI co-simulation slave so a co-simulation master — a power-systems or control bench such as Dymola, OpenModelica or FMPy — can drive the SPO controller as a block: it sets the measured state and the set point, calls do_step, and reads back the proposed control.

The slave, the modelDescription.xml generator and the .fmu packager are pure NumPy and fully exercised in-process by driving the slave the way a master would. They emit a conformant FMI 3.0 model interface; a C-ABI binary shim is not shipped, so loading the package inside a third-party FMI tool needs a Python-backed FMI runtime that reconstructs the model from resources/model.json — the review-only model and its evidence are produced here.

The reverse, import direction is :func:cosimulate: a co-simulation master that drives the controller slave against a plant supplied as a step callable, closing the loop. An external plant FMU plugs in by wrapping its FMI runtime (for example fmpy) as that callable, so no FMI runtime dependency is imposed here either.

References

• Modelica Association 2024, Functional Mock-up Interface Specification 3.0.

Classes

FMIVariable dataclass

FMIVariable(
    name: str,
    value_reference: int,
    causality: str,
    start: float | None = None,
)

A scalar FMI 3.0 Float64 model variable.

Parameters

name : str The variable name, a valid FMI identifier. value_reference : int The handle a co-simulation master uses to get or set the variable. causality : str "input" or "output". start : float | None The required start value for inputs; None for outputs.

CoSimulationSlave

CoSimulationSlave(
    controller: KoopmanMPCController,
    *,
    model_name: str = "scpn_koopman_mpc",
)

An FMI 3.0 co-simulation slave wrapping a Koopman MPC controller.

The slave mirrors the FMI co-simulation lifecycle — set inputs, do_step, get outputs — and computes the control by solving the MPC at each step.

Parameters

controller : KoopmanMPCController The fitted Koopman MPC controller to expose. model_name : str The FMI model name.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def __init__(
    self, controller: KoopmanMPCController, *, model_name: str = "scpn_koopman_mpc"
) -> None:
    self._controller = controller
    self.model_name = model_name
    self.state_dim = int(controller.predictor.state_dim)
    self.input_dim = int(controller.predictor.input_dim)
    self.variables = _model_variables(self.state_dim, self.input_dim)
    self._state = np.zeros(self.state_dim, dtype=np.float64)
    self._reference = np.zeros(self.state_dim, dtype=np.float64)
    self._control = np.zeros(self.input_dim, dtype=np.float64)
    self._previous_input = np.zeros(self.input_dim, dtype=np.float64)
    self._by_reference = {var.value_reference: var for var in self.variables}

Methods:

enter_initialization_mode

enter_initialization_mode() -> None

Reset the slave to its start values for a fresh co-simulation run.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def enter_initialization_mode(self) -> None:
    """Reset the slave to its start values for a fresh co-simulation run."""
    self._state = np.zeros(self.state_dim, dtype=np.float64)
    self._reference = np.zeros(self.state_dim, dtype=np.float64)
    self._control = np.zeros(self.input_dim, dtype=np.float64)
    self._previous_input = np.zeros(self.input_dim, dtype=np.float64)

exit_initialization_mode

exit_initialization_mode() -> None

Compute the initial control output from the start inputs.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def exit_initialization_mode(self) -> None:
    """Compute the initial control output from the start inputs."""
    self._solve()

set_float64

set_float64(
    value_references: list[int], values: list[float]
) -> None

Set input variables addressed by their value references.

Parameters

value_references : list[int] The handles of the variables to set. values : list[float] The values, one per reference.

Raises

ValueError If the lengths differ, a reference is unknown, or it is not an input.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def set_float64(self, value_references: list[int], values: list[float]) -> None:
    """Set input variables addressed by their value references.

    Parameters
    ----------
    value_references : list[int]
        The handles of the variables to set.
    values : list[float]
        The values, one per reference.

    Raises
    ------
    ValueError
        If the lengths differ, a reference is unknown, or it is not an input.
    """
    if len(value_references) != len(values):
        raise ValueError("value_references and values must have equal length")
    for reference, value in zip(value_references, values, strict=True):
        variable = self._lookup(reference)
        if variable.causality != "input":
            raise ValueError(f"value reference {reference} is not an input")
        if reference < self.state_dim:
            self._state[reference] = float(value)
        else:
            self._reference[reference - self.state_dim] = float(value)

get_float64

get_float64(value_references: list[int]) -> list[float]

Get any variables addressed by their value references.

Parameters

value_references : list[int] The handles of the variables to read.

Returns

list[float] The current values, one per reference.

Raises

ValueError If a reference is unknown.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def get_float64(self, value_references: list[int]) -> list[float]:
    """Get any variables addressed by their value references.

    Parameters
    ----------
    value_references : list[int]
        The handles of the variables to read.

    Returns
    -------
    list[float]
        The current values, one per reference.

    Raises
    ------
    ValueError
        If a reference is unknown.
    """
    values: list[float] = []
    for reference in value_references:
        variable = self._lookup(reference)
        if variable.causality == "output":
            values.append(float(self._control[reference - _OUTPUT_VREF_BASE]))
        elif reference < self.state_dim:
            values.append(float(self._state[reference]))
        else:
            values.append(float(self._reference[reference - self.state_dim]))
    return values

do_step

do_step(
    current_communication_point: float,
    communication_step_size: float,
) -> None

Advance the co-simulation by solving the MPC for the current inputs.

Parameters

current_communication_point : float The master's current time; recorded for the lifecycle contract. communication_step_size : float The communication step; the controller's own sample period governs the internal prediction, so the step is accepted as given.

Raises

ValueError If the communication step size is negative.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def do_step(
    self, current_communication_point: float, communication_step_size: float
) -> None:
    """Advance the co-simulation by solving the MPC for the current inputs.

    Parameters
    ----------
    current_communication_point : float
        The master's current time; recorded for the lifecycle contract.
    communication_step_size : float
        The communication step; the controller's own sample period governs
        the internal prediction, so the step is accepted as given.

    Raises
    ------
    ValueError
        If the communication step size is negative.
    """
    if communication_step_size < 0.0:
        raise ValueError("communication_step_size must be non-negative")
    self._solve()
    self._previous_input = self._control.copy()

terminate

terminate() -> None

End the co-simulation; the slave holds no external resources.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def terminate(self) -> None:
    """End the co-simulation; the slave holds no external resources."""

Functions:

generate_model_description

generate_model_description(slave: CoSimulationSlave) -> str

Render the FMI 3.0 modelDescription.xml for a slave.

Parameters

slave : CoSimulationSlave The slave whose model interface to describe.

Returns

str The modelDescription.xml document.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def generate_model_description(slave: CoSimulationSlave) -> str:
    """Render the FMI 3.0 ``modelDescription.xml`` for a slave.

    Parameters
    ----------
    slave : CoSimulationSlave
        The slave whose model interface to describe.

    Returns
    -------
    str
        The ``modelDescription.xml`` document.
    """
    root = ElementTree.Element(
        "fmiModelDescription",
        {
            "fmiVersion": "3.0",
            "modelName": slave.model_name,
            "instantiationToken": _instantiation_token(
                slave.model_name, slave.variables
            ),
            "description": "SCPN Koopman MPC controller (review-only proposal)",
            "generationTool": "SCPN Phase Orchestrator",
        },
    )
    ElementTree.SubElement(
        root,
        "CoSimulation",
        {
            "modelIdentifier": slave.model_name,
            "canHandleVariableCommunicationStepSize": "true",
        },
    )
    model_variables = ElementTree.SubElement(root, "ModelVariables")
    for variable in slave.variables:
        attributes = {
            "name": variable.name,
            "valueReference": str(variable.value_reference),
            "causality": variable.causality,
            "variability": "continuous",
        }
        if variable.start is not None:
            attributes["start"] = repr(variable.start)
        ElementTree.SubElement(model_variables, "Float64", attributes)
    model_structure = ElementTree.SubElement(root, "ModelStructure")
    for variable in slave.variables:
        if variable.causality == "output":
            reference = {"valueReference": str(variable.value_reference)}
            ElementTree.SubElement(model_structure, "Output", reference)
            ElementTree.SubElement(model_structure, "InitialUnknown", reference)
    ElementTree.indent(root)
    return '<?xml version="1.0" encoding="UTF-8"?>\n' + ElementTree.tostring(
        root, encoding="unicode"
    )

write_fmu

write_fmu(
    slave: CoSimulationSlave, path: str | Path
) -> Path

Package a slave as a .fmu archive (model interface + resources).

The archive carries the conformant modelDescription.xml and a resources/model.json describing the controller, the self-contained model a Python-backed FMI runtime reconstructs. No C-ABI binary shim is shipped, so a third-party FMI tool needs such a runtime to load the archive.

Parameters

slave : CoSimulationSlave The slave to package. path : str | pathlib.Path Destination .fmu path.

Returns

pathlib.Path The written archive path.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def write_fmu(slave: CoSimulationSlave, path: str | Path) -> Path:
    """Package a slave as a ``.fmu`` archive (model interface + resources).

    The archive carries the conformant ``modelDescription.xml`` and a
    ``resources/model.json`` describing the controller, the self-contained model
    a Python-backed FMI runtime reconstructs. No C-ABI binary shim is shipped,
    so a third-party FMI tool needs such a runtime to load the archive.

    Parameters
    ----------
    slave : CoSimulationSlave
        The slave to package.
    path : str | pathlib.Path
        Destination ``.fmu`` path.

    Returns
    -------
    pathlib.Path
        The written archive path.
    """
    destination = Path(path)
    resources = {
        "model_name": slave.model_name,
        "state_dim": slave.state_dim,
        "input_dim": slave.input_dim,
        "state_matrix": slave._controller.predictor.state_matrix.tolist(),
        "input_matrix": slave._controller.predictor.input_matrix.tolist(),
        "output_matrix": slave._controller.predictor.output_matrix.tolist(),
        "horizon": slave._controller.config.horizon,
    }
    with zipfile.ZipFile(destination, "w", zipfile.ZIP_DEFLATED) as archive:
        archive.writestr("modelDescription.xml", generate_model_description(slave))
        archive.writestr("resources/model.json", json.dumps(resources, indent=2))
    return destination

cosimulate

cosimulate(
    controller: CoSimulationSlave,
    plant_step: PlantStep,
    *,
    initial_state: FloatArray,
    steps: int,
    dt: float,
    reference: FloatArray | None = None,
) -> FloatArray

Run a co-simulation master coupling the controller slave with a plant.

This is the import/master direction: SPO drives a plant model in co-simulation. Each step writes the plant state to the controller's state inputs, advances the controller by one MPC step, reads its control output and applies it to the plant, then advances the plant. The plant is any step callable (state, control, dt) -> next_state; an external plant FMU plugs in by wrapping its FMI runtime (for example fmpy) as such a callable, so no FMI runtime dependency is imposed here.

Parameters

controller : CoSimulationSlave The FMI controller slave to drive. plant_step : Callable[[numpy.ndarray, numpy.ndarray, float], numpy.ndarray] Advances the plant by dt under the applied control. initial_state : numpy.ndarray The plant's initial state x_0 of shape (n,). steps : int Number of co-simulation steps. dt : float The communication step size. reference : numpy.ndarray | None The controller set point of shape (n,); defaults to the origin.

Returns

numpy.ndarray The closed-loop plant-state trajectory of shape (steps + 1, n).

Raises

ValueError If the initial state length, step count, or dt are inconsistent.

Source code in src/scpn_phase_orchestrator/adapters/fmi_cosimulation.py

def cosimulate(
    controller: CoSimulationSlave,
    plant_step: PlantStep,
    *,
    initial_state: FloatArray,
    steps: int,
    dt: float,
    reference: FloatArray | None = None,
) -> FloatArray:
    """Run a co-simulation master coupling the controller slave with a plant.

    This is the import/master direction: SPO drives a plant model in
    co-simulation. Each step writes the plant state to the controller's state
    inputs, advances the controller by one MPC step, reads its control output and
    applies it to the plant, then advances the plant. The plant is any step
    callable ``(state, control, dt) -> next_state``; an external plant FMU plugs
    in by wrapping its FMI runtime (for example ``fmpy``) as such a callable, so
    no FMI runtime dependency is imposed here.

    Parameters
    ----------
    controller : CoSimulationSlave
        The FMI controller slave to drive.
    plant_step : Callable[[numpy.ndarray, numpy.ndarray, float], numpy.ndarray]
        Advances the plant by ``dt`` under the applied control.
    initial_state : numpy.ndarray
        The plant's initial state ``x_0`` of shape ``(n,)``.
    steps : int
        Number of co-simulation steps.
    dt : float
        The communication step size.
    reference : numpy.ndarray | None
        The controller set point of shape ``(n,)``; defaults to the origin.

    Returns
    -------
    numpy.ndarray
        The closed-loop plant-state trajectory of shape ``(steps + 1, n)``.

    Raises
    ------
    ValueError
        If the initial state length, step count, or ``dt`` are inconsistent.
    """
    state_dim = controller.state_dim
    state = np.ascontiguousarray(np.asarray(initial_state, dtype=np.float64).ravel())
    if state.shape[0] != state_dim:
        raise ValueError("initial_state length must match the controller state")
    if steps < 1:
        raise ValueError("steps must be at least 1")
    if dt < 0.0:
        raise ValueError("dt must be non-negative")

    set_point = (
        np.zeros(state_dim, dtype=np.float64)
        if reference is None
        else np.asarray(reference, dtype=np.float64).ravel()
    )
    controller.enter_initialization_mode()
    controller.set_float64(
        list(range(state_dim, 2 * state_dim)), [float(v) for v in set_point]
    )
    controller.exit_initialization_mode()

    state_references = list(range(state_dim))
    control_references = [_OUTPUT_VREF_BASE + j for j in range(controller.input_dim)]
    trajectory = [state.copy()]
    time = 0.0
    for _ in range(steps):
        controller.set_float64(state_references, [float(v) for v in state])
        controller.do_step(time, dt)
        control = np.asarray(
            controller.get_float64(control_references), dtype=np.float64
        )
        state = np.ascontiguousarray(
            np.asarray(plant_step(state, control, dt), dtype=np.float64).ravel()
        )
        trajectory.append(state.copy())
        time += dt
    return np.asarray(trajectory, dtype=np.float64)

Adapter selection guidance

Choose adapter layers by failure tolerance and change management profile:

Deployment pattern	Recommended adapter set	Primary constraint
Internal production control loop	control or hardware adapters only where bounded actuators are required	bounded actuation and deterministic replay
Observability-first rollout	opentelemetry, metrics_exporter, prometheus	low-latency visibility before actuation
Cross-repo data exchange	fusion_core_bridge, scpn_control_bridge, neurocore_bridge	schema compatibility and replayability
Hardware field trial	modbus_tls, hardware_io with explicit opt-in flags	physical safety and rollback path
Research and co-compilation	hybrid_cocompiler, quantum_control_bridge	review-only policy and explicit scope notes

spo doctor reports the package-local FMI and hybrid co-compiler review/export surfaces as optional adapter diagnostics. A warning there means the package is missing an expected local adapter export; it does not mean SPO is connected to a live FMI runtime, QPU, neuromorphic backend, or actuator.

Every adapter contributes a conversion boundary. Production changes should only depend on adapters that are covered by active parity and boundary tests for the target release.

Security and quality boundary

Adapter boundaries should remain explicit in runtime runbooks:

• validate all external payloads before deriving phase or coupling state,
• keep non-production adapters out of critical paths by default,
• keep adapter version, endpoint, and mode in audit metadata so post-hoc reviews can identify where control decisions changed.

That boundary allows teams to reuse the same internal core while keeping external dependencies isolated from safety-critical decision flow.