harden USB telemetry transport

tools/trikke_protocol.py

@@ -84,6 +84,75 @@ class Frame:
     raw: bytes
 
 
+@dataclass
+class IntegrityTracker:
+    packet_gap_count: int = 0
+    packet_reset_count: int = 0
+    sample_gap_count: int = 0
+    sample_reset_count: int = 0
+    timing_anomaly_count: int = 0
+    timestamp_saturation_frame_count: int = 0
+    accel_stale_count: int = 0
+    accel_overrun_count: int = 0
+    gyro_stale_count: int = 0
+    gyro_overrun_count: int = 0
+    final_dropped_sample_count: int = 0
+    final_loop_overrun_count: int = 0
+    _previous_packet_sequence: int | None = None
+    _previous_sample_sequence: int | None = None
+    _previous_timestamp_us: int | None = None
+
+    @staticmethod
+    def _classify_sequence(
+        previous: int,
+        current: int,
+    ) -> tuple[int, int]:
+        expected = (previous + 1) & 0xFFFFFFFF
+        forward_distance = (current - expected) & 0xFFFFFFFF
+        if forward_distance == 0:
+            return (0, 0)
+        if forward_distance < 0x80000000:
+            return (forward_distance, 0)
+        return (0, 1)
+
+    def observe(self, frame: Frame) -> None:
+        if self._previous_packet_sequence is not None:
+            gaps, resets = self._classify_sequence(
+                self._previous_packet_sequence, frame.packet_sequence
+            )
+            self.packet_gap_count += gaps
+            self.packet_reset_count += resets
+        self._previous_packet_sequence = frame.packet_sequence
+
+        if frame.flags & PACKET_FLAG_TIMESTAMP_DELTA_SATURATED:
+            self.timestamp_saturation_frame_count += 1
+        self.final_dropped_sample_count = frame.dropped_sample_count
+        self.final_loop_overrun_count = frame.loop_overrun_count
+
+        for sample in frame.samples:
+            if self._previous_sample_sequence is not None:
+                gaps, resets = self._classify_sequence(
+                    self._previous_sample_sequence, sample.sequence
+                )
+                self.sample_gap_count += gaps
+                self.sample_reset_count += resets
+            if (
+                self._previous_timestamp_us is not None
+                and sample.timestamp_us - self._previous_timestamp_us != 10_000
+            ):
+                self.timing_anomaly_count += 1
+            if not sample.accel_status & 0x80:
+                self.accel_stale_count += 1
+            if sample.accel_status & 0x01:
+                self.accel_overrun_count += 1
+            if not sample.gyro_status & 0x08:
+                self.gyro_stale_count += 1
+            if sample.gyro_status & 0x80:
+                self.gyro_overrun_count += 1
+            self._previous_sample_sequence = sample.sequence
+            self._previous_timestamp_us = sample.timestamp_us
+
+
 def _lround(value: float) -> int:
     """Match C lroundf: nearest integer, halfway cases away from zero."""
     return math.floor(value + 0.5) if value >= 0 else math.ceil(value - 0.5)