68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Iterable
|
|
|
|
from h8536.bench_connect_lcd import BenchLogger, FrameDetector, format_frame, label_frame
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RxFrame:
|
|
frame: bytes
|
|
label: str
|
|
|
|
|
|
class SerialLink:
|
|
"""Thin serial-port wrapper with checksum-resync frame detection."""
|
|
|
|
def __init__(
|
|
self,
|
|
device: Any,
|
|
logger: BenchLogger,
|
|
*,
|
|
sync_mode: str = "checksum",
|
|
) -> None:
|
|
self.device = device
|
|
self.logger = logger
|
|
self.detector = FrameDetector(sync_mode=sync_mode)
|
|
|
|
def reset_input(self) -> None:
|
|
self.device.reset_input_buffer()
|
|
self.detector = FrameDetector(sync_mode=self.detector.sync_mode)
|
|
|
|
def read_available(self) -> list[RxFrame]:
|
|
waiting = getattr(self.device, "in_waiting", 0)
|
|
data = self.device.read(waiting or 1)
|
|
if not data:
|
|
return []
|
|
|
|
dropped_before = self.detector.dropped_bytes
|
|
self.logger.chunk("RX", data)
|
|
frames = [RxFrame(frame, label) for frame, label in self.detector.feed(data)]
|
|
for item in frames:
|
|
self.logger.event(f"DETECT {item.label} {format_frame(item.frame)}")
|
|
dropped_now = self.detector.dropped_bytes - dropped_before
|
|
if dropped_now:
|
|
self.logger.event(
|
|
f"RESYNC dropped_bytes={dropped_now} total_dropped={self.detector.dropped_bytes} "
|
|
f"buffered={len(self.detector.buffer)}"
|
|
)
|
|
return frames
|
|
|
|
def send(self, frame: bytes, label: str) -> None:
|
|
self.device.write(frame)
|
|
self.device.flush()
|
|
self.logger.chunk("TX", frame)
|
|
self.logger.event(f"SENT {label} {format_frame(frame)}")
|
|
|
|
def labels(self) -> dict[str, int]:
|
|
return dict(self.detector.labels)
|
|
|
|
|
|
def label_for_frame(frame: bytes) -> str:
|
|
return label_frame(frame)
|
|
|
|
|
|
def format_frames(frames: Iterable[bytes]) -> str:
|
|
return " | ".join(format_frame(frame) for frame in frames)
|