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>
220 lines
7.1 KiB
Python
220 lines
7.1 KiB
Python
import sys
|
||
import time
|
||
import json
|
||
import signal
|
||
import argparse
|
||
|
||
from backend.network_backend import NetworkBackend
|
||
from backend.serial_backend import SerialBackend
|
||
from backend.file_backend import FileBackend, source_id_from_path
|
||
from msg.frame_logger import FrameLogger
|
||
from msg.replay_pacer import ReplayPacer
|
||
from transceiver import Transceiver
|
||
|
||
from nmea.packetizer import NmeaPacketizer
|
||
from nmea.messages import Gll, Gsa, Gsv, Gga, Rmc
|
||
from ubx.ubx import frame_create
|
||
from ubx.msg_types import *
|
||
from ubx.packetizer import UbxPacketizer
|
||
from ubx.messages import Sig, Sat, RawX, UbxMsg
|
||
from ubx.listener import UbxListener
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Argument parsing
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_args():
|
||
ap = argparse.ArgumentParser(description='GNSS receiver client')
|
||
ap.add_argument('--project', metavar='FILE',
|
||
help='Load config from JSON project file')
|
||
ap.add_argument('--save', metavar='FILE',
|
||
help='Save current config as JSON project file')
|
||
ap.add_argument('--log-dir', default='./log',
|
||
help='Frame log directory (default: ./log)')
|
||
ap.add_argument('--delay', type=float, default=0.1,
|
||
help='File replay inter-frame delay in seconds, 0–10 (default: 1.0)')
|
||
ap.add_argument('--network', nargs=3, action='append',
|
||
metavar=('NAME', 'HOST', 'PORT'),
|
||
help='Add a network source')
|
||
ap.add_argument('--serial', nargs=3, action='append',
|
||
metavar=('NAME', 'DEVICE', 'BAUD'),
|
||
help='Add a serial source')
|
||
ap.add_argument('--file', action='append', metavar='PATH',
|
||
help='Add a log file source for replay')
|
||
return ap.parse_args()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Config (de)serialisation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def args_to_config(args) -> dict:
|
||
sources = []
|
||
for name, host, port in (args.network or []):
|
||
sources.append({'type': 'network', 'name': name,
|
||
'host': host, 'port': int(port)})
|
||
for name, device, baud in (args.serial or []):
|
||
sources.append({'type': 'serial', 'name': name,
|
||
'device': device, 'baudrate': int(baud)})
|
||
for path in (args.file or []):
|
||
sources.append({'type': 'file', 'path': path})
|
||
return {'log_dir': args.log_dir, 'delay': args.delay, 'sources': sources}
|
||
|
||
|
||
def load_config(path: str) -> dict:
|
||
with open(path) as f:
|
||
return json.load(f)
|
||
|
||
|
||
def save_config(config: dict, path: str):
|
||
with open(path, 'w') as f:
|
||
json.dump(config, f, indent=2)
|
||
print(f'Config saved to {path}')
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Wiring helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _make_listeners(name: str):
|
||
nmea = [Gll(name, 'GN'), Gsa(name, 'GN'), Gsv(name, 'GP'),
|
||
Gga(name, 'GP'), Rmc(name, 'GN'), Gga(name, 'GN')]
|
||
ubx = [UbxListener(Sig()), UbxListener(Sat()), UbxListener(RawX()),
|
||
UbxListener(UbxMsg(UBX_CLASS_ACK, UBX_ID_ACK_ACK)),
|
||
UbxListener(UbxMsg(UBX_CLASS_ACK, UBX_ID_ACK_NACK))]
|
||
return nmea, ubx
|
||
|
||
|
||
def wire_source(name: str, backend, log_dir: str | None,
|
||
loggers: list, transceivers: dict, pacer_delay: float = 0.0):
|
||
xcvr = Transceiver(name)
|
||
backend.register_xcvr(xcvr)
|
||
transceivers[name] = xcvr
|
||
|
||
if log_dir:
|
||
lg = FrameLogger(name, log_dir)
|
||
loggers.append(lg)
|
||
xcvr.register_listener(lg)
|
||
|
||
nmea_pkt = NmeaPacketizer()
|
||
ubx_pkt = UbxPacketizer()
|
||
xcvr.register_listener(nmea_pkt)
|
||
xcvr.register_listener(ubx_pkt)
|
||
|
||
nmea_rx, ubx_rx = _make_listeners(name)
|
||
|
||
if pacer_delay > 0:
|
||
nmea_pacer = ReplayPacer(pacer_delay)
|
||
ubx_pacer = ReplayPacer(pacer_delay)
|
||
nmea_pkt.register_listener(nmea_pacer)
|
||
ubx_pkt.register_listener(ubx_pacer)
|
||
for rx in nmea_rx:
|
||
nmea_pacer.register_listener(rx)
|
||
for rx in ubx_rx:
|
||
ubx_pacer.register_listener(rx)
|
||
else:
|
||
for rx in nmea_rx:
|
||
nmea_pkt.register_listener(rx)
|
||
for rx in ubx_rx:
|
||
ubx_pkt.register_listener(rx)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Signal handler
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def handler(signum, _frame, backends: list, loggers: list):
|
||
print(f'Signal {signal.Signals(signum).name} – shutting down')
|
||
for be in backends:
|
||
be.stop()
|
||
be.disconnect()
|
||
for lg in loggers:
|
||
lg.close()
|
||
sys.exit(0)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Entry point
|
||
# ---------------------------------------------------------------------------
|
||
|
||
if __name__ == '__main__':
|
||
args = parse_args()
|
||
|
||
config = load_config(args.project) if args.project else args_to_config(args)
|
||
|
||
if args.save:
|
||
save_config(config, args.save)
|
||
|
||
sources = config.get('sources', [])
|
||
if not sources:
|
||
print('No sources specified. Use --network, --serial, --file or --project.')
|
||
sys.exit(1)
|
||
|
||
log_dir = config.get('log_dir', './log')
|
||
delay = config.get('delay', 1.0)
|
||
|
||
backends = []
|
||
loggers = []
|
||
transceivers = {}
|
||
|
||
# Network sources → one shared NetworkBackend
|
||
net_sources = [s for s in sources if s['type'] == 'network']
|
||
if net_sources:
|
||
net_be = NetworkBackend([{'name': s['name'], 'host': s['host'], 'port': s['port']}
|
||
for s in net_sources])
|
||
backends.append(net_be)
|
||
for s in net_sources:
|
||
wire_source(s['name'], net_be, log_dir, loggers, transceivers)
|
||
|
||
# Serial sources → one SerialBackend each
|
||
for s in [s for s in sources if s['type'] == 'serial']:
|
||
ser_be = SerialBackend(s['device'], s.get('baudrate', 115200))
|
||
backends.append(ser_be)
|
||
wire_source(s['name'], ser_be, log_dir, loggers, transceivers)
|
||
|
||
# File sources → one shared FileBackend (no secondary logging)
|
||
file_sources = [s for s in sources if s['type'] == 'file']
|
||
if file_sources:
|
||
file_be = FileBackend([s['path'] for s in file_sources])
|
||
backends.append(file_be)
|
||
for s in file_sources:
|
||
name = source_id_from_path(s['path'])
|
||
if name:
|
||
wire_source(name, file_be, log_dir=None, loggers=loggers,
|
||
transceivers=transceivers, pacer_delay=delay)
|
||
|
||
signal.signal(signal.SIGINT,
|
||
lambda sig, frm: handler(sig, frm, backends, loggers))
|
||
|
||
for be in backends:
|
||
be.connect()
|
||
be.start()
|
||
|
||
# Poll loop – only meaningful for network/serial sources
|
||
net_names = [s['name'] for s in net_sources]
|
||
if net_names:
|
||
while True:
|
||
try:
|
||
for name in net_names:
|
||
transceivers[name].send(frame_create(UBX_CLASS_NAV, UBX_ID_CLOCK))
|
||
transceivers[name].send(frame_create(UBX_CLASS_NAV, UBX_ID_NAV_SIG))
|
||
transceivers[name].send(frame_create(UBX_CLASS_NAV, UBX_ID_NAV_SAT))
|
||
transceivers[name].send(frame_create(UBX_CLASS_RXM, UBX_ID_RXM_RAW))
|
||
except ValueError:
|
||
break
|
||
time.sleep(1)
|
||
elif file_sources:
|
||
# Wait until all file threads finish
|
||
for be in backends:
|
||
if isinstance(be, FileBackend):
|
||
be.join()
|
||
else:
|
||
# Serial-only: block until interrupted
|
||
signal.pause()
|
||
|
||
for be in backends:
|
||
be.stop()
|
||
be.disconnect()
|
||
for lg in loggers:
|
||
lg.close() |