es_sfgtools.novatel_tools.rangea_parser module

RANGEA ASCII Log Parser

This module provides Python functions to parse NovAtel RANGEA ASCII log strings into structured observation data, inspired by the Go-lang GNSS tools implementation using novatelascii.DeserializeRANGEA and observation.Epoch.

The RANGEA log contains GNSS pseudorange, carrier phase, Doppler, and C/N0 measurements for all tracked satellites across multiple constellations.

class es_sfgtools.novatel_tools.rangea_parser.GNSSEpoch(*, time: datetime, gps_week: int, gps_seconds: float, satellites: Dict[Tuple[int, int], Satellite] = None, receiver_status: str = '', num_observations: int = 0)

Bases: BaseModel

A GNSS observation epoch containing all satellite measurements at one time.

This is the Python equivalent of Go’s observation.Epoch structure. An epoch represents all GNSS observations recorded at a single instant, typically at the receiver’s measurement rate (e.g., 1 Hz, 10 Hz).

time

UTC timestamp of the epoch

Type:: datetime.datetime

gps_week

GPS week number

Type:: int

gps_seconds

Seconds into the GPS week

Type:: float

satellites

Dictionary mapping (system, prn) tuple to Satellite

Type:: Dict[Tuple[int, int], es_sfgtools.novatel_tools.rangea_parser.Satellite]

receiver_status

Raw receiver status word from header

Type:: str

num_observations

Total number of observation records

Type:: int

add_satellite(sat: Satellite) → None: Add or update a satellite in this epoch.

get_satellite(system: GNSSSystem, prn: int) → Satellite | None: Get a satellite by system and PRN.

get_systems() → List[GNSSSystem]: Return list of GNSS systems present in this epoch.

gps_seconds: float

gps_week: int

model_computed_fields = {'satellite_count': ComputedFieldInfo(wrapped_property=<property object>, return_type=<class 'int'>, alias=None, alias_priority=None, title=None, field_title_generator=None, description='Return the number of unique satellites in this epoch.', deprecated=None, examples=None, json_schema_extra=None, repr=True)}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config = {'frozen': False}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields = {'gps_seconds': FieldInfo(annotation=float, required=True), 'gps_week': FieldInfo(annotation=int, required=True), 'num_observations': FieldInfo(annotation=int, required=False, default=0), 'receiver_status': FieldInfo(annotation=str, required=False, default=''), 'satellites': FieldInfo(annotation=Dict[Tuple[int, int], Satellite], required=False, default_factory=dict), 'time': FieldInfo(annotation=datetime, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

num_observations: int

receiver_status: str

property satellite_count: int: Return the number of unique satellites in this epoch.

satellites: Dict[Tuple[int, int], Satellite]

time: datetime

class es_sfgtools.novatel_tools.rangea_parser.GNSSSystem(value)

Bases: IntEnum

GNSS constellation identifiers from NovAtel channel tracking status.

BEIDOU = 5

GALILEO = 3

GLONASS = 1

GPS = 0

NAVIC = 7

QZSS = 6

SBAS = 2

class es_sfgtools.novatel_tools.rangea_parser.Observation(*, signal_type: int, pseudorange: float, pseudorange_std: float, carrier_phase: float, carrier_phase_std: float, doppler: float, cn0: float, locktime: float, tracking_status: int, half_cycle_ambiguity: bool = False, phase_lock: bool = True, code_lock: bool = True, parity_known: bool = True)

Bases: BaseModel

A single GNSS observation for one signal from one satellite.

This corresponds to a single observation record within a RANGEA message, containing pseudorange, carrier phase, Doppler, and signal quality metrics.

carrier_phase: float

carrier_phase_std: float

cn0: float

code_lock: bool

doppler: float

half_cycle_ambiguity: bool

locktime: float

model_computed_fields = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config = {'frozen': False}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields = {'carrier_phase': FieldInfo(annotation=float, required=True, title='Carrier Phase', description='Accumulated Doppler range (ADR) in cycles'), 'carrier_phase_std': FieldInfo(annotation=float, required=True, title='Carrier Phase Std', description='Carrier phase standard deviation in cycles'), 'cn0': FieldInfo(annotation=float, required=True, title='C/N0', description='Carrier-to-noise density ratio in dB-Hz'), 'code_lock': FieldInfo(annotation=bool, required=False, default=True, title='Code Lock', description='True if code is locked'), 'doppler': FieldInfo(annotation=float, required=True, title='Doppler', description='Doppler frequency shift in Hz'), 'half_cycle_ambiguity': FieldInfo(annotation=bool, required=False, default=False, title='Half Cycle Ambiguity', description='True if half-cycle ambiguity is present'), 'locktime': FieldInfo(annotation=float, required=True, title='Lock Time', description='Continuous tracking time in seconds'), 'parity_known': FieldInfo(annotation=bool, required=False, default=True, title='Parity Known', description='True if parity is known (for navigation data)'), 'phase_lock': FieldInfo(annotation=bool, required=False, default=True, title='Phase Lock', description='True if phase is locked'), 'pseudorange': FieldInfo(annotation=float, required=True, title='Pseudorange', description='Pseudorange measurement in meters'), 'pseudorange_std': FieldInfo(annotation=float, required=True, title='Pseudorange Std', description='Pseudorange standard deviation in meters'), 'signal_type': FieldInfo(annotation=int, required=True, title='Signal Type Identifier'), 'tracking_status': FieldInfo(annotation=int, required=True, title='Tracking Status', description='Raw 32-bit channel tracking status word')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

parity_known: bool

phase_lock: bool

pseudorange: float

pseudorange_std: float

signal_type: int

tracking_status: int

class es_sfgtools.novatel_tools.rangea_parser.Satellite(*, system: GNSSSystem, prn: int, fcn: int = 0, observations: Dict[int, Observation] = None)

Bases: BaseModel

GNSS satellite with all its observations.

A satellite may have multiple observations for different signals (e.g., GPS satellite might have L1CA, L2C, and L5 observations).

system

GNSS constellation (GPS, GLONASS, Galileo, etc.)

Type:: es_sfgtools.novatel_tools.rangea_parser.GNSSSystem

prn

Satellite PRN number (or slot for GLONASS)

Type:: int

fcn

GLONASS frequency channel number (-7 to +6), 0 for other systems

Type:: int

observations

Dictionary mapping signal type to Observation

Type:: Dict[int, es_sfgtools.novatel_tools.rangea_parser.Observation]

add_observation(obs: Observation) → None: Add an observation for a specific signal type.

fcn: int

model_computed_fields = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config = {'frozen': False}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields = {'fcn': FieldInfo(annotation=int, required=False, default=0), 'observations': FieldInfo(annotation=Dict[int, Observation], required=False, default_factory=dict), 'prn': FieldInfo(annotation=int, required=True), 'system': FieldInfo(annotation=GNSSSystem, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

observations: Dict[int, Observation]

prn: int

system: GNSSSystem

class es_sfgtools.novatel_tools.rangea_parser.SignalType(value)

Bases: IntEnum

Common GNSS signal types (simplified mapping).

B1I = 0

B2I = 2

B3I = 6

E1 = 2

E5A = 12

E5B = 17

L1C = 17

L1CA = 0

L2C = 9

L2P = 5

L5Q = 14

es_sfgtools.novatel_tools.rangea_parser.deserialize_rangea(rangea_string: str) → GNSSEpoch

Parse a NovAtel RANGEA ASCII log string into an Epoch object.

This function is the Python equivalent of the Go code:: rangea, err := novatelascii.DeserializeRANGEA(m.Data) epoch, err := rangea.SerializeGNSSEpoch(m.Time())
RANGEA Format:: #RANGEA,<header>;num_obs,<obs1>,…,<obsN>*checksum
Each observation has 10 fields:: prn, glo_freq, psr, psr_std, adr, adr_std, dopp, cn0, locktime, ch_tr_status

Parameters:: rangea_string – Complete RANGEA ASCII log string including header and checksum
Returns:: Epoch object containing all parsed satellite observations
Raises:: ValueError – If the string cannot be parsed as a valid RANGEA message

Example

>>> rangea = "#RANGEA,USB2,0,73.5,FINESTEERING,2379,414835.000,..."
>>> epoch = deserialize_rangea(rangea)
>>> print(f"Epoch time: {epoch.time}, satellites: {epoch.satellite_count}")

es_sfgtools.novatel_tools.rangea_parser.epoch_to_dict(epoch: GNSSEpoch) → dict

Convert an Epoch object to a dictionary for serialization.

Parameters:: epoch – Epoch object to convert
Returns:: Dictionary representation suitable for JSON serialization

es_sfgtools.novatel_tools.rangea_parser.extract_rangea_from_qcpin(source: str | Path) → List[GNSSEpoch]

Extract and parse all RANGEA logs from a QC PIN file.

This function loads a QC PIN JSON file and searches through it for NOV_RANGE observations containing raw RANGEA strings, parses them into GNSSEpoch objects, and returns all unique epochs.

The JSON structure is expected to have entries like:

{: “interrogation”: {“observations”: {“NOV_RANGE”: {“raw”: “#RANGEA,…”, “time”: {…}}}}, “007BE1”: {“observations”: {“NOV_RANGE”: {“raw”: “#RANGEA,…”, “time”: {…}}}}, …

}

Parameters:: source – Path to the QC PIN file in JSON format
Returns:: List of unique GNSSEpoch objects, deduplicated by GPS week/seconds. Returns empty list if file cannot be read or contains no valid RANGEA logs.

Example

>>> epochs = extract_rangea_from_qcpin("/path/to/file.pin")
>>> print(f"Found {len(epochs)} unique epochs")

es_sfgtools.novatel_tools.rangea_parser.extract_rangea_strings_from_qcpin(source: str | Path) → List[str]

Extract raw RANGEA strings from a QC PIN file.

This function loads a QC PIN JSON file and searches through it for NOV_RANGE observations containing raw RANGEA strings, returning a list of all found RANGEA strings without parsing them into epochs.

Parameters:: source – Path to the QC PIN file in JSON format
Returns:: List of raw RANGEA strings found in the file. Returns empty list if file cannot be read or contains no valid RANGEA logs.