Skip to content

catalog module

Search-result workspace helpers.

SearchResult dataclass

Represent reusable search results.

Parameters:

Name Type Description Default
items List[Any]

Search result items.

required
sensor Optional[str]

Optional sensor name.

None
query Dict[str, Any]

Optional query metadata.

<factory>
gdf Any

Optional GeoDataFrame returned by search backends.

None
Source code in hypercoast/catalog.py
@dataclass
class SearchResult:
    """Represent reusable search results.

    Args:
        items: Search result items.
        sensor: Optional sensor name.
        query: Optional query metadata.
        gdf: Optional GeoDataFrame returned by search backends.
    """

    items: List[Any]
    sensor: Optional[str] = None
    query: Dict[str, Any] = field(default_factory=dict)
    gdf: Any = None

    @classmethod
    def from_result(
        cls,
        result: Any,
        sensor: Optional[str] = None,
        query: Optional[Dict[str, Any]] = None,
    ) -> "SearchResult":
        """Create a workspace from a backend search result.

        Args:
            result: List of items or ``(items, gdf)`` tuple.
            sensor: Optional sensor name.
            query: Optional query metadata.

        Returns:
            SearchResult: Search result workspace.
        """
        gdf = None
        items = result
        if isinstance(result, tuple) and len(result) >= 2:
            items, gdf = result[0], result[1]
        if isinstance(items, dict):
            items = items.get("items", [items])
        return cls(list(items or []), sensor=sensor, query=query or {}, gdf=gdf)

    @classmethod
    def from_json(cls, path: str | Path) -> "SearchResult":
        """Load search results from JSON.

        Args:
            path: JSON file path.

        Returns:
            SearchResult: Loaded search result workspace.
        """
        with open(path, encoding="utf-8") as src:
            payload = json.load(src)
        return cls(
            items=payload.get("items", []),
            sensor=payload.get("sensor"),
            query=payload.get("query", {}),
        )

    def to_json(self, path: str | Path) -> str:
        """Save search results to JSON.

        Args:
            path: Output JSON path.

        Returns:
            str: Output path.
        """
        output = Path(path)
        output.parent.mkdir(parents=True, exist_ok=True)
        payload = {
            "sensor": self.sensor,
            "query": self.query,
            "items": self.items,
        }
        with open(output, "w", encoding="utf-8") as dst:
            json.dump(payload, dst, default=str, indent=2)
        return str(output)

    def to_dataframe(self) -> pd.DataFrame:
        """Return search items as a pandas DataFrame.

        Returns:
            pd.DataFrame: Tabular search results.
        """
        if self.gdf is not None:
            return pd.DataFrame(self.gdf)
        rows = [_item_summary(item) for item in self.items]
        return pd.DataFrame(rows)

    def to_csv(self, path: str | Path) -> str:
        """Save search results to CSV.

        Args:
            path: Output CSV path.

        Returns:
            str: Output path.
        """
        output = Path(path)
        output.parent.mkdir(parents=True, exist_ok=True)
        self.to_dataframe().to_csv(output, index=False)
        return str(output)

    def to_file(self, path: str | Path, driver: Optional[str] = None) -> str:
        """Save geospatial search results when GeoDataFrame output is available.

        Args:
            path: Output file path.
            driver: Optional GeoPandas output driver.

        Returns:
            str: Output path.
        """
        if self.gdf is None:
            suffix = Path(path).suffix.lower()
            if suffix == ".csv":
                return self.to_csv(path)
            return self.to_json(path)
        output = Path(path)
        output.parent.mkdir(parents=True, exist_ok=True)
        kwargs = {"driver": driver} if driver else {}
        self.gdf.to_file(output, **kwargs)
        return str(output)

    def filter(self, **properties: Any) -> "SearchResult":
        """Filter items by exact top-level or property values.

        Args:
            **properties: Values to match.

        Returns:
            SearchResult: Filtered result workspace.
        """
        items = [
            item
            for item in self.items
            if all(_item_value(item, key) == value for key, value in properties.items())
        ]
        return SearchResult(items=items, sensor=self.sensor, query=self.query)

    def download(self, out_dir: Optional[str] = None, **kwargs: Any) -> Any:
        """Download items with the registered sensor downloader.

        Args:
            out_dir: Optional output directory.
            **kwargs: Downloader keyword arguments.

        Returns:
            Any: Downloader result.
        """
        if not self.sensor:
            raise ValueError("SearchResult.sensor is required for download.")
        return download_sensor(self.sensor, self.items, out_dir=out_dir, **kwargs)

download(self, out_dir=None, **kwargs)

Download items with the registered sensor downloader.

Parameters:

Name Type Description Default
out_dir Optional[str]

Optional output directory.

None
**kwargs Any

Downloader keyword arguments.

{}

Returns:

Type Description
Any

Downloader result.

Source code in hypercoast/catalog.py
def download(self, out_dir: Optional[str] = None, **kwargs: Any) -> Any:
    """Download items with the registered sensor downloader.

    Args:
        out_dir: Optional output directory.
        **kwargs: Downloader keyword arguments.

    Returns:
        Any: Downloader result.
    """
    if not self.sensor:
        raise ValueError("SearchResult.sensor is required for download.")
    return download_sensor(self.sensor, self.items, out_dir=out_dir, **kwargs)

filter(self, **properties)

Filter items by exact top-level or property values.

Parameters:

Name Type Description Default
**properties Any

Values to match.

{}

Returns:

Type Description
SearchResult

Filtered result workspace.

Source code in hypercoast/catalog.py
def filter(self, **properties: Any) -> "SearchResult":
    """Filter items by exact top-level or property values.

    Args:
        **properties: Values to match.

    Returns:
        SearchResult: Filtered result workspace.
    """
    items = [
        item
        for item in self.items
        if all(_item_value(item, key) == value for key, value in properties.items())
    ]
    return SearchResult(items=items, sensor=self.sensor, query=self.query)

from_json(path) classmethod

Load search results from JSON.

Parameters:

Name Type Description Default
path str | Path

JSON file path.

required

Returns:

Type Description
SearchResult

Loaded search result workspace.

Source code in hypercoast/catalog.py
@classmethod
def from_json(cls, path: str | Path) -> "SearchResult":
    """Load search results from JSON.

    Args:
        path: JSON file path.

    Returns:
        SearchResult: Loaded search result workspace.
    """
    with open(path, encoding="utf-8") as src:
        payload = json.load(src)
    return cls(
        items=payload.get("items", []),
        sensor=payload.get("sensor"),
        query=payload.get("query", {}),
    )

from_result(result, sensor=None, query=None) classmethod

Create a workspace from a backend search result.

Parameters:

Name Type Description Default
result Any

List of items or (items, gdf) tuple.

required
sensor Optional[str]

Optional sensor name.

None
query Optional[Dict[str, Any]]

Optional query metadata.

None

Returns:

Type Description
SearchResult

Search result workspace.

Source code in hypercoast/catalog.py
@classmethod
def from_result(
    cls,
    result: Any,
    sensor: Optional[str] = None,
    query: Optional[Dict[str, Any]] = None,
) -> "SearchResult":
    """Create a workspace from a backend search result.

    Args:
        result: List of items or ``(items, gdf)`` tuple.
        sensor: Optional sensor name.
        query: Optional query metadata.

    Returns:
        SearchResult: Search result workspace.
    """
    gdf = None
    items = result
    if isinstance(result, tuple) and len(result) >= 2:
        items, gdf = result[0], result[1]
    if isinstance(items, dict):
        items = items.get("items", [items])
    return cls(list(items or []), sensor=sensor, query=query or {}, gdf=gdf)

to_csv(self, path)

Save search results to CSV.

Parameters:

Name Type Description Default
path str | Path

Output CSV path.

required

Returns:

Type Description
str

Output path.

Source code in hypercoast/catalog.py
def to_csv(self, path: str | Path) -> str:
    """Save search results to CSV.

    Args:
        path: Output CSV path.

    Returns:
        str: Output path.
    """
    output = Path(path)
    output.parent.mkdir(parents=True, exist_ok=True)
    self.to_dataframe().to_csv(output, index=False)
    return str(output)

to_dataframe(self)

Return search items as a pandas DataFrame.

Returns:

Type Description
pd.DataFrame

Tabular search results.

Source code in hypercoast/catalog.py
def to_dataframe(self) -> pd.DataFrame:
    """Return search items as a pandas DataFrame.

    Returns:
        pd.DataFrame: Tabular search results.
    """
    if self.gdf is not None:
        return pd.DataFrame(self.gdf)
    rows = [_item_summary(item) for item in self.items]
    return pd.DataFrame(rows)

to_file(self, path, driver=None)

Save geospatial search results when GeoDataFrame output is available.

Parameters:

Name Type Description Default
path str | Path

Output file path.

required
driver Optional[str]

Optional GeoPandas output driver.

None

Returns:

Type Description
str

Output path.

Source code in hypercoast/catalog.py
def to_file(self, path: str | Path, driver: Optional[str] = None) -> str:
    """Save geospatial search results when GeoDataFrame output is available.

    Args:
        path: Output file path.
        driver: Optional GeoPandas output driver.

    Returns:
        str: Output path.
    """
    if self.gdf is None:
        suffix = Path(path).suffix.lower()
        if suffix == ".csv":
            return self.to_csv(path)
        return self.to_json(path)
    output = Path(path)
    output.parent.mkdir(parents=True, exist_ok=True)
    kwargs = {"driver": driver} if driver else {}
    self.gdf.to_file(output, **kwargs)
    return str(output)

to_json(self, path)

Save search results to JSON.

Parameters:

Name Type Description Default
path str | Path

Output JSON path.

required

Returns:

Type Description
str

Output path.

Source code in hypercoast/catalog.py
def to_json(self, path: str | Path) -> str:
    """Save search results to JSON.

    Args:
        path: Output JSON path.

    Returns:
        str: Output path.
    """
    output = Path(path)
    output.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "sensor": self.sensor,
        "query": self.query,
        "items": self.items,
    }
    with open(output, "w", encoding="utf-8") as dst:
        json.dump(payload, dst, default=str, indent=2)
    return str(output)

load_search_result(path)

Load a search-result workspace from JSON.

Parameters:

Name Type Description Default
path str | Path

JSON file path.

required

Returns:

Type Description
SearchResult

Loaded search result workspace.

Source code in hypercoast/catalog.py
def load_search_result(path: str | Path) -> SearchResult:
    """Load a search-result workspace from JSON.

    Args:
        path: JSON file path.

    Returns:
        SearchResult: Loaded search result workspace.
    """
    return SearchResult.from_json(path)