add UBX-MON-VER support and surface device identification in CLI/GUI
Add the Ver message parser and a fixed UbxQuery (was unused and had a broken semaphore/state bug) to poll MON-VER once on connect. The CLI prints hw/sw version and extensions; the GUI polls it per receiver, pre-fills the read-only ID field with a connection counter and rewrites it from the MOD= extension once the device answers, and propagates the resolved display name to the Sky and Plot tabs so they no longer show the stale placeholder ID. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XM4FNUjhu9x8df5Dp1hHvd
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
from ubx.messages.msg import UbxMsg
|
||||
from ubx.messages.nav import Sig, Sat
|
||||
from ubx.messages.rxm import RawX
|
||||
from ubx.messages.rxm import RawX
|
||||
from ubx.messages.mon import Ver
|
||||
@@ -0,0 +1,32 @@
|
||||
import struct
|
||||
from ubx.messages.msg import UbxMsg
|
||||
|
||||
class Ver(UbxMsg):
|
||||
def __init__(self):
|
||||
UbxMsg.__init__(self, 0x0A, 0x04)
|
||||
|
||||
class Group1:
|
||||
@classmethod
|
||||
def from_bytes(cls, data: bytes):
|
||||
obj = Ver.Group1()
|
||||
obj.extension = data[0:30].split(b"\x00", 1)[0].decode("ascii", errors="replace")
|
||||
return obj.__dict__
|
||||
|
||||
@classmethod
|
||||
def from_bytes(cls, data: bytes):
|
||||
obj = Ver()
|
||||
obj.sw_version = data[0:30].split(b"\x00", 1)[0].decode("ascii", errors="replace")
|
||||
obj.hw_version = data[30:40].split(b"\x00", 1)[0].decode("ascii", errors="replace")
|
||||
obj.extensions = []
|
||||
offset = 40
|
||||
remain = len(data) - offset
|
||||
group_size = 30
|
||||
n = offset
|
||||
while remain >= group_size:
|
||||
extension = cls.Group1.from_bytes(data[n:n + group_size])
|
||||
n += group_size
|
||||
remain -= group_size
|
||||
obj.extensions.append(extension)
|
||||
if remain:
|
||||
print(f"Ver.from_bytes: {remain} trailing bytes unparsed (payload size mismatch)")
|
||||
return obj.__dict__
|
||||
+8
-13
@@ -3,30 +3,25 @@ from ubx.ubx import frame_parse, frame_create
|
||||
from ubx.messages.msg import UbxMsg
|
||||
from transceiver import Transceiver
|
||||
from ubx.listener import UbxListener
|
||||
from threading import BoundedSemaphore
|
||||
from threading import Event
|
||||
|
||||
class UbxQuery(UbxListener):
|
||||
def __init__(self, transceiver: Transceiver, msg: UbxMsg):
|
||||
UbxListener.__init__(self, msg)
|
||||
self.transceiver = transceiver
|
||||
self.sema = BoundedSemaphore()
|
||||
self.result = None
|
||||
self.event = Event()
|
||||
|
||||
def on_recv(self, msg: MsgContainer):
|
||||
if self.is_msg(msg):
|
||||
data, hdr = frame_parse(msg.data)
|
||||
msg_parsed = self.msg.from_bytes(data)
|
||||
self.msg = msg_parsed
|
||||
self.sema.release()
|
||||
self.result = self.msg.from_bytes(data)
|
||||
self.event.set()
|
||||
|
||||
def poll(self):
|
||||
self.event.clear()
|
||||
frame = frame_create(self.msg.mclass, self.msg.mid, b'')
|
||||
self.transceiver.send(frame)
|
||||
self.sema.acquire()
|
||||
print(self.msg)
|
||||
|
||||
def query(self, data: bytes):
|
||||
frame = frame_create(0x01, 0x35, b'')
|
||||
self.transceiver.send(frame)
|
||||
self.sema.acquire()
|
||||
print(self.msg)
|
||||
self.event.wait()
|
||||
return self.result
|
||||
|
||||
|
||||
Reference in New Issue
Block a user