jensandClaude Sonnet 5
b3447a28c5
A struct size-mismatch or any other parse error in a listener could propagate uncaught out of the socket-reading thread and silently kill it. Catch broadly in the event loop and per-listener in call_listener, and make the NAV-SAT/NAV-SIG/RXM-RAWX group parsers tolerate a payload length that isn't an exact multiple of the group size instead of crashing on a short struct.unpack. Also wire NetworkBackend's socket-loss detection through to ReceiverManager (via a queued signal, since the callback fires from the backend thread) so the GUI reflects a peer-initiated disconnect instead of leaving the row stuck showing "connected". Verified against a real ZED-X20P: it drops the TCP session on its own after ~20-28s regardless of traffic; the GUI now marks the receiver disconnected and frees its ID for reconnecting. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AzmCaNjDb3TAqPpTwunKTY
51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
import struct
|
|
from ubx.messages.msg import UbxMsg
|
|
|
|
class RawX(UbxMsg):
|
|
def __init__(self):
|
|
UbxMsg.__init__(self, 0x02, 0x15)
|
|
|
|
class Group1:
|
|
@classmethod
|
|
def from_bytes(cls, data: bytes):
|
|
obj = RawX.Group1()
|
|
obj.pr_mes = struct.unpack("<d", data[0:8])[0]
|
|
obj.cp_mes = struct.unpack("<d", data[8:16])[0]
|
|
obj.do_mes = struct.unpack("<f", data[16:20])[0]
|
|
obj.gnss_id = struct.unpack("B", data[20:21])[0]
|
|
obj.sv_id = struct.unpack("B", data[21:22])[0]
|
|
obj.sig_id = struct.unpack("B", data[22:23])[0]
|
|
obj.freq_id = struct.unpack("B", data[23:24])[0]
|
|
obj.lock_time = struct.unpack("<H", data[24:26])[0]
|
|
obj.cno = struct.unpack("B", data[26:27])[0]
|
|
obj.pr_std_dev = struct.unpack("B", data[27:28])[0]
|
|
obj.cp_std_dev = struct.unpack("B", data[28:29])[0]
|
|
obj.do_std_dev = struct.unpack("B", data[29:30])[0]
|
|
obj.do_trk_stat = struct.unpack("B", data[30:31])[0]
|
|
obj.reserved1 = struct.unpack("B", data[31:32])[0]
|
|
return obj.__dict__
|
|
|
|
@classmethod
|
|
def from_bytes(cls, data: bytes):
|
|
obj = RawX()
|
|
obj.rcv_tow = struct.unpack("<d", data[0:8])[0]
|
|
obj.week = struct.unpack("<H", data[8:10])[0]
|
|
obj.leap_s = struct.unpack("b", data[10:11])[0]
|
|
obj.num_meas = struct.unpack("B", data[11:12])[0]
|
|
obj.rec_stat = struct.unpack("B", data[12:13])[0]
|
|
obj.version = struct.unpack("B", data[13:14])[0]
|
|
obj.reserved0 = struct.unpack("BB", data[14:16])[0]
|
|
obj.meas = []
|
|
offset = 16
|
|
remain = len(data) - offset
|
|
group_size = 32
|
|
n = offset
|
|
while remain >= group_size:
|
|
group = cls.Group1.from_bytes(data[n:n + 32])
|
|
n += group_size
|
|
remain -= group_size
|
|
obj.meas.append(group)
|
|
if remain:
|
|
print(f"RawX.from_bytes: {remain} trailing bytes unparsed (payload size mismatch)")
|
|
return obj.__dict__
|