canvod-readers¶

Purpose¶

The canvod-readers package provides validated parsers for GNSS observation data. It transforms raw receiver files into analysis-ready xarray Datasets, serving as the data ingestion layer for GNSS Transmissometry (GNSS-T) analysis.

Supported Formats at a Glance¶

Design¶

Data flow¶

graph TD
    A1["RINEX v3 File (.rnx)"] --> B1["Rnxv3Obs (+ SP3/CLK)"]
    A2["SBF File (.sbf)"] --> B2["SbfReader"]
    B1 --> C["validate_dataset()"]
    B2 --> C
    C --> D["`**xarray.Dataset**
    epoch x sid`"]
    B2 --> E["`**Metadata Dataset**
    DOP, PVT, theta, phi`"]
    D --> F["Downstream Analysis"]
    E --> F

Contract-Based Design¶

All readers implement the GNSSDataReader base class — a Pydantic BaseModel + ABC that provides file path validation, model configuration, and a consistent interface:

from pydantic import BaseModel, ConfigDict, field_validator
from abc import ABC, abstractmethod
import xarray as xr

class GNSSDataReader(BaseModel, ABC):
    """Base class for all GNSS data format readers."""

    model_config = ConfigDict(arbitrary_types_allowed=True)
    fpath: Path  # Validated at construction time

    @abstractmethod
    def to_ds(self, **kwargs) -> xr.Dataset:
        """Convert to xarray.Dataset (epoch × sid)."""

    @abstractmethod
    def iter_epochs(self):
        """Iterate through epochs."""

    @property
    @abstractmethod
    def file_hash(self) -> str:
        """SHA-256 hash for deduplication."""

    def to_ds_and_auxiliary(
        self, **kwargs
    ) -> tuple[xr.Dataset, dict[str, xr.Dataset]]:
        """Single-pass scan: obs dataset + any auxiliary datasets.

        Default returns empty aux dict.
        SbfReader overrides for one-pass binary decode.
        """
        return self.to_ds(**kwargs), {}

Subclasses only need to inherit from GNSSDataReader — no separate BaseModel import, no fpath field, no file validation boilerplate.

Full architecture

Usage Examples¶

from canvod.readers import Rnxv3Obs

reader = Rnxv3Obs(fpath="station.25o")
ds = reader.to_ds(keep_data_vars=["SNR"])

# Filter L-band signals
l_band = ds.where(ds.band.isin(["L1", "L2", "L5"]), drop=True)

from canvod.readers.sbf import SbfReader

reader = SbfReader(fpath="rref001a00.25_")
obs_ds, aux = reader.to_ds_and_auxiliary(keep_data_vars=["SNR"])
meta_ds = aux["sbf_obs"]

# Polar angle filter: elevation ≥ 20°
snr_filtered = obs_ds["SNR"].where(meta_ds["theta"] <= 70)

ds = reader.to_ds()

for system in ["G", "R", "E", "C"]:
    sys_ds = ds.where(ds.system == system, drop=True)
    mean_snr = sys_ds.SNR.mean(dim=["epoch", "sid"])
    print(f"{system}: {mean_snr:.2f} dB")

from canvodpy import ReaderFactory

# By name (works for all registered readers)
reader = ReaderFactory.create("rinex3", fpath="station.25o")

# Auto-detect RINEX v2/v3 from file header
reader = ReaderFactory.create_from_file("station.25o")

# Both produce identical (epoch × sid) datasets
ds = reader.to_ds(keep_data_vars=["SNR"])

import xarray as xr
from pathlib import Path

datasets = [
    Rnxv3Obs(fpath=f).to_ds(keep_data_vars=["SNR"])
    for f in sorted(Path("/data/").glob("*.rnx"))
]

time_series = xr.concat(datasets, dim="epoch")

Key Components¶

Performance¶

Single-Pass Parser¶

Rnxv3Obs uses a single-pass parser (_create_dataset_single_pass) that pre-computes the full Signal ID (SID) space from the RINEX header and fills pre-allocated NumPy arrays in one pass over the file. This avoids the overhead of:

• Per-observation object allocation — inline string parsing (_parse_obs_fast) replaces Pydantic model instantiation
• Repeated signal ID lookups — a pre-built lookup table maps (SV, obs_code) → array index directly
• Redundant header re-parsing — SIDs are derived once from header metadata via _precompute_sids_from_header()

Tips¶