test(pps): analyze loop stability and telemetry

soc/software/pps_python/adaptive_pps.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 from dataclasses import dataclass
 from enum import IntEnum
 import math
-from typing import Iterable
+from typing import Iterable, Mapping
 
 
 TRACKING_HZ = 125_000_000
@@ -242,17 +242,7 @@ class AdaptiveTrace:
 
     def one_second_edge_jitter_rms(self, *, after_s: float = 0.0) -> float:
         """Return 1 s PPS-edge jitter including the 125 MHz output grid."""
-        selected = {
-            round(time_s, 9): (
-                round(-phase_s * TRACKING_HZ) / TRACKING_HZ
-            )
-            for time_s, phase_s, locked in zip(
-                self.time_s, self.phase_error_s, self.locked
-            )
-            if locked
-            and time_s >= after_s
-            and abs(time_s - round(time_s)) < 1.0e-9
-        }
+        selected = dict(self.integer_second_edge_errors(after_s=after_s))
         differences = [
             edge_error - selected[round(time_s - 1.0, 9)]
             for time_s, edge_error in selected.items()
@@ -264,6 +254,31 @@ class AdaptiveTrace:
             sum(value * value for value in differences) / len(differences)
         )
 
+    def integer_second_edge_errors(
+        self,
+        *,
+        after_s: float = 0.0,
+        locked_only: bool = True,
+    ) -> list[tuple[float, float]]:
+        """Return physical PPS edge error samples on the 125 MHz grid.
+
+        Period jitter alone hides low-frequency motion.  Exposing the signed
+        edge error lets stability analyses report absolute RMS, peak-to-peak
+        wander and moving-average wander alongside cycle-to-cycle jitter.
+        """
+        return [
+            (
+                round(time_s, 9),
+                round(-phase_s * TRACKING_HZ) / TRACKING_HZ,
+            )
+            for time_s, phase_s, locked in zip(
+                self.time_s, self.phase_error_s, self.locked
+            )
+            if (locked or not locked_only)
+            and time_s >= after_s
+            and abs(time_s - round(time_s)) < 1.0e-9
+        ]
+
     def phase_rms(self, *, after_s: float = 0.0) -> float:
         values = [
             phase_s
@@ -319,6 +334,10 @@ class FixedPointAdaptivePps:
         software_filter: bool = True,
         receiver_clock_bias_m: float = 1200.0,
         receiver_clock_drift_mps: float = 475.0,
+        profiles: Mapping[Profile, LoopProfile] | None = None,
+        acquire_profile: LoopProfile | None = None,
+        software_drift_gain: float = 1.0 / 4.0,
+        software_fallback_drift_gain: float = 1.0 / 16.0,
     ) -> None:
         self.update_period_s = update_period_s
         self.oscillator_fraction = oscillator_ppm * 1.0e-6
@@ -326,9 +345,24 @@ class FixedPointAdaptivePps:
         self.software_filter = software_filter
         self.receiver_clock_bias_m = receiver_clock_bias_m
         self.receiver_clock_drift_mps = receiver_clock_drift_mps
+        self.profiles = dict(PROFILES if profiles is None else profiles)
+        missing_profiles = set(Profile) - self.profiles.keys()
+        if missing_profiles:
+            raise ValueError(f"missing loop profiles: {sorted(missing_profiles)}")
+        self.acquire_profile = (
+            ACQUIRE_PROFILE if acquire_profile is None else acquire_profile
+        )
+        if not 0.0 < software_drift_gain <= 1.0:
+            raise ValueError("software_drift_gain must be in (0, 1]")
+        if not 0.0 < software_fallback_drift_gain <= 1.0:
+            raise ValueError(
+                "software_fallback_drift_gain must be in (0, 1]"
+            )
+        self.software_drift_gain = software_drift_gain
+        self.software_fallback_drift_gain = software_fallback_drift_gain
         self.selector = QualitySelector()
         self.profile = Profile.HOLDOVER if adaptive else Profile.PRECISION
-        self.parameters = PROFILES[self.profile]
+        self.parameters = self.profiles[self.profile]
         self.kp_q46, self.ki_q46, self.max_slew_q32 = (
             self.parameters.fixed_coefficients(update_period_s)
         )
@@ -376,10 +410,10 @@ class FixedPointAdaptivePps:
 
     def _select_loop_parameters(self) -> None:
         parameters = (
-            ACQUIRE_PROFILE
+            self.acquire_profile
             if self.adaptive and
                self.profile != Profile.HOLDOVER and not self.locked
-            else PROFILES[self.profile]
+            else self.profiles[self.profile]
         )
         if parameters == self.parameters:
             return
@@ -541,15 +575,16 @@ class FixedPointAdaptivePps:
             + self.parameters.software_bias_gain * innovation_m
         )
         if observation.clock_drift_valid:
-            self.software_clock_drift_mps += 0.25 * (
+            self.software_clock_drift_mps += self.software_drift_gain * (
                 measured_drift_mps - self.software_clock_drift_mps
             )
         else:
             if self.software_updates < self.parameters.software_settle_updates:
                 self.software_clock_drift_mps = fallback_drift_mps
             else:
-                self.software_clock_drift_mps += (1.0 / 16.0) * (
-                    fallback_drift_mps - self.software_clock_drift_mps
+                self.software_clock_drift_mps += (
+                    self.software_fallback_drift_gain *
+                    (fallback_drift_mps - self.software_clock_drift_mps)
                 )
         self.software_previous_measurement_m = measured_bias_m
         self.software_error_s = -(
@@ -705,6 +740,10 @@ def run_observations(
     software_filter: bool = True,
     receiver_clock_bias_m: float = 1200.0,
     receiver_clock_drift_mps: float = 475.0,
+    profiles: Mapping[Profile, LoopProfile] | None = None,
+    acquire_profile: LoopProfile | None = None,
+    software_drift_gain: float = 1.0 / 4.0,
+    software_fallback_drift_gain: float = 1.0 / 16.0,
 ) -> AdaptiveTrace:
     model = FixedPointAdaptivePps(
         oscillator_ppm=oscillator_ppm,
@@ -712,6 +751,10 @@ def run_observations(
         software_filter=software_filter,
         receiver_clock_bias_m=receiver_clock_bias_m,
         receiver_clock_drift_mps=receiver_clock_drift_mps,
+        profiles=profiles,
+        acquire_profile=acquire_profile,
+        software_drift_gain=software_drift_gain,
+        software_fallback_drift_gain=software_fallback_drift_gain,
     )
     for observation in observations:
         model.update(observation)