Skip to content

Add packet error rate evaluation (PER) and SIGINT handling to BER mode #194

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sneakers-the-rat merged 1 commit into from
May 6, 2026
Merged

Add packet error rate evaluation (PER) and SIGINT handling to BER mode #194

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
163 changes: 96 additions & 67 deletions mio/devices/stream/ber.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,8 @@ def prbs15_ber(
header_cls: type[StreamBufferHeader] = StreamBufferHeader,
) -> dict[str, float | int]:
"""
Measure bit-error-rate (BER) on the communication link using PRBS-15
(pseudo-random binary sequence; standard pattern for link tests).
Measure bit-error-rate (BER) and packet-error-rate (PER) on the communication link
using PRBS-15 (pseudo-random binary sequence; standard pattern for link tests).

Unlike typical continuous-stream BER tests, this preserves the buffer framing of
normal image capture and substitutes PRBS-15 for the pixel payload, so the
Expand All @@ -30,11 +30,19 @@ def prbs15_ber(
(dummies, merged-buffer tails) are excluded so they don't inflate the count.
Errors and bits accumulate across up to ``n_buffers`` buffers.

PER is computed from the buffer_count range observed: any gap between the first
and last buffer_count (beyond what was received) is treated as a dropped packet,
and any received buffer with non-zero errors counts as an errored packet.

KeyboardInterrupt during the run returns the partial result with ``aborted``
set to ``"interrupted"``.

Returns
-------
dict
Run summary: ``buffers`` received, ``bits`` and ``errors`` compared,
cumulative ``ber``, ``buffer_count`` range, and per-window snapshots.
cumulative ``ber``, ``per`` with packet counts, ``buffer_count`` range,
per-window snapshots, and ``aborted`` reason if the run did not complete.
"""

def prbs15_bytes(n: int, seed: int) -> bytes:
Expand All @@ -51,6 +59,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
Expand All @@ -59,78 +68,98 @@ 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

logger.info("BER capture starting, target=%d buffers", n_buffers)

for buf in exact_iter(serial_buffer_queue.get, None):
header_data, payload_u8 = header_cls.from_buffer(buf, config)
if payload_u8.size == 0:
continue

# Trim to pixel_count; bytes beyond are dummies or random artifacts, not PRBS.
n_prbs = header_data.pixel_count
if n_prbs <= 0 or n_prbs > payload_u8.size:
continue
payload_u8 = payload_u8[:n_prbs]

# buffer_count seeds the PRBS.
buffer_count = header_data.buffer_count
seed = (buffer_count & 0x7FFF) or 1
exp = np.frombuffer(prbs15_bytes(payload_u8.size, seed), dtype=np.uint8)

diff = np.bitwise_xor(payload_u8, exp)
errors = int(np.unpackbits(diff).sum())
bits = payload_u8.size * 8

total_errors += errors
total_bits += bits
window_errors += errors
window_bits += bits
got += 1
if first_buffer_count is None:
first_buffer_count = buffer_count
if window_first_buffer_count is None:
window_first_buffer_count = buffer_count
last_buffer_count = buffer_count

if got % log_every == 0:
running = total_errors / total_bits if total_bits else float("nan")
window_ber = window_errors / window_bits if window_bits else float("nan")
logger.info(
"BER progress: %d/%d buffers, bits=%d errors=%d "
"cumulative_ber=%.6g window_ber=%.6g",
got,
n_buffers,
total_bits,
total_errors,
running,
window_ber,
)
windows.append(
{
"buffer_index_end": got,
"buffer_count_start": window_first_buffer_count,
"buffer_count_end": buffer_count,
"bits": window_bits,
"errors": window_errors,
"window_ber": window_ber,
"cumulative_ber": running,
}
)
window_errors = 0
window_bits = 0
window_first_buffer_count = None

if got >= n_buffers:
break
aborted: str | None = None

logger.info(f"BER capture starting, target={n_buffers} buffers")

try:
for buf in exact_iter(serial_buffer_queue.get, None):
header_data, payload_u8 = header_cls.from_buffer(buf, config)
if payload_u8.size == 0:
continue

# Trim to pixel_count; bytes beyond are dummies or random artifacts, not PRBS.
n_prbs = header_data.pixel_count
if n_prbs <= 0 or n_prbs > payload_u8.size:
continue
payload_u8 = payload_u8[:n_prbs]

# buffer_count seeds the PRBS.
buffer_count = header_data.buffer_count
seed = (buffer_count & 0x7FFF) or 1
exp = np.frombuffer(prbs15_bytes(payload_u8.size, seed), dtype=np.uint8)

diff = np.bitwise_xor(payload_u8, exp)
errors = int(np.unpackbits(diff).sum())
bits = payload_u8.size * 8

total_errors += errors
total_bits += bits
if errors > 0:
total_errored_buffers += 1
window_errors += errors
window_bits += bits
got += 1
if first_buffer_count is None:
first_buffer_count = buffer_count
if window_first_buffer_count is None:
window_first_buffer_count = buffer_count
last_buffer_count = buffer_count

if got % log_every == 0:
running = total_errors / total_bits if total_bits else float("nan")
window_ber = window_errors / window_bits if window_bits else float("nan")
logger.info(
f"BER progress: {got}/{n_buffers} buffers, "
f"bits={total_bits} errors={total_errors} "
f"cumulative_ber={running:.6g} window_ber={window_ber:.6g}"
)
windows.append(
{
"buffer_index_end": got,
"buffer_count_start": window_first_buffer_count,
"buffer_count_end": buffer_count,
"bits": window_bits,
"errors": window_errors,
"window_ber": window_ber,
"cumulative_ber": running,
}
)
window_errors = 0
window_bits = 0
window_first_buffer_count = None

if got >= n_buffers:
break
except KeyboardInterrupt:
logger.warning(f"BER capture interrupted by user (got {got}/{n_buffers} buffers)")
aborted = "interrupted"

ber = (total_errors / total_bits) if total_bits else float("nan")
if first_buffer_count is not None and last_buffer_count is not None and got > 0:
expected_buffers = last_buffer_count - first_buffer_count + 1
dropped_buffers = max(expected_buffers - got, 0)
per = (
(dropped_buffers + total_errored_buffers) / expected_buffers
if expected_buffers
else float("nan")
)
else:
expected_buffers = 0
dropped_buffers = 0
per = float("nan")

return {
"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,
"expected_buffers": expected_buffers,
"dropped_buffers": dropped_buffers,
"per": per,
"aborted": aborted,
"windows": windows,
}
21 changes: 14 additions & 7 deletions mio/devices/stream/device.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,28 +87,35 @@ def _ber_mode(
result = prbs15_ber(
serial_buffer_queue, self.config, self.logger, target_buffers, self.header_cls
)
aborted = result["aborted"]
status = f"aborted ({aborted})" if aborted else "complete"
self.logger.info(
"BER test complete: buffers=%d bits=%d errors=%d ber=%.6g",
result["buffers"],
result["bits"],
result["errors"],
result["ber"],
f"BER test {status}: "
f"buffers={result['buffers']}/{result['expected_buffers']} "
f"dropped={result['dropped_buffers']} errored={result['errored_buffers']} "
f"bits={result['bits']} errors={result['errors']} "
f"ber={result['ber']:.6g} per={result['per']:.6g}"
)
if ber_output:
summary = {
"prbs": "PRBS-15 (x^15+x^14+1, MSB-first), " "seed=(buffer_count & 0x7FFF) or 1",
"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"],
"expected_buffers": result["expected_buffers"],
"dropped_buffers": result["dropped_buffers"],
"errored_buffers": result["errored_buffers"],
"bits": result["bits"],
"errors": result["errors"],
"ber": result["ber"],
"per": result["per"],
"aborted": aborted,
"windows": result["windows"],
}
with open(ber_output, "w") as f:
json.dump(summary, f, indent=2, default=float)
self.logger.info("BER results written to %s", ber_output)
self.logger.info(f"BER results written to {ber_output}")

def capture(
self,
Expand Down
Loading