pygnssutils/examples/process_rxmsfrbx_frames.py at e9fcad7b0ce6d895b553945adbb449cbf0ab1a1b · semuconsulting/pygnssutils · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
"""
process_rxmsfrbx_frames.py

Demo script to illustrate acquisition and collation of
raw NAV subframes from a UBX RXM-SFRBX data log using
the pygnssutils.RawNav utility class.

This collates the following subframes for each SV
into a single RawNav object.

 - subframe 1: clock corrections, sv health, etc.
 - subframes 2 & 3: ephemerides
 - subframe 4 page 18: (optional) ionospheric and time corrections

(NB: subframe data dictionaries only currently defined
for GPS LNAV, but others can be added)

python3 process_rxmsfrbx_frames.py infile=pygpsdata-rxmsfrbx.log

Created on 21 Apr 2026

:author: semuadmin (Steve Smith)
:copyright: semuadmin © 2026
:license: BSD 3-Clause
"""

# pylint: disable=no-member

from sys import argv

from pyubx2 import UBXMessage

from pygnssutils import GPS, GNSSReader, RawNav, RINEXProcessingError
from pygnssutils.rinex_subframes_gps import (
    GPS_LNAV_SUBFRAME_1,
    GPS_LNAV_SUBFRAME_2,
    GPS_LNAV_SUBFRAME_3,
    GPS_LNAV_SUBFRAME_4_P18,
)

SFR1 = 1
SFR2 = 2
SFR3 = 4
SFR4 = 8
TARGET_SFR = SFR1 | SFR2 | SFR3 # | SFR4


def main(**kwargs):
    """
    Main routine.
    """

    infile = kwargs["infile"]

    tot = 0
    sfr = 0
    err = 0
    lnavs = {}
    navs = {}
    with open(infile, "rb") as stream:
        gnr = GNSSReader(stream)
        for _, parsed in gnr:
            if not isinstance(parsed, UBXMessage):
                continue
            if parsed.identity != "RXM-SFRBX":  # UBX RXM-SFRBX (raw subframes) only
                continue
            tot += 1

            sfrdata = RawNav.process_rxm_sfrbx(parsed)
            if sfrdata.get("gnss", "") != GPS or sfrdata.get("sigid", "") not in ("1C","2L"):
                continue

            gnss = sfrdata["gnss"]
            svid = sfrdata["svid"]
            sigid = sfrdata["sigid"]
            subframeid = sfrdata["subframeid"]
            svcode = sfrdata.get("svcode", 0)
            subframe = sfrdata["subframe"]

            try:
                navs[(gnss, svid)] = navs.get((gnss, svid), RawNav(gnss, svid, sigid))
                nav = navs[(gnss, svid)]
                if subframeid == 1:  # clock parameters, sv health, etc.
                    wn = subframe >> 230 & 0b1111111111
                    toc = (subframe >> 66 & 0b1111111111111111) * 16
                    tow = subframe >> 253 & 0b11111111111111111
                    # print(f"{tow=}, {wn=}, {toc=}")
                    nav.parse(subframe, GPS_LNAV_SUBFRAME_1)
                elif subframeid == 2:  # ephemerides
                    nav.parse(subframe, GPS_LNAV_SUBFRAME_2)
                elif subframeid == 3:  # ephemerides
                    nav.parse(subframe, GPS_LNAV_SUBFRAME_3)
                elif subframeid == 4:
                    if svcode == 56:  # page 18, ionospheric corrections
                        nav.parse(subframe, GPS_LNAV_SUBFRAME_4_P18)
                if nav.sfracq & 0b111 == TARGET_SFR:
                    frame = navs.pop((gnss, svid))
                    print(f"{str(frame)}\n")
                    lnavs[svid] = lnavs.get(svid, 0) + 1
                sfr += 1
            except RINEXProcessingError:
                err += 1

    if tot:
        print(f"Total RXM-SFRBX records: {tot}, errors: {err} ({err*100/tot:.0f}%)")
        print(f"Total GPS LNAV subrames processed: {sfr}")
        print(
            (
                f"Total GPS LNAV full frames acquired (sfracq={TARGET_SFR}): "
                f"{sum(lnavs.values())}"
            )
        )
        print(
            (
                f"Breakdown of GPS LNAV frames by SV: {sorted(lnavs.items())}   "
                f"Sum: {sum(lnavs.values())}"
            )
        )
    else:
        print("No RXM-SFRBX records in file")


if __name__ == "__main__":

    main(**dict(arg.split("=") for arg in argv[1:]))