Source code for coincident.search.main

from __future__ import annotations

from typing import Any

import geopandas as gpd
import rustac

# Used to access formatters
from pystac_client.item_search import ItemSearch as _ItemSearch

from coincident.datasets import _alias_to_Dataset
from coincident.datasets.general import Dataset
from coincident.overlaps import subset_by_minimum_area
from coincident.search import neon_api, opentopo_api, stac, wesm

_pystac_client = _ItemSearch("no_url")






def _validate_temporal_bounds(
    dataset: Dataset,
    datetime: str | list[str] | None = None,
) -> None:
    """
    Validates the temporal bounds of a dataset against a given datetime range.

    Parameters
    ----------
    dataset
        The dataset object containing start and end attributes.
    datetime
        The datetime range to validate against the dataset's temporal bounds.
        It can be a single string, a list of strings, or None. The datetime
        range should be in a format supported by pystac_client.

    Raises
    ------
    ValueError
        If the requested search start date is after the dataset's end date or
        if the requested search end date is before the dataset's start date.
    """
    if datetime is not None:
        # Use pystac_client to parse all supported datetime inputs
        start, end = _pystac_client._format_datetime(datetime).split("/")

        if start != ".." and dataset.end is not None:
            search_start = gpd.pd.to_datetime(start)
            dataset_end = gpd.pd.to_datetime(dataset.end, utc=True)
            if search_start > dataset_end:
                message = f"Requested search start {search_start} > {dataset.alias} end date: {dataset_end}"
                raise ValueError(message)

        if end != ".." and dataset.start is not None:
            search_end = gpd.pd.to_datetime(end)
            dataset_start = gpd.pd.to_datetime(dataset.start, utc=True)
            if search_end < dataset_start:
                message = f"Requested search end {search_end} < {dataset.alias} start date: {dataset_start}"
                raise ValueError(message)


def _validate_spatial_bounds(
    intersects: gpd.GeoSeries,
) -> None:
    """
    Validate that the specified area of interest (AOI) is type GeoDataFrame or GeoSeries

    Parameters
    ----------
    intersects
        A GeoSeries containing the area of interest (AOI) geometry.

    Raises
    ------
    ValueError
        If the AOI is not type GeoDataFrame or GeoSeries
    """
    if not isinstance(intersects, gpd.GeoDataFrame | gpd.GeoSeries):
        message = f"intersects value must be a GeoDataFrame or GeoSeries, not {type(intersects)}"
        raise ValueError(message)
    if len(intersects) > 1:
        message = "GeoDataFrame contains multiple geometries, search requires a single geometry"
        raise ValueError(message)