jensandClaude Sonnet 4.6
ff686b64f7
- ConnectionConfig: new file_path and delay fields - ConnectionRow: File page with path input, Browse button, delay spinner (0–10 s), and auto-fill of receiver-ID from filename - ReceiverManager.connect_receiver: FileBackend wired for 'file' type; UBX poll timer skipped for file sources - FileBackend.register_xcvr: fallback to first unoccupied entry when name does not match filename pattern Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
271 lines
8.9 KiB
Python
271 lines
8.9 KiB
Python
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 backend.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
|
||
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 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 backend.serial_backend import SerialBackend
|
||
from backend.file_backend import FileBackend
|
||
|
||
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}]
|
||
)
|
||
elif config.conn_type == 'serial':
|
||
backend = SerialBackend(config.serial_port, config.baud_rate)
|
||
else:
|
||
backend = FileBackend([config.file_path], delay=config.delay)
|
||
|
||
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
|
||
|
||
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
|
||
|
||
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)
|