added gui
This commit is contained in:
@@ -0,0 +1,261 @@
|
||||
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',
|
||||
}
|
||||
GNSS_SHORT = {0: 'G', 1: 'S', 2: 'E', 3: 'C', 4: 'I', 5: 'J', 6: 'R'}
|
||||
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
|
||||
}
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user