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 |
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)