canvod.auxiliary API Reference¶

@dataclass(config=ConfigDict(arbitrary_types_allowed=True))
class Sp3File(AuxFile):
    """Handler for SP3 orbit files with multi-product support.

    Now supports all IGS analysis centers via product registry:
    COD, GFZ, ESA, JPL, IGS, WHU, GRG, SHA

    Notes
    -----
    This is a Pydantic dataclass with `arbitrary_types_allowed=True`.

    Attributes
    ----------
    date : str
        String in YYYYDOY format.
    agency : str
        Agency code (e.g., "COD", "GFZ", "ESA").
    product_type : str
        Product type ("final", "rapid").
    ftp_server : str
        Base URL for downloads.
    local_dir : Path
        Local storage directory.
    add_velocities : bool | None, default True
        Whether to compute velocities.
    dimensionless : bool | None, default True
        Whether to strip units (store magnitudes only).
    product_spec : ProductSpec | None, optional
        Product specification resolved from the registry.
    """

    date: str
    agency: str
    product_type: str
    ftp_server: str
    local_dir: Path
    add_velocities: bool | None = True
    dimensionless: bool | None = True
    product_spec: ProductSpec | None = None

    def __post_init__(self) -> None:
        """Initialize with product validation."""
        self.file_type = ["orbit"]
        self.local_dir = Path(self.local_dir)
        self.local_dir.mkdir(parents=True, exist_ok=True)

        # Validate product exists in registry
        self.product_spec = get_product_spec(self.agency, self.product_type)

        super().__post_init__()

    def get_interpolation_strategy(self) -> Interpolator:
        """Get appropriate interpolation strategy for SP3 files."""
        config = Sp3Config(
            use_velocities=bool(self.add_velocities),
            fallback_method="linear",
        )
        return Sp3InterpolationStrategy(config=config)

    def generate_filename_based_on_type(self) -> Path:
        """Generate filename using product registry.

        Pattern: {PREFIX}_{YYYYDOY}0000_{DURATION}_{SAMPLING}_ORB.SP3

        Example: COD0MGXFIN_20240150000_01D_05M_ORB.SP3
        """
        assert self.product_spec is not None, (
            "product_spec must be set via \
            __post_init__"
        )
        prefix = self.product_spec.prefix
        duration = self.product_spec.duration
        sampling = self.product_spec.sampling_rate

        return Path(f"{prefix}_{self.date}0000_{duration}_{sampling}_ORB.SP3")

    def download_aux_file(self) -> None:
        """Download SP3 file, trying all servers from the product registry.

        Servers are tried in priority order from products.toml. Auth-required
        servers are skipped when no credentials are configured. Warnings are
        printed when falling back to alternate servers.

        Raises
        ------
        RuntimeError
            If download fails from all available servers.
        ValueError
            If GPS week calculation fails.
        """
        assert self.product_spec is not None, (
            "product_spec must be set via \
            __post_init__"
        )
        orbit_file = self.generate_filename_based_on_type()
        gps_week = get_gps_week_from_filename(orbit_file)

        ftp_path = self.product_spec.ftp_path_pattern.format(
            gps_week=gps_week,
            file=f"{orbit_file}.gz",
        )
        destination = self.local_dir / orbit_file
        file_info = {
            "gps_week": gps_week,
            "filename": orbit_file,
            "type": "orbit",
            "agency": self.agency,
        }

        self.download_with_fallback(ftp_path, destination, file_info, self.product_spec)
        print(f"Downloaded orbit file for {self.agency} on date {self.date}")

    def read_file(self) -> xr.Dataset:
        """Read and validate SP3 file.

        Returns
        -------
        xr.Dataset
            Dataset with satellite positions (X, Y, Z) in meters.

        Raises
        ------
        FileNotFoundError
            If file does not exist.
        ValueError
            If validation fails.
        """
        # Use dedicated parser
        assert self.fpath is not None, "fpath must be set before calling read_file"
        parser = Sp3Parser(self.fpath, dimensionless=bool(self.dimensionless))
        dataset = parser.parse()

        # Validate format
        validator = Sp3Validator(dataset, self.fpath)
        result = validator.validate()

        if not result.is_valid:
            raise ValueError(f"SP3 validation failed:\n{result.summary()}")

        # Add metadata
        dataset = self._add_metadata(dataset)

        # Compute velocities if requested
        if self.add_velocities:
            dataset = self.compute_velocity(dataset)

        return dataset

    def _add_metadata(self, ds: xr.Dataset) -> xr.Dataset:
        """Add file-level metadata to dataset."""
        assert self.fpath is not None, "fpath must be set before calling _add_metadata"
        assert self.product_spec is not None, (
            "product_spec must be set before \
            calling _add_metadata"
        )
        ds.attrs = {
            "file": str(self.fpath.name),
            "agency": self.agency,
            "agency_name": self.product_spec.agency_name,
            "product_type": self.product_type,
            "ftp_server": self.ftp_server,
            "date": self.date,
            "sampling_rate": self.product_spec.sampling_rate,
            "duration": self.product_spec.duration,
        }
        return ds

    def compute_velocity(self, ds: xr.Dataset) -> xr.Dataset:
        """Compute satellite velocities from position data.

        Uses central differences for interior points, forward/backward
        differences for endpoints.

        Parameters
        ----------
        ds : xr.Dataset
            Dataset with X, Y, Z coordinates.

        Returns
        -------
        xr.Dataset
            Dataset augmented with Vx, Vy, Vz velocities.
        """
        # Calculate time step
        time_diffs = np.diff(ds["epoch"].values)
        dt = np.median(time_diffs).astype("timedelta64[s]").astype(float)

        # Initialize velocity arrays
        vx = np.zeros_like(ds["X"].values)
        vy = np.zeros_like(ds["Y"].values)
        vz = np.zeros_like(ds["Z"].values)

        # Central difference for interior points
        vx[1:-1] = (ds["X"].values[2:] - ds["X"].values[:-2]) / (2 * dt)
        vy[1:-1] = (ds["Y"].values[2:] - ds["Y"].values[:-2]) / (2 * dt)
        vz[1:-1] = (ds["Z"].values[2:] - ds["Z"].values[:-2]) / (2 * dt)

        # Forward difference for first point
        vx[0] = (ds["X"].values[1] - ds["X"].values[0]) / dt
        vy[0] = (ds["Y"].values[1] - ds["Y"].values[0]) / dt
        vz[0] = (ds["Z"].values[1] - ds["Z"].values[0]) / dt

        # Backward difference for last point
        vx[-1] = (ds["X"].values[-1] - ds["X"].values[-2]) / dt
        vy[-1] = (ds["Y"].values[-1] - ds["Y"].values[-2]) / dt
        vz[-1] = (ds["Z"].values[-1] - ds["Z"].values[-2]) / dt

        # Add units if needed
        if not self.dimensionless:
            vx = vx * (UREG.meter / UREG.second)
            vy = vy * (UREG.meter / UREG.second)
            vz = vz * (UREG.meter / UREG.second)

        # Add to dataset
        ds = ds.assign(
            Vx=(("epoch", "sv"), vx),
            Vy=(("epoch", "sv"), vy),
            Vz=(("epoch", "sv"), vz),
        )

        # Add velocity attributes
        for var, attrs in self._get_velocity_attributes(dt).items():
            if var in ds:
                ds[var].attrs = attrs

        return ds

    def _get_velocity_attributes(
        self,
        dt: float,
    ) -> dict[str, dict[str, str | float]]:
        """Get standardized attributes for velocity variables."""
        base_attrs = {
            "units": "m/s",
            "computation_method": "central_difference",
            "time_step": float(dt),
            "reference_frame": "ECEF",
        }

        return {
            "Vx": {
                "long_name": "x-component of velocity",
                "standard_name": "v_x",
                "short_name": "v_x",
                "axis": "v_x",
                **base_attrs,
            },
            "Vy": {
                "long_name": "y-component of velocity",
                "standard_name": "v_y",
                "short_name": "v_y",
                "axis": "v_y",
                **base_attrs,
            },
            "Vz": {
                "long_name": "z-component of velocity",
                "standard_name": "v_z",
                "short_name": "v_z",
                "axis": "v_z",
                **base_attrs,
            },
        }

@dataclass(config=ConfigDict(arbitrary_types_allowed=True))
class ClkFile(AuxFile):
    """Handler for GNSS clock files in CLK format.

    This class reads and processes clock offset files containing satellite clock
    corrections. It handles the parsing of CLK format files and provides the data
    in xarray Dataset format.

    Supports multiple analysis centers via product registry with proper FTP paths
    and filename conventions.

    Notes
    -----
    This is a Pydantic dataclass with `arbitrary_types_allowed=True`.

    Attributes
    ----------
    date : str
        String in YYYYDOY format representing the start date.
    agency : str
        Analysis center identifier (e.g., "COD", "GFZ").
    product_type : str
        Product type ("final", "rapid", "ultrarapid").
    ftp_server : str
        Base URL for file downloads.
    local_dir : Path
        Local storage directory.
    dimensionless : bool | None, default True
        If True, outputs magnitude-only values (no units attached).
    """

    date: str
    agency: str
    product_type: str
    ftp_server: str
    local_dir: Path
    dimensionless: bool | None = True

    def __post_init__(self) -> None:
        """Initialize CLK file handler."""
        self.file_type = ["clock"]
        self.local_dir = Path(self.local_dir)
        self.local_dir.mkdir(parents=True, exist_ok=True)
        super().__post_init__()

    def get_interpolation_strategy(self) -> Interpolator:
        """Get appropriate interpolation strategy for CLK files."""
        config = ClockConfig(
            window_size=9,
            jump_threshold=1e-6,
        )
        return ClockInterpolationStrategy(config=config)

    def generate_filename_based_on_type(self) -> Path:
        """Generate standard CLK filename using product registry.

        Uses product registry to get correct prefix for the agency/product combination.
        Filename format: {PREFIX}_{YYYYDOY}0000_01D_30S_CLK.CLK

        Returns
        -------
        Path
            Filename according to CLK conventions.

        Raises
        ------
        ValueError
            If agency/product combination not in registry.
        """
        # Get product spec from registry
        spec = get_product_spec(self.agency, self.product_type)

        # CLK files use 30S sampling for most products
        sampling = "30S"  # Could be made configurable via registry

        return Path(f"{spec.prefix}_{self.date}0000_01D_{sampling}_CLK.CLK")

    def download_aux_file(self) -> None:
        """Download CLK file, trying all servers from the product registry.

        Servers are tried in priority order from products.toml. Auth-required
        servers are skipped when no credentials are configured. Warnings are
        printed when falling back to alternate servers.

        Raises
        ------
        RuntimeError
            If file cannot be downloaded from any available server.
        ValueError
            If GPS week calculation fails.
        """
        clock_file = self.generate_filename_based_on_type()
        gps_week = get_gps_week_from_filename(clock_file)

        spec = get_product_spec(self.agency, self.product_type)
        ftp_path = spec.ftp_path_pattern.format(
            gps_week=gps_week,
            file=f"{clock_file}.gz",
        )
        destination = self.local_dir / clock_file
        file_info = {
            "gps_week": gps_week,
            "filename": clock_file,
            "type": "clock",
            "agency": self.agency,
        }

        self.download_with_fallback(ftp_path, destination, file_info, spec)
        print(f"Downloaded clock file for {self.agency} on date {self.date}")

    def read_file(self) -> xr.Dataset:
        """Read and parse CLK file into xarray Dataset.

        Uses modular parser for data extraction and validator for quality checks.
        Applies unit conversion from microseconds to seconds.

        Returns
        -------
        xr.Dataset
            Clock offsets with dimensions (epoch, sv). Values are in seconds
            (or dimensionless if specified).
        """
        # Parse file using modular parser
        assert self.fpath is not None, "fpath must be set before calling read_file"
        epochs, satellites, clock_offsets = parse_clk_file(self.fpath)

        # Convert units (microseconds → seconds)
        clock_offsets = (UREG.microsecond * clock_offsets).to("s")

        if self.dimensionless:
            clock_offsets = clock_offsets.magnitude

        # Create dataset
        ds = xr.Dataset(
            data_vars={"clock_offset": (("epoch", "sv"), clock_offsets)},
            coords={
                "epoch": np.array(epochs, dtype="datetime64[ns]"),
                "sv": np.array(satellites),
            },
        )

        # Validate and add metadata
        validation = validate_clk_dataset(ds)
        ds = self._prepare_dataset(ds, validation)

        return ds

    def _prepare_dataset(self, ds: xr.Dataset, validation: dict) -> xr.Dataset:
        """Add metadata and attributes to dataset.

        Parameters
        ----------
        ds : xr.Dataset
            Dataset from parsing.
        validation : dict
            Validation results dictionary.

        Returns
        -------
        xr.Dataset
            Dataset with complete metadata.
        """
        assert self.fpath is not None, (
            "fpath must be set \
            before calling _prepare_dataset"
        )
        ds.attrs = {
            "file": str(self.fpath.name),
            "agency": self.agency,
            "product_type": self.product_type,
            "ftp_server": self.ftp_server,
            "date": self.date,
            "valid_data_percent": validation["valid_data_percent"],
            "num_epochs": validation["num_epochs"],
            "num_satellites": validation["num_satellites"],
        }

        # Add variable attributes
        ds.clock_offset.attrs = {
            "long_name": "Satellite clock offset",
            "standard_name": "clock_offset",
            "units": "seconds",
            "description": "Clock correction for each satellite",
        }

        return ds

@dataclass(config=ConfigDict(arbitrary_types_allowed=True))
class AuxFile(ABC):
    """Abstract base class for GNSS auxiliary files (SP3, CLK, IONEX, etc.).

    This class provides two ways to create instances:
    1. from_datetime_date(): Create from a datetime.date object and metadata
    2. from_file(): Create directly from an existing file path

    The class handles both newly downloaded files and existing local files,
    maintaining consistent behavior regardless of how the instance is created.

    FTP Server Configuration:
    ------------------------
    - user_email: Optional email for NASA CDDIS authentication
      - If None: Uses ESA FTP server exclusively (no authentication required)
      - If provided: Enables NASA CDDIS as fallback server (requires registration)
      - To enable CDDIS, set nasa_earthdata_acc_mail in config/processing.yaml

    Notes
    -----
    This is a Pydantic dataclass with `arbitrary_types_allowed=True`, and
    it uses `ABC` to define required subclass hooks.
    """

    date: str
    agency: str
    product_type: str
    ftp_server: str
    local_dir: Path
    file_type: list[str] = Field(default_factory=list)
    fpath: Path | None = None
    user_email: str | None = None
    downloader: FileDownloader | None = None
    _data: xr.Dataset | None = Field(default=None, init=False)

    def __post_init__(self) -> None:
        """Initialize after dataclass creation.

        Sets up paths, downloader, and verifies local file existence.
        """
        if not self.file_type:
            self.file_type = ["unknown"]
        self.local_dir = Path(self.local_dir)
        self.local_dir.mkdir(parents=True, exist_ok=True)

        if self.downloader is None:
            self.downloader = FtpDownloader(user_email=self.user_email)

        self.fpath = self.check_file_exists()

    @classmethod
    def from_datetime_date(
        cls,
        date: datetime.date,
        agency: str,
        product_type: str,
        ftp_server: str,
        local_dir: Path,
        **kwargs: Any,
    ) -> AuxFile:
        """Create an AuxFile instance from a datetime.date.

        Parameters
        ----------
        date : datetime.date
            Date for the desired auxiliary file.
        agency : str
            Agency providing the data (e.g., "COD", "IGS").
        product_type : str
            Product type ("final", "rapid", "ultrarapid").
        ftp_server : str
            Base URL for file downloads.
        local_dir : Path
            Directory for storing files locally.
        **kwargs : Any
            Extra keyword arguments for subclass construction.

        Returns
        -------
        AuxFile
            A new instance of the AuxFile subclass.
        """
        yyyydoy = YYYYDOY.from_date(date=date).to_str()
        return cls(
            date=yyyydoy,
            agency=agency,
            product_type=product_type,
            ftp_server=ftp_server,
            local_dir=local_dir,
            **kwargs,
        )

    @classmethod
    def from_file(cls, fpath: Path | str, **kwargs: Any) -> AuxFile:
        """Create an AuxFile instance from an existing file path.

        Parameters
        ----------
        fpath : Path
            Path to the existing GNSS file.
        **kwargs : Any
            Extra keyword arguments for subclass construction.

        Returns
        -------
        AuxFile
            A new instance of the AuxFile subclass.

        Raises
        ------
        FileNotFoundError
            If the specified file does not exist.
        """
        fpath = Path(fpath)
        if not fpath.exists():
            raise FileNotFoundError(f"File not found: {fpath}")

        fname = fpath.name
        agency = fname[0:3]
        yyyydoy = fname.split("_")[1][0:7]

        return cls(
            date=yyyydoy,
            agency=agency,
            product_type="final",
            ftp_server="N/A",
            local_dir=fpath.parent,
            fpath=fpath,
            **kwargs,
        )

    def download_file(
        self,
        url: str,
        destination: Path,
        file_info: dict | None = None,
    ) -> Path:
        """Download a file using the configured downloader.

        Parameters
        ----------
        url : str
            Download URL.
        destination : Path
            Local file destination.
        file_info : dict, optional
            Extra info passed to the downloader.

        Returns
        -------
        Path
            Path to the downloaded file.
        """
        if self.downloader is None:
            raise RuntimeError("No downloader is configured")
        return self.downloader.download(url, destination, file_info)

    def download_with_fallback(
        self,
        ftp_path: str,
        destination: Path,
        file_info: dict | None = None,
        product_spec: Any | None = None,
    ) -> Path:
        """Download a file, trying all servers from the product spec in priority order.

        Iterates through the product's FTP server list (from products.toml),
        skipping servers that require authentication when no credentials are
        configured. Warns when falling back to an alternate server. Raises a
        clear error when all servers fail.

        Parameters
        ----------
        ftp_path : str
            Server-relative path (e.g. ``/gnss/products/2345/FILE.gz``).
        destination : Path
            Local file destination.
        file_info : dict, optional
            Extra context passed to the downloader.
        product_spec : ProductSpec, optional
            Product specification with ``ftp_servers`` list. If None, falls
            back to single-server download using ``self.ftp_server``.

        Returns
        -------
        Path
            Path to the downloaded file.

        Raises
        ------
        RuntimeError
            If download fails from all available servers.
        """
        if self.downloader is None:
            raise RuntimeError("No downloader is configured")

        # Fall back to legacy single-server path when no product spec
        if product_spec is None or not product_spec.ftp_servers:
            url = f"{self.ftp_server}{ftp_path}"
            return self.downloader.download(url, destination, file_info)

        # Sort servers by priority (lowest number = highest priority)
        servers = sorted(product_spec.ftp_servers, key=lambda s: s.priority)

        # Filter out auth-required servers when no credentials
        available_servers = []
        skipped_servers = []
        for server in servers:
            if server.requires_auth and self.user_email is None:
                skipped_servers.append(server)
            else:
                available_servers.append(server)

        if not available_servers:
            server_names = ", ".join(s.url for s in skipped_servers)
            raise RuntimeError(
                f"No FTP servers available for {product_spec.prefix}.\n"
                f"All servers require authentication: {server_names}\n"
                f"Set nasa_earthdata_acc_mail in config/processing.yaml to "
                f"enable NASA CDDIS access.\n"
                f"Register at: https://urs.earthdata.nasa.gov/users/new"
            )

        errors: list[str] = []
        for i, server in enumerate(available_servers):
            url = f"{server.url.rstrip('/')}{ftp_path}"
            try:
                result = self.downloader.download(url, destination, file_info)
                if i > 0:
                    # We got here via fallback — log which server succeeded
                    print(
                        f"  ✓ Download succeeded from fallback server: "
                        f"{server.description or server.url}"
                    )
                return result
            except Exception as e:
                if self.downloader._is_network_error(e):
                    raise RuntimeError(
                        "No internet connection detected. "
                        "Please check your network and try again."
                    ) from e

                server_label = server.description or server.url
                error_msg = f"{server_label}: {e!s}"
                errors.append(error_msg)
                print(f"  ⚠ Download failed from {server_label}: {e!s}")

                # Warn about fallback if there are more servers to try
                if i < len(available_servers) - 1:
                    next_server = available_servers[i + 1]
                    next_label = next_server.description or next_server.url
                    print(f"  → Trying fallback server: {next_label}")

        # All servers exhausted
        product_label = f"{product_spec.agency_code}/{product_spec.product_type}"
        error_detail = "\n".join(f"  - {e}" for e in errors)
        if skipped_servers:
            skip_detail = "\n".join(
                f"  - {s.description or s.url} (requires auth)" for s in skipped_servers
            )
            auth_hint = (
                f"\n\nSkipped servers (no credentials):\n{skip_detail}\n"
                f"Set nasa_earthdata_acc_mail in config/processing.yaml "
                f"to enable these servers."
            )
        else:
            auth_hint = ""

        raise RuntimeError(
            f"Failed to download {product_label} from all available servers.\n"
            f"\nFile: {destination.name}\n"
            f"\nServer errors:\n{error_detail}"
            f"{auth_hint}"
        )

    @abstractmethod
    def read_file(self) -> xr.Dataset:
        """Read and parse the auxiliary file.

        Returns
        -------
        xr.Dataset
            Parsed dataset representation of the file.
        """
        pass

    @abstractmethod
    def get_interpolation_strategy(self) -> Interpolator:
        """Get the interpolation strategy for this file type."""
        pass

    @property
    def data(self) -> xr.Dataset:
        """Access the file's data, loading it if necessary."""
        if self._data is None:
            self._data = self.read_file()
            strategy = self.get_interpolation_strategy()
            self._data.attrs["interpolator_config"] = strategy.to_attrs()
        return self._data

    def check_file_exists(self) -> Path:
        """Verify file exists locally or download it if needed.

        Returns
        -------
        Path
            Local file path.
        """
        filename = self.generate_filename_based_on_type()
        file_path = self.local_dir / filename
        if not file_path.exists():
            print(f"File {file_path} does not exist. Downloading...")
            self.download_aux_file()
        else:
            print(f"File {file_path} exists.")
        return file_path

    @abstractmethod
    def generate_filename_based_on_type(self) -> Path:
        """Generate the appropriate filename for this type of auxiliary file."""
        pass

    @abstractmethod
    def download_aux_file(self) -> None:
        """Download the auxiliary file from the specified FTP server."""
        pass

@dataclass(config=ConfigDict(arbitrary_types_allowed=True))
class Interpolator(ABC):
    """Abstract base class for interpolation strategies.

    Notes
    -----
    This is a Pydantic dataclass with `arbitrary_types_allowed=True` and
    uses `ABC` to define required interpolation hooks.
    """

    config: InterpolatorConfig

    @abstractmethod
    def interpolate(self, ds: xr.Dataset, target_epochs: np.ndarray) -> xr.Dataset:
        """Interpolate dataset to match target epochs.

        Parameters
        ----------
        ds : xr.Dataset
            Source dataset with (epoch, sid) dimensions.
        target_epochs : np.ndarray
            Target epoch grid (datetime64).

        Returns
        -------
        xr.Dataset
            Interpolated dataset at target epochs.
        """
        pass

    def to_attrs(self) -> dict[str, Any]:
        """Convert interpolator to attrs-compatible dictionary."""
        return {
            "interpolator_type": self.__class__.__name__,
            "config": self.config.to_dict(),
        }

class InterpolatorConfig:
    """Base class for interpolator configuration."""

    def to_dict(self) -> dict[str, Any]:
        """Convert config to dictionary for attrs storage."""
        return dict(vars(self))