canvod.adapters API Reference¶

class ObservationsAdapter(_Adapter, ABC):
    """Observations: the tool's observation files <-> canvodpy observations."""

    @abstractmethod
    def reader(self, path: Path) -> GNSSDataReader:
        """A canvodpy reader of one of the tool's observation files."""

    @abstractmethod
    def convert_from_canvodpy(self, obs_ds: xr.Dataset) -> xr.Dataset:
        """canvodpy observations in the tool's data structure."""

    @abstractmethod
    def save(self, tool_ds: xr.Dataset, path: Path) -> None:
        """Write a dataset in the tool's data structure to its file format."""

    def import_file(self, path: Path | str, keep_data_vars: list[str] | None = None) -> xr.Dataset:
        """Read one of the tool's observation files as canvodpy observations.

        Parameters
        ----------
        path : Path or str
            The tool's observation file.
        keep_data_vars : list of str, optional
            Data variables to keep and require; ``None`` keeps all and
            requires ``SNR`` (:func:`canvod.readers.validate_dataset`).

        Returns
        -------
        xarray.Dataset
            canvodpy observations with provenance attributes.

        Raises
        ------
        ValueError
            If the result does not meet the observations contract.
        """
        path = Path(path)
        ds = self.reader(path).to_ds(keep_data_vars=keep_data_vars)
        ds.attrs.update(Provenance.now(self.tool, "import", path.name).to_attrs())
        validate_dataset(ds, required_vars=keep_data_vars)
        return ds

    def export_file(self, obs_ds: xr.Dataset, path: Path | str, source: str) -> Path:
        """Write canvodpy observations as one of the tool's files.

        Parameters
        ----------
        obs_ds : xarray.Dataset
            canvodpy observations.
        path : Path or str
            File to write.
        source : str
            What ``obs_ds`` is, recorded as provenance (e.g. a store group).

        Returns
        -------
        Path
            ``path``.
        """
        path = Path(path)
        tool_ds = self.convert_from_canvodpy(obs_ds)
        tool_ds.attrs.update(Provenance.now(self.tool, "export", source).to_attrs())
        self.save(tool_ds, path)
        return path

class VodAdapter(_Adapter, ABC):
    """VOD: the tool's VOD files <-> canvodpy VOD."""

    @abstractmethod
    def open(self, path: Path) -> xr.Dataset:
        """Read one of the tool's VOD files in its own data structure."""

    @abstractmethod
    def save(self, tool_ds: xr.Dataset, path: Path) -> None:
        """Write a dataset in the tool's data structure to its file format."""

    @abstractmethod
    def convert_to_canvodpy(self, tool_ds: xr.Dataset) -> xr.Dataset:
        """VOD in the tool's data structure as canvodpy VOD."""

    @abstractmethod
    def convert_from_canvodpy(self, vod_ds: xr.Dataset) -> xr.Dataset:
        """canvodpy VOD in the tool's data structure."""

    def import_dataset(self, tool_ds: xr.Dataset, source: str) -> xr.Dataset:
        """The tool's VOD as canvodpy VOD.

        Parameters
        ----------
        tool_ds : xarray.Dataset
            VOD in the tool's data structure.
        source : str
            What ``tool_ds`` is, recorded as provenance (e.g. a file name).

        Returns
        -------
        xarray.Dataset
            canvodpy VOD with provenance attributes.

        Raises
        ------
        ValueError
            If the result does not meet the VOD dataset contract.
        """
        ds = self.convert_to_canvodpy(tool_ds)
        ds.attrs.update(Provenance.now(self.tool, "import", source).to_attrs())
        validate_vod_dataset(ds)
        return ds

    def import_file(self, path: Path | str) -> xr.Dataset:
        """Read one of the tool's VOD files as canvodpy VOD.

        See :meth:`import_dataset`.
        """
        path = Path(path)
        return self.import_dataset(self.open(path), source=path.name)

    def export_dataset(self, vod_ds: xr.Dataset, source: str) -> xr.Dataset:
        """canvodpy VOD in the tool's data structure, with provenance.

        Parameters
        ----------
        vod_ds : xarray.Dataset
            canvodpy VOD.
        source : str
            What ``vod_ds`` is, recorded as provenance (e.g. a store group).
        """
        tool_ds = self.convert_from_canvodpy(vod_ds)
        tool_ds.attrs.update(Provenance.now(self.tool, "export", source).to_attrs())
        return tool_ds

    def export_file(self, vod_ds: xr.Dataset, path: Path | str, source: str) -> Path:
        """Write canvodpy VOD as one of the tool's VOD files.

        See :meth:`export_dataset`. Returns ``path``.
        """
        path = Path(path)
        self.save(self.export_dataset(vod_ds, source), path)
        return path

class Tool(BaseModel):
    """A program canvodpy exchanges data with.

    Parameters
    ----------
    name : str
        Short name, also the model layer of imported VOD in the VOD store
        (e.g. ``"gnssvod"``).
    distribution : str
        Name of its installed Python distribution, to record its version.
    url : str
        Where to find it.
    """

    model_config = ConfigDict(frozen=True)

    name: str
    distribution: str
    url: str

class Provenance(BaseModel):
    """Which tool data was converted from or to, by what, when.

    Parameters
    ----------
    tool : str
        Name of the other tool.
    tool_url : str
        Where to find it.
    tool_version : str
        Its installed version at conversion time, or :data:`NOT_INSTALLED`.
    adapter_version : str
        Version of canvod-adapters.
    direction : {"import", "export"}
        ``"import"``: from the tool into canvodpy; ``"export"``: the reverse.
    source : str
        What was converted, e.g. a file name or a VOD store group.
    created : datetime
        When (UTC).
    """

    model_config = ConfigDict(frozen=True)

    tool: str
    tool_url: str
    tool_version: str
    adapter_version: str
    direction: Literal["import", "export"]
    source: str
    created: datetime

    @classmethod
    def now(cls, tool: Tool, direction: Literal["import", "export"], source: str) -> Provenance:
        """Provenance of a conversion running now, with installed versions."""
        return cls(
            tool=tool.name,
            tool_url=tool.url,
            tool_version=package_version(tool.distribution),
            adapter_version=package_version("canvod-adapters"),
            direction=direction,
            source=source,
            created=datetime.now(UTC),
        )

    def to_attrs(self) -> dict[str, str]:
        """Dataset attributes, each name prefixed with :data:`ATTR_PREFIX`."""
        return {
            f"{ATTR_PREFIX}{key}": str(value)
            for key, value in self.model_dump(mode="json").items()
        }

    @classmethod
    def from_attrs(cls, attrs: dict[str, Any]) -> Provenance:
        """Read provenance back from dataset attributes.

        Raises
        ------
        pydantic.ValidationError
            If the attributes hold no complete provenance.
        """
        return cls.model_validate(
            {
                key.removeprefix(ATTR_PREFIX): value
                for key, value in attrs.items()
                if key.startswith(ATTR_PREFIX)
            }
        )

class GnssvodObservations(ObservationsAdapter):
    """gnssvod observation files <-> canvodpy observations.

    Parameters
    ----------
    time_system : str
        Time scale of the epochs of the gnssvod files to import, as in the
        RINEX files they came from (e.g. ``"GPS"``); see
        :class:`~canvod.adapters.gnssvod.reader.GnssvodObsReader`.

    Examples
    --------
    >>> adapter = GnssvodObservations(time_system="GPS")
    >>> obs = adapter.import_file("ROSA_obs.nc")
    >>> adapter.export_file(obs, "back_obs.nc", source="ROSA_obs.nc")
    """

    tool: ClassVar[Tool] = GNSSVOD

    time_system: str

    def reader(self, path: Path) -> GnssvodObsReader:
        return GnssvodObsReader(fpath=path, time_system=self.time_system)

    def convert_from_canvodpy(self, obs_ds: xr.Dataset) -> xr.Dataset:
        return to_gnssvod_observations(obs_ds)

    def save(self, tool_ds: xr.Dataset, path: Path) -> None:
        tool_ds.to_netcdf(path)

class GnssvodVod(VodAdapter):
    """gnssvod VOD files <-> canvodpy VOD.

    A gnssvod VOD file is the result of ``gnssvod.calc_vod`` for one
    station pair (``DataFrame.to_xarray()``) written as NetCDF: dims
    ``Epoch``, ``SV``, one variable per band, ``Azimuth``/``Elevation``.

    Parameters
    ----------
    bands : dict[str, list[str]]
        gnssvod's ``bands`` argument: band variable to the signal-strength
        observation codes ``calc_vod`` merges into it, e.g.
        ``{"VOD_L1": ["S1C", "S1W"], "VOD_L2": ["S2W"]}``. Importing gives
        each band's VOD the sids of its code if the band has one code
        (``"S2W"`` -> ``"G01|L2|W"``), and of its RINEX 2 form if it merges
        several (``"S1"`` -> ``"G01|L1|u"``, canvodpy's marker for a band
        whose tracking code is not known). Exporting merges canvodpy's VOD
        the same way as ``calc_vod``, a merged band also taking the sids
        an import gives it, so imported VOD exports unchanged.

    Notes
    -----
    ``calc_vod`` reports the reference receiver's ``Azimuth``/
    ``Elevation``; canvodpy reports the canopy receiver's. Imported VOD
    keeps gnssvod's angles.

    Examples
    --------
    >>> adapter = GnssvodVod(bands={"VOD_L1": ["S1C", "S1W"]})
    >>> vod = adapter.import_file("vod_canopy_vs_reference.nc")
    """

    tool: ClassVar[Tool] = GNSSVOD

    bands: dict[str, list[str]]

    @field_validator("bands")
    @classmethod
    def _check_bands(cls, bands: dict[str, list[str]]) -> dict[str, list[str]]:
        if not bands:
            raise ValueError("bands is empty")
        for band, codes in bands.items():
            if not codes:
                raise ValueError(f"band {band!r} has no observation codes")
            for code in codes:
                if len(code) not in (2, 3) or code[0] != "S" or code[1] not in _FREQUENCY_DIGITS:
                    raise ValueError(
                        f"band {band!r}: {code!r} is not a signal-strength "
                        "observation code (e.g. 'S1C', or 'S1' in RINEX 2)"
                    )
            if len({code[1] for code in codes}) > 1:
                raise ValueError(f"band {band!r} merges codes of different frequencies: {codes}")
        imported = [cls._import_code(codes) for codes in bands.values()]
        if len(set(imported)) < len(imported):
            raise ValueError(
                f"two bands would import to the same signals: {dict(zip(bands, imported, strict=True))}"
            )
        return bands

    @staticmethod
    def _import_code(codes: list[str]) -> str:
        """Observation code the sids of an imported band get."""
        return codes[0] if len(codes) == 1 else codes[0][:2]

    def open(self, path: Path) -> xr.Dataset:
        return _netcdf(path)

    def save(self, tool_ds: xr.Dataset, path: Path) -> None:
        tool_ds.to_netcdf(path)

    def convert_to_canvodpy(self, tool_ds: xr.Dataset) -> xr.Dataset:
        return from_gnssvod_vod(
            tool_ds,
            {band: self._import_code(codes) for band, codes in self.bands.items()},
        )

    def convert_from_canvodpy(self, vod_ds: xr.Dataset) -> xr.Dataset:
        # A merged band also takes the sids an import gives it ("S1" ->
        # "G01|L1|u"), so imported VOD exports as it came in
        bands = {
            band: sorted({*codes, self._import_code(codes)}) for band, codes in self.bands.items()
        }
        return to_gnssvod_vod(vod_ds, bands)

class GnssvodObsReader(GNSSDataReader):
    """Reads the observation NetCDF files ``gnssvod.preprocess`` writes.

    A canvodpy reader like the RINEX and SBF readers: its datasets meet
    the reader contract (:func:`canvod.readers.validate_dataset`), carry
    the file hash for deduplication and the global attributes of every
    canvodpy reader.

    Parameters
    ----------
    fpath : Path
        gnssvod observation file (dims ``Epoch``, ``SV``).
    time_system : str
        Time scale of the file's epochs, which gnssvod keeps as in the
        RINEX file (its ``TIME OF FIRST OBS`` record), e.g. ``"GPS"``.
        Recorded on the epoch coordinate, like the other readers do.
    """

    time_system: str

    @property
    def source_format(self) -> str:
        return "gnssvod"

    @cached_property
    def _gnssvod_ds(self) -> xr.Dataset:
        with xr.open_dataset(self.fpath) as ds:
            return ds.load()

    @cached_property
    def _file_hash(self) -> str:
        return file_hash(self.fpath)

    @property
    def file_hash(self) -> str:
        return self._file_hash

    def to_ds(
        self,
        keep_data_vars: list[str] | None = None,
        **kwargs: object,
    ) -> xr.Dataset:
        """The file's observations as a canvodpy dataset.

        Parameters
        ----------
        keep_data_vars : list of str, optional
            Data variables to keep (e.g. ``["SNR"]``); ``None`` keeps all.
        **kwargs
            Not used; accepted for the reader interface.

        Raises
        ------
        ValueError
            If the result does not meet the reader contract.
        """
        ds = from_gnssvod_observations(self._gnssvod_ds)
        if keep_data_vars is not None:
            ds = ds[[v for v in keep_data_vars if v in ds.data_vars]]
        ds["epoch"].attrs.update(epoch_coord_attrs(self.time_system))
        ds.attrs.update(self._build_attrs())
        validate_dataset(ds, required_vars=keep_data_vars)
        return ds

    def iter_epochs(self) -> Iterator[xr.Dataset]:
        """The file's observations, one epoch at a time."""
        ds = self.to_ds()
        for i in range(ds.sizes["epoch"]):
            yield ds.isel(epoch=i)

    def _epochs(self) -> np.ndarray:
        return np.sort(self._gnssvod_ds["Epoch"].values)

    @property
    def start_time(self) -> datetime:
        return self._epochs()[0].astype("datetime64[us]").item()

    @property
    def end_time(self) -> datetime:
        return self._epochs()[-1].astype("datetime64[us]").item()

    @property
    def num_epochs(self) -> int:
        return int(self._gnssvod_ds.sizes["Epoch"])

    @property
    def systems(self) -> list[str]:
        return sorted({str(sv)[0] for sv in self._gnssvod_ds["SV"].values})

    @property
    def num_satellites(self) -> int:
        return int(self._gnssvod_ds.sizes["SV"])