canvod.utils API Reference¶

class CanvodConfig(BaseModel):
    """
    Complete canvodpy configuration.

    This is the top-level configuration object that combines all
    configuration sections. It's fully serializable and can be used
    for local development (YAML files) or API-based configuration.
    """

    processing: ProcessingConfig
    sites: SitesConfig
    sids: SidsConfig

    model_config = {"extra": "forbid"}  # Catch typos in config files!

    @property
    def nasa_earthdata_acc_mail(self) -> str | None:
        """Return the configured NASA Earthdata email for CDDIS authentication.

        Returns
        -------
        str | None
            NASA Earthdata email address.
        """
        return self.processing.credentials.nasa_earthdata_acc_mail

class MetadataConfig(BaseModel):
    """Metadata to be written to processed files.

    Notes
    -----
    This is a Pydantic model for configuration validation.
    """

    author: str = Field(..., description="Author name")
    email: EmailStr = Field(..., description="Author email")
    orcid: str | None = Field(None, description="ORCID identifier")
    institution: str = Field(..., description="Institution name")
    institution_ror: str | None = Field(None, description="ROR identifier")
    department: str | None = Field(None, description="Department name")
    research_group: str | None = Field(
        None,
        description="Research group name",
    )
    website: str | None = Field(
        None,
        description="Institution/group website",
    )
    license: str | None = Field(None, description="SPDX license identifier")
    publisher: str | None = Field(None, description="Publisher name")
    publisher_url: str | None = Field(None, description="Publisher URL")
    naming_authority: str | None = Field(None, description="Naming authority URI")

    def to_attrs_dict(self) -> dict[str, str]:
        """Convert to a dictionary for xarray attributes.

        Returns
        -------
        dict[str, str]
            Metadata as xarray-compatible attributes.
        """
        attrs = {
            "author": self.author,
            "email": self.email,
            "institution": self.institution,
        }
        if self.department:
            attrs["department"] = self.department
        if self.research_group:
            attrs["research_group"] = self.research_group
        if self.website:
            attrs["website"] = self.website
        return attrs

class ProcessingConfig(BaseModel):
    """Complete processing configuration."""

    metadata: MetadataConfig
    credentials: CredentialsConfig = Field(
        default_factory=CredentialsConfig,
        description="Credentials for external data services",
    )
    aux_data: AuxDataConfig = Field(default_factory=AuxDataConfig)
    processing: ProcessingParams = Field(default_factory=ProcessingParams)
    compression: CompressionConfig = Field(default_factory=CompressionConfig)
    icechunk: IcechunkConfig = Field(default_factory=IcechunkConfig)
    storage: StorageConfig = Field(default_factory=StorageConfig)
    logging: LoggingConfig = Field(default_factory=LoggingConfig)
    preprocessing: PreprocessingConfig = Field(
        default_factory=PreprocessingConfig,
    )
    references: ReferencesConfig = Field(
        default_factory=ReferencesConfig,
        description="Publication and funding references",
    )

class SiteConfig(BaseModel):
    """Research site configuration."""

    gnss_site_data_root: str = Field(
        ..., description="Root directory for site GNSS data"
    )
    description: str | None = Field(None, description="Site description")
    country: str | None = Field(None, description="Country code (ISO 3166-1)")
    latitude: float | None = Field(None, description="WGS84 latitude")
    longitude: float | None = Field(None, description="WGS84 longitude")
    altitude_m: float | None = Field(None, description="Altitude in meters")
    receivers: dict[str, ReceiverConfig] = Field(..., description="Site receivers")
    vod_analyses: dict[str, VodAnalysisConfig] | None = Field(
        None,
        description="VOD analysis pairs",
    )
    naming: dict | None = Field(
        None,
        description="Naming configuration (validated by canvod-virtualiconvname package)",
    )

    @model_validator(mode="after")
    def validate_scs_from_targets(self) -> SiteConfig:
        """Validate that scs_from entries reference existing canopy receivers."""
        canopy_names = self.get_canopy_receiver_names()
        for name, cfg in self.receivers.items():
            if cfg.type != "reference" or cfg.scs_from is None:
                continue
            if isinstance(cfg.scs_from, str) and cfg.scs_from == "all":
                continue
            targets = cfg.scs_from if isinstance(cfg.scs_from, list) else [cfg.scs_from]
            for target in targets:
                if target not in canopy_names:
                    msg = (
                        f"Receiver '{name}' scs_from references '{target}' "
                        f"which is not a canopy receiver. "
                        f"Available canopy receivers: {canopy_names}"
                    )
                    raise ValueError(msg)
        return self

    def get_base_path(self) -> Path:
        """Get gnss_site_data_root as a Path.

        Returns
        -------
        Path
            Site data root directory as a Path object.
        """
        return Path(self.gnss_site_data_root)

    def get_canopy_receiver_names(self) -> list[str]:
        """Get names of all canopy receivers.

        Returns
        -------
        list[str]
            Canopy receiver names.
        """
        return [name for name, cfg in self.receivers.items() if cfg.type == "canopy"]

    def resolve_scs_from(self, receiver_name: str) -> list[str]:
        """Resolve scs_from for a reference receiver to a list of canopy names.

        Parameters
        ----------
        receiver_name : str
            Name of the reference receiver.

        Returns
        -------
        list[str]
            List of canopy receiver names for SCS computation.
        """
        cfg = self.receivers[receiver_name]
        if cfg.type != "reference":
            msg = f"resolve_scs_from only applies to reference receivers, got '{cfg.type}'"
            raise ValueError(msg)
        if cfg.scs_from == "all":
            return self.get_canopy_receiver_names()
        if isinstance(cfg.scs_from, list):
            return cfg.scs_from
        # Single canopy name as string
        if cfg.scs_from is None:
            msg = f"Receiver '{receiver_name}' has no scs_from configured"
            raise ValueError(msg)
        return [cfg.scs_from]

    def get_reference_canopy_pairs(self) -> list[tuple[str, str]]:
        """Expand scs_from into (reference_name, canopy_name) pairs.

        Returns
        -------
        list[tuple[str, str]]
            List of (reference_name, canopy_name) pairs.
        """
        pairs = []
        for name, cfg in self.receivers.items():
            if cfg.type != "reference":
                continue
            for canopy_name in self.resolve_scs_from(name):
                pairs.append((name, canopy_name))
        return pairs

class SitesConfig(BaseModel):
    """All research sites."""

    sites: dict[str, SiteConfig]

    @field_validator("sites")
    @classmethod
    def validate_at_least_one_site(
        cls,
        v: dict[str, SiteConfig],
    ) -> dict[str, SiteConfig]:
        """Warn if no sites are defined.

        Parameters
        ----------
        v : dict[str, SiteConfig]
            Sites dictionary to validate.

        Returns
        -------
        dict[str, SiteConfig]
            Validated sites dictionary.
        """
        if not v:
            import warnings

            warnings.warn(
                "No research sites defined in sites.yaml. Run: just config-init",
                UserWarning,
                stacklevel=2,
            )
        return v

class SidsConfig(BaseModel):
    """Signal ID configuration."""

    mode: Literal["all", "preset", "custom"] = Field(
        "all",
        description="SID selection mode",
    )
    preset: str | None = Field(
        None,
        description="Preset name when mode=preset",
    )
    custom_sids: list[str] = Field(
        default_factory=list,
        description="Custom SID list when mode=custom",
    )

    @field_validator("preset")
    @classmethod
    def validate_preset_when_mode_preset(
        cls,
        v: str | None,
        info: ValidationInfo,
    ) -> str | None:
        """Ensure preset is set when mode is preset.

        Parameters
        ----------
        v : str | None
            Preset name.
        info : ValidationInfo
            Pydantic validation info.

        Returns
        -------
        str | None
            Preset value if valid.
        """
        mode = info.data.get("mode")
        if mode == "preset" and not v:
            msg = "preset must be specified when mode is 'preset'"
            raise ValueError(msg)
        return v

    def get_sids(self) -> list[str] | None:
        """Get the effective SID list.

        Returns
        -------
        list[str] | None
            None if mode is "all" (keep all SIDs), otherwise a SID list.
        """
        if self.mode == "all":
            return None
        if self.mode == "preset":
            return self._get_preset_sids()
        # CUSTOM
        return self.custom_sids

    def _get_preset_sids(self) -> list[str]:
        """Load the preset SID list.

        Returns
        -------
        list[str]
            Preset SID list.
        """
        # TODO: Implement preset loading from package defaults
        # For now, return empty list
        return []

@total_ordering
@dataclass
class YYYYDOY:
    """Year and day-of-year (DOY) date representation.

    Used throughout canvodpy for GNSS data organization and file naming.

    Notes
    -----
    This is a Pydantic dataclass and uses ``total_ordering`` for comparisons.

    Attributes
    ----------
    year : int
        The year (e.g., 2025).
    doy : int
        Day of year (1-366).
    date : datetime.date
        Calculated calendar date.
    yydoy : str
        Short format (YYDDD, e.g., "25001").

    Examples
    --------
    >>> # From components
    >>> d = YYYYDOY(year=2025, doy=1)
    >>> d.to_str()
    '2025001'

    >>> # From date object
    >>> import datetime
    >>> d = YYYYDOY.from_date(datetime.date(2025, 1, 1))
    >>> d.doy
    '001'

    >>> # From string
    >>> d = YYYYDOY.from_str("2025001")
    >>> d.year
    2025

    >>> # From short string (YYDDD)
    >>> d = YYYYDOY.from_yydoy_str("25001")
    >>> d.to_str()
    '2025001'

    """

    year: int = Field(..., ge=1)
    doy: int = Field(..., ge=1, le=366)
    yydoy: str | None = None
    date: datetime.date | None = None

    def __post_init__(self) -> None:
        """Calculate derived fields after initialization.

        Returns
        -------
        None
        """
        doy_int = int(self.doy)
        self._validate_doy(doy_int)
        self.date = self._calculate_date()
        self.yydoy = f"{str(self.year)[-2:]}{doy_int:03}"

    def __repr__(self) -> str:
        """Return the developer-focused representation.

        Returns
        -------
        str
            Detailed representation of the date.
        """
        return (
            f"YYYYDOY(year={self.year}, doy={self.doy}, date={self.date}, "
            f"yydoy={self.yydoy}, gps_week={self.gps_week}, "
            f"gps_day_of_week={self.gps_day_of_week})"
        )

    def __str__(self) -> str:
        """Return the user-facing date string.

        Returns
        -------
        str
            Date string in YYYYDDD format.
        """
        return self.to_str()

    def __eq__(self, other: object) -> bool:
        """Compare equality based on the date string.

        Parameters
        ----------
        other : object
            Object to compare with.

        Returns
        -------
        bool
            True if the other object is an equal YYYYDOY instance.
        """
        if not isinstance(other, YYYYDOY):
            return False
        return self.to_str() == other.to_str()

    def __lt__(self, other: object) -> bool:
        """Compare ordering for sorting.

        Parameters
        ----------
        other : object
            Object to compare with.

        Returns
        -------
        bool
            True if this date is earlier than the other date.
        """
        if not isinstance(other, YYYYDOY):
            return NotImplemented
        if self.date is None or other.date is None:
            return self.to_str() < other.to_str()
        return self.date < other.date

    def __hash__(self) -> int:
        """Return the hash of the date string.

        Returns
        -------
        int
            Hash value for use in sets and dicts.
        """
        return hash(self.to_str())

    def _calculate_date(self) -> datetime.date:
        """Calculate the calendar date from year and DOY.

        Returns
        -------
        datetime.date
            Calculated calendar date.
        """
        return datetime.date(self.year, 1, 1) + datetime.timedelta(
            days=int(self.doy) - 1
        )

    @staticmethod
    def _validate_doy(doy: int) -> None:
        """Validate day of year is in range [1, 366].

        Parameters
        ----------
        doy : int
            Day of year value to validate.

        Returns
        -------
        None
        """
        if not 1 <= doy <= 366:
            raise ValueError(f"Day of year (DOY) must be in range [1, 366], got {doy}")

    @classmethod
    def from_date(cls, date: datetime.date) -> YYYYDOY:
        """Create from datetime.date object.

        Parameters
        ----------
        date : datetime.date
            Calendar date

        Returns
        -------
        YYYYDOY
        """
        if isinstance(date, datetime.datetime):
            date = date.date()
        year = date.year
        doy = (date - datetime.date(year, 1, 1)).days + 1
        cls._validate_doy(doy)
        return cls(year=year, doy=doy)

    @classmethod
    def from_str(cls, yyyydoy: str | int) -> YYYYDOY:
        """Create from YYYYDDD string.

        Parameters
        ----------
        yyyydoy : str or int
            Date in YYYYDDD format (e.g., "2025001" or 2025001)

        Returns
        -------
        YYYYDOY
        """
        if isinstance(yyyydoy, int):
            yyyydoy = str(yyyydoy)
        if len(yyyydoy) != 7:
            raise ValueError(f"Invalid format. Expected 'YYYYDDD', got '{yyyydoy}'")
        year = int(yyyydoy[:4])
        doy = int(yyyydoy[4:])
        cls._validate_doy(doy)
        jan_first = datetime.datetime(year, 1, 1)
        final_date = jan_first + datetime.timedelta(days=doy - 1)
        return cls.from_date(final_date.date())

    @classmethod
    def from_int(cls, yyyydoy: int) -> YYYYDOY:
        """Create from YYYYDDD integer.

        Parameters
        ----------
        yyyydoy : int
            Date as integer (e.g., 2025001)

        Returns
        -------
        YYYYDOY
        """
        return cls.from_str(str(yyyydoy))

    @classmethod
    def from_yydoy_str(cls, yydoy: str) -> YYYYDOY:
        """Create from YYDDD short string.

        Assumes current millennium (20XX).

        Parameters
        ----------
        yydoy : str
            Short date format (e.g., "25001" for 2025 DOY 001)

        Returns
        -------
        YYYYDOY
        """
        current_millennium = str(datetime.datetime.now().year)[0:2]
        return cls.from_str(f"{current_millennium}{yydoy}")

    def to_str(self) -> str:
        """Convert to YYYYDDD string.

        Returns
        -------
        str
            Date in YYYYDDD format (e.g., "2025001").

        """
        return f"{self.year}{int(self.doy):03}"

    @staticmethod
    def gpsweekday(
        input_date: datetime.datetime | datetime.date | str,
        is_datetime: bool = False,
    ) -> tuple[int, int]:
        """
        Calculate GPS week number and day of week from a given date.

        GPS time started on January 6, 1980.

        Parameters
        ----------
        input_date : datetime.datetime, datetime.date, or str
            The date to calculate GPS week for. If string, format: "dd-mm-yyyy"
        is_datetime : bool, optional
            Whether input_date is a datetime object (default: False)

        Returns
        -------
        tuple[int, int]
            (GPS week number, day of week)
        """
        gps_start_date = datetime.date(1980, 1, 6)

        # Convert string to date if needed
        if not is_datetime and isinstance(input_date, str):
            input_date = datetime.datetime.strptime(input_date, "%d-%m-%Y").date()
        elif isinstance(input_date, datetime.datetime):
            input_date = input_date.date()

        if isinstance(input_date, datetime.date):
            # Calculate weeks and days since GPS epoch
            return divmod((input_date - gps_start_date).days, 7)
        raise TypeError(f"Unsupported input_date type: {type(input_date)!r}")

    @property
    def gps_week(self) -> int:
        """GPS week number.

        Returns
        -------
        int
            GPS week number since GPS epoch (1980-01-06).
        """
        if self.date is None:
            raise ValueError("Date is not initialized")
        return self.gpsweekday(self.date)[0]

    @property
    def gps_day_of_week(self) -> int:
        """GPS day of week (0=Sunday, 6=Saturday).

        Returns
        -------
        int
            Day of week where 0=Sunday, 6=Saturday.
        """
        if self.date is None:
            raise ValueError("Date is not initialized")
        return self.gpsweekday(self.date)[1]

@dataclass
class YYDOY:
    """
    Two-digit year + day of year (DOY) date representation.

    This is a compact format often used in GNSS filenames.

    Examples
    --------
    >>> date = YYDOY.from_str("25001")  # 2025, day 1
    """

    @classmethod
    def from_str(cls, yydoy: str) -> YYYYDOY:
        """
        Convert two-digit year DOY string to four-digit YYYYDOY.

        Parameters
        ----------
        yydoy : str
            Two-digit year + DOY (e.g., "25001" for 2025-01-01)

        Returns
        -------
        YYYYDOY
            Four-digit year DOY object
        """
        date_str = f"20{yydoy}"
        return YYYYDOY.from_str(date_str)