Monitor — Merge Window

MergeWindowMonitor is the PHA-C.4 runtime gate for deciding when a moving-frame phase cluster has actually merged. It combines two predicates: wrapped phase dispersion around a reference phase and axial spatial dispersion around a reference point. The monitor emits lock_achieved=True only after both predicates remain inside tolerance for the configured number of consecutive samples.

What it is for

The monitor is intended for PHA-C pipelines where phase dynamics and physical or abstract position are both meaningful. Examples include moving-frame UPDE runs, coalescence studies, chamber or conveyor synchronization, plasma or beam alignment experiments, robotic swarms, and digital-twin cells where a phase lock alone is not enough evidence that the population has spatially merged.

Contract

• Phase dispersion: max_i |wrap(theta_i - theta_ref)| <= phase_tol_rad.
• Spatial dispersion: max_i |z_i - z_ref| <= spatial_tol_m.
• Signed margins: phase_margin_rad = phase_tol_rad - phase_dispersion_rad and spatial_margin_m = spatial_tol_m - spatial_dispersion_m.
• Signed-margin replay tolerance: MERGE_WINDOW_MARGIN_REPLAY_TOLERANCE.
• Consecutive gate: both predicates must pass for required_consecutive_samples.
• Default tolerances: phase_tol_rad=0.01, spatial_tol_m=0.002, required_consecutive_samples=3.
• Named profiles multiply the reviewed baseline: baseline_1x, buffer_3x, and review_5x.
• Public scalar and vector inputs must contain finite real numeric evidence before conversion. Boolean, complex, numeric-string, and broken array-protocol payloads fail closed; real numeric object arrays remain valid.
• A directly constructed tolerance profile must replay its reviewed name, multiplier, baseline, and resolved tolerances. A directly constructed MergeReport independently validates finite scalar fields, non-negative dispersions, canonical booleans/counts, signed-margin lock semantics, and the current joint-lock/count relationship before it can be serialised.
• Evidence boundary: benchmark timings are local regression evidence unless run under the documented isolated-core benchmark protocol.

import numpy as np
from scpn_phase_orchestrator.monitor.merge_window import MergeWindowMonitor

monitor = MergeWindowMonitor(
    phase_tol_rad=0.01,
    spatial_tol_m=0.002,
    required_consecutive_samples=3,
)

for t in range(3):
    report = monitor.evaluate(
        np.array([0.0, 0.004, -0.005]),
        np.array([0.0, 0.001, -0.0015]),
        t=float(t),
    )

assert report.lock_achieved

Tolerance profiles

PHA-C and MIF/FRC review lanes often need to separate the reviewed baseline window from wider diagnostic buffers. resolve_merge_window_tolerance_profile keeps that boundary explicit:

from scpn_phase_orchestrator.monitor.merge_window import (
    MergeWindowMonitor,
    resolve_merge_window_tolerance_profile,
)

profile = resolve_merge_window_tolerance_profile("buffer_3x")
assert profile.spatial_tol_m == 0.006

monitor = MergeWindowMonitor(tolerance_profile="buffer_3x")

The default baseline is 0.01 rad and 0.002 m. Passing explicit phase_tol_rad or spatial_tol_m with a profile treats those values as the baseline before applying the multiplier.

Polyglot surfaces

The Python monitor is the public runtime reference. Rust, Go, Julia, and Mojo source-contract adapter modules are present for parity gates and downstream accelerator wiring. The benchmark gate records all declared backend slots and labels the local workstation timing data as non-isolated evidence. Adapter parity includes the signed margin fields, so a backend cannot pass with only boolean lock evidence. The benchmark payload also publishes phase_margin_equation_validated, spatial_margin_equation_validated, signed_margin_equations_validated, and margin_replay_tolerance; the gate fails unless every declared backend row proves both signed-margin equations. The shared accelerator validator checks raw MergeReport fields before parity comparison: numeric fields must be finite real non-boolean scalars, lock fields must be plain booleans, consecutive counts must be non-negative integers, and the comparison tolerance itself must be finite and non-negative.

uv run python benchmarks/merge_window_benchmark.py --parity-gate

Event/state handoff

When the merge-window report must cross into MIF, Studio, audit replay, or another downstream PHA-C lane, use build_pha_c_handoff_record(...). It binds the merge report to the source phase and position digests, adds signed margin and order-parameter evidence, and fixes the non-actuating claim boundary for later review.

from scpn_phase_orchestrator.upde.pha_c_handoff import (
    build_pha_c_handoff_record,
)

handoff = build_pha_c_handoff_record(
    phases,
    positions,
    phase_tol_rad=0.01,
    spatial_tol_m=0.002,
    required_consecutive_samples=3,
)

Operational role

• This monitor is the gate where “mostly synchronized” becomes “merged” in a replayable way.
• Signed-margin outputs are what operators use to tune recovery aggressiveness without guessing how close the signal was to the boundary.
• The handoff record keeps review lanes deterministic: MIF, Studio, and audit replay all receive the same lock/evidence contract.

What this means for operations

Merge-window gates are most useful when teams need a binary operational decision (lock_achieved) plus a continuous confidence context (margins).

The signed-margin design makes recovery tuning possible without guesswork:

• phase_margin_rad and spatial_margin_m quantify how far the current state is from the lock boundary.
• required_consecutive_samples filters transients and reduces false merge-on-spike behaviour.

In production review lanes, that combination is what allows teams to avoid both premature lock declarations and excessive delay in returning to active control.

merge_window

Phase-and-space merge-window lock monitor.

The PHA-C moving-frame lane tracks phase theta and axial position z for candidate merger/coalescence events. A merge is accepted only when both the wrapped phase dispersion and the axial spatial dispersion remain inside their reviewed tolerances for a configured number of consecutive samples.

Classes

MergeWindowToleranceProfile dataclass

MergeWindowToleranceProfile(
    name: str,
    phase_tol_rad: float,
    spatial_tol_m: float,
    multiplier: float,
    baseline_phase_tol_rad: float,
    baseline_spatial_tol_m: float,
)

Resolved phase and spatial tolerances for a PHA-C merge window.

Methods:

__post_init__

__post_init__() -> None

Validate and normalise the resolved named-profile evidence.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def __post_init__(self) -> None:
    """Validate and normalise the resolved named-profile evidence."""
    name = _validate_profile_name(self.name)
    phase_tol = _validate_tolerance(self.phase_tol_rad, name="phase_tol_rad")
    spatial_tol = _validate_tolerance(self.spatial_tol_m, name="spatial_tol_m")
    multiplier = _validate_positive_scalar(self.multiplier, name="multiplier")
    baseline_phase = _validate_tolerance(
        self.baseline_phase_tol_rad,
        name="baseline_phase_tol_rad",
    )
    baseline_spatial = _validate_tolerance(
        self.baseline_spatial_tol_m,
        name="baseline_spatial_tol_m",
    )
    expected_multiplier = MERGE_WINDOW_TOLERANCE_PROFILE_MULTIPLIERS[name]
    if multiplier != expected_multiplier:
        raise ValueError(
            "name and multiplier must match the reviewed tolerance profile"
        )
    expected_phase_tol = baseline_phase * multiplier
    expected_spatial_tol = baseline_spatial * multiplier
    if not np.isclose(
        phase_tol,
        expected_phase_tol,
        rtol=8.0 * np.finfo(np.float64).eps,
        atol=0.0,
    ):
        raise ValueError(
            "phase_tol_rad must equal baseline_phase_tol_rad * multiplier"
        )
    if not np.isclose(
        spatial_tol,
        expected_spatial_tol,
        rtol=8.0 * np.finfo(np.float64).eps,
        atol=0.0,
    ):
        raise ValueError(
            "spatial_tol_m must equal baseline_spatial_tol_m * multiplier"
        )
    object.__setattr__(self, "name", name)
    object.__setattr__(self, "phase_tol_rad", phase_tol)
    object.__setattr__(self, "spatial_tol_m", spatial_tol)
    object.__setattr__(self, "multiplier", multiplier)
    object.__setattr__(self, "baseline_phase_tol_rad", baseline_phase)
    object.__setattr__(self, "baseline_spatial_tol_m", baseline_spatial)

to_dict

to_dict() -> dict[str, float | str]

Return a JSON-safe tolerance-profile payload.

Returns

dict[str, float | str] Return a JSON-safe tolerance-profile payload.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def to_dict(self) -> dict[str, float | str]:
    """Return a JSON-safe tolerance-profile payload.

    Returns
    -------
    dict[str, float | str]
        Return a JSON-safe tolerance-profile payload.
    """
    return merge_window_tolerance_profile_to_dict(self)

MergeReport dataclass

MergeReport(
    t: float,
    phase_dispersion_rad: float,
    spatial_dispersion_m: float,
    phase_margin_rad: float,
    spatial_margin_m: float,
    phase_locked: bool,
    spatial_locked: bool,
    lock_achieved: bool,
    consecutive_lock_samples: int,
)

Audit-ready merge-window state for one sampled instant.

Attributes

t: Sample timestamp in the caller's runtime units.
phase_dispersion_rad: Maximum wrapped distance to the reference phase.
spatial_dispersion_m: Maximum axial distance to the reference point.
phase_margin_rad: Signed distance from phase tolerance to dispersion.
spatial_margin_m: Signed distance from spatial tolerance to dispersion.
phase_locked: True when phase margin is non-negative.
spatial_locked: True when spatial margin is non-negative.
lock_achieved: True after the required consecutive joint-lock count.
consecutive_lock_samples: Current consecutive joint-lock count.

Methods:

__post_init__

__post_init__() -> None

Validate and normalise directly constructed merge evidence.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def __post_init__(self) -> None:
    """Validate and normalise directly constructed merge evidence."""
    timestamp = _validate_real_scalar(self.t, name="t")
    phase_dispersion = _validate_tolerance(
        self.phase_dispersion_rad,
        name="phase_dispersion_rad",
    )
    spatial_dispersion = _validate_tolerance(
        self.spatial_dispersion_m,
        name="spatial_dispersion_m",
    )
    phase_margin = _validate_real_scalar(
        self.phase_margin_rad,
        name="phase_margin_rad",
    )
    spatial_margin = _validate_real_scalar(
        self.spatial_margin_m,
        name="spatial_margin_m",
    )
    phase_locked = _validate_plain_bool(self.phase_locked, name="phase_locked")
    spatial_locked = _validate_plain_bool(
        self.spatial_locked,
        name="spatial_locked",
    )
    lock_achieved = _validate_plain_bool(
        self.lock_achieved,
        name="lock_achieved",
    )
    consecutive = _validate_sample_count(
        self.consecutive_lock_samples,
        name="consecutive_lock_samples",
        minimum=0,
    )
    if phase_locked is not (phase_margin >= 0.0):
        raise ValueError("phase_locked must match the sign of phase_margin_rad")
    if spatial_locked is not (spatial_margin >= 0.0):
        raise ValueError("spatial_locked must match the sign of spatial_margin_m")
    joint_lock = phase_locked and spatial_locked
    if (joint_lock and consecutive == 0) or (not joint_lock and consecutive != 0):
        raise ValueError(
            "consecutive_lock_samples must be positive exactly when jointly locked"
        )
    if lock_achieved and not joint_lock:
        raise ValueError("lock_achieved requires current phase and spatial lock")
    object.__setattr__(self, "t", timestamp)
    object.__setattr__(self, "phase_dispersion_rad", phase_dispersion)
    object.__setattr__(self, "spatial_dispersion_m", spatial_dispersion)
    object.__setattr__(self, "phase_margin_rad", phase_margin)
    object.__setattr__(self, "spatial_margin_m", spatial_margin)
    object.__setattr__(self, "phase_locked", phase_locked)
    object.__setattr__(self, "spatial_locked", spatial_locked)
    object.__setattr__(self, "lock_achieved", lock_achieved)
    object.__setattr__(self, "consecutive_lock_samples", consecutive)

to_dict

to_dict() -> dict[str, float | int | bool]

Return a JSON-safe representation for audit and benchmark records.

Returns

dict[str, float | int | bool] Return a JSON-safe representation for audit and benchmark records.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def to_dict(self) -> dict[str, float | int | bool]:
    """Return a JSON-safe representation for audit and benchmark records.

    Returns
    -------
    dict[str, float | int | bool]
        Return a JSON-safe representation for audit and benchmark records.
    """
    return merge_window_report_to_dict(self)

MergeWindowMonitor

MergeWindowMonitor(
    *,
    phase_tol_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_tol_m: object = DEFAULT_SPATIAL_TOL_M,
    required_consecutive_samples: object = 3,
    tolerance_profile: object | None = None,
)

Stateful consecutive-sample gate for PHA-C merge events.

Initialise the stateful merge gate.

Parameters

phase_tol_rad : object Baseline phase tolerance in radians. spatial_tol_m : object Baseline spatial tolerance in metres. required_consecutive_samples : object Positive joint-lock sample count required for acceptance. tolerance_profile : object | None Reviewed named tolerance profile, or None for explicit values.

Raises

ValueError If a tolerance, count, or named-profile contract is invalid.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def __init__(
    self,
    *,
    phase_tol_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_tol_m: object = DEFAULT_SPATIAL_TOL_M,
    required_consecutive_samples: object = 3,
    tolerance_profile: object | None = None,
) -> None:
    """Initialise the stateful merge gate.

    Parameters
    ----------
    phase_tol_rad : object
        Baseline phase tolerance in radians.
    spatial_tol_m : object
        Baseline spatial tolerance in metres.
    required_consecutive_samples : object
        Positive joint-lock sample count required for acceptance.
    tolerance_profile : object | None
        Reviewed named tolerance profile, or ``None`` for explicit values.

    Raises
    ------
    ValueError
        If a tolerance, count, or named-profile contract is invalid.
    """
    self.tolerance_profile = None
    if tolerance_profile is None:
        self.phase_tol_rad = _validate_tolerance(
            phase_tol_rad,
            name="phase_tol_rad",
        )
        self.spatial_tol_m = _validate_tolerance(
            spatial_tol_m,
            name="spatial_tol_m",
        )
    else:
        profile = resolve_merge_window_tolerance_profile(
            tolerance_profile,
            phase_baseline_rad=phase_tol_rad,
            spatial_baseline_m=spatial_tol_m,
        )
        self.tolerance_profile = profile
        self.phase_tol_rad = profile.phase_tol_rad
        self.spatial_tol_m = profile.spatial_tol_m
    self.required_consecutive_samples = _validate_sample_count(
        required_consecutive_samples,
        name="required_consecutive_samples",
        minimum=1,
    )
    self._consecutive_lock_samples = 0

Attributes

consecutive_lock_samples property

consecutive_lock_samples: int

Current consecutive joint-lock count.

Returns

int Current consecutive joint-lock count.

Methods:

reset

reset() -> None

Reset the consecutive joint-lock counter.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def reset(self) -> None:
    """Reset the consecutive joint-lock counter."""
    self._consecutive_lock_samples = 0

evaluate

evaluate(
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
) -> MergeReport

Evaluate one sample and update the consecutive joint-lock counter.

Parameters

phases : ArrayLike Oscillator phases in radians, shape (N,). positions : ArrayLike Absolute axial coordinates per oscillator, shape (N,). t : object Absolute time of the sample in seconds. reference_phase : object Reference phase for the lock criterion, in radians. reference_point : object Reference axial coordinate for the spatial-margin criterion.

Returns

MergeReport The merge-window report with the updated lock counter.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def evaluate(
    self,
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
) -> MergeReport:
    """Evaluate one sample and update the consecutive joint-lock counter.

    Parameters
    ----------
    phases : ArrayLike
        Oscillator phases in radians, shape ``(N,)``.
    positions : ArrayLike
        Absolute axial coordinates per oscillator, shape ``(N,)``.
    t : object
        Absolute time of the sample in seconds.
    reference_phase : object
        Reference phase for the lock criterion, in radians.
    reference_point : object
        Reference axial coordinate for the spatial-margin criterion.

    Returns
    -------
    MergeReport
        The merge-window report with the updated lock counter.
    """
    report = evaluate_merge_window(
        phases,
        positions,
        t=t,
        reference_phase=reference_phase,
        reference_point=reference_point,
        phase_tol_rad=self.phase_tol_rad,
        spatial_tol_m=self.spatial_tol_m,
        required_consecutive_samples=self.required_consecutive_samples,
        prior_consecutive_lock_samples=self._consecutive_lock_samples,
    )
    self._consecutive_lock_samples = report.consecutive_lock_samples
    return report

__call__

__call__(
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
) -> MergeReport

Alias for :meth:evaluate for monitor-pipeline call sites.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def __call__(
    self,
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
) -> MergeReport:
    """Alias for :meth:`evaluate` for monitor-pipeline call sites."""
    return self.evaluate(
        phases,
        positions,
        t=t,
        reference_phase=reference_phase,
        reference_point=reference_point,
    )

Functions:

resolve_merge_window_tolerance_profile

resolve_merge_window_tolerance_profile(
    tolerance_profile: object,
    *,
    phase_baseline_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_baseline_m: object = DEFAULT_SPATIAL_TOL_M,
) -> MergeWindowToleranceProfile

Resolve a named PHA-C tolerance profile into numeric tolerances.

Parameters

tolerance_profile : object Named tolerance profile, or None for the baseline. phase_baseline_rad : object Baseline phase tolerance in radians. spatial_baseline_m : object Baseline spatial tolerance in metres.

Returns

MergeWindowToleranceProfile The resolved numeric tolerance profile.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def resolve_merge_window_tolerance_profile(
    tolerance_profile: object,
    *,
    phase_baseline_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_baseline_m: object = DEFAULT_SPATIAL_TOL_M,
) -> MergeWindowToleranceProfile:
    """Resolve a named PHA-C tolerance profile into numeric tolerances.

    Parameters
    ----------
    tolerance_profile : object
        Named tolerance profile, or ``None`` for the baseline.
    phase_baseline_rad : object
        Baseline phase tolerance in radians.
    spatial_baseline_m : object
        Baseline spatial tolerance in metres.

    Returns
    -------
    MergeWindowToleranceProfile
        The resolved numeric tolerance profile.
    """
    if isinstance(tolerance_profile, MergeWindowToleranceProfile):
        return tolerance_profile
    name = _validate_profile_name(tolerance_profile)
    phase_baseline = _validate_tolerance(
        phase_baseline_rad,
        name="phase_baseline_rad",
    )
    spatial_baseline = _validate_tolerance(
        spatial_baseline_m,
        name="spatial_baseline_m",
    )
    multiplier = MERGE_WINDOW_TOLERANCE_PROFILE_MULTIPLIERS[name]
    return MergeWindowToleranceProfile(
        name=name,
        phase_tol_rad=phase_baseline * multiplier,
        spatial_tol_m=spatial_baseline * multiplier,
        multiplier=multiplier,
        baseline_phase_tol_rad=phase_baseline,
        baseline_spatial_tol_m=spatial_baseline,
    )

evaluate_merge_window

evaluate_merge_window(
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
    phase_tol_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_tol_m: object = DEFAULT_SPATIAL_TOL_M,
    required_consecutive_samples: object = 3,
    prior_consecutive_lock_samples: object = 0,
    tolerance_profile: object | None = None,
) -> MergeReport

Evaluate one PHA-C merge-window sample.

Phase lock is max_i |wrap(theta_i - theta_ref)| <= phase_tol_rad. Spatial lock is max_i |z_i - z_ref| <= spatial_tol_m. The combined lock counter increments only when both predicates pass; otherwise it resets to zero. lock_achieved becomes true once the counter reaches required_consecutive_samples.

Parameters

phases : ArrayLike Oscillator phases in radians, shape (N,). positions : ArrayLike Absolute axial coordinates per oscillator, shape (N,). t : object Absolute time of the sample in seconds. reference_phase : object Reference phase for the lock criterion, in radians. reference_point : object Reference axial coordinate for the spatial-margin criterion. phase_tol_rad : object Phase lock tolerance in radians. spatial_tol_m : object Spatial lock tolerance in metres. required_consecutive_samples : object Consecutive in-tolerance samples required to declare lock. prior_consecutive_lock_samples : object Consecutive lock-sample count carried in from a prior window. tolerance_profile : object | None Named tolerance profile, or None for the baseline.

Returns

MergeReport The merge-window evaluation report for the sample.

Raises

ValueError If any input is invalid.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def evaluate_merge_window(
    phases: ArrayLike,
    positions: ArrayLike,
    *,
    t: object = 0.0,
    reference_phase: object = 0.0,
    reference_point: object = 0.0,
    phase_tol_rad: object = DEFAULT_PHASE_TOL_RAD,
    spatial_tol_m: object = DEFAULT_SPATIAL_TOL_M,
    required_consecutive_samples: object = 3,
    prior_consecutive_lock_samples: object = 0,
    tolerance_profile: object | None = None,
) -> MergeReport:
    """Evaluate one PHA-C merge-window sample.

    Phase lock is ``max_i |wrap(theta_i - theta_ref)| <= phase_tol_rad``.
    Spatial lock is ``max_i |z_i - z_ref| <= spatial_tol_m``. The combined lock
    counter increments only when both predicates pass; otherwise it resets to
    zero. ``lock_achieved`` becomes true once the counter reaches
    ``required_consecutive_samples``.

    Parameters
    ----------
    phases : ArrayLike
        Oscillator phases in radians, shape ``(N,)``.
    positions : ArrayLike
        Absolute axial coordinates per oscillator, shape ``(N,)``.
    t : object
        Absolute time of the sample in seconds.
    reference_phase : object
        Reference phase for the lock criterion, in radians.
    reference_point : object
        Reference axial coordinate for the spatial-margin criterion.
    phase_tol_rad : object
        Phase lock tolerance in radians.
    spatial_tol_m : object
        Spatial lock tolerance in metres.
    required_consecutive_samples : object
        Consecutive in-tolerance samples required to declare lock.
    prior_consecutive_lock_samples : object
        Consecutive lock-sample count carried in from a prior window.
    tolerance_profile : object | None
        Named tolerance profile, or ``None`` for the baseline.

    Returns
    -------
    MergeReport
        The merge-window evaluation report for the sample.

    Raises
    ------
    ValueError
        If any input is invalid.
    """
    phase_vector = _as_float_vector(phases, name="phases")
    position_vector = _as_float_vector(positions, name="positions")
    if position_vector.shape != phase_vector.shape:
        raise ValueError("positions must have the same one-dimensional shape as phases")

    timestamp = _validate_real_scalar(t, name="t")
    phase_reference = _validate_real_scalar(reference_phase, name="reference_phase")
    spatial_reference = _validate_real_scalar(reference_point, name="reference_point")
    if tolerance_profile is None:
        phase_tol = _validate_tolerance(phase_tol_rad, name="phase_tol_rad")
        spatial_tol = _validate_tolerance(spatial_tol_m, name="spatial_tol_m")
    else:
        profile = resolve_merge_window_tolerance_profile(
            tolerance_profile,
            phase_baseline_rad=phase_tol_rad,
            spatial_baseline_m=spatial_tol_m,
        )
        phase_tol = profile.phase_tol_rad
        spatial_tol = profile.spatial_tol_m
    required = _validate_sample_count(
        required_consecutive_samples,
        name="required_consecutive_samples",
        minimum=1,
    )
    prior = _validate_sample_count(
        prior_consecutive_lock_samples,
        name="prior_consecutive_lock_samples",
        minimum=0,
    )

    phase_dispersion = _phase_dispersion_rad(phase_vector, phase_reference)
    spatial_dispersion = _spatial_dispersion_m(position_vector, spatial_reference)
    phase_margin = phase_tol - phase_dispersion
    spatial_margin = spatial_tol - spatial_dispersion
    phase_locked = phase_margin >= 0.0
    spatial_locked = spatial_margin >= 0.0
    consecutive = prior + 1 if phase_locked and spatial_locked else 0
    return MergeReport(
        t=timestamp,
        phase_dispersion_rad=phase_dispersion,
        spatial_dispersion_m=spatial_dispersion,
        phase_margin_rad=phase_margin,
        spatial_margin_m=spatial_margin,
        phase_locked=bool(phase_locked),
        spatial_locked=bool(spatial_locked),
        lock_achieved=bool(consecutive >= required),
        consecutive_lock_samples=consecutive,
    )

merge_window_report_to_dict

merge_window_report_to_dict(
    report: MergeReport,
) -> dict[str, float | int | bool]

Convert a :class:MergeReport into a JSON-safe dictionary.

Parameters

report : MergeReport The merge-window report to serialise.

Returns

dict[str, float | int | bool] The JSON-safe merge-window report dictionary.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def merge_window_report_to_dict(report: MergeReport) -> dict[str, float | int | bool]:
    """Convert a :class:`MergeReport` into a JSON-safe dictionary.

    Parameters
    ----------
    report : MergeReport
        The merge-window report to serialise.

    Returns
    -------
    dict[str, float | int | bool]
        The JSON-safe merge-window report dictionary.
    """
    return {
        "t": float(report.t),
        "phase_dispersion_rad": float(report.phase_dispersion_rad),
        "spatial_dispersion_m": float(report.spatial_dispersion_m),
        "phase_margin_rad": float(report.phase_margin_rad),
        "spatial_margin_m": float(report.spatial_margin_m),
        "phase_locked": bool(report.phase_locked),
        "spatial_locked": bool(report.spatial_locked),
        "lock_achieved": bool(report.lock_achieved),
        "consecutive_lock_samples": int(report.consecutive_lock_samples),
    }

merge_window_tolerance_profile_to_dict

merge_window_tolerance_profile_to_dict(
    profile: MergeWindowToleranceProfile,
) -> dict[str, float | str]

Convert a resolved tolerance profile into a JSON-safe dictionary.

Parameters

profile : MergeWindowToleranceProfile The resolved tolerance profile to serialise.

Returns

dict[str, float | str] The JSON-safe tolerance-profile dictionary.

Source code in src/scpn_phase_orchestrator/monitor/merge_window.py

def merge_window_tolerance_profile_to_dict(
    profile: MergeWindowToleranceProfile,
) -> dict[str, float | str]:
    """Convert a resolved tolerance profile into a JSON-safe dictionary.

    Parameters
    ----------
    profile : MergeWindowToleranceProfile
        The resolved tolerance profile to serialise.

    Returns
    -------
    dict[str, float | str]
        The JSON-safe tolerance-profile dictionary.
    """
    return {
        "name": str(profile.name),
        "phase_tol_rad": float(profile.phase_tol_rad),
        "spatial_tol_m": float(profile.spatial_tol_m),
        "multiplier": float(profile.multiplier),
        "baseline_phase_tol_rad": float(profile.baseline_phase_tol_rad),
        "baseline_spatial_tol_m": float(profile.baseline_spatial_tol_m),
    }