add a few more types

.vscode/extensions.json

@@ -1,5 +1,5 @@
 {
     "recommendations": [
-        "ms-python.black-formatter"
+        "charliermarsh.ruff"
     ]
-}
+}

src/eodash_catalog/endpoints.py

@@ -18,7 +18,7 @@
 from eodash_catalog.stac_handling import (
     add_collection_information,
     add_example_info,
-    get_or_create_collection,
+    get_or_create_collection_and_times,
 )
 from eodash_catalog.thumbnails import generate_thumbnail
 from eodash_catalog.utils import (
@@ -29,7 +29,9 @@
 
 
 def process_STAC_Datacube_Endpoint(config, endpoint, data, catalog):
-    collection, _ = get_or_create_collection(catalog, data["Name"], data, config, endpoint)
+    collection, _ = get_or_create_collection_and_times(
+        catalog, data["Name"], data, config, endpoint
+    )
     add_visualization_info(collection, data, endpoint)
 
     stac_endpoint_url = endpoint["EndPoint"]
@@ -80,7 +82,9 @@ def process_STAC_Datacube_Endpoint(config, endpoint, data, catalog):
 
 def handle_STAC_based_endpoint(config, endpoint, data, catalog, options, headers=None):
     if "Locations" in data:
-        root_collection, _ = get_or_create_collection(catalog, data["Name"], data, config, endpoint)
+        root_collection, _ = get_or_create_collection_and_times(
+            catalog, data["Name"], data, config, endpoint
+        )
         for location in data["Locations"]:
             if "FilterDates" in location:
                 collection = process_STACAPI_Endpoint(
@@ -171,7 +175,7 @@ def process_STACAPI_Endpoint(
 ):
     if headers is None:
         headers = {}
-    collection, _ = get_or_create_collection(
+    collection, _ = get_or_create_collection_and_times(
         catalog, endpoint["CollectionId"], data, config, endpoint
     )
     # add_visualization_info(collection, data, endpoint)
@@ -261,7 +265,9 @@ def handle_VEDA_endpoint(config, endpoint, data, catalog, options):
 
 
 def handle_collection_only(config, endpoint, data, catalog):
-    collection, times = get_or_create_collection(catalog, data["Name"], data, config, endpoint)
+    collection, times = get_or_create_collection_and_times(
+        catalog, data["Name"], data, config, endpoint
+    )
     if len(times) > 0 and not endpoint.get("Disable_Items"):
         for t in times:
             item = Item(
@@ -280,15 +286,17 @@ def handle_collection_only(config, endpoint, data, catalog):
 def handle_SH_WMS_endpoint(config, endpoint, data, catalog):
     # create collection and subcollections (based on locations)
     if "Locations" in data:
-        root_collection, _ = get_or_create_collection(catalog, data["Name"], data, config, endpoint)
+        root_collection, _ = get_or_create_collection_and_times(
+            catalog, data["Name"], data, config, endpoint
+        )
         for location in data["Locations"]:
             # create  and populate location collections based on times
             # TODO: Should we add some new description per location?
             location_config = {
                 "Title": location["Name"],
                 "Description": "",
             }
-            collection, _ = get_or_create_collection(
+            collection, _ = get_or_create_collection_and_times(
                 catalog, location["Identifier"], location_config, config, endpoint
             )
             collection.extra_fields["endpointtype"] = endpoint["Name"]
@@ -324,7 +332,7 @@ def handle_SH_WMS_endpoint(config, endpoint, data, catalog):
     return root_collection
 
 
-def handle_xcube_endpoint(config, endpoint, data, catalog):
+def handle_xcube_endpoint(config, endpoint, data: dict, catalog):
     root_collection = process_STAC_Datacube_Endpoint(
         config=config,
         endpoint=endpoint,
@@ -336,8 +344,8 @@ def handle_xcube_endpoint(config, endpoint, data, catalog):
     return root_collection
 
 
-def handle_GeoDB_endpoint(config, endpoint, data, catalog):
-    collection, _ = get_or_create_collection(
+def handle_GeoDB_endpoint(config, endpoint, data: dict, catalog):
+    collection, _ = get_or_create_collection_and_times(
         catalog, endpoint["CollectionId"], data, config, endpoint
     )
     select = "?select=aoi,aoi_id,country,city,time"
@@ -437,7 +445,9 @@ def handle_SH_endpoint(config, endpoint, data, catalog, options):
 
 
 def handle_WMS_endpoint(config, endpoint, data, catalog, wmts=False):
-    collection, times = get_or_create_collection(catalog, data["Name"], data, config, endpoint)
+    collection, times = get_or_create_collection_and_times(
+        catalog, data["Name"], data, config, endpoint
+    )
     spatial_extent = collection.extent.spatial.to_dict().get("bbox", [-180, -90, 180, 90])[0]
     if endpoint.get("Type") != "OverwriteTimes" or not endpoint.get("OverwriteBBox"):
         # some endpoints allow "narrowed-down" capabilities per-layer, which we utilize to not

src/eodash_catalog/generate_indicators.py

@@ -14,6 +14,7 @@
 from pystac import (
     Catalog,
     CatalogType,
+    Collection,
     Summaries,
 )
 from pystac.layout import TemplateLayoutStrategy
@@ -34,7 +35,7 @@
     add_base_overlay_info,
     add_collection_information,
     add_extra_fields,
-    get_or_create_collection,
+    get_or_create_collection_and_times,
 )
 from eodash_catalog.utils import (
     RaisingThread,
@@ -46,10 +47,22 @@
 load_dotenv()
 
 
-def process_catalog_file(file_path: str, options):
+@dataclass
+class Options:
+    catalogspath: str
+    collectionspath: str
+    indicatorspath: str
+    outputpath: str
+    vd: bool
+    ni: bool
+    tn: bool
+    collections: list[str]
+
+
+def process_catalog_file(file_path: str, options: Options):
     print("Processing catalog:", file_path)
     with open(file_path) as f:
-        config = yaml.load(f, Loader=SafeLoader)
+        config: dict = yaml.load(f, Loader=SafeLoader)
 
         if len(options.collections) > 0:
             # create only catalogs containing the passed collections
@@ -153,11 +166,13 @@ def extract_indicator_info(parent_collection):
     parent_collection.summaries = Summaries(summaries)
 
 
-def process_indicator_file(config, file_path, catalog: Catalog, options):
+def process_indicator_file(config: dict, file_path: str, catalog: Catalog, options: Options):
     with open(file_path) as f:
         print("Processing indicator:", file_path)
-        data = yaml.load(f, Loader=SafeLoader)
-        parent_indicator, _ = get_or_create_collection(catalog, data["Name"], data, config)
+        data: dict = yaml.load(f, Loader=SafeLoader)
+        parent_indicator, _ = get_or_create_collection_and_times(
+            catalog, data["Name"], data, config, {}
+        )
         if "Collections" in data:
             for collection in data["Collections"]:
                 process_collection_file(
@@ -182,10 +197,12 @@ def process_indicator_file(config, file_path, catalog: Catalog, options):
         add_to_catalog(parent_indicator, catalog, None, data)
 
 
-def process_collection_file(config, file_path, catalog, options):
+def process_collection_file(
+    config: dict, file_path: str, catalog: Catalog | Collection, options: Options
+):
     print("Processing collection:", file_path)
     with open(file_path) as f:
-        data = yaml.load(f, Loader=SafeLoader)
+        data: dict = yaml.load(f, Loader=SafeLoader)
         if "Resources" in data:
             for resource in data["Resources"]:
                 if "EndPoint" in resource:
@@ -219,7 +236,9 @@ def process_collection_file(config, file_path, catalog, options):
                         raise Exception("No collection generated")
         elif "Subcollections" in data:
             # if no endpoint is specified we check for definition of subcollections
-            parent_collection, _ = get_or_create_collection(catalog, data["Name"], data, config)
+            parent_collection, _ = get_or_create_collection_and_times(
+                catalog, data["Name"], data, config, {}
+            )
 
             locations = []
             countries = []
@@ -267,7 +286,7 @@ def process_collection_file(config, file_path, catalog, options):
                         tmp_catalog,
                         options,
                     )
-                    links = tmp_catalog.get_child(sub_coll_def["Identifier"]).get_links()
+                    links = tmp_catalog.get_child(sub_coll_def["Identifier"]).get_links()  # type: ignore
                     for link in links:
                         # extract summary information
                         if "city" in link.extra_fields:
@@ -332,18 +351,6 @@ def add_to_catalog(collection, catalog, endpoint, data):
     return link
 
 
-@dataclass
-class Options:
-    catalogspath: str
-    collectionspath: str
-    indicatorspath: str
-    outputpath: str
-    vd: bool
-    ni: bool
-    tn: bool
-    collections: list[str]
-
-
 @click.command()
 @click.option(
     "--catalog",

src/eodash_catalog/stac_handling.py

@@ -18,15 +18,14 @@
 from eodash_catalog.utils import generateDateIsostringsFromInterval
 
 
-def get_or_create_collection(catalog, collection_id, data, config, endpoint=None):
+def get_or_create_collection_and_times(catalog, collection_id, data, config, endpoint=None):
     # Check if collection already in catalog
     for collection in catalog.get_collections():
         if collection.id == collection_id:
             return collection, []
     # If none found create a new one
-    spatial_extent = [-180.0, -90.0, 180.0, 90.0]
-    if endpoint and endpoint.get("OverwriteBBox"):
-        spatial_extent = endpoint.get("OverwriteBBox")
+    spatial_extent = endpoint.get("OverwriteBBox", [-180.0, -90.0, 180.0, 90.0])
+
     spatial_extent = SpatialExtent(
         [
             spatial_extent,