jensandClaude Sonnet 4.6
dce570460e
ReplayPacer (MsgListener+MsgTalker) sits between packetizer output and downstream listeners; sleeps after each complete NMEA/UBX frame before forwarding. FileBackend no longer owns the delay — it reads as fast as possible and the pacer provides back-pressure on the same thread. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
import re
|
|
from threading import Thread
|
|
|
|
from backend.a_backend import ABackend
|
|
from msg.container import MsgContainer
|
|
|
|
_FILENAME_RE = re.compile(r'\d{4}_\d{2}_\d{2}_\d{2}_\d{2}_\d{2}_(.+)\.log$')
|
|
_CHUNK = 64
|
|
|
|
|
|
def source_id_from_path(path: str) -> str | None:
|
|
m = _FILENAME_RE.search(path)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
class FileBackend(ABackend):
|
|
def __init__(self, files: list[str]):
|
|
self._entries = [{'path': p, 'source_id': source_id_from_path(p), 'xcvr': None, 'file': None, 'thread': None}
|
|
for p in files]
|
|
self._running = False
|
|
|
|
def register_xcvr(self, xcvr):
|
|
for e in self._entries:
|
|
if e['source_id'] == xcvr.name.strip():
|
|
e['xcvr'] = xcvr
|
|
xcvr.on_register(self)
|
|
return
|
|
for e in self._entries:
|
|
if e['xcvr'] is None:
|
|
e['xcvr'] = xcvr
|
|
xcvr.on_register(self)
|
|
return
|
|
raise Exception(f"No available log file entry for \"{xcvr.name}\"")
|
|
|
|
def connect(self):
|
|
for e in self._entries:
|
|
e['file'] = open(e['path'], 'rb')
|
|
|
|
def start(self):
|
|
self._running = True
|
|
for e in self._entries:
|
|
e['thread'] = Thread(target=self._run, args=(e,), daemon=True)
|
|
e['thread'].start()
|
|
|
|
def stop(self):
|
|
self._running = False
|
|
for e in self._entries:
|
|
if e['thread'] and e['thread'].is_alive():
|
|
e['thread'].join(timeout=2.0)
|
|
|
|
def join(self):
|
|
for e in self._entries:
|
|
if e['thread']:
|
|
e['thread'].join()
|
|
|
|
def disconnect(self):
|
|
for e in self._entries:
|
|
if e['file']:
|
|
e['file'].close()
|
|
e['file'] = None
|
|
|
|
def _run(self, entry: dict):
|
|
xcvr = entry['xcvr']
|
|
f = entry['file']
|
|
while self._running:
|
|
data = f.read(_CHUNK)
|
|
if not data:
|
|
break
|
|
if xcvr:
|
|
xcvr.on_recv(MsgContainer(data, xcvr)) |