fix(gps): correct UBX NAV-SVINFO and NAV-SAT satellite decoding

python/PiFinder/gps_ubx.py

@@ -32,14 +32,18 @@ async def process_messages(
         elif msg_class == "NAV-SVINFO" and not got_sat_update:
             # Fallback satellite info if NAV-SAT not available
             if "nSat" in msg:
-                # uSat is also in the message but contains stale info
                 sats_seen = msg["nSat"]
+                sats_used = msg["uSat"]
                 sats[0] = sats_seen
+                sats[1] = sats_used
                 gps_queue.put(("satellites", tuple(sats)))
-                logger.debug("Number of sats (SVINFO) seen: %i", sats_seen)
+                logger.debug(
+                    "Number of sats (SVINFO) seen: %i, used: %i", sats_seen, sats_used
+                )
 
         elif msg_class == "NAV-SAT":
             # Preferred satellite info source - not seen in the current pifinder gps versions
+            got_sat_update = True
             sats_seen = msg["nSat"]
             sats_used = sum(
                 1 for sat in msg.get("satellites", []) if sat.get("used", False)
@@ -95,6 +99,9 @@ async def process_messages(
                 logger.debug(f"TIMEGPS message does not qualify: {msg}")
 
         elif msg_class == "NAV-PVT":
+            if "numSV" in msg:
+                sats[1] = msg["numSV"]
+                gps_queue.put(("satellites", tuple(sats)))
             if all(k in msg for k in ["lat", "lon", "altHAE", "hAcc", "vAcc"]):
                 if not gps_locked and msg["hAcc"] < MAX_GPS_ERROR:
                     gps_locked = True

python/PiFinder/gps_ubx_parser.py

@@ -14,6 +14,11 @@
 
 logger = logging.getLogger("GPS.parser")
 
+# u-blox quality indicator (qualityInd / flags bits 0-2) value at which the
+# signal is code locked; below this the receiver is still searching and any
+# reported C/N0 is an unconfirmed acquisition candidate.
+QUALITY_CODE_LOCKED = 4
+
 
 class UBXClass(IntEnum):
     NAV = 0x01
@@ -312,28 +317,27 @@ def _parse_nav_sat(self, data: bytes) -> dict:
             gnssId = data[offset]
             svId = data[offset + 1]
             cno = data[offset + 2]
-            elev = data[offset + 3]
-            azim = int.from_bytes(data[offset + 4 : offset + 6], "little")
-            flags = data[
-                offset + 8
-            ]  # Warning this is a 4 byte field of flags, we're only using the first byte
-            # lowest 3 bits are a quality indicator and according to
-            # https://portal.u-blox.com/s/question/0D52p000097B0bFCAS/interpretation-of-signal-quality-indicator-in-ubxnavsat
-            # the 0-7 values from an ordered scale. So taking 3 as the threshold below.q
-            satellites.append(
-                {
-                    "id": svId,
-                    "system": gnssId,
-                    "signal": cno,
-                    "elevation": elev,
-                    "azimuth": azim,
-                    "used": (flags & 0x07) > 3,  # lowest 3 bits are used for the status
-                    "flags": flags & 0x07,
-                }
-            )
+            elev = int.from_bytes(data[offset + 3 : offset + 4], "little", signed=True)
+            azim = int.from_bytes(data[offset + 4 : offset + 6], "little", signed=True)
+            flags = data[offset + 8]  # X4 bitfield, only the first byte is needed here:
+            # bits 0-2 are the quality indicator, bit 3 is svUsed
+            # NAV-SAT lists every known satellite, including acquisition
+            # candidates with an estimated C/N0; only count tracked signals
+            if (flags & 0x07) >= QUALITY_CODE_LOCKED:
+                satellites.append(
+                    {
+                        "id": svId,
+                        "system": gnssId,
+                        "signal": cno,
+                        "elevation": elev,
+                        "azimuth": azim,
+                        "used": bool(flags & 0x08),
+                        "quality": flags & 0x07,
+                    }
+                )
         result = {
             "class": "NAV-SAT",
-            "nSat": sum(1 for sat in satellites),
+            "nSat": len(satellites),
             "satellites": satellites,
         }
         logger.debug(f"NAV-SAT result: {result}")
@@ -358,18 +362,23 @@ def _parse_nav_svinfo(self, data: bytes) -> dict:
                 logger.warning(f"SVINFO: Message truncated at satellite {i}")
                 break
 
-            svid = data[offset]
-            flags = data[offset + 1]
-            quality = data[offset + 2]
-            cno = data[offset + 3]
-            elev = data[offset + 4]
-            azim = int.from_bytes(data[offset + 6 : offset + 8], "little")
+            # Repeated block layout: chn, svid, flags, quality, cno, elev, azim, prRes
+            svid = data[offset + 1]
+            flags = data[offset + 2]
+            quality = data[offset + 3]
+            cno = data[offset + 4]
+            elev = int.from_bytes(data[offset + 5 : offset + 6], "little", signed=True)
+            azim = int.from_bytes(data[offset + 6 : offset + 8], "little", signed=True)
 
             is_used = bool(flags & 0x01)
             if is_used:
                 used_sats += 1
 
-            if cno > 0:
+            # During cold-start acquisition the receiver reports estimated
+            # C/N0 for candidates it is still searching for; counting those
+            # makes the seen count start high and sink as they fail to
+            # confirm. Only quality >= 4 (code locked) is really tracked.
+            if quality >= QUALITY_CODE_LOCKED:
                 satellites.append(
                     {
                         "id": svid,

python/tests/test_gps_ubx_parser.py

@@ -0,0 +1,100 @@
+import struct
+
+import pytest
+
+from PiFinder.gps_ubx_parser import UBXParser
+
+
+def make_svinfo_channel(chn, svid, flags, quality, cno, elev, azim, prres=0):
+    """Build one 12-byte UBX-NAV-SVINFO repeated block."""
+    return struct.pack("<BBBBBbhi", chn, svid, flags, quality, cno, elev, azim, prres)
+
+
+def make_svinfo_payload(channels):
+    header = struct.pack("<IBBH", 1000, len(channels), 0, 0)
+    return header + b"".join(make_svinfo_channel(*ch) for ch in channels)
+
+
+@pytest.fixture
+def parser():
+    return UBXParser.__new__(UBXParser)
+
+
+@pytest.mark.unit
+def test_svinfo_field_alignment(parser):
+    # chn, svid, flags (bit0 = used in fix), quality, cno, elev, azim
+    payload = make_svinfo_payload(
+        [
+            (4, 17, 0x0D, 4, 27, 45, 180),
+            (2, 13, 0x1C, 4, 15, -5, 300),
+        ]
+    )
+    result = parser._parse_nav_svinfo(payload)
+
+    assert result["class"] == "NAV-SVINFO"
+    assert result["nSat"] == 2
+    assert result["uSat"] == 1
+
+    sat13, sat17 = result["satellites"]
+    assert sat17["id"] == 17
+    assert sat17["signal"] == 27
+    assert sat17["elevation"] == 45
+    assert sat17["azimuth"] == 180
+    assert sat17["used"] is True
+
+    assert sat13["id"] == 13
+    assert sat13["signal"] == 15
+    assert sat13["elevation"] == -5
+    assert sat13["used"] is False
+
+
+@pytest.mark.unit
+def test_svinfo_only_code_locked_counted_as_seen(parser):
+    # Idle channels (e.g. SBAS) and cold-start acquisition candidates
+    # (quality < 4 with an estimated cno) must not inflate the seen count.
+    payload = make_svinfo_payload(
+        [
+            (0, 14, 0x0D, 4, 26, 30, 90),
+            (7, 25, 0x00, 2, 9, 0, 0),
+            (11, 120, 0x10, 1, 0, 0, 0),
+            (5, 193, 0x10, 1, 0, 0, 0),
+        ]
+    )
+    result = parser._parse_nav_svinfo(payload)
+
+    assert result["nSat"] == 1
+    assert result["uSat"] == 1
+    assert result["satellites"][0]["id"] == 14
+
+
+@pytest.mark.unit
+def test_svinfo_too_short(parser):
+    assert "error" in parser._parse_nav_svinfo(b"\x00" * 4)
+
+
+def make_nav_sat_block(gnss_id, sv_id, cno, elev, azim, flags):
+    return struct.pack("<BBBbhhI", gnss_id, sv_id, cno, elev, azim, 0, flags)
+
+
+def make_nav_sat_payload(svs):
+    header = struct.pack("<IBBH", 1000, 1, len(svs), 0)
+    return header + b"".join(make_nav_sat_block(*sv) for sv in svs)
+
+
+@pytest.mark.unit
+def test_nav_sat_used_from_svused_bit(parser):
+    # flags bits 0-2 = quality indicator, bit 3 = svUsed
+    payload = make_nav_sat_payload(
+        [
+            (0, 17, 27, 45, 180, 0x0C),  # quality 4, used
+            (0, 13, 15, -5, 300, 0x04),  # quality 4, tracked but not used
+            (0, 25, 9, 0, 0, 0x02),  # acquisition candidate: not seen
+            (6, 3, 0, 0, 0, 0x01),  # searching, no signal: not seen
+        ]
+    )
+    result = parser._parse_nav_sat(payload)
+
+    assert result["nSat"] == 2
+    sat17, sat13 = result["satellites"]
+    assert (sat17["id"], sat17["used"], sat17["quality"]) == (17, True, 4)
+    assert (sat13["id"], sat13["used"], sat13["elevation"]) == (13, False, -5)