K-LD7 Timing and PRF Probe Implementation Plan¶

import importlib.util
from pathlib import Path

SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "hardware-test" / "probe_kld7_timing.py"
spec = importlib.util.spec_from_file_location("probe_kld7_timing", SCRIPT)
probe = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(probe)


def test_build_packet_uppercases_command_and_packs_length():
    packet = probe.build_packet("gnfd", (0x21).to_bytes(4, "little"))

    assert packet == b"GNFD\x04\x00\x00\x00!\x00\x00\x00"


def test_validate_command_rejects_non_four_byte_command():
    assert probe.validate_probe_command("ABC", "") == "command must be exactly 4 ASCII characters"


def test_validate_command_rejects_non_uppercase_command():
    assert probe.validate_probe_command("test", "") == "command must be uppercase ASCII"


def test_validate_command_rejects_odd_hex_payload():
    assert probe.validate_probe_command("TST1", "abc") == "hex payload must have an even number of characters"

#!/usr/bin/env python3
"""Guarded K-LD7 timing and protocol probe."""

from __future__ import annotations

import argparse
import json
import statistics
import struct
import sys
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import BinaryIO, Iterable, Optional

try:
    import serial
    from serial.tools.list_ports import comports
except ImportError:  # pragma: no cover - exercised by operator environment
    serial = None
    comports = None


DEFAULT_BAUD = 3_000_000
DEFAULT_START_BAUD = 115_200
SUPPORTED_BAUD_RATES = [115_200, 460_800, 921_600, 2_000_000, 3_000_000]
FRAME_CODES = {
    "RADC": 0x01,
    "RFFT": 0x02,
    "PDAT": 0x04,
    "TDAT": 0x08,
    "DDAT": 0x10,
    "DONE": 0x20,
}
DOCUMENTED_WRITE_COMMANDS = {
    "RBFR", "RSPI", "RRAI", "THOF", "TRFT", "VISU", "MIRA", "MARA",
    "MIAN", "MAAN", "MISP", "MASP", "DEDI", "RATH", "ANTH", "SPTH",
    "DIG1", "DIG2", "DIG3", "HOLD", "MIDE", "MIDS",
}
DESTRUCTIVE_COMMANDS = {"RFSE"}


def build_packet(command: str, payload: bytes = b"") -> bytes:
    cmd = command.upper().encode("ascii")
    if len(cmd) != 4:
        raise ValueError("command must be exactly 4 ASCII characters")
    return struct.pack("<4sI", cmd, len(payload)) + payload


def validate_probe_command(command: str, hex_payload: str) -> Optional[str]:
    try:
        raw = command.encode("ascii")
    except UnicodeEncodeError:
        return "command must be ASCII"
    if len(raw) != 4:
        return "command must be exactly 4 ASCII characters"
    if command != command.upper():
        return "command must be uppercase ASCII"
    if len(hex_payload) % 2:
        return "hex payload must have an even number of characters"
    try:
        bytes.fromhex(hex_payload)
    except ValueError:
        return "hex payload must be valid hexadecimal"
    return None

class FakeSerial:
    def __init__(self, chunks):
        self.chunks = list(chunks)
        self.writes = []
        self.timeout = 0.2
        self.baudrate = 115200

    def read(self, n):
        if not self.chunks:
            return b""
        chunk = self.chunks.pop(0)
        if len(chunk) > n:
            self.chunks.insert(0, chunk[n:])
            return chunk[:n]
        return chunk

    def write(self, data):
        self.writes.append(data)
        return len(data)

    def flush(self):
        return None

    def reset_input_buffer(self):
        return None

    def close(self):
        return None


def test_read_packet_handles_split_header_and_payload():
    payload = b"\x00\x01\x02\x03"
    header = b"DONE" + len(payload).to_bytes(4, "little")
    fake = FakeSerial([header[:3], header[3:8], payload[:1], payload[1:]])
    protocol = probe.KLD7Protocol.__new__(probe.KLD7Protocol)
    protocol.port = fake

    packet = protocol.read_packet()

    assert packet.code == "DONE"
    assert packet.payload == payload
    assert packet.payload_bytes == 4


def test_summarize_measurements_counts_done_gaps():
    packets = [
        probe.PacketRecord(code="DONE", payload_bytes=4, complete_monotonic=1.0, done_frame=10),
        probe.PacketRecord(code="RADC", payload_bytes=3072, complete_monotonic=1.1, read_duration_ms=10.0),
        probe.PacketRecord(code="DONE", payload_bytes=4, complete_monotonic=2.0, done_frame=12),
        probe.PacketRecord(code="RADC", payload_bytes=3072, complete_monotonic=2.1, read_duration_ms=20.0),
    ]

    summary = probe.summarize_packets(packets, duration_s=2.0)

    assert summary["radc_frames"] == 2
    assert summary["done_frames"] == 2
    assert summary["done_frame_gaps"] == 1
    assert summary["effective_radc_hz"] == 1.0
    assert summary["read_duration_ms_p95"] == 20.0

@dataclass
class PacketRecord:
    code: str
    payload_bytes: int
    command: Optional[str] = None
    response_code: Optional[int] = None
    send_monotonic: Optional[float] = None
    first_byte_monotonic: Optional[float] = None
    header_complete_monotonic: Optional[float] = None
    complete_monotonic: Optional[float] = None
    read_duration_ms: Optional[float] = None
    done_frame: Optional[int] = None
    error: Optional[str] = None
    payload: bytes = field(default=b"", repr=False)


class KLD7Protocol:
    def __init__(self, port_path: str, baud: int = DEFAULT_BAUD, timeout: float = 0.2):
        if serial is None:
            raise RuntimeError("pyserial is required for hardware probing")
        self.port_path = port_path
        self.baud = baud
        self.port = serial.Serial(
            port=port_path,
            baudrate=DEFAULT_START_BAUD,
            parity=serial.PARITY_EVEN,
            stopbits=1,
            timeout=timeout,
        )

    def _read_exact(self, n: int) -> tuple[bytes, Optional[float]]:
        buf = b""
        first_byte_at = None
        deadline = time.monotonic() + max(float(getattr(self.port, "timeout", 0.2) or 0.2), 0.2)
        while len(buf) < n:
            chunk = self.port.read(n - len(buf))
            if chunk:
                if first_byte_at is None:
                    first_byte_at = time.monotonic()
                buf += chunk
                continue
            if time.monotonic() >= deadline:
                break
            time.sleep(0.001)
        return buf, first_byte_at

    def read_packet(self) -> PacketRecord:
        started = time.monotonic()
        header, first_byte_at = self._read_exact(8)
        header_complete = time.monotonic()
        if len(header) != 8:
            return PacketRecord(
                code="",
                payload_bytes=0,
                first_byte_monotonic=first_byte_at,
                header_complete_monotonic=header_complete,
                complete_monotonic=header_complete,
                error=f"short header read: got {len(header)} of 8 bytes",
            )
        raw_code, length = struct.unpack("<4sI", header)
        code = raw_code.decode("ascii", errors="replace")
        payload = b""
        payload_first = None
        if length:
            payload, payload_first = self._read_exact(length)
        complete = time.monotonic()
        error = None
        if len(payload) != length:
            error = f"short payload read: got {len(payload)} of {length} bytes"
        done_frame = None
        if code == "DONE" and len(payload) == 4:
            done_frame = int.from_bytes(payload, "little", signed=False)
        return PacketRecord(
            code=code,
            payload_bytes=length,
            first_byte_monotonic=first_byte_at or payload_first,
            header_complete_monotonic=header_complete,
            complete_monotonic=complete,
            read_duration_ms=(complete - (first_byte_at or started)) * 1000.0,
            done_frame=done_frame,
            error=error,
            payload=payload,
        )


def _percentile(values: list[float], percentile: float) -> Optional[float]:
    if not values:
        return None
    values = sorted(values)
    index = min(len(values) - 1, max(0, round((percentile / 100.0) * (len(values) - 1))))
    return values[index]


def summarize_packets(packets: list[PacketRecord], duration_s: float) -> dict:
    radc_packets = [p for p in packets if p.code == "RADC" and not p.error]
    done_packets = [p for p in packets if p.code == "DONE" and not p.error]
    done_frames = [p.done_frame for p in done_packets if p.done_frame is not None]
    gaps = 0
    for previous, current in zip(done_frames, done_frames[1:]):
        if current > previous + 1:
            gaps += current - previous - 1
    read_durations = [p.read_duration_ms for p in radc_packets if p.read_duration_ms is not None]
    errors: dict[str, int] = {}
    for packet in packets:
        if packet.error:
            errors[packet.error] = errors.get(packet.error, 0) + 1
    return {
        "duration_s": duration_s,
        "radc_frames": len(radc_packets),
        "done_frames": len(done_packets),
        "effective_radc_hz": round(len(radc_packets) / duration_s, 3) if duration_s > 0 else 0.0,
        "effective_done_hz": round(len(done_packets) / duration_s, 3) if duration_s > 0 else 0.0,
        "done_frame_gaps": gaps,
        "read_duration_ms_mean": statistics.mean(read_durations) if read_durations else None,
        "read_duration_ms_p50": statistics.median(read_durations) if read_durations else None,
        "read_duration_ms_p95": _percentile(read_durations, 95),
        "errors": errors,
    }

def test_unsafe_probe_requires_output():
    parser = probe.build_parser()
    args = parser.parse_args(["--port", "/dev/null", "--unsafe-probe", "--probe-command", "TEST"])

    assert probe.validate_args(args) == ["--unsafe-probe requires --output so probe activity is auditable"]


def test_rfse_requires_factory_reset_flag():
    parser = probe.build_parser()
    args = parser.parse_args([
        "--port", "/dev/null",
        "--output", "/tmp/probe.jsonl",
        "--unsafe-probe",
        "--probe-command", "RFSE",
    ])

    assert probe.validate_args(args) == ["RFSE is refused unless --allow-factory-reset is set"]


def test_parse_frame_mask_combines_known_flags():
    assert probe.parse_frame_mask("RADC,DONE") == 0x21

def parse_frame_mask(value: str) -> int:
    mask = 0
    for raw_name in value.split(","):
        name = raw_name.strip().upper()
        if not name:
            continue
        if name not in FRAME_CODES:
            raise argparse.ArgumentTypeError(f"unknown frame type {name!r}")
        mask |= FRAME_CODES[name]
    return mask


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Probe K-LD7 RADC timing and guarded commands.")
    parser.add_argument("--port")
    parser.add_argument("--baud", type=int, default=DEFAULT_BAUD, choices=SUPPORTED_BAUD_RATES)
    parser.add_argument("--duration", type=float, default=10.0)
    parser.add_argument("--frame-mask", default="RADC,DONE")
    parser.add_argument("--rspi-sweep", action="store_true")
    parser.add_argument("--rrai", type=int)
    parser.add_argument("--rbfr", type=int)
    parser.add_argument("--output", type=Path)
    parser.add_argument("--unsafe-probe", action="store_true")
    parser.add_argument("--probe-command", action="append", default=[])
    parser.add_argument("--allow-factory-reset", action="store_true")
    parser.add_argument("--no-restore-params", action="store_true")
    return parser


def validate_args(args: argparse.Namespace) -> list[str]:
    errors = []
    if args.unsafe_probe and not args.output:
        errors.append("--unsafe-probe requires --output so probe activity is auditable")
    if args.probe_command and not args.unsafe_probe:
        errors.append("--probe-command requires --unsafe-probe")
    for command_spec in args.probe_command:
        command, _, hex_payload = command_spec.partition(":")
        error = validate_probe_command(command, hex_payload)
        if error:
            errors.append(f"{command_spec}: {error}")
        if command.upper() in DESTRUCTIVE_COMMANDS and not args.allow_factory_reset:
            errors.append("RFSE is refused unless --allow-factory-reset is set")
    return errors

    def send_command(self, command: str, payload: bytes = b"") -> PacketRecord:
        sent = time.monotonic()
        self.port.reset_input_buffer()
        self.port.write(build_packet(command, payload))
        self.port.flush()
        packet = self.read_packet()
        packet.command = command.upper()
        packet.send_monotonic = sent
        if packet.code == "RESP" and packet.payload:
            packet.response_code = packet.payload[0]
        return packet

    def request_frame(self, frame_mask: int) -> list[PacketRecord]:
        records = [self.send_command("GNFD", int(frame_mask).to_bytes(4, "little", signed=True))]
        expected = bin(frame_mask).count("1")
        for _ in range(expected):
            record = self.read_packet()
            records.append(record)
            if record.code == "DONE":
                break
        return records

    def close(self) -> None:
        try:
            self.port.write(build_packet("GBYE"))
            self.port.flush()
        finally:
            self.port.close()

def measure(protocol: KLD7Protocol, frame_mask: int, duration_s: float) -> list[PacketRecord]:
    records = []
    deadline = time.monotonic() + duration_s
    while time.monotonic() < deadline:
        records.extend(protocol.request_frame(frame_mask))
    return records


def write_jsonl(path: Path, records: Iterable[PacketRecord], summary: dict) -> None:
    with path.open("w", encoding="utf-8") as handle:
        for record in records:
            data = asdict(record)
            data.pop("payload", None)
            handle.write(json.dumps({"type": "packet", **data}, sort_keys=True) + "\n")
        handle.write(json.dumps({"type": "summary", **summary}, sort_keys=True) + "\n")


def main(argv: Optional[list[str]] = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    errors = validate_args(args)
    if errors:
        for error in errors:
            print(f"error: {error}", file=sys.stderr)
        return 2
    frame_mask = parse_frame_mask(args.frame_mask)
    if not args.port:
        print("error: --port is required for the first implementation", file=sys.stderr)
        return 2
    protocol = KLD7Protocol(args.port, baud=args.baud)
    records = []
    try:
        records = measure(protocol, frame_mask, args.duration)
    finally:
        protocol.close()
    summary = summarize_packets(records, args.duration)
    print(json.dumps(summary, indent=2, sort_keys=True))
    if args.output:
        write_jsonl(args.output, records, summary)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())