added UBX:RXM:RAWX

network_backend.py

@@ -19,6 +19,7 @@ from queue import Queue
 from ubx.ubx import frame_create
 from ubx.msg import UbxMsg
 from ubx.messages.nav import Sig
+from ubx.messages.rxm import RawX
 
 class NetworkBackend(ABackend):
 	def __init__(self, servers: list[dict]):
@@ -148,6 +149,8 @@ if __name__ == "__main__":
 		]
 		ubx_rx_list = [
 			UbxReceiver(Sig(), 0x01, 0x43),
+			UbxReceiver(RawX(), 0x02, 0x15),
+
 			UbxReceiver(UbxMsg(), 0x05, 0x01),
 			UbxReceiver(UbxMsg(), 0x05, 0x00),
 		]
@@ -198,6 +201,9 @@ if __name__ == "__main__":
 				# Poll UBX-NAV-SIG (0x01 0x43)
 				frame = frame_create(0x01, 0x43, b'')
 				transceiver[grm].send(frame)
+				# Poll UBX-RXM-RAWX (0x02 0x15)
+				frame = frame_create(0x02, 0x15, b'')
+				transceiver[grm].send(frame)
 		except ValueError:
 			break
 

ubx/messages/nav.py

@@ -27,11 +27,13 @@ class Sig(ubx.msg.UbxMsg):
 		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
+		offset = 8
+		remain = len(data) - offset
+		group_size = 16
+		n = offset
 		while remain:
-			signal = cls.Group1.from_bytes(data[n:n + 16])
-			n += 16
-			remain -= 16
+			signal = cls.Group1.from_bytes(data[n:n + group_size])
+			n += group_size
+			remain -= group_size
 			obj.signals.append(signal)
 		return obj.__dict__

ubx/messages/rxm.py

@@ -0,0 +1,45 @@
+import struct
+import ubx.msg
+
+class RawX(ubx.msg.UbxMsg):
+	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 = cls.Group1.from_bytes(data[n:n + 32])
+			n += group_size
+			remain -= group_size
+			obj.meas.append(group)
+		return obj.__dict__