Interfaces¶

def build_collection_asset_filters(
    collections_config: Mapping[str, Sequence[str]] | Sequence[str] | None,
) -> tuple[list[str], dict[str, set[str] | None]]:
    """Derive collection list and per-collection asset filters from configuration mapping or sequence.

    Args:
        collections_config: Mapping of collection -> list of asset/channel keys,
            or sequence of collection names.

    Returns:
        A ``(collections, asset_filters)`` tuple.
    """
    if collections_config is None:
        return [], {}

    if isinstance(collections_config, Mapping):
        asset_filters: dict[str, set[str] | None] = {}
        for coll, vars_list in collections_config.items():
            if vars_list and "*" not in vars_list:
                asset_filters[coll] = set(str(v) for v in vars_list)
            else:
                asset_filters[coll] = None
        return list(collections_config.keys()), asset_filters

    collections = list(dict.fromkeys(collections_config))
    return collections, {coll: None for coll in collections}

def empty_asset_result() -> GeoDataFrame[AssetSchema]:
    """Return an empty GeoDataFrame with AssetSchema columns."""
    import geopandas as gpd

    columns = list(AssetSchema.to_schema().columns.keys())
    if "geometry" not in columns:
        columns.append("geometry")
    gdf = gpd.GeoDataFrame(columns=columns, geometry="geometry")
    return cast(GeoDataFrame[AssetSchema], AssetSchema.validate(gdf))

def infer_dataset_time_bounds(ds: xr.Dataset) -> xr.Dataset:
    """Infer and set the start and end time bounds in the dataset's attributes.

    If a ``time`` coordinate is present, uses its minimum and maximum values.
    Otherwise, leaves the dataset attributes unchanged.

    Args:
        ds: The xarray.Dataset.

    Returns:
        The dataset with inferred time bounds set in its attributes (if possible).
    """
    import pandas as pd

    if "time" in ds.coords:
        times = ds.coords["time"].values
        if len(times) > 0:
            ds.attrs["start_time"] = pd.Timestamp(times.min()).to_pydatetime()
            ds.attrs["end_time"] = pd.Timestamp(times.max()).to_pydatetime()
    return ds

def normalize_geometry_input(
    value: BaseGeometry | dict[str, Any] | str | Path | None,
) -> BaseGeometry | None:
    """Normalize a geometry input into a Shapely ``BaseGeometry``.

    Supports:

    * ``BaseGeometry`` instances (returned unchanged)
    * GeoJSON-like ``dict``
    * A ``str`` or ``Path`` pointing to a GeoJSON file (``.geojson`` or ``.json``)

    Args:
        value: Geometry input to normalize.

    Returns:
        A Shapely geometry, or ``None`` if the input was ``None``.

    Raises:
        ValueError: If the input cannot be parsed into a geometry.
    """
    if value is None:
        return None

    if isinstance(value, BaseGeometry):
        return value

    if isinstance(value, Path):
        return _geometry_from_geojson_path(value)

    if isinstance(value, str):
        if _looks_like_geojson_path(value):
            return _geometry_from_geojson_path(Path(value))
        raise ValueError(
            "Invalid geometry input type: str. "
            "Expected a path to a GeoJSON file (ending in .geojson or .json) "
            "or a file-system path containing a separator."
        )

    if isinstance(value, dict):
        return shape(value)

    raise ValueError(
        f"Invalid geometry input type: {type(value).__name__}. "
        "Expected BaseGeometry, GeoJSON dict, or path to a GeoJSON file."
    )

def resolve_callable(val: Any) -> Any:
    """Resolve a callable from a string, dict target, or direct callable.

    Args:
        val: The value to resolve.

    Returns:
        The resolved callable.
    """
    import importlib

    if callable(val):
        return val

    if isinstance(val, str):
        if ":" in val:
            module_name, func_name = val.split(":", 1)
        else:
            module_name, func_name = val.rsplit(".", 1)
        module = importlib.import_module(module_name)
        func = getattr(module, func_name)
        return func

    if isinstance(val, dict) or (hasattr(val, "keys") and hasattr(val, "get")):
        d = dict(val)
        target = d.pop("_target_", None)
        if not target:
            raise ValueError(
                "Dictionary-based configuration must include a '_target_' key."
            )

        func = resolve_callable(target)
        kwargs = {k: v for k, v in d.items() if not k.startswith("_")}
        if kwargs:
            return partial(func, **kwargs)
        return func

    raise ValueError(f"Cannot resolve callable from type: {type(val).__name__}")

def set_dataset_time_bounds(
    ds: xr.Dataset, start_time: datetime, end_time: datetime
) -> xr.Dataset:
    """Set the start and end time bounds in the dataset's attributes.

    Args:
        ds: The xarray.Dataset.
        start_time: The start time.
        end_time: The end time.

    Returns:
        The dataset with time bounds set in its attributes.
    """
    ds.attrs["start_time"] = start_time
    ds.attrs["end_time"] = end_time
    return ds

def update_callable(callable_obj: Any, **kwargs: Any) -> Any:
    """Update a callable (standard function or partial) with new keywords.

    Args:
        callable_obj: The callable to update.
        **kwargs: The keyword arguments to update it with.

    Returns:
        A new callable with updated keyword arguments.
    """
    if isinstance(callable_obj, partial):
        new_keywords = {**callable_obj.keywords, **kwargs}
        return partial(callable_obj.func, *callable_obj.args, **new_keywords)
    if callable(callable_obj):
        return partial(callable_obj, **kwargs)
    return partial(resolve_callable(callable_obj), **kwargs)

def validate_aereo_dataset(
    ds: Any,
    *,
    require_crs: bool = True,
    require_dims: Sequence[str] | None = ("band", "y", "x"),
) -> None:
    """Validate that *ds* conforms to the AEREO xarray conventions.

    Args:
        ds: The dataset to validate.
        require_crs: If True, ensure ``ds.rio.crs`` is set.
        require_dims: If given, ensure all listed dimensions exist.

    Raises:
        ValueError: If any convention is violated.
        ImportError: If ``rioxarray`` is not installed and *require_crs* is True.
    """
    if not isinstance(ds, xr.Dataset):
        raise ValueError(f"Expected xarray.Dataset, got {type(ds).__name__}")

    if require_crs:
        _import_rioxarray()
        # Access the rio accessor to trigger its import side-effects
        if ds.rio.crs is None:
            raise ValueError(
                "xarray.Dataset must have a CRS set via rioxarray (ds.rio.crs)"
            )

    if require_dims:
        missing = [d for d in require_dims if d not in ds.dims]
        if missing:
            raise ValueError(f"xarray.Dataset missing required dimensions: {missing}")