update uni_handler

semuadmin · semuadmin · commit 22132136802b · 2026-02-10T09:58:10.000Z
diff --git a/src/pygpsclient/globals.py b/src/pygpsclient/globals.py
@@ -423,11 +423,31 @@
     "PVTGeodetic7": "RTK FIXED",
     "PVTGeodetic8": "RTK FLOAT",
     "PVTGeodetic10": "PPP",
+    "BESTNAV0": "NO FIX",
+    "BESTNAV1": "FIXEDPOS",  # FIXEDPOS Position fixed by the FIX POSITION command
+    "BESTNAV2": "FIXEDHMSL",  # FIXEDHEIGHT Not supported currently
+    "BESTNAV8": "DOPP",  # DOPPLER_VELOCITY Velocity computed using instantaneous Doppler
+    "BESTNAV16": "3D",  # SINGLE Single point positioning
+    "BESTNAV17": "RTK FLOAT",  # SRDIFF Pseudorange differential solution
+    "BESTNAV18": "SBAS",  # SBAS SBAS positioning
+    "BESTNAV32": "RTK FLOAT",  # L1_FLOAT L1 float solution
+    "BESTNAV33": "RTK FLOAT",  # IONOFREE_FLOAT Ionosphere-free float solution
+    "BESTNAV34": "RTK FLOAT",  # NARROW_FLOAT Narrow-lane float solution
+    "BESTNAV48": "RTK FIXED",  # L1_INT L1 fixed solution
+    "BESTNAV49": "RTK FIXED",  # WIDE_INT Wide-lane fixed solution
+    "BESTNAV50": "RTK FIXED",  # NARROW_INT Narrow-lane fixed solution
+    "BESTNAV52": "DR",  # INS Inertial navigation solution
+    "BESTNAV53": "3D+DR",  # INS_PSRSP Integrated solution of INS and single point pos
+    "BESTNAV54": "RTK+DR",  # INS_PSRDIFF Integrated solution of INS and pseudorange diff pos
+    "BESTNAV55": "RTK FLOAT+DR",  # INS_RTKFLOAT Integrated solution of INS and RTK float
+    "BESTNAV56": "RTK FIXED+DR",  # INS_RTKFIXED Integrated solution of INS and RTK fix
+    "BESTNAV68": "RTK FIXED+DR",  # INS_RTKFIXED Integrated solution of INS and RTK fix
+    "BESTNAV69": "PPP",  # PPP
 }
 """
 Map of fix values to descriptions.
-The keys in this map are a concatenation of NMEA/UBX
-message identifier and attribute value e.g.
+The keys in this map are a concatenation of
+message identity and attribute value e.g.
 GGA1: GGA + quality = 1
 NAV-STATUS3: NAV-STATUS + gpsFix = 3
 (valid for NMEA >=4)
diff --git a/src/pygpsclient/helpers.py b/src/pygpsclient/helpers.py
@@ -42,7 +42,7 @@
 from typing import Any, Literal
 
 from pygnssutils import version as PGVERSION
-from pynmeagps import WGS84_SMAJ_AXIS, NMEAMessage, haversine
+from pynmeagps import WGS84_SMAJ_AXIS, NMEAMessage, haversine, leapsecond
 from pynmeagps import version as NMEAVERSION
 from pyqgc import version as QGCVERSION
 from pyrtcm import version as RTCMVERSION
@@ -1447,20 +1447,24 @@ def valid_geom(geom: str) -> bool:
     return regexgeom.match(geom) is not None
 
 
-def wnotow2date(wno: int, tow: int) -> datetime:
+def wnotow2date(wno: int, tow: int, ls: int | NoneType = None) -> datetime:
     """
-    Get datetime from GPS Week number (Wno) and Time of Week (Tow).
+    Get datetime from GPS Week number (Wno), Time of Week (Tow)
+    and leapsecond offset.
 
     GPS Epoch 0 = 6th Jan 1980
 
     :param int wno: week number
     :param int tow: time of week
+    :param int ls: leapsecond offset
     :return: datetime
     :rtype: datetime
     """
 
+    if ls is None:
+        ls = leapsecond(datetime.now())
     dat = GPSEPOCH0 + timedelta(days=wno * 7)
-    dat += timedelta(seconds=tow)
+    dat += timedelta(seconds=tow - ls)
     return dat
 
 
diff --git a/src/pygpsclient/init_presets.py b/src/pygpsclient/init_presets.py
@@ -101,7 +101,7 @@
         "Unicore UM98n Save Current Configuration to NVM CONFIRM; saveconfig",
         "Unicore UM98n Force COLD reset CONFIRM; reset",
         "Unicore UM98n Enable NMEA on current port; gpgsa 1; gpgga 1; gpgll 1; gpgsv 4; gpvtg 1; gprmc 1",
-        "Unicore UM98n Enable UNI on current port; PVTSLNB 1; SATELLITEB 4; BESTNAV 1; STADOP 1",
+        "Unicore UM98n Enable UNI on current port; PVTSLNB 1; SATSINFOB 4; BESTNAV 1; STADOP 1",
         "Unicore UM98n Query Version; version",
         "Unicore UM98n Query Config; config",
         "Unicore UM98n Stop All Output on COM1; unlog COM1",
diff --git a/src/pygpsclient/skyview_frame.py b/src/pygpsclient/skyview_frame.py
@@ -110,12 +110,9 @@ def update_frame(self):
         """
 
         data = self.__app.gnss_status.gsv_data
-        show_unused = self.__app.configuration.get("unusedsat_b")
         siv = len(data)
-        siv = siv if show_unused else siv - unused_sats(data)
         sel = sum(1 for (_, _, ele, _, _, _) in data.values() if ele not in ("", None))
-        # ignore if cno are all zero and 'show_unused' is not set,
-        # or if elevation values are all null
+        # ignore if elevation values are all null
         if siv <= 0 or sel <= 0:
             return
 
@@ -125,11 +122,7 @@ def update_frame(self):
         for val in sorted(data.values(), key=lambda x: x[4]):  # sort by ascending C/N0
             try:
                 gnssId, prn, ele, azi, cno, _ = val
-                if (
-                    (cno == 0 and not show_unused)
-                    or ele in ("", None)
-                    or azi in ("", None)
-                ):
+                if ele in ("", None) or azi in ("", None):
                     continue
                 x, y = self._canvas.d2xy(int(azi), int(ele))
                 _, ol_col = GNSS_LIST[gnssId]
diff --git a/src/pygpsclient/uni_handler.py b/src/pygpsclient/uni_handler.py
@@ -1,8 +1,6 @@
 """
 uni_handler.py
 
-WORK IN PROGRESS - AWAITING PARSER DEVELOPMENT
-
 Unicore GNSS Protocol handler - handles all incoming UNI messages
 
 Parses individual UNI (Unicore UM98n GNSS) messages (using pyunignss library)
@@ -18,6 +16,31 @@
 """
 
 import logging
+from time import time
+
+from pyunigps import UNIMessage
+
+from pygpsclient.helpers import fix2desc, wnotow2date
+
+SATSINFO_GNSSID = {
+    0: 0,  # GPS
+    1: 6,  # GLONASS
+    2: 1,  # SBAS
+    3: 2,  # GAL
+    4: 3,  # BDS
+    5: 5,  # QZSS
+    6: 7,  # IRNSS
+}
+
+SATELLITE_GNSSID = {
+    0: 0,  # GPS
+    1: 6,  #  GLONASS
+    2: 1,  #  SBAS
+    5: 2,  #  GALILEO
+    6: 3,  #  BEIDOU
+    7: 5,  #  QZSS
+    9: 7,  #  NAVIC
+}
 
 
 class UNIHandler:
@@ -40,7 +63,7 @@ def __init__(self, app):
         self._parsed_data = None
 
     # pylint: disable=unused-argument
-    def process_data(self, raw_data: bytes, parsed_data: object):
+    def process_data(self, raw_data: bytes, parsed_data: UNIMessage):
         """
         Process relevant UNI message types
 
@@ -49,5 +72,169 @@ def process_data(self, raw_data: bytes, parsed_data: object):
         """
 
         if raw_data is None:
-            pass
+            return
         # self.logger.debug(f"data received {parsed_data.identity}")
+        self.__app.gnss_status.utc = wnotow2date(
+            parsed_data.wno, int(parsed_data.tow / 1000), parsed_data.leapsecond
+        )  # datetime.time
+        if parsed_data.identity in ("BESTNAV", "BESTNAVH"):
+            self._process_BESTNAV(parsed_data)
+        elif parsed_data.identity in ("PVTSLT",):
+            self._process_PVTSLT(parsed_data)
+        elif parsed_data.identity in (
+            "ADRNAV",
+            "ADRNAVH",
+            "PPPNAV",
+            "SPPNAV",
+            "SPPNAVH",
+        ):
+            self._process_ADRNAV(parsed_data)
+        elif parsed_data.identity in ("SATSINFO",):
+            self._process_SATSINFO(parsed_data)
+        elif parsed_data.identity in ("SATELLITE",):
+            self._process_SATELLITE(parsed_data)
+        elif parsed_data.identity in ("STADOP", "ADRDOP", "PPPDOP"):
+            self._process_STADOP(parsed_data)
+
+    def _process_pos(self, lat: float, lon: float, hmsl: float, undulation: float):
+        """
+        Process lat/lon/hmsl/hae.
+
+        :param float lat: lat
+        :param float lon: lon
+        :param float hmsl: hmsl in meters
+        :param float undulation: separation in meters
+        """
+
+        self.__app.gnss_status.lat = lat
+        self.__app.gnss_status.lon = lon
+        self.__app.gnss_status.alt = hmsl
+        self.__app.gnss_status.hae = hmsl + undulation
+
+    def _process_fix(self, postype: int):
+        """
+        Process fix type.
+
+        :param int postype: attribute representing fix type
+        """
+
+        self.__app.gnss_status.fix = fix2desc("BESTNAV", postype)
+        self.__app.gnss_status.diff_corr = ("RTK" in self.__app.gnss_status.fix) or (
+            "PPP" in self.__app.gnss_status.fix
+        )
+
+    def _process_BESTNAV(self, data: UNIMessage):
+        """
+        Process BESTNAV sentence -  Navigation position velocity time solution.
+
+        :param UNIMessage data: BESTNAV parsed message
+        """
+
+        self._process_pos(data.lat, data.lon, data.hmsl, data.undulation)
+        self._process_fix(data.postype)
+        self.__app.gnss_status.sip = data.numsolnsvs
+        self.__app.gnss_status.diff_age = data.diffage
+        self.__app.gnss_status.speed = data.horspd
+        self.__app.gnss_status.track = data.trkgnd
+        self.__app.gnss_status.diff_station = data.stationid
+
+    def _process_PVTSLT(self, data: UNIMessage):
+        """
+        Process PVTSLT sentence -  Navigation position velocity time solution.
+
+        :param UNIMessage data: PVTSLT parsed message
+        """
+
+        self._process_pos(
+            data.psrposlat, data.psrposlon, data.psrposhmsl, data.undulation
+        )
+        self._process_fix(data.psrpostype)
+        self.__app.gnss_status.sip = data.psrpossolnsvs
+        self.__app.gnss_status.speed = data.psrvelground
+        self.__app.gnss_status.track = data.headingdegree
+        self.__app.gnss_status.pdop = data.pdop
+        self.__app.gnss_status.hdop = data.hdop
+        self.__app.gnss_status.diff_age = data.bestpos_diffage
+
+    def _process_ADRNAV(self, data: UNIMessage):
+        """
+        Process ADRNAV, PPPNAV, SPPNAV sentences.
+
+        :param UNIMessage data: ADRNAV/PPPNAV/SPPNAV parsed message
+        """
+
+        self._process_pos(data.lat, data.lon, data.hmsl, data.undulation)
+        self._process_fix(data.postype)
+
+    def _process_SATSINFO(self, data: UNIMessage):
+        """
+        Process SATSINFO sentences - Space Vehicle Information for all
+        GNSS constellations.
+
+        :param UNIMessage data: SATSINFO parsed message
+        """
+
+        self.__app.gnss_status.gsv_data = {}
+        num_siv = int(data.numsat)
+        now = time()
+
+        for i in range(num_siv):
+            idx = f"_{i+1:02d}"
+            gnssId = SATSINFO_GNSSID[getattr(data, "sysstatus" + idx + "_01")]
+            svid = getattr(data, "prn" + idx)
+            elev = getattr(data, "elev" + idx)
+            azim = getattr(data, "azi" + idx)
+            cno = getattr(data, "cno" + idx + "_01")
+            self.__app.gnss_status.gsv_data[(gnssId, svid)] = (
+                gnssId,
+                svid,
+                elev,
+                azim,
+                cno,
+                now,
+            )
+
+        self.__app.gnss_status.siv = len(self.__app.gnss_status.gsv_data)
+
+    def _process_SATELLITE(self, data: UNIMessage):
+        """
+        Process SATELLITE sentences - Space Vehicle Information for a
+        specific GNSS constellation (7 in total).
+
+        :param UNIMessage data: SATSINFO parsed message
+        """
+
+        gnssId = SATELLITE_GNSSID[getattr(data, "gnss")]
+        for gnss, prn in list(self.__app.gnss_status.gsv_data.keys()):
+            if gnss == gnssId:
+                self.__app.gnss_status.gsv_data.pop((gnss, prn))
+
+        num_siv = int(data.numsat)
+        now = time()
+        for i in range(num_siv):
+            idx = f"_{i+1:02d}"
+            svid = getattr(data, "prn" + idx)
+            elev = getattr(data, "elv" + idx)
+            azim = getattr(data, "azi" + idx)
+            cno = 0  # cno not available from this message
+            self.__app.gnss_status.gsv_data[(gnssId, svid)] = (
+                gnssId,
+                svid,
+                elev,
+                azim,
+                cno,
+                now,
+            )
+
+        self.__app.gnss_status.siv = len(self.__app.gnss_status.gsv_data)
+
+    def _process_STADOP(self, data: UNIMessage):
+        """
+        Process STADOP/ADRDOP/PPPDOP sentences - DOP Information.
+
+        :param UNIMessage data: STADOP/ADRDOP/PPPDOP parsed message
+        """
+
+        self.__app.gnss_status.pdop = data.pdop
+        self.__app.gnss_status.hdop = data.hdop
+        self.__app.gnss_status.vdop = data.vdop
diff --git a/tests/test_static.py b/tests/test_static.py
@@ -406,17 +406,17 @@ def testwnotow2date(self):
             (2263, 518400),
         ]
         dats = [
-            "2023-01-01 00:00:00",
-            "2005-11-05 00:00:00",
-            "2020-08-20 00:00:00",
-            "2014-03-16 00:00:00",
-            "2023-05-21 00:00:00",
-            "2023-05-27 00:00:00",
+            "2022-12-31 23:59:42",
+            "2005-11-04 23:59:43",
+            "2020-08-19 23:59:44",
+            "2014-03-15 23:59:45",
+            "2023-05-20 23:59:46",
+            "2023-05-26 23:59:47",
         ]
         for i, (wno, tow) in enumerate(vals):
-            self.assertEqual(str(wnotow2date(wno, tow)), dats[i])
-        wno, tow = date2wnotow(datetime(2020, 4, 12))
-        self.assertEqual(wnotow2date(wno, tow), datetime(2020, 4, 12))
+            dat = wnotow2date(wno, tow, 18-i)
+            # print(f'"{dat}",')
+            self.assertEqual(str(dat), dats[i])
 
     def testbitsval(self):
         bits = [(7, 1), (8, 8), (22, 2), (24, 4), (40, 16)]
@@ -428,13 +428,11 @@ def testbitsval(self):
             self.assertEqual(res, EXPECTED_RESULT[i])
 
     def testparserxm(self):
-        EXPECTED_RESULT = [
-            ("0c00", datetime(1988, 3, 1, 7, 40)),
-            ("290900", datetime(1988, 7, 4, 2, 40)),
-        ]
+        EXPECTED_RESULT = [('0c00', datetime(1988, 3, 1, 7, 39, 42)), ('290900', datetime(1988, 7, 4, 2, 39, 42))]
         RXM_SPARTNKEY = b"\xb5b\x026\x19\x00\x01\x02\x00\x00\x00\x02+\x00\xd0Y\xc8\r\x00\x03+\x00\x00\xdfl\x0e\x0c\x00)\t\x00D;"
         msg = UBXReader.parse(RXM_SPARTNKEY)
         res = parse_rxmspartnkey(msg)
+        # print(f'"{res}",')
         self.assertEqual(res, EXPECTED_RESULT)
 
     def testmapqcompress(self):
