Skip to content

UPDE Doppler Engine

DopplerEngine is the PHA-C.2 kinematic phase-dynamics surface for moving oscillator systems where relative velocity shifts the effective natural frequency before phase coupling is applied. It is intended for counter-moving plasmoids, moving sensor swarms, acoustic/clock synchronisation under motion, and any domain where phase lock depends on velocity-corrected detuning rather than static omega alone.

Mathematical contract

For each outer integration step the engine resolves omega(t) and scalarises velocity into one value per oscillator. The effective frequency is

omega_eff_i(t) = omega_i(t) + D_i(t)
D_i(t) = s * sum_j |K_ij| * (v_i - v_j) / (|v_i| + epsilon) / sum_j |K_ij|

where s is doppler_strength, epsilon is doppler_epsilon, and rows with no active coupling receive zero Doppler correction. The row normalisation keeps doppler_strength independent of graph degree while the active K_nm topology still determines which relative velocities are physically coupled.

Scalar velocities use the signed values directly. Vector velocities are reduced to scalar speeds by Euclidean norm unless velocity_axis is supplied, in which case velocities are projected onto the normalised axis so counter-propagating motion keeps its sign.

Public API

import numpy as np
from scpn_phase_orchestrator.upde import DopplerEngine

knm = np.array([[0.0, 5.0], [5.0, 0.0]])
velocities = np.array([300.0, -300.0])
omega = np.array([-2.0, 2.0])

engine = DopplerEngine(
    2,
    omega=omega,
    k_nm=knm,
    alpha=0.0,
    dt=1.0e-3,
    velocities=velocities,
    solver="euler",
)
phases = engine.run(n_steps=2_000)
print(phases, engine.doppler_term)

velocities may be a fixed array or a callable velocities(t) -> array. omega may use the same fixed/callable forms supported by UPDEEngine.

Backend contract

Doppler integration is implemented as a schedule-backed UPDE run:

  1. Resolve omega_schedule[step, i].
  2. Resolve velocity_schedule[step, i].
  3. Compute graph-weighted D_i.
  4. Integrate one UPDE outer step with omega_eff = omega + D.

The source contract exists for Python, Rust/PyO3, Go, Julia, and Mojo. The benchmark gate records unavailable optional runtimes explicitly rather than silently skipping parity evidence.

PYTHONPATH=src python benchmarks/upde_doppler_benchmark.py --parity-gate

Committed local benchmark artefacts are regression/parity evidence only unless rerun with the repository benchmark-isolation protocol.

Failure boundaries

DopplerEngine fails closed on:

  • non-finite or non-real phases, omega schedules, velocities, K_nm, or alpha;
  • boolean and object-dtype numeric aliases;
  • non-zero self-coupling diagonal in K_nm;
  • non-positive doppler_epsilon;
  • malformed scalar or vector velocity shapes;
  • backend outputs outside [0, 2*pi).

Operational use case

  • Doppler correction is used when velocity mismatch can destabilise the intended phase order before a policy loop can react.
  • The row-normalised coupling form keeps response characteristics comparable across changing communication density and moving graph topologies.
  • In review mode, compare doppler_term trends with regime transitions so you can separate physical transport effects from control-induced desynchronization.

Operational context

The Doppler surface makes motion effects explicit instead of forcing operators to absorb them as unexplained model error. That reduces false positives in review lanes that otherwise appear as sudden synchronization loss.

For fleets, moving swarms, or rotating hardware, this surface enables consistent comparisons by normalising relative-velocity effects by coupling structure.

Because output includes the per-step doppler_term, teams can attribute recovery latency either to transport effects or to control actuation.

DopplerEngine

DopplerEngine(
    n: int,
    omega: object,
    k_nm: object,
    alpha: object = 0.0,
    dt: float = 0.01,
    velocities: object
    | Callable[[float], object]
    | None = None,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1e-09,
    solver: str = "rk45",
    phases: object | None = None,
    velocity_axis: object | None = None,
    t0: float = 0.0,
)

Bases: UPDEEngine

Stateful UPDE engine with graph-weighted Doppler velocity correction.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def __init__(
    self,
    n: int,
    omega: object,
    k_nm: object,
    alpha: object = 0.0,
    dt: float = 0.01,
    velocities: object | Callable[[float], object] | None = None,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1.0e-9,
    solver: str = "rk45",
    phases: object | None = None,
    velocity_axis: object | None = None,
    t0: float = 0.0,
) -> None:
    if velocities is None:
        raise ValueError("velocities are required for DopplerEngine")
    super().__init__(n, dt=dt, method=solver, omega=omega, t0=t0)
    self.k_nm = _validate_knm(k_nm, n=n)
    self.alpha_matrix = _validate_alpha(alpha, n=n)
    self._velocity_source = velocities
    self._velocity_axis = velocity_axis
    self.doppler_strength = _finite_float(doppler_strength, name="doppler_strength")
    self.doppler_epsilon = _finite_float(doppler_epsilon, name="doppler_epsilon")
    if self.doppler_epsilon <= 0.0:
        raise ValueError("doppler_epsilon must be positive")
    if phases is None:
        self.phases = np.zeros(n, dtype=np.float64)
    else:
        self.phases = _validate_phases(phases, n=n)
    self.velocity_current = self._velocity_for_step(self.time)
    self._doppler_term = doppler_term(
        self.velocity_current,
        self.k_nm,
        doppler_strength=self.doppler_strength,
        doppler_epsilon=self.doppler_epsilon,
    )

Attributes

doppler_term property

doppler_term: FloatArray

Most recently applied Doppler correction vector.

Returns

FloatArray Most recently applied Doppler correction vector.

Methods:

step

step(
    phases: object | None = None,
    omegas: object | None = None,
    knm: object | None = None,
    zeta: float = 0.0,
    psi: float = 0.0,
    alpha: object | None = None,
) -> FloatArray

Advance one Doppler-corrected UPDE step.

Parameters

phases : object | None Oscillator phases in radians, shape (N,). omegas : object | None Natural frequencies in rad/s, shape (N,). knm : object | None Coupling matrix K_nm, shape (N, N). zeta : float External drive strength ζ. psi : float External drive reference phase Ψ in radians. alpha : object | None Phase-lag matrix in radians, shape (N, N), or None for no lag.

Returns

FloatArray The phases after one Doppler-corrected UPDE step.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def step(
    self,
    phases: object | None = None,
    omegas: object | None = None,
    knm: object | None = None,
    zeta: float = 0.0,
    psi: float = 0.0,
    alpha: object | None = None,
) -> FloatArray:
    """Advance one Doppler-corrected UPDE step.

    Parameters
    ----------
    phases : object | None
        Oscillator phases in radians, shape ``(N,)``.
    omegas : object | None
        Natural frequencies in rad/s, shape ``(N,)``.
    knm : object | None
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    zeta : float
        External drive strength ``ζ``.
    psi : float
        External drive reference phase ``Ψ`` in radians.
    alpha : object | None
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.

    Returns
    -------
    FloatArray
        The phases after one Doppler-corrected UPDE step.
    """
    p = self.phases if phases is None else _validate_phases(phases, n=self._n)
    k = self.k_nm if knm is None else _validate_knm(knm, n=self._n)
    a = self.alpha_matrix if alpha is None else _validate_alpha(alpha, n=self._n)
    omega_source = self._omega_source if omegas is None else omegas
    omega = self._omega_at(omega_source, self._time)
    self.velocity_current = self._velocity_for_step(self._time)
    self._doppler_term = doppler_term(
        self.velocity_current,
        k,
        doppler_strength=self.doppler_strength,
        doppler_epsilon=self.doppler_epsilon,
    )
    out = super().step(p, omega + self._doppler_term, k, zeta, psi, a)
    self.phases = np.ascontiguousarray(out, dtype=np.float64)
    return self.phases.copy()

run

run(
    phases: object | None = None,
    omegas: object | None = None,
    knm: object | None = None,
    zeta: float = 0.0,
    psi: float = 0.0,
    alpha: object | None = None,
    n_steps: int = 1,
) -> FloatArray

Run n_steps of the Doppler-corrected UPDE dynamics.

Parameters

phases : object | None Oscillator phases in radians, shape (N,). omegas : object | None Natural frequencies in rad/s, shape (N,). knm : object | None Coupling matrix K_nm, shape (N, N). zeta : float External drive strength ζ. psi : float External drive reference phase Ψ in radians. alpha : object | None Phase-lag matrix in radians, shape (N, N), or None for no lag. n_steps : int Number of integration steps to run.

Returns

FloatArray The final phases after n_steps Doppler-corrected steps.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def run(
    self,
    phases: object | None = None,
    omegas: object | None = None,
    knm: object | None = None,
    zeta: float = 0.0,
    psi: float = 0.0,
    alpha: object | None = None,
    n_steps: int = 1,
) -> FloatArray:
    """Run ``n_steps`` of the Doppler-corrected UPDE dynamics.

    Parameters
    ----------
    phases : object | None
        Oscillator phases in radians, shape ``(N,)``.
    omegas : object | None
        Natural frequencies in rad/s, shape ``(N,)``.
    knm : object | None
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    zeta : float
        External drive strength ``ζ``.
    psi : float
        External drive reference phase ``Ψ`` in radians.
    alpha : object | None
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    n_steps : int
        Number of integration steps to run.

    Returns
    -------
    FloatArray
        The final phases after ``n_steps`` Doppler-corrected steps.
    """
    steps = _validate_positive_step_count(n_steps, name="n_steps")
    p = self.phases if phases is None else _validate_phases(phases, n=self._n)
    k = self.k_nm if knm is None else _validate_knm(knm, n=self._n)
    a = self.alpha_matrix if alpha is None else _validate_alpha(alpha, n=self._n)
    omega_source = self._omega_source if omegas is None else omegas
    omega_schedule = self._omega_schedule(omega_source, steps)
    velocity_schedule = self._velocity_schedule(steps)
    out = doppler_run(
        p,
        omega_schedule,
        k,
        a,
        velocity_schedule,
        self.doppler_strength,
        self.doppler_epsilon,
        zeta,
        psi,
        self._dt,
        self._method,
        1,
        self._atol,
        self._rtol,
    )
    _effective, terms = _effective_omega_schedule(
        omega_schedule,
        velocity_schedule,
        k,
        doppler_strength=self.doppler_strength,
        doppler_epsilon=self.doppler_epsilon,
    )
    self._omega_current = omega_schedule[-1].copy()
    self.velocity_current = velocity_schedule[-1]
    self._doppler_term = terms[-1]
    self._time += steps * self._dt
    self.phases = np.ascontiguousarray(out, dtype=np.float64)
    return self.phases.copy()

doppler_term

doppler_term(
    velocities: object,
    knm: object,
    *,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1e-09,
    velocity_axis: object | None = None,
) -> FloatArray

Return the graph-weighted Doppler correction for each oscillator.

Parameters

velocities : object Per-oscillator axial velocities, shape (N,). knm : object Coupling matrix K_nm, shape (N, N). doppler_strength : float Doppler coupling-correction strength. doppler_epsilon : float Numerical floor guarding the Doppler denominator. velocity_axis : object | None Optional unit axis along which velocity is projected, or None.

Returns

FloatArray The graph-weighted Doppler correction per oscillator.

Raises

ValueError If the velocity or coupling inputs are non-finite or mismatched.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def doppler_term(
    velocities: object,
    knm: object,
    *,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1.0e-9,
    velocity_axis: object | None = None,
) -> FloatArray:
    """Return the graph-weighted Doppler correction for each oscillator.

    Parameters
    ----------
    velocities : object
        Per-oscillator axial velocities, shape ``(N,)``.
    knm : object
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    doppler_strength : float
        Doppler coupling-correction strength.
    doppler_epsilon : float
        Numerical floor guarding the Doppler denominator.
    velocity_axis : object | None
        Optional unit axis along which velocity is projected, or ``None``.

    Returns
    -------
    FloatArray
        The graph-weighted Doppler correction per oscillator.

    Raises
    ------
    ValueError
        If the velocity or coupling inputs are non-finite or mismatched.
    """
    k_raw = _reject_non_real_array(knm, name="knm")
    if k_raw.ndim != 2 or k_raw.shape[0] != k_raw.shape[1]:
        raise ValueError("knm shape must be (n, n)")
    n = int(k_raw.shape[0])
    k = _validate_knm(k_raw, n=n)
    speed = scalarise_velocities(velocities, n=n, velocity_axis=velocity_axis)
    strength = _finite_float(doppler_strength, name="doppler_strength")
    epsilon = _finite_float(doppler_epsilon, name="doppler_epsilon")
    if epsilon <= 0.0:
        raise ValueError("doppler_epsilon must be positive")

    weights = np.abs(k)
    np.fill_diagonal(weights, 0.0)
    row_mass = weights.sum(axis=1)
    relative = (speed[:, None] - speed[None, :]) / (np.abs(speed)[:, None] + epsilon)
    weighted = np.sum(weights * relative, axis=1)
    term = np.divide(
        weighted, row_mass, out=np.zeros(n, dtype=np.float64), where=row_mass > 0.0
    )
    return np.ascontiguousarray(strength * term, dtype=np.float64)

doppler_run

doppler_run(
    phases: object,
    omega_schedule: object,
    knm: object,
    alpha: object,
    velocity_schedule: object,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1e-09,
    zeta: float = 0.0,
    psi: float = 0.0,
    dt: float = 0.01,
    method: str = "rk45",
    n_substeps: int = 1,
    atol: float = 1e-06,
    rtol: float = 0.001,
    *,
    backend: str = "auto",
) -> FloatArray

Run a Doppler-corrected UPDE schedule through the selected backend.

Parameters

phases : object Oscillator phases in radians, shape (N,). omega_schedule : object Per-step natural-frequency vectors, shape (n_steps, N). knm : object Coupling matrix K_nm, shape (N, N). alpha : object Phase-lag matrix in radians, shape (N, N), or None for no lag. velocity_schedule : object Per-step axial velocity vectors, shape (n_steps, N). doppler_strength : float Doppler coupling-correction strength. doppler_epsilon : float Numerical floor guarding the Doppler denominator. zeta : float External drive strength ζ. psi : float External drive reference phase Ψ in radians. dt : float Integration step size. method : str Integration method (euler, rk4, or rk45). n_substeps : int Number of inner substeps per outer step. atol : float Absolute tolerance for the adaptive (rk45) integrator. rtol : float Relative tolerance for the adaptive (rk45) integrator. backend : str Name of the compute backend to run.

Returns

FloatArray The final phases after running the Doppler schedule on the selected backend.

Raises

ImportError If the selected backend's runtime is unavailable. ValueError If the schedule contract is invalid; the underlying backend error propagates when every backend fails.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def doppler_run(
    phases: object,
    omega_schedule: object,
    knm: object,
    alpha: object,
    velocity_schedule: object,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1.0e-9,
    zeta: float = 0.0,
    psi: float = 0.0,
    dt: float = 0.01,
    method: str = "rk45",
    n_substeps: int = 1,
    atol: float = 1.0e-6,
    rtol: float = 1.0e-3,
    *,
    backend: str = "auto",
) -> FloatArray:
    """Run a Doppler-corrected UPDE schedule through the selected backend.

    Parameters
    ----------
    phases : object
        Oscillator phases in radians, shape ``(N,)``.
    omega_schedule : object
        Per-step natural-frequency vectors, shape ``(n_steps, N)``.
    knm : object
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    alpha : object
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    velocity_schedule : object
        Per-step axial velocity vectors, shape ``(n_steps, N)``.
    doppler_strength : float
        Doppler coupling-correction strength.
    doppler_epsilon : float
        Numerical floor guarding the Doppler denominator.
    zeta : float
        External drive strength ``ζ``.
    psi : float
        External drive reference phase ``Ψ`` in radians.
    dt : float
        Integration step size.
    method : str
        Integration method (``euler``, ``rk4``, or ``rk45``).
    n_substeps : int
        Number of inner substeps per outer step.
    atol : float
        Absolute tolerance for the adaptive (rk45) integrator.
    rtol : float
        Relative tolerance for the adaptive (rk45) integrator.
    backend : str
        Name of the compute backend to run.

    Returns
    -------
    FloatArray
        The final phases after running the Doppler schedule on the selected backend.

    Raises
    ------
    ImportError
        If the selected backend's runtime is unavailable.
    ValueError
        If the schedule contract is invalid; the underlying backend error propagates
        when every backend fails.
    """
    validated = validate_doppler_backend_inputs(
        phases,
        omega_schedule,
        knm,
        alpha,
        velocity_schedule,
        doppler_strength,
        doppler_epsilon,
        zeta,
        psi,
        dt,
        method,
        n_substeps,
        atol,
        rtol,
    )
    (
        p,
        omega,
        k,
        a,
        velocities,
        strength,
        epsilon,
        zeta_f,
        psi_f,
        dt_f,
        _n_steps,
        method_s,
        n_substeps_i,
        atol_f,
        rtol_f,
    ) = validated
    backends = _backend_map()
    if backend != "auto" and backend not in backends:
        raise ImportError(f"Doppler backend {backend!r} is not available")
    order = _BACKEND_ORDER if backend == "auto" else (backend,)
    last_error: Exception | None = None
    for name in order:
        fn = backends.get(name)
        if fn is None:
            continue
        try:
            out = fn(
                p,
                omega,
                k,
                a,
                velocities,
                strength,
                epsilon,
                zeta_f,
                psi_f,
                dt_f,
                method_s,
                n_substeps_i,
                atol_f,
                rtol_f,
            )
            return validate_doppler_backend_output(out, n=int(p.size))
        except (AttributeError, ImportError) as exc:
            last_error = exc
            continue
    if backend != "auto" and last_error is not None:
        raise last_error
    return doppler_run_python(
        p,
        omega,
        k,
        a,
        velocities,
        strength,
        epsilon,
        zeta_f,
        psi_f,
        dt_f,
        method_s,
        n_substeps_i,
        atol_f,
        rtol_f,
    )

API documentation

doppler

Doppler-corrected Kuramoto UPDE integration.

DopplerEngine augments the standard Sakaguchi-Kuramoto UPDE with a relative-velocity correction. For oscillator i the correction is the coupling-graph weighted relative velocity

D_i = s * mean_j(|K_ij| * (v_i - v_j) / (|v_i| + eps))

where the mean is normalised by the active absolute coupling row. Normalising by the row mass keeps the Doppler strength independent of graph degree while still respecting the active coupling topology.

Classes

DopplerEngine

DopplerEngine(
    n: int,
    omega: object,
    k_nm: object,
    alpha: object = 0.0,
    dt: float = 0.01,
    velocities: object
    | Callable[[float], object]
    | None = None,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1e-09,
    solver: str = "rk45",
    phases: object | None = None,
    velocity_axis: object | None = None,
    t0: float = 0.0,
)

Bases: UPDEEngine

Stateful UPDE engine with graph-weighted Doppler velocity correction.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def __init__(
    self,
    n: int,
    omega: object,
    k_nm: object,
    alpha: object = 0.0,
    dt: float = 0.01,
    velocities: object | Callable[[float], object] | None = None,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1.0e-9,
    solver: str = "rk45",
    phases: object | None = None,
    velocity_axis: object | None = None,
    t0: float = 0.0,
) -> None:
    if velocities is None:
        raise ValueError("velocities are required for DopplerEngine")
    super().__init__(n, dt=dt, method=solver, omega=omega, t0=t0)
    self.k_nm = _validate_knm(k_nm, n=n)
    self.alpha_matrix = _validate_alpha(alpha, n=n)
    self._velocity_source = velocities
    self._velocity_axis = velocity_axis
    self.doppler_strength = _finite_float(doppler_strength, name="doppler_strength")
    self.doppler_epsilon = _finite_float(doppler_epsilon, name="doppler_epsilon")
    if self.doppler_epsilon <= 0.0:
        raise ValueError("doppler_epsilon must be positive")
    if phases is None:
        self.phases = np.zeros(n, dtype=np.float64)
    else:
        self.phases = _validate_phases(phases, n=n)
    self.velocity_current = self._velocity_for_step(self.time)
    self._doppler_term = doppler_term(
        self.velocity_current,
        self.k_nm,
        doppler_strength=self.doppler_strength,
        doppler_epsilon=self.doppler_epsilon,
    )
Attributes
doppler_term property
doppler_term: FloatArray

Most recently applied Doppler correction vector.

Returns

FloatArray Most recently applied Doppler correction vector.

Methods:
step
step(
    phases: object | None = None,
    omegas: object | None = None,
    knm: object | None = None,
    zeta: float = 0.0,
    psi: float = 0.0,
    alpha: object | None = None,
) -> FloatArray

Advance one Doppler-corrected UPDE step.

Parameters

phases : object | None Oscillator phases in radians, shape (N,). omegas : object | None Natural frequencies in rad/s, shape (N,). knm : object | None Coupling matrix K_nm, shape (N, N). zeta : float External drive strength ζ. psi : float External drive reference phase Ψ in radians. alpha : object | None Phase-lag matrix in radians, shape (N, N), or None for no lag.

Returns

FloatArray The phases after one Doppler-corrected UPDE step.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def step(
    self,
    phases: object | None = None,
    omegas: object | None = None,
    knm: object | None = None,
    zeta: float = 0.0,
    psi: float = 0.0,
    alpha: object | None = None,
) -> FloatArray:
    """Advance one Doppler-corrected UPDE step.

    Parameters
    ----------
    phases : object | None
        Oscillator phases in radians, shape ``(N,)``.
    omegas : object | None
        Natural frequencies in rad/s, shape ``(N,)``.
    knm : object | None
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    zeta : float
        External drive strength ``ζ``.
    psi : float
        External drive reference phase ``Ψ`` in radians.
    alpha : object | None
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.

    Returns
    -------
    FloatArray
        The phases after one Doppler-corrected UPDE step.
    """
    p = self.phases if phases is None else _validate_phases(phases, n=self._n)
    k = self.k_nm if knm is None else _validate_knm(knm, n=self._n)
    a = self.alpha_matrix if alpha is None else _validate_alpha(alpha, n=self._n)
    omega_source = self._omega_source if omegas is None else omegas
    omega = self._omega_at(omega_source, self._time)
    self.velocity_current = self._velocity_for_step(self._time)
    self._doppler_term = doppler_term(
        self.velocity_current,
        k,
        doppler_strength=self.doppler_strength,
        doppler_epsilon=self.doppler_epsilon,
    )
    out = super().step(p, omega + self._doppler_term, k, zeta, psi, a)
    self.phases = np.ascontiguousarray(out, dtype=np.float64)
    return self.phases.copy()
run
run(
    phases: object | None = None,
    omegas: object | None = None,
    knm: object | None = None,
    zeta: float = 0.0,
    psi: float = 0.0,
    alpha: object | None = None,
    n_steps: int = 1,
) -> FloatArray

Run n_steps of the Doppler-corrected UPDE dynamics.

Parameters

phases : object | None Oscillator phases in radians, shape (N,). omegas : object | None Natural frequencies in rad/s, shape (N,). knm : object | None Coupling matrix K_nm, shape (N, N). zeta : float External drive strength ζ. psi : float External drive reference phase Ψ in radians. alpha : object | None Phase-lag matrix in radians, shape (N, N), or None for no lag. n_steps : int Number of integration steps to run.

Returns

FloatArray The final phases after n_steps Doppler-corrected steps.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def run(
    self,
    phases: object | None = None,
    omegas: object | None = None,
    knm: object | None = None,
    zeta: float = 0.0,
    psi: float = 0.0,
    alpha: object | None = None,
    n_steps: int = 1,
) -> FloatArray:
    """Run ``n_steps`` of the Doppler-corrected UPDE dynamics.

    Parameters
    ----------
    phases : object | None
        Oscillator phases in radians, shape ``(N,)``.
    omegas : object | None
        Natural frequencies in rad/s, shape ``(N,)``.
    knm : object | None
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    zeta : float
        External drive strength ``ζ``.
    psi : float
        External drive reference phase ``Ψ`` in radians.
    alpha : object | None
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    n_steps : int
        Number of integration steps to run.

    Returns
    -------
    FloatArray
        The final phases after ``n_steps`` Doppler-corrected steps.
    """
    steps = _validate_positive_step_count(n_steps, name="n_steps")
    p = self.phases if phases is None else _validate_phases(phases, n=self._n)
    k = self.k_nm if knm is None else _validate_knm(knm, n=self._n)
    a = self.alpha_matrix if alpha is None else _validate_alpha(alpha, n=self._n)
    omega_source = self._omega_source if omegas is None else omegas
    omega_schedule = self._omega_schedule(omega_source, steps)
    velocity_schedule = self._velocity_schedule(steps)
    out = doppler_run(
        p,
        omega_schedule,
        k,
        a,
        velocity_schedule,
        self.doppler_strength,
        self.doppler_epsilon,
        zeta,
        psi,
        self._dt,
        self._method,
        1,
        self._atol,
        self._rtol,
    )
    _effective, terms = _effective_omega_schedule(
        omega_schedule,
        velocity_schedule,
        k,
        doppler_strength=self.doppler_strength,
        doppler_epsilon=self.doppler_epsilon,
    )
    self._omega_current = omega_schedule[-1].copy()
    self.velocity_current = velocity_schedule[-1]
    self._doppler_term = terms[-1]
    self._time += steps * self._dt
    self.phases = np.ascontiguousarray(out, dtype=np.float64)
    return self.phases.copy()

Functions:

scalarise_velocities

scalarise_velocities(
    velocities: object,
    *,
    n: int,
    velocity_axis: object | None = None,
) -> FloatArray

Convert scalar or vector velocities to one signed scalar per oscillator.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def scalarise_velocities(
    velocities: object,
    *,
    n: int,
    velocity_axis: object | None = None,
) -> FloatArray:
    """Convert scalar or vector velocities to one signed scalar per oscillator."""
    arr = _reject_non_real_array(velocities, name="velocities")
    if arr.shape == (n,):
        return arr
    if arr.ndim == 2 and arr.shape[0] == n:
        if velocity_axis is None:
            return np.ascontiguousarray(np.linalg.norm(arr, axis=1), dtype=np.float64)
        axis = _normalise_axis(velocity_axis, dimension=int(arr.shape[1]))
        return np.ascontiguousarray(arr @ axis, dtype=np.float64)
    raise ValueError("velocities must have shape (n,) or (n, d)")

validate_doppler_backend_output

validate_doppler_backend_output(
    value: object, *, n: int
) -> FloatArray

Validate Doppler backend output before returning it to callers.

Parameters

value : object Backend-produced oscillator phases in radians, shape (N,). n : int Expected oscillator count.

Returns

FloatArray Contiguous float64 phase vector in the principal [0, 2*pi) branch.

Raises

ValueError If the backend output is non-finite, has the wrong shape, or leaves the principal phase branch.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def validate_doppler_backend_output(value: object, *, n: int) -> FloatArray:
    """Validate Doppler backend output before returning it to callers.

    Parameters
    ----------
    value : object
        Backend-produced oscillator phases in radians, shape ``(N,)``.
    n : int
        Expected oscillator count.

    Returns
    -------
    FloatArray
        Contiguous ``float64`` phase vector in the principal ``[0, 2*pi)`` branch.

    Raises
    ------
    ValueError
        If the backend output is non-finite, has the wrong shape, or leaves the
        principal phase branch.
    """
    out = _validate_phases(value, n=n)
    if np.any(out < 0.0) or np.any(out >= _TWO_PI):
        raise ValueError("Doppler backend output phases must be in [0, 2*pi)")
    return np.ascontiguousarray(out, dtype=np.float64)

doppler_term

doppler_term(
    velocities: object,
    knm: object,
    *,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1e-09,
    velocity_axis: object | None = None,
) -> FloatArray

Return the graph-weighted Doppler correction for each oscillator.

Parameters

velocities : object Per-oscillator axial velocities, shape (N,). knm : object Coupling matrix K_nm, shape (N, N). doppler_strength : float Doppler coupling-correction strength. doppler_epsilon : float Numerical floor guarding the Doppler denominator. velocity_axis : object | None Optional unit axis along which velocity is projected, or None.

Returns

FloatArray The graph-weighted Doppler correction per oscillator.

Raises

ValueError If the velocity or coupling inputs are non-finite or mismatched.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def doppler_term(
    velocities: object,
    knm: object,
    *,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1.0e-9,
    velocity_axis: object | None = None,
) -> FloatArray:
    """Return the graph-weighted Doppler correction for each oscillator.

    Parameters
    ----------
    velocities : object
        Per-oscillator axial velocities, shape ``(N,)``.
    knm : object
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    doppler_strength : float
        Doppler coupling-correction strength.
    doppler_epsilon : float
        Numerical floor guarding the Doppler denominator.
    velocity_axis : object | None
        Optional unit axis along which velocity is projected, or ``None``.

    Returns
    -------
    FloatArray
        The graph-weighted Doppler correction per oscillator.

    Raises
    ------
    ValueError
        If the velocity or coupling inputs are non-finite or mismatched.
    """
    k_raw = _reject_non_real_array(knm, name="knm")
    if k_raw.ndim != 2 or k_raw.shape[0] != k_raw.shape[1]:
        raise ValueError("knm shape must be (n, n)")
    n = int(k_raw.shape[0])
    k = _validate_knm(k_raw, n=n)
    speed = scalarise_velocities(velocities, n=n, velocity_axis=velocity_axis)
    strength = _finite_float(doppler_strength, name="doppler_strength")
    epsilon = _finite_float(doppler_epsilon, name="doppler_epsilon")
    if epsilon <= 0.0:
        raise ValueError("doppler_epsilon must be positive")

    weights = np.abs(k)
    np.fill_diagonal(weights, 0.0)
    row_mass = weights.sum(axis=1)
    relative = (speed[:, None] - speed[None, :]) / (np.abs(speed)[:, None] + epsilon)
    weighted = np.sum(weights * relative, axis=1)
    term = np.divide(
        weighted, row_mass, out=np.zeros(n, dtype=np.float64), where=row_mass > 0.0
    )
    return np.ascontiguousarray(strength * term, dtype=np.float64)

validate_doppler_backend_inputs

validate_doppler_backend_inputs(
    phases: object,
    omega_schedule: object,
    knm: object,
    alpha: object,
    velocity_schedule: object,
    doppler_strength: float,
    doppler_epsilon: float,
    zeta: float,
    psi: float,
    dt: float,
    method: str,
    n_substeps: int,
    atol: float,
    rtol: float,
) -> tuple[
    FloatArray,
    FloatArray,
    FloatArray,
    FloatArray,
    FloatArray,
    float,
    float,
    float,
    float,
    float,
    int,
    str,
    int,
    float,
    float,
]

Validate the backend-neutral Doppler schedule contract.

Parameters

phases : object Oscillator phases in radians, shape (N,). omega_schedule : object Per-step natural-frequency vectors, shape (n_steps, N). knm : object Coupling matrix K_nm, shape (N, N). alpha : object Phase-lag matrix in radians, shape (N, N), or None for no lag. velocity_schedule : object Per-step axial velocity vectors, shape (n_steps, N). doppler_strength : float Doppler coupling-correction strength. doppler_epsilon : float Numerical floor guarding the Doppler denominator. zeta : float External drive strength ζ. psi : float External drive reference phase Ψ in radians. dt : float Integration step size. method : str Integration method (euler, rk4, or rk45). n_substeps : int Number of inner substeps per outer step. atol : float Absolute tolerance for the adaptive (rk45) integrator. rtol : float Relative tolerance for the adaptive (rk45) integrator.

Returns

tuple[FloatArray, FloatArray, FloatArray, FloatArray, FloatArray, float, float, float, float, float, int, str, int, float, float] The validated, canonicalised Doppler schedule contract tuple.

Raises

ValueError If any schedule array is non-finite or has an inconsistent shape.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def validate_doppler_backend_inputs(
    phases: object,
    omega_schedule: object,
    knm: object,
    alpha: object,
    velocity_schedule: object,
    doppler_strength: float,
    doppler_epsilon: float,
    zeta: float,
    psi: float,
    dt: float,
    method: str,
    n_substeps: int,
    atol: float,
    rtol: float,
) -> tuple[
    FloatArray,
    FloatArray,
    FloatArray,
    FloatArray,
    FloatArray,
    float,
    float,
    float,
    float,
    float,
    int,
    str,
    int,
    float,
    float,
]:
    """Validate the backend-neutral Doppler schedule contract.

    Parameters
    ----------
    phases : object
        Oscillator phases in radians, shape ``(N,)``.
    omega_schedule : object
        Per-step natural-frequency vectors, shape ``(n_steps, N)``.
    knm : object
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    alpha : object
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    velocity_schedule : object
        Per-step axial velocity vectors, shape ``(n_steps, N)``.
    doppler_strength : float
        Doppler coupling-correction strength.
    doppler_epsilon : float
        Numerical floor guarding the Doppler denominator.
    zeta : float
        External drive strength ``ζ``.
    psi : float
        External drive reference phase ``Ψ`` in radians.
    dt : float
        Integration step size.
    method : str
        Integration method (``euler``, ``rk4``, or ``rk45``).
    n_substeps : int
        Number of inner substeps per outer step.
    atol : float
        Absolute tolerance for the adaptive (rk45) integrator.
    rtol : float
        Relative tolerance for the adaptive (rk45) integrator.

    Returns
    -------
    tuple[FloatArray, FloatArray, FloatArray, FloatArray, FloatArray, float, float,
    float, float, float, int, str, int, float, float]
        The validated, canonicalised Doppler schedule contract tuple.

    Raises
    ------
    ValueError
        If any schedule array is non-finite or has an inconsistent shape.
    """
    p_raw = _reject_non_real_array(phases, name="phases")
    if p_raw.ndim != 1 or p_raw.size < 1:
        raise ValueError("phases must be a non-empty vector")
    n = int(p_raw.size)
    p = _validate_phases(p_raw, n=n)
    omega = _validate_schedule(omega_schedule, n=n, name="omega_schedule")
    velocities = _validate_schedule(velocity_schedule, n=n, name="velocity_schedule")
    if velocities.shape[0] != omega.shape[0]:
        raise ValueError("velocity_schedule step count must match omega_schedule")
    k = _validate_knm(knm, n=n)
    a = _validate_alpha(alpha, n=n)
    strength = _finite_float(doppler_strength, name="doppler_strength")
    epsilon = _finite_float(doppler_epsilon, name="doppler_epsilon")
    if epsilon <= 0.0:
        raise ValueError("doppler_epsilon must be positive")
    zeta_f = _finite_float(zeta, name="zeta")
    psi_f = _finite_float(psi, name="psi")
    dt_f = _finite_float(dt, name="dt")
    if dt_f <= 0.0:
        raise ValueError("dt must be positive")
    method_s = _validate_method(method)
    n_substeps_i = _validate_positive_step_count(n_substeps, name="n_substeps")
    atol_f = _finite_float(atol, name="atol")
    rtol_f = _finite_float(rtol, name="rtol")
    if atol_f <= 0.0 or rtol_f <= 0.0:
        raise ValueError("atol and rtol must be positive")
    return (
        p,
        omega,
        k,
        a,
        velocities,
        strength,
        epsilon,
        zeta_f,
        psi_f,
        dt_f,
        int(omega.shape[0]),
        method_s,
        n_substeps_i,
        atol_f,
        rtol_f,
    )

doppler_run_python

doppler_run_python(
    phases: object,
    omega_schedule: object,
    knm: object,
    alpha: object,
    velocity_schedule: object,
    doppler_strength: float,
    doppler_epsilon: float,
    zeta: float,
    psi: float,
    dt: float,
    method: str = "rk45",
    n_substeps: int = 1,
    atol: float = 1e-06,
    rtol: float = 0.001,
) -> FloatArray

Run the Doppler-corrected UPDE schedule in the Python reference path.

Parameters

phases : object Oscillator phases in radians, shape (N,). omega_schedule : object Per-step natural-frequency vectors, shape (n_steps, N). knm : object Coupling matrix K_nm, shape (N, N). alpha : object Phase-lag matrix in radians, shape (N, N), or None for no lag. velocity_schedule : object Per-step axial velocity vectors, shape (n_steps, N). doppler_strength : float Doppler coupling-correction strength. doppler_epsilon : float Numerical floor guarding the Doppler denominator. zeta : float External drive strength ζ. psi : float External drive reference phase Ψ in radians. dt : float Integration step size. method : str Integration method (euler, rk4, or rk45). n_substeps : int Number of inner substeps per outer step. atol : float Absolute tolerance for the adaptive (rk45) integrator. rtol : float Relative tolerance for the adaptive (rk45) integrator.

Returns

FloatArray The final phases after running the Doppler schedule on the Python path.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def doppler_run_python(
    phases: object,
    omega_schedule: object,
    knm: object,
    alpha: object,
    velocity_schedule: object,
    doppler_strength: float,
    doppler_epsilon: float,
    zeta: float,
    psi: float,
    dt: float,
    method: str = "rk45",
    n_substeps: int = 1,
    atol: float = 1.0e-6,
    rtol: float = 1.0e-3,
) -> FloatArray:
    """Run the Doppler-corrected UPDE schedule in the Python reference path.

    Parameters
    ----------
    phases : object
        Oscillator phases in radians, shape ``(N,)``.
    omega_schedule : object
        Per-step natural-frequency vectors, shape ``(n_steps, N)``.
    knm : object
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    alpha : object
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    velocity_schedule : object
        Per-step axial velocity vectors, shape ``(n_steps, N)``.
    doppler_strength : float
        Doppler coupling-correction strength.
    doppler_epsilon : float
        Numerical floor guarding the Doppler denominator.
    zeta : float
        External drive strength ``ζ``.
    psi : float
        External drive reference phase ``Ψ`` in radians.
    dt : float
        Integration step size.
    method : str
        Integration method (``euler``, ``rk4``, or ``rk45``).
    n_substeps : int
        Number of inner substeps per outer step.
    atol : float
        Absolute tolerance for the adaptive (rk45) integrator.
    rtol : float
        Relative tolerance for the adaptive (rk45) integrator.

    Returns
    -------
    FloatArray
        The final phases after running the Doppler schedule on the Python path.
    """
    (
        p,
        omega,
        k,
        a,
        velocities,
        strength,
        epsilon,
        zeta_f,
        psi_f,
        dt_f,
        _n_steps,
        method_s,
        n_substeps_i,
        atol_f,
        rtol_f,
    ) = validate_doppler_backend_inputs(
        phases,
        omega_schedule,
        knm,
        alpha,
        velocity_schedule,
        doppler_strength,
        doppler_epsilon,
        zeta,
        psi,
        dt,
        method,
        n_substeps,
        atol,
        rtol,
    )
    effective, _terms = _effective_omega_schedule(
        omega,
        velocities,
        k,
        doppler_strength=strength,
        doppler_epsilon=epsilon,
    )
    return upde_run_omega_schedule_python(
        p,
        effective,
        k,
        a,
        zeta_f,
        psi_f,
        dt_f,
        method_s,
        n_substeps_i,
        atol_f,
        rtol_f,
    )

doppler_run

doppler_run(
    phases: object,
    omega_schedule: object,
    knm: object,
    alpha: object,
    velocity_schedule: object,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1e-09,
    zeta: float = 0.0,
    psi: float = 0.0,
    dt: float = 0.01,
    method: str = "rk45",
    n_substeps: int = 1,
    atol: float = 1e-06,
    rtol: float = 0.001,
    *,
    backend: str = "auto",
) -> FloatArray

Run a Doppler-corrected UPDE schedule through the selected backend.

Parameters

phases : object Oscillator phases in radians, shape (N,). omega_schedule : object Per-step natural-frequency vectors, shape (n_steps, N). knm : object Coupling matrix K_nm, shape (N, N). alpha : object Phase-lag matrix in radians, shape (N, N), or None for no lag. velocity_schedule : object Per-step axial velocity vectors, shape (n_steps, N). doppler_strength : float Doppler coupling-correction strength. doppler_epsilon : float Numerical floor guarding the Doppler denominator. zeta : float External drive strength ζ. psi : float External drive reference phase Ψ in radians. dt : float Integration step size. method : str Integration method (euler, rk4, or rk45). n_substeps : int Number of inner substeps per outer step. atol : float Absolute tolerance for the adaptive (rk45) integrator. rtol : float Relative tolerance for the adaptive (rk45) integrator. backend : str Name of the compute backend to run.

Returns

FloatArray The final phases after running the Doppler schedule on the selected backend.

Raises

ImportError If the selected backend's runtime is unavailable. ValueError If the schedule contract is invalid; the underlying backend error propagates when every backend fails.

Source code in src/scpn_phase_orchestrator/upde/doppler.py
def doppler_run(
    phases: object,
    omega_schedule: object,
    knm: object,
    alpha: object,
    velocity_schedule: object,
    doppler_strength: float = 1.0,
    doppler_epsilon: float = 1.0e-9,
    zeta: float = 0.0,
    psi: float = 0.0,
    dt: float = 0.01,
    method: str = "rk45",
    n_substeps: int = 1,
    atol: float = 1.0e-6,
    rtol: float = 1.0e-3,
    *,
    backend: str = "auto",
) -> FloatArray:
    """Run a Doppler-corrected UPDE schedule through the selected backend.

    Parameters
    ----------
    phases : object
        Oscillator phases in radians, shape ``(N,)``.
    omega_schedule : object
        Per-step natural-frequency vectors, shape ``(n_steps, N)``.
    knm : object
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    alpha : object
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    velocity_schedule : object
        Per-step axial velocity vectors, shape ``(n_steps, N)``.
    doppler_strength : float
        Doppler coupling-correction strength.
    doppler_epsilon : float
        Numerical floor guarding the Doppler denominator.
    zeta : float
        External drive strength ``ζ``.
    psi : float
        External drive reference phase ``Ψ`` in radians.
    dt : float
        Integration step size.
    method : str
        Integration method (``euler``, ``rk4``, or ``rk45``).
    n_substeps : int
        Number of inner substeps per outer step.
    atol : float
        Absolute tolerance for the adaptive (rk45) integrator.
    rtol : float
        Relative tolerance for the adaptive (rk45) integrator.
    backend : str
        Name of the compute backend to run.

    Returns
    -------
    FloatArray
        The final phases after running the Doppler schedule on the selected backend.

    Raises
    ------
    ImportError
        If the selected backend's runtime is unavailable.
    ValueError
        If the schedule contract is invalid; the underlying backend error propagates
        when every backend fails.
    """
    validated = validate_doppler_backend_inputs(
        phases,
        omega_schedule,
        knm,
        alpha,
        velocity_schedule,
        doppler_strength,
        doppler_epsilon,
        zeta,
        psi,
        dt,
        method,
        n_substeps,
        atol,
        rtol,
    )
    (
        p,
        omega,
        k,
        a,
        velocities,
        strength,
        epsilon,
        zeta_f,
        psi_f,
        dt_f,
        _n_steps,
        method_s,
        n_substeps_i,
        atol_f,
        rtol_f,
    ) = validated
    backends = _backend_map()
    if backend != "auto" and backend not in backends:
        raise ImportError(f"Doppler backend {backend!r} is not available")
    order = _BACKEND_ORDER if backend == "auto" else (backend,)
    last_error: Exception | None = None
    for name in order:
        fn = backends.get(name)
        if fn is None:
            continue
        try:
            out = fn(
                p,
                omega,
                k,
                a,
                velocities,
                strength,
                epsilon,
                zeta_f,
                psi_f,
                dt_f,
                method_s,
                n_substeps_i,
                atol_f,
                rtol_f,
            )
            return validate_doppler_backend_output(out, n=int(p.size))
        except (AttributeError, ImportError) as exc:
            last_error = exc
            continue
    if backend != "auto" and last_error is not None:
        raise last_error
    return doppler_run_python(
        p,
        omega,
        k,
        a,
        velocities,
        strength,
        epsilon,
        zeta_f,
        psi_f,
        dt_f,
        method_s,
        n_substeps_i,
        atol_f,
        rtol_f,
    )