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 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=1.0, 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): 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) 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], delay=delay) 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) 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()