jensandClaude Sonnet 5
eb18b1485f
NetworkBackend's connect/send/lost/error prints always used the connect-time registration name, so GUI console output kept showing the placeholder counter ID even after MON-VER resolved the real device name. Add a display_name per socket (defaults to name, kept separate from the routing key used by find_by_name/register_xcvr) and have the GUI update it once MON-VER responds. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XM4FNUjhu9x8df5Dp1hHvd
350 lines
12 KiB
Python
350 lines
12 KiB
Python
import sys, os, json, dataclasses, time, threading
|
||
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 backend.network_backend import NetworkBackend
|
||
from ubx.ubx import frame_create
|
||
from ubx.query import UbxQuery
|
||
from ubx.messages import Ver
|
||
|
||
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
|
||
|
||
|
||
def _device_id_from_ver(ver: dict) -> str:
|
||
for ext in ver.get('extensions', []):
|
||
s = ext.get('extension', '')
|
||
if s.startswith('MOD='):
|
||
return s[len('MOD='):]
|
||
return ver.get('hw_version', '')
|
||
|
||
|
||
@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
|
||
file_path: str = ''
|
||
delay: float = 1.0
|
||
|
||
def display_str(self) -> str:
|
||
if self.conn_type == 'tcp':
|
||
return f"{self.receiver_id} [{self.host}:{self.port}]"
|
||
if self.conn_type == 'serial':
|
||
return f"{self.receiver_id} [{self.serial_port} @ {self.baud_rate}]"
|
||
return f"{self.receiver_id} [file: {os.path.basename(self.file_path)}]"
|
||
|
||
def endpoint_key(self):
|
||
if self.conn_type == 'tcp':
|
||
return ('tcp', self.host, self.port)
|
||
if self.conn_type == 'serial':
|
||
return ('serial', self.serial_port, self.baud_rate)
|
||
return ('file', self.file_path)
|
||
|
||
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
|
||
self.device_id: str = ''
|
||
self.sw_version: str = ''
|
||
|
||
|
||
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
|
||
version_update = pyqtSignal(str, str, str) # rid, device_id, sw_version
|
||
receiver_added = pyqtSignal(str)
|
||
receiver_removed = pyqtSignal(str)
|
||
_connection_lost = pyqtSignal(str) # internal: backend thread -> GUI thread
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self._states: dict = {}
|
||
self._configs: dict = {}
|
||
self._lru: list = self._load_lru()
|
||
self._connection_lost.connect(self._handle_connection_lost)
|
||
self.version_update.connect(self._on_version)
|
||
|
||
# ── 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.endpoint_key()
|
||
self._lru = [c for c in self._lru if c.endpoint_key() != 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 backend.serial_backend import SerialBackend
|
||
from backend.file_backend import FileBackend
|
||
from msg.replay_pacer import ReplayPacer
|
||
|
||
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}],
|
||
on_disconnect=lambda name, r=rid: self._connection_lost.emit(r),
|
||
)
|
||
elif config.conn_type == 'serial':
|
||
backend = SerialBackend(config.serial_port, config.baud_rate)
|
||
else:
|
||
backend = FileBackend([config.file_path])
|
||
|
||
backend.register_xcvr(xcvr)
|
||
|
||
nmea_pkt = NmeaPacketizer()
|
||
ubx_pkt = UbxPacketizer()
|
||
xcvr.register_listener(nmea_pkt)
|
||
xcvr.register_listener(ubx_pkt)
|
||
|
||
if config.conn_type == 'file' and config.delay > 0:
|
||
nmea_sink = ReplayPacer(config.delay)
|
||
ubx_sink = ReplayPacer(config.delay)
|
||
nmea_pkt.register_listener(nmea_sink)
|
||
ubx_pkt.register_listener(ubx_sink)
|
||
else:
|
||
nmea_sink = nmea_pkt
|
||
ubx_sink = ubx_pkt
|
||
|
||
ubx_sink.register_listener(
|
||
NavSatQtListener(lambda sats, r=rid: self._on_sat(r, sats))
|
||
)
|
||
nmea_sink.register_listener(
|
||
GsaQtListener(rid, lambda p, t, r=rid: self._on_pdop(r, p, t))
|
||
)
|
||
|
||
ver_query = None
|
||
if config.conn_type != 'file':
|
||
ver_query = UbxQuery(xcvr, Ver())
|
||
ubx_sink.register_listener(ver_query)
|
||
|
||
backend.connect()
|
||
backend.start()
|
||
|
||
state.backend = backend
|
||
state.transceiver = xcvr
|
||
state.connected = True
|
||
|
||
if config.conn_type != 'file':
|
||
timer = QTimer(self)
|
||
timer.setInterval(1000)
|
||
timer.timeout.connect(lambda r=rid: self._poll_ubx(r))
|
||
timer.start()
|
||
state.poll_timer = timer
|
||
|
||
threading.Thread(target=self._poll_version, args=(rid, ver_query),
|
||
daemon=True).start()
|
||
|
||
self._push_lru(config)
|
||
self.connection_changed.emit(rid, True)
|
||
|
||
def _poll_version(self, rid: str, query: UbxQuery):
|
||
ver = query.poll()
|
||
if ver is None:
|
||
return
|
||
self.version_update.emit(rid, _device_id_from_ver(ver), ver['sw_version'])
|
||
|
||
def _handle_connection_lost(self, rid: str):
|
||
state = self._states.get(rid)
|
||
if not state or not state.connected:
|
||
return
|
||
print(f"Receiver '{self.display_name(rid)}': connection lost")
|
||
self.remove_receiver(rid)
|
||
|
||
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()
|
||
state.device_id = ''
|
||
state.sw_version = ''
|
||
self.connection_changed.emit(rid, False)
|
||
self.version_update.emit(rid, '', '')
|
||
|
||
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_version_info(self, rid: str):
|
||
s = self._states.get(rid)
|
||
return (s.device_id, s.sw_version) if s else ('', '')
|
||
|
||
def display_name(self, rid: str) -> str:
|
||
s = self._states.get(rid)
|
||
return (s.device_id if s and s.device_id else 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)
|
||
|
||
def _on_version(self, rid: str, device_id: str, sw_version: str):
|
||
state = self._states.get(rid)
|
||
if state:
|
||
state.device_id = device_id
|
||
state.sw_version = sw_version
|
||
set_name = getattr(state.backend, 'set_display_name', None)
|
||
if device_id and callable(set_name):
|
||
set_name(rid, device_id)
|
||
config = self._configs.get(rid)
|
||
if config and device_id:
|
||
config.receiver_id = device_id
|
||
self._push_lru(config)
|