diff --git a/python/PiFinder/gps_ubx.py b/python/PiFinder/gps_ubx.py index dc56bf07..95301495 100644 --- a/python/PiFinder/gps_ubx.py +++ b/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 diff --git a/python/PiFinder/gps_ubx_parser.py b/python/PiFinder/gps_ubx_parser.py index f0bb7da4..0c41f41b 100644 --- a/python/PiFinder/gps_ubx_parser.py +++ b/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, diff --git a/python/tests/test_gps_ubx_parser.py b/python/tests/test_gps_ubx_parser.py new file mode 100644 index 00000000..c4265d7e --- /dev/null +++ b/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("