import sys, os, json, dataclasses, time from dataclasses import dataclass from PyQt5.QtCore import QObject, QTimer, pyqtSignal _root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) _gui = os.path.dirname(os.path.abspath(__file__)) for _p in [_root, _gui]: if _p not in sys.path: sys.path.insert(0, _p) from transceiver import Transceiver from nmea.packetizer import NmeaPacketizer from ubx.packetizer import UbxPacketizer from network_backend import NetworkBackend from ubx.ubx import frame_create GNSS_NAMES = { 0: 'GPS', 1: 'SBAS', 2: 'Galileo', 3: 'BeiDou', 4: 'IMES', 5: 'QZSS', 6: 'GLONASS', 7: 'NavIC' } GNSS_SHORT = {0: 'G', 1: 'S', 2: 'E', 3: 'C', 4: 'I', 5: 'J', 6: 'R', 7: 'N'} GNSS_COLORS = { 0: '#2196F3', # GPS – blue 1: '#9E9E9E', # SBAS – gray 2: '#4CAF50', # Galileo – green 3: '#FF9800', # BeiDou – orange 4: '#607D8B', # IMES – blue-gray 5: '#9C27B0', # QZSS – purple 6: '#F44336', # GLONASS – red 7: '#F4A0A0', # NavIC – rose } LRU_PATH = os.path.expanduser('~/.nmea_client_gui.json') LRU_MAX = 20 @dataclass class SatelliteData: gnss_id: int sv_id: int cno: float elevation: float azimuth: float used_in_fix: bool def sat_label(self) -> str: name = GNSS_NAMES.get(self.gnss_id, f'SYS{self.gnss_id}') return f"{name}-{self.sv_id}" @dataclass class ConnectionConfig: receiver_id: str conn_type: str = 'tcp' host: str = '' port: int = 8721 serial_port: str = '' baud_rate: int = 115200 def display_str(self) -> str: if self.conn_type == 'tcp': return f"{self.receiver_id} [{self.host}:{self.port}]" return f"{self.receiver_id} [{self.serial_port} @ {self.baud_rate}]" def to_dict(self): return dataclasses.asdict(self) @classmethod def from_dict(cls, d: dict) -> 'ConnectionConfig': valid = {f.name for f in dataclasses.fields(cls)} return cls(**{k: v for k, v in d.items() if k in valid}) class _ReceiverState: def __init__(self, rid: str): self.receiver_id = rid self.satellites: dict = {} # (gnss_id, sv_id) -> SatelliteData self.pdop: float = 0.0 self.connected: bool = False self.backend = None self.transceiver = None self.poll_timer = None class ReceiverManager(QObject): satellite_update = pyqtSignal(str, list) # rid, [SatelliteData] pdop_update = pyqtSignal(str, float, float) # rid, pdop, timestamp connection_changed = pyqtSignal(str, bool) # rid, connected receiver_added = pyqtSignal(str) receiver_removed = pyqtSignal(str) def __init__(self): super().__init__() self._states: dict = {} self._configs: dict = {} self._lru: list = self._load_lru() # ── LRU ────────────────────────────────────────────────────────────────── def _load_lru(self) -> list: try: with open(LRU_PATH) as f: return [ConnectionConfig.from_dict(d) for d in json.load(f).get('lru', [])] except Exception: return [] def _save_lru(self): try: with open(LRU_PATH, 'w') as f: json.dump({'lru': [c.to_dict() for c in self._lru[:LRU_MAX]]}, f, indent=2) except Exception: pass def _push_lru(self, config: ConnectionConfig): key = config.display_str() self._lru = [c for c in self._lru if c.display_str() != key] self._lru.insert(0, config) self._lru = self._lru[:LRU_MAX] self._save_lru() def get_lru(self) -> list: return list(self._lru) # ── Receiver lifecycle ──────────────────────────────────────────────────── def add_receiver(self, config: ConnectionConfig): rid = config.receiver_id if rid in self._states: return self._configs[rid] = config self._states[rid] = _ReceiverState(rid) self.receiver_added.emit(rid) def update_config(self, config: ConnectionConfig): rid = config.receiver_id if rid in self._states and not self._states[rid].connected: self._configs[rid] = config def remove_receiver(self, rid: str): self.disconnect_receiver(rid) self._states.pop(rid, None) self._configs.pop(rid, None) self.receiver_removed.emit(rid) def connect_receiver(self, rid: str): from listeners import NavSatQtListener, GsaQtListener from backends import SerialBackend config = self._configs.get(rid) state = self._states.get(rid) if not config or not state or state.connected: return xcvr = Transceiver(rid) if config.conn_type == 'tcp': backend = NetworkBackend( [{'name': rid, 'host': config.host, 'port': config.port}] ) else: backend = SerialBackend(config.serial_port, config.baud_rate) backend.register_xcvr(xcvr) nmea_pkt = NmeaPacketizer() ubx_pkt = UbxPacketizer() xcvr.register_listener(nmea_pkt) xcvr.register_listener(ubx_pkt) ubx_pkt.register_listener( NavSatQtListener(lambda sats, r=rid: self._on_sat(r, sats)) ) nmea_pkt.register_listener( GsaQtListener(rid, lambda p, t, r=rid: self._on_pdop(r, p, t)) ) backend.connect() backend.start() state.backend = backend state.transceiver = xcvr state.connected = True # Poll UBX messages every second (receiver only sends on request by default) timer = QTimer(self) timer.setInterval(1000) timer.timeout.connect(lambda r=rid: self._poll_ubx(r)) timer.start() state.poll_timer = timer self._push_lru(config) self.connection_changed.emit(rid, True) def disconnect_receiver(self, rid: str): state = self._states.get(rid) if not state or not state.connected: return if state.poll_timer: state.poll_timer.stop() state.poll_timer = None try: state.backend.stop() state.backend.disconnect() except Exception: pass state.connected = False state.backend = None state.transceiver = None state.satellites.clear() self.connection_changed.emit(rid, False) def _poll_ubx(self, rid: str): state = self._states.get(rid) if not state or not state.connected or not state.transceiver: return try: for cls_id, msg_id in [(0x01, 0x43), # NAV-SIG (0x01, 0x35), # NAV-SAT (0x02, 0x15)]: # RXM-RAWX state.transceiver.send(frame_create(cls_id, msg_id)) except Exception: pass # ── Queries ─────────────────────────────────────────────────────────────── def get_receiver_ids(self) -> list: return list(self._states.keys()) def is_connected(self, rid: str) -> bool: s = self._states.get(rid) return s.connected if s else False def get_config(self, rid: str): return self._configs.get(rid) def get_satellites(self, rids: list) -> list: result = [] for rid in rids: state = self._states.get(rid) if state: for sat in state.satellites.values(): result.append((rid, sat)) return result # ── Internal signal emission (called from backend threads) ───────────── def _on_sat(self, rid: str, sats: list): state = self._states.get(rid) if not state: return for s in sats: state.satellites[(s.gnss_id, s.sv_id)] = s self.satellite_update.emit(rid, list(state.satellites.values())) def _on_pdop(self, rid: str, pdop: float, ts: float): state = self._states.get(rid) if not state: return state.pdop = pdop self.pdop_update.emit(rid, pdop, ts)