added UBX:NAV:SIG message

network_backend.py

@@ -17,6 +17,8 @@ from nmea.RMC import Rmc
 from threading import Thread
 from queue import Queue
 from ubx.ubx import frame_create
+from ubx.msg import UbxMsg
+from ubx.messages.nav import Sig
 
 class NetworkBackend(ABackend):
 	def __init__(self, servers: list[dict]):
@@ -145,10 +147,9 @@ if __name__ == "__main__":
 			Gga(grm, 'GN'),
 		]
 		ubx_rx_list = [
-#			UbxReceiver('UBX-NAV-SIG', 0x01, 0x43),
-			UbxReceiver('UBX-ACK-ACK', 0x05, 0x01),
-			UbxReceiver('UBX-ACK-NACK', 0x05, 0x00),
-			UbxReceiver('UBX:default'),
+			UbxReceiver(Sig(), 0x01, 0x43),
+			UbxReceiver(UbxMsg(), 0x05, 0x01),
+			UbxReceiver(UbxMsg(), 0x05, 0x00),
 		]
 
 		# Topology
@@ -194,6 +195,9 @@ if __name__ == "__main__":
 				# Poll UBX-NAV-CLOCK (0x01 0x22)
 				frame = frame_create(0x01, 0x22, b'')
 				transceiver[grm].send(frame)
+				# Poll UBX-NAV-SIG (0x01 0x43)
+				frame = frame_create(0x01, 0x43, b'')
+				transceiver[grm].send(frame)
 		except ValueError:
 			break
 

receiver/ubx_receiver.py

@@ -1,17 +1,19 @@
 from msg_container import MsgContainer
 from msg_sink import MsgSink
 from ubx.ubx import frame_parse, frame_parse_hdr
+from ubx.msg import UbxMsg
 
 class UbxReceiver(MsgSink):
-	def __init__(self, name: str, mclass: int|None = None, mid: int|None = None):
+	def __init__(self, parser: UbxMsg, mclass: int|None = None, mid: int|None = None):
 		MsgSink.__init__(self)
-		self.name = name
+		self.parser = parser
 		self.mclass = mclass
 		self.mid = mid
 
 	def on_recv(self, msg: MsgContainer):
 		data, hdr = frame_parse(msg.data)
-		print(f"{msg.name}:{self.name}:{hdr.mclass}:{hdr.mid}:data({len(data)}):{data}")
+		msg_parsed = self.parser.from_bytes(data)
+		print(f"{msg.name}:{hdr.mclass}:{hdr.mid}:data({len(data)}):{msg_parsed}")
 
 	def is_msg(self, msg: MsgContainer):
 		hdr = frame_parse_hdr(msg.data)

ubx/messages/nav.py

@@ -0,0 +1,37 @@
+import struct
+import ubx.msg
+
+class Sig(ubx.msg.UbxMsg):
+	class Group1:
+		@classmethod
+		def from_bytes(cls, data: bytes):
+			obj = Sig.Group1()
+			obj.gnss_id = struct.unpack("B", data[0:1])[0]
+			obj.sv_id = struct.unpack("B", data[1:2])[0]
+			obj.sig_id = struct.unpack("B", data[2:3])[0]
+			obj.freq_id = struct.unpack("B", data[3:4])[0]
+			obj.pr_res = struct.unpack("<H", data[4:6])[0]
+			obj.cno = struct.unpack("B", data[6:7])[0]
+			obj.quality_ind = struct.unpack("B", data[7:8])[0]
+			obj.corr_source = struct.unpack("B", data[8:9])[0]
+			obj.iono_model = struct.unpack("B", data[9:10])[0]
+			obj.sig_flags = struct.unpack("<H", data[10:12])[0]
+			obj.reserved1 = struct.unpack("BBBB", data[12:16])[0]
+			return obj.__dict__
+
+	@classmethod
+	def from_bytes(cls, data: bytes):
+		obj = Sig()
+		obj.iTOW = struct.unpack("<L", data[0:4])[0]
+		obj.version = struct.unpack("B", data[4:5])[0]
+		obj.num_signals = struct.unpack("B", data[5:6])[0]
+		obj.reserved = struct.unpack("BB", data[6:8])[0]
+		obj.signals = []
+		remain = len(data) - 8
+		n = 0
+		while remain:
+			signal = cls.Group1.from_bytes(data[n:n + 16])
+			n += 16
+			remain -= 16
+			obj.signals.append(signal)
+		return obj.__dict__

ubx/msg.py

@@ -0,0 +1,7 @@
+
+
+class UbxMsg:
+
+	@classmethod
+	def from_bytes(cls, data: bytes):
+		pass