canvod.filemap API Reference¶

@dataclass(frozen=True)
class VirtualFile:
    """Physical file mapped to its canVOD conventional name."""

    physical_path: Path
    conventional_name: CanVODFilename

    @property
    def canonical_str(self) -> str:
        """The conventional filename as a string."""
        return self.conventional_name.name

    def open(self, mode: str = "rb"):
        """Open the physical file."""
        return self.physical_path.open(mode)

class NamingRecipe(BaseModel):
    """A user-defined mapping from arbitrary filenames to canVOD names.

    Serialize to YAML for sharing, to JSON for API transport.
    """

    name: str = Field(description="Recipe identifier (e.g. 'rosalia_reference')")
    description: str = ""

    # Canonical identity
    site: SiteId
    agency: AgencyId
    receiver_number: int = Field(ge=1, le=99)
    receiver_type: Literal["reference", "canopy"] = "canopy"
    sampling: Duration = "05S"
    period: Duration = "15M"
    content: ContentCode = "AA"
    file_type: Literal["rnx", "sbf", "ubx", "nmea"] = "rnx"

    # File discovery
    glob: str = Field(description="Glob pattern to find files (e.g. '*.??o', '*.rinex')")

    # Field extraction: sequential left-to-right
    # Each entry is a single-key dict: {field_name: width}
    fields: list[dict[str, int]] = Field(
        description="Sequential field extraction. Each entry: {field_name: width}"
    )

    @model_validator(mode="after")
    def _validate_fields(self) -> NamingRecipe:
        for i, entry in enumerate(self.fields):
            if len(entry) != 1:
                msg = f"fields[{i}]: each entry must be a single key-value pair, got {entry}"
                raise ValueError(msg)
            field_name = next(iter(entry))
            width = entry[field_name]
            if field_name not in KNOWN_FIELDS:
                msg = f"fields[{i}]: unknown field '{field_name}'. Known: {sorted(KNOWN_FIELDS)}"
                raise ValueError(msg)
            if not isinstance(width, int) or width < 1:
                msg = f"fields[{i}]: width must be a positive integer, got {width}"
                raise ValueError(msg)
        return self

    # -- Parsing ---------------------------------------------------------------

    def parse_filename(self, filename: str) -> dict[str, str | int]:
        """Extract fields from a physical filename.

        Parameters
        ----------
        filename
            Bare filename (no directory components).

        Returns
        -------
        dict
            Extracted field values. Integer fields (year, doy, etc.) are
            returned as ``int``.  ``hour_letter`` is returned as ``str``.
            ``skip`` fields are not included.

        Raises
        ------
        ValueError
            If the filename is too short for the field spec.
        """
        pos = 0
        result: dict[str, str | int] = {}

        for entry in self.fields:
            field_name = next(iter(entry))
            width = entry[field_name]

            if pos + width > len(filename):
                msg = (
                    f"Filename {filename!r} too short: need {width} chars "
                    f"at position {pos} for '{field_name}', "
                    f"but only {len(filename) - pos} remain"
                )
                raise ValueError(msg)

            raw = filename[pos : pos + width]
            pos += width

            if field_name == "skip":
                continue
            elif field_name == "hour_letter":
                result["hour_letter"] = raw
            else:
                try:
                    result[field_name] = int(raw)
                except ValueError:
                    msg = f"Cannot parse '{field_name}' as integer from {raw!r} in {filename!r}"
                    raise ValueError(msg) from None

        return result

    def to_virtual_file(self, file_path: Path) -> VirtualFile:
        """Map a physical file to a VirtualFile using this recipe.

        Parameters
        ----------
        file_path
            Path to the physical file.

        Returns
        -------
        VirtualFile
            The mapped virtual file.

        Raises
        ------
        ValueError
            If the filename cannot be parsed.
        """
        parsed = self.parse_filename(file_path.name)

        def _require_int(parsed_key: str) -> int:
            value = parsed[parsed_key]
            if isinstance(value, int):
                return value
            msg = (
                f"Recipe '{self.name}': expected integer field '{parsed_key}' "
                f"for {file_path.name!r}, got {value!r}"
            )
            raise ValueError(msg)

        def _require_str(parsed_key: str) -> str:
            value = parsed[parsed_key]
            if isinstance(value, str):
                return value
            msg = (
                f"Recipe '{self.name}': expected string field '{parsed_key}' "
                f"for {file_path.name!r}, got {value!r}"
            )
            raise ValueError(msg)

        # Resolve year
        if "year" in parsed:
            year = _require_int("year")
        elif "yy" in parsed:
            year = resolve_year_from_yy(_require_int("yy"))
        else:
            raise ValueError(
                f"Recipe '{self.name}': no 'year' or 'yy' field "
                f"in parsed result for {file_path.name!r}"
            )

        # Resolve DOY (from doy directly, or from month+day)
        if "doy" in parsed:
            doy = _require_int("doy")
        elif "month" in parsed and "day" in parsed:
            from datetime import date

            doy = date(year, _require_int("month"), _require_int("day")).timetuple().tm_yday
        else:
            raise ValueError(
                f"Recipe '{self.name}': no 'doy' or 'month'+'day' fields "
                f"in parsed result for {file_path.name!r}"
            )

        # Resolve hour
        if "hour" in parsed:
            hour = _require_int("hour")
        elif "hour_letter" in parsed:
            hour = hour_letter_to_int(_require_str("hour_letter"))
        else:
            hour = 0

        # Resolve minute
        minute = _require_int("minute") if "minute" in parsed else 0

        # Daily file: no hour in the name, or the RINEX 2 daily session "0"
        # (hourly sessions are "a"-"x", so one recipe matches both)
        period = self.period
        has_hour = any("hour" in e or "hour_letter" in e for e in self.fields)
        if parsed.get("hour_letter") == "0" or (hour == 0 and minute == 0 and not has_hour):
            period = "01D"

        rx_type = (
            ReceiverType.REFERENCE if self.receiver_type == "reference" else ReceiverType.ACTIVE
        )

        conventional = CanVODFilename(
            site=self.site,
            receiver_type=rx_type,
            receiver_number=self.receiver_number,
            agency=self.agency,
            year=year,
            doy=doy,
            hour=hour,
            minute=minute,
            period=period,
            sampling=self.sampling,
            content=self.content,
            file_type=FileType(self.file_type),
        )

        return VirtualFile(physical_path=file_path, conventional_name=conventional)

    @property
    def expected_length(self) -> int:
        """Total number of characters consumed by the field spec."""
        return sum(next(iter(e.values())) for e in self.fields)

    def matches(self, filename: str) -> bool:
        """Check if a filename can be parsed by this recipe."""
        if len(filename) != self.expected_length:
            return False
        try:
            self.parse_filename(filename)
            return True
        except ValueError:
            return False

    # -- Serialization ---------------------------------------------------------

    def to_yaml(self) -> str:
        """Serialize to YAML string."""
        return yaml.dump(
            self.model_dump(mode="json"),
            default_flow_style=False,
            sort_keys=False,
            allow_unicode=True,
        )

    def save(self, path: Path) -> None:
        """Write recipe to a YAML file."""
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(self.to_yaml(), encoding="utf-8")

    @classmethod
    def from_yaml(cls, text: str) -> NamingRecipe:
        """Load from a YAML string."""
        data = yaml.safe_load(text)
        return cls.model_validate(data)

    @classmethod
    def load(cls, path: Path) -> NamingRecipe:
        """Load from a YAML file."""
        text = path.read_text(encoding="utf-8")
        return cls.from_yaml(text)

class RecipeNotFoundError(FileNotFoundError):
    """No recipe file exists for a site and recipe name."""

class NamingRecipe(BaseModel):
    """A user-defined mapping from arbitrary filenames to canVOD names.

    Serialize to YAML for sharing, to JSON for API transport.
    """

    name: str = Field(description="Recipe identifier (e.g. 'rosalia_reference')")
    description: str = ""

    # Canonical identity
    site: SiteId
    agency: AgencyId
    receiver_number: int = Field(ge=1, le=99)
    receiver_type: Literal["reference", "canopy"] = "canopy"
    sampling: Duration = "05S"
    period: Duration = "15M"
    content: ContentCode = "AA"
    file_type: Literal["rnx", "sbf", "ubx", "nmea"] = "rnx"

    # File discovery
    glob: str = Field(description="Glob pattern to find files (e.g. '*.??o', '*.rinex')")

    # Field extraction: sequential left-to-right
    # Each entry is a single-key dict: {field_name: width}
    fields: list[dict[str, int]] = Field(
        description="Sequential field extraction. Each entry: {field_name: width}"
    )

    @model_validator(mode="after")
    def _validate_fields(self) -> NamingRecipe:
        for i, entry in enumerate(self.fields):
            if len(entry) != 1:
                msg = f"fields[{i}]: each entry must be a single key-value pair, got {entry}"
                raise ValueError(msg)
            field_name = next(iter(entry))
            width = entry[field_name]
            if field_name not in KNOWN_FIELDS:
                msg = f"fields[{i}]: unknown field '{field_name}'. Known: {sorted(KNOWN_FIELDS)}"
                raise ValueError(msg)
            if not isinstance(width, int) or width < 1:
                msg = f"fields[{i}]: width must be a positive integer, got {width}"
                raise ValueError(msg)
        return self

    # -- Parsing ---------------------------------------------------------------

    def parse_filename(self, filename: str) -> dict[str, str | int]:
        """Extract fields from a physical filename.

        Parameters
        ----------
        filename
            Bare filename (no directory components).

        Returns
        -------
        dict
            Extracted field values. Integer fields (year, doy, etc.) are
            returned as ``int``.  ``hour_letter`` is returned as ``str``.
            ``skip`` fields are not included.

        Raises
        ------
        ValueError
            If the filename is too short for the field spec.
        """
        pos = 0
        result: dict[str, str | int] = {}

        for entry in self.fields:
            field_name = next(iter(entry))
            width = entry[field_name]

            if pos + width > len(filename):
                msg = (
                    f"Filename {filename!r} too short: need {width} chars "
                    f"at position {pos} for '{field_name}', "
                    f"but only {len(filename) - pos} remain"
                )
                raise ValueError(msg)

            raw = filename[pos : pos + width]
            pos += width

            if field_name == "skip":
                continue
            elif field_name == "hour_letter":
                result["hour_letter"] = raw
            else:
                try:
                    result[field_name] = int(raw)
                except ValueError:
                    msg = f"Cannot parse '{field_name}' as integer from {raw!r} in {filename!r}"
                    raise ValueError(msg) from None

        return result

    def to_virtual_file(self, file_path: Path) -> VirtualFile:
        """Map a physical file to a VirtualFile using this recipe.

        Parameters
        ----------
        file_path
            Path to the physical file.

        Returns
        -------
        VirtualFile
            The mapped virtual file.

        Raises
        ------
        ValueError
            If the filename cannot be parsed.
        """
        parsed = self.parse_filename(file_path.name)

        def _require_int(parsed_key: str) -> int:
            value = parsed[parsed_key]
            if isinstance(value, int):
                return value
            msg = (
                f"Recipe '{self.name}': expected integer field '{parsed_key}' "
                f"for {file_path.name!r}, got {value!r}"
            )
            raise ValueError(msg)

        def _require_str(parsed_key: str) -> str:
            value = parsed[parsed_key]
            if isinstance(value, str):
                return value
            msg = (
                f"Recipe '{self.name}': expected string field '{parsed_key}' "
                f"for {file_path.name!r}, got {value!r}"
            )
            raise ValueError(msg)

        # Resolve year
        if "year" in parsed:
            year = _require_int("year")
        elif "yy" in parsed:
            year = resolve_year_from_yy(_require_int("yy"))
        else:
            raise ValueError(
                f"Recipe '{self.name}': no 'year' or 'yy' field "
                f"in parsed result for {file_path.name!r}"
            )

        # Resolve DOY (from doy directly, or from month+day)
        if "doy" in parsed:
            doy = _require_int("doy")
        elif "month" in parsed and "day" in parsed:
            from datetime import date

            doy = date(year, _require_int("month"), _require_int("day")).timetuple().tm_yday
        else:
            raise ValueError(
                f"Recipe '{self.name}': no 'doy' or 'month'+'day' fields "
                f"in parsed result for {file_path.name!r}"
            )

        # Resolve hour
        if "hour" in parsed:
            hour = _require_int("hour")
        elif "hour_letter" in parsed:
            hour = hour_letter_to_int(_require_str("hour_letter"))
        else:
            hour = 0

        # Resolve minute
        minute = _require_int("minute") if "minute" in parsed else 0

        # Daily file: no hour in the name, or the RINEX 2 daily session "0"
        # (hourly sessions are "a"-"x", so one recipe matches both)
        period = self.period
        has_hour = any("hour" in e or "hour_letter" in e for e in self.fields)
        if parsed.get("hour_letter") == "0" or (hour == 0 and minute == 0 and not has_hour):
            period = "01D"

        rx_type = (
            ReceiverType.REFERENCE if self.receiver_type == "reference" else ReceiverType.ACTIVE
        )

        conventional = CanVODFilename(
            site=self.site,
            receiver_type=rx_type,
            receiver_number=self.receiver_number,
            agency=self.agency,
            year=year,
            doy=doy,
            hour=hour,
            minute=minute,
            period=period,
            sampling=self.sampling,
            content=self.content,
            file_type=FileType(self.file_type),
        )

        return VirtualFile(physical_path=file_path, conventional_name=conventional)

    @property
    def expected_length(self) -> int:
        """Total number of characters consumed by the field spec."""
        return sum(next(iter(e.values())) for e in self.fields)

    def matches(self, filename: str) -> bool:
        """Check if a filename can be parsed by this recipe."""
        if len(filename) != self.expected_length:
            return False
        try:
            self.parse_filename(filename)
            return True
        except ValueError:
            return False

    # -- Serialization ---------------------------------------------------------

    def to_yaml(self) -> str:
        """Serialize to YAML string."""
        return yaml.dump(
            self.model_dump(mode="json"),
            default_flow_style=False,
            sort_keys=False,
            allow_unicode=True,
        )

    def save(self, path: Path) -> None:
        """Write recipe to a YAML file."""
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(self.to_yaml(), encoding="utf-8")

    @classmethod
    def from_yaml(cls, text: str) -> NamingRecipe:
        """Load from a YAML string."""
        data = yaml.safe_load(text)
        return cls.model_validate(data)

    @classmethod
    def load(cls, path: Path) -> NamingRecipe:
        """Load from a YAML file."""
        text = path.read_text(encoding="utf-8")
        return cls.from_yaml(text)

class RecipeNotFoundError(FileNotFoundError):
    """No recipe file exists for a site and recipe name."""

@dataclass(frozen=True)
class VirtualFile:
    """Physical file mapped to its canVOD conventional name."""

    physical_path: Path
    conventional_name: CanVODFilename

    @property
    def canonical_str(self) -> str:
        """The conventional filename as a string."""
        return self.conventional_name.name

    def open(self, mode: str = "rb"):
        """Open the physical file."""
        return self.physical_path.open(mode)