wip with query

Author: jeafreezy

README.md

@@ -25,6 +25,7 @@ catalogs:
       username: youruser
       password: yourpass
 ```
+# load from the environment variable - SUPERSTAC_CATALOG_CONFIG
 
 # Todo - auth configuration documentation.
 

superstac/__init__.py

@@ -1 +1,7 @@
+from superstac.catalog_registry import (
+    register_catalog,
+    get_catalog_registry,
+    load_catalogs_from_config,
+)
 
+__all__ = ["register_catalog", "get_catalog_registry", "load_catalogs_from_config"]

superstac/catalog.py

@@ -98,12 +98,12 @@ def get_available_catalogs(
         return available
 
     def load_catalogs_from_config(
-        self, file: Union[str, Path, None] = None
+        self, config: Union[str, Path, None] = None
     ) -> Dict[str, CatalogEntry]:
         """Load catalogs from configuration file.
 
         Args:
-            file (Union[str, Path, None], optional): Path to the configuration file. Defaults to None.
+            config (Union[str, Path, None], optional): Path to the configuration file. Defaults to None.
 
         Raises:
             CatalogConfigFileNotFound: Raised when the catalog config file is not founds.
@@ -114,11 +114,12 @@ def load_catalogs_from_config(
             Dict[str, CatalogEntry]: The registered catalogs.
         """
         logger.info("Loading catalogs from configuration file.")
-        if file is None:
+        if config is None:
+            # todo. - make public and environment variable
             base_dir = Path(__file__).parent
-            file = base_dir / ".superstac.yml"
+            config = base_dir / ".superstac.yml"
 
-        path = Path(file).expanduser().resolve()
+        path = Path(config).expanduser().resolve()
         logger.debug(f"Resolved config path: {path}")
 
         if not path.exists():
@@ -161,14 +162,3 @@ def load_catalogs_from_config(
                 logger.warning(f"Failed to register catalog '{name}': {e}")
         logger.info("All catalogs loaded and registered.")
         return self.catalogs
-
-
-## TEST
-
-
-if __name__ == "__main__":
-    # cm = CatalogManager()
-    # cm.register_catalog(name="My Catalog", url="https://example.com/stac")
-    # print(cm.load_catalogs_from_config())
-    # print(cm.get_available_catalogs())
-    ...

superstac/catalog_registry.py

@@ -0,0 +1,35 @@
+"""SuperSTAC Catalog Registry"""
+
+from typing import Union
+from superstac.catalog import CatalogManager
+
+
+_catalog_registry = CatalogManager()
+
+
+def get_catalog_registry() -> CatalogManager:
+    """
+    Returns the singleton CatalogManager instance.
+    """
+    return _catalog_registry
+
+
+def register_catalog(*args, **kwargs):
+    """
+    Shortcut to register a catalog globally.
+    """
+    return _catalog_registry.register_catalog(*args, **kwargs)
+
+
+def load_catalogs_from_config(file: Union[str, None] = None):
+    """
+    Loads catalogs from YAML into the global registry.
+    """
+    return _catalog_registry.load_catalogs_from_config(file)
+
+
+def clear_registry():
+    """
+    Optional: Reset the registry, mainly for testing.
+    """
+    _catalog_registry.catalogs.clear()

superstac/enums.py

@@ -1,3 +1,5 @@
+"""SuperSTAC Enums"""
+
 from enum import Enum
 
 

superstac/exceptions.py

@@ -1,3 +1,6 @@
+"""SuperSTAC Exceptions"""
+
+
 class InvalidCatalogSchemaError(Exception):
     """Raised when an invalid catalog schema is provided."""
 

superstac/query.py

@@ -0,0 +1,86 @@
+"""SuperSTAC Query Manager"""
+
+from typing import List, Optional, Dict, Any
+from pystac_client import Client
+from pystac import Item
+from superstac._logging import logger
+from superstac.utils import BAND_MAP
+
+
+# todo - support auth and headers for client.open() in query_catalog_with_pystac param.
+# test resolving band alias
+
+
+def query_catalog_with_pystac(
+    name: str,
+    url: str,
+    required_assets: Optional[List[str]] = None,
+    max_items: int = 100,
+    **search_kwargs: Any,
+) -> List[Dict[str, Any]]:
+    """
+    Search a single catalog using pystac-client and normalize the assets.
+
+    Args:
+        name: Unique catalog name.
+        url: STAC API URL.
+        required_assets: Band aliases to filter and rename assets.
+        max_items: Maximum number of results.
+        **search_kwargs: All valid parameters supported by pystac-client's search().
+
+    Returns:
+        List of STAC items (as dicts), normalized with filtered assets and catalog_name.
+    """
+    try:
+        client = Client.open(url)
+        logger.debug(f"Querying STAC with: {search_kwargs.items()}")
+        search = client.search(max_items=max_items, **search_kwargs)
+        items: List[Item] = list(search.items())
+
+        collection_id = None
+        collections = search_kwargs.get("collections")
+        if isinstance(collections, list) and collections:
+            collection_id = collections[0]
+        elif isinstance(collections, str):
+            collection_id = collections
+
+        band_map = BAND_MAP.get(collection_id, {})
+        normalized = []
+
+        for item in items:
+            item_assets = item.assets
+            normalized_assets = {}
+
+            for alias, actual in band_map.items():
+                if not required_assets or alias in required_assets:
+                    if actual in item_assets:
+                        normalized_assets[alias] = item_assets[actual].to_dict()
+
+            if normalized_assets:
+                item_dict = item.to_dict()
+                item_dict["assets"] = normalized_assets
+                item_dict["catalog_name"] = name
+                normalized.append(item_dict)
+
+        logger.info(
+            f"[{name}] Returned {len(normalized)} items for collection '{collection_id}'"
+        )
+        return normalized
+
+    except Exception as e:
+        logger.warning(f"[{name}] Catalog query failed: {e}")
+        return []
+
+
+if __name__ == "__main__":
+    results = query_catalog_with_pystac(
+        name="aws",
+        url="https://earth-search.aws.element84.com/v1",
+        collections=["sentinel-2-l2a"],
+        bbox=[6.0, 49.0, 7.0, 50.0],
+        datetime="2024-01-01/2024-01-31",
+        query={"eo:cloud_cover": {"lt": 20}},
+        required_assets=["red", "nir"],
+        sortby=[{"field": "properties.datetime", "direction": "desc"}],
+    )
+    print(results)

superstac/search.py

@@ -1,3 +1,5 @@
+"""SuperSTAC utils"""
+
 import uuid
 
 
@@ -13,3 +15,12 @@ def compute_catalog_id(name: str, url: str) -> str:
     """
     uid = uuid.uuid4().hex[:10]
     return f"{name.lower().replace(' ', '_')}_{uid}"
+
+
+# Band map between Element 84's STAC and Planetary Computer
+# allow users to provide band map ?
+BAND_MAP = {
+    "sentinel-2-l2a": {"red": "B04", "green": "B03", "blue": "B02", "nir": "B08"},
+    "landsat-8": {"red": "SR_B4", "green": "SR_B3", "blue": "SR_B2", "nir": "SR_B5"},
+    "modis": {"red": "sur_refl_b01", "nir": "sur_refl_b02"},
+}