PER measurement

mio/devices/stream/device.py

@@ -303,7 +303,10 @@ def _parse_header(self, buffer: bytes) -> tuple[StreamBufferHeader, np.ndarray]:
         return header_data, payload
 
     def prbs15_ber(
-        self, serial_buffer_queue: multiprocessing.Queue, n_buffers: int = 100
+        self,
+        serial_buffer_queue: multiprocessing.Queue,
+        n_buffers: int = 100,
+        timeout_s: float | None = 10.0,
     ) -> dict[str, float | int]:
         """
         Measure bit-error-rate (BER) on the communication link using PRBS-15
@@ -340,6 +343,7 @@ def prbs15_bytes(n: int, seed: int) -> bytes:
 
         total_bits = 0
         total_errors = 0
+        total_errored_buffers = 0
         window_bits = 0
         window_errors = 0
         got = 0
@@ -348,10 +352,42 @@ def prbs15_bytes(n: int, seed: int) -> bytes:
         first_buffer_count: int | None = None
         last_buffer_count: int | None = None
         window_first_buffer_count: int | None = None
+        aborted: str | None = None
 
-        self.logger.info("BER capture starting, target=%d buffers", n_buffers)
+        self.logger.info(
+            "BER capture starting, target=%d buffers, timeout=%s",
+            n_buffers,
+            f"{timeout_s:.1f}s" if timeout_s is not None else "none",
+        )
 
-        for buf in exact_iter(serial_buffer_queue.get, None):
+        while True:
+            try:
+                buf = (
+                    serial_buffer_queue.get(timeout=timeout_s)
+                    if timeout_s is not None
+                    else serial_buffer_queue.get()
+                )
+            except queue.Empty:
+                self.logger.warning(
+                    "BER capture timed out after %.1fs with no buffer received "
+                    "(got %d/%d buffers)",
+                    timeout_s,
+                    got,
+                    n_buffers,
+                )
+                aborted = "timeout"
+                break
+            except KeyboardInterrupt:
+                self.logger.warning(
+                    "BER capture interrupted by user (got %d/%d buffers)",
+                    got,
+                    n_buffers,
+                )
+                aborted = "interrupted"
+                self.terminate.set()
+                break
+            if buf is None:
+                break
             header_data, payload_u8 = self._parse_header(buf)
             if payload_u8.size == 0:
                 continue
@@ -373,6 +409,8 @@ def prbs15_bytes(n: int, seed: int) -> bytes:
 
             total_errors += errors
             total_bits += bits
+            if errors > 0:
+                total_errored_buffers += 1
             window_errors += errors
             window_bits += bits
             got += 1
@@ -418,10 +456,12 @@ def prbs15_bytes(n: int, seed: int) -> bytes:
             "buffers": got,
             "bits": total_bits,
             "errors": total_errors,
+            "errored_buffers": total_errored_buffers,
             "ber": ber,
             "buffer_count_start": first_buffer_count,
             "buffer_count_end": last_buffer_count,
             "windows": windows,
+            "aborted": aborted,
         }
 
     def _ber_mode(
@@ -435,23 +475,56 @@ def _ber_mode(
         """
         target_buffers = self.config.runtime.ber_test_n_buffers
         result = self.prbs15_ber(serial_buffer_queue, target_buffers)
+
+        buffers_received = result["buffers"]
+        bits = result["bits"]
+        errors = result["errors"]
+        errored_buffers = result["errored_buffers"]
+        buf_start = result["buffer_count_start"]
+        buf_end = result["buffer_count_end"]
+        aborted = result["aborted"]
+
+        if buf_start is not None and buf_end is not None and buffers_received > 0:
+            expected_buffers = buf_end - buf_start + 1
+            dropped_buffers = max(expected_buffers - buffers_received, 0)
+            per = (
+                (dropped_buffers + errored_buffers) / expected_buffers
+                if expected_buffers
+                else float("nan")
+            )
+        else:
+            expected_buffers = 0
+            dropped_buffers = 0
+            per = float("nan")
+
         self.logger.info(
-            "BER test complete: buffers=%d bits=%d errors=%d ber=%.6g",
-            result["buffers"],
-            result["bits"],
-            result["errors"],
+            "BER test %s: buffers=%d/%d dropped=%d errored=%d "
+            "bits=%d errors=%d ber=%.6g per=%.6g",
+            f"aborted ({aborted})" if aborted else "complete",
+            buffers_received,
+            expected_buffers,
+            dropped_buffers,
+            errored_buffers,
+            bits,
+            errors,
             result["ber"],
+            per,
         )
         if ber_output:
             summary = {
                 "prbs": "PRBS-15 (x^15+x^14+1, MSB-first), " "seed=(buffer_count & 0x7FFF) or 1",
                 "target_buffers": target_buffers,
-                "buffers_received": result["buffers"],
-                "buffer_count_start": result["buffer_count_start"],
-                "buffer_count_end": result["buffer_count_end"],
-                "bits": result["bits"],
-                "errors": result["errors"],
+                "buffers_received": buffers_received,
+                "buffer_count_start": buf_start,
+                "buffer_count_end": buf_end,
+                "expected_buffers": expected_buffers,
+                "dropped_buffers": dropped_buffers,
+                "errored_buffers": errored_buffers,
+                "bits": bits,
+                "errors": errors,
                 "ber": result["ber"],
+                "per": per,
+                "aborted": aborted,
                 "windows": result["windows"],
             }
             with open(ber_output, "w") as f:

tests/test_devices/test_stream.py

@@ -345,6 +345,11 @@ def test_ber_measurement(tmp_path, set_okdev_input):
     assert result["ber"] == pytest.approx(0.2936758910, rel=1e-9)
     assert result["buffer_count_start"] == 207576
     assert result["buffer_count_end"] == 207695
+    assert result["expected_buffers"] == 120
+    assert result["dropped_buffers"] == 0
+    assert result["errored_buffers"] == 120
+    assert result["per"] == pytest.approx(1.0, rel=1e-12)
+    assert result["aborted"] is None
 
     # Mid-run window (logged every 100 buffers)
     assert len(result["windows"]) == 1