release(v0.3.4)

release(v0.3.4)

docs/CHANGELOG.md

@@ -4,6 +4,8 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.3.4] - Unreleased
+
 ## [0.3.3] - 2024-12-13
 
 ### Fixed

earthdaily/__init__.py

@@ -7,7 +7,7 @@
 # to hide warnings from rioxarray or nano seconds conversion
 # warnings.filterwarnings("ignore")
 
-__version__ = "0.3.3"
+__version__ = "0.3.4"
 
 
 def EarthDataStore(

earthdaily/earthdatastore/__init__.py

@@ -28,46 +28,135 @@
 logging.getLogger("earthdaily-earthdatastore")
 
 
-def post_query_items(items, query):
-    """Applies query to items fetched from the STAC catalog.
+def apply_single_condition(
+    item_value, condition_op: str, condition_value: [any, list[any]]
+) -> bool:
+    """
+    Apply a single comparison condition to an item's property value.
 
     Parameters
     ----------
-    items : list
-        List of items
-    query : dict
-        Query to post
+    item_value : any
+        The value of the property in the item.
+    condition_op : str
+        The comparison operator (e.g., 'lt', 'gt', 'eq').
+    condition_value : [any, list[any]]
+        The value or list of values to compare against.
 
     Returns
     -------
-    items : ItemCollection
-        filtered items
+    bool
+        True if the condition is met, False otherwise.
     """
-    items_ = []
-    for idx, item in enumerate(items):
-        queries_results = 0
-        for k, v in query.items():
-            if k not in item.properties.keys():
-                continue
-            for v_op, v_val in v.items():
-                if isinstance(v_val, list):
-                    results = 0
-                    for v_val_ in v_val:
-                        operation = operator.__dict__[v_op](item.properties[k], v_val_)
-
-                        if operation:
-                            results += 1
-                    if results == len(v_val):
-                        queries_results += 1
-                else:
-                    operation = operator.__dict__[v_op](item.properties[k], v_val)
-                    if operation:
-                        queries_results += 1
-        if queries_results == len(query.keys()):
-            items_.append(item)
-
-    items = ItemCollection(items_)
-    return items
+    # Ensure condition_value is always a list
+    values = condition_value if isinstance(condition_value, list) else [condition_value]
+
+    # Get the comparison function from the operator module
+    op_func = operator.__dict__.get(condition_op)
+    if not op_func:
+        raise ValueError(f"Unsupported operator: {condition_op}")
+
+    # Check if any value meets the condition
+    return any(op_func(item_value, val) for val in values)
+
+
+def validate_property_condition(
+    item: any, property_name: str, conditions: dict[str, any]
+) -> bool:
+    """
+    Validate if an item meets all conditions for a specific property.
+
+    Parameters
+    ----------
+    item : any
+        The STAC item to check.
+    property_name : str
+        The name of the property to validate.
+    conditions : dict[str, any]
+        Dictionary of conditions to apply to the property.
+
+    Returns
+    -------
+    bool
+        True if all conditions are met, False otherwise.
+    """
+    # Check if the property exists in the item
+    if property_name not in item.properties:
+        return False
+
+    # Check each condition for the property
+    return all(
+        apply_single_condition(
+            item.properties.get(property_name), condition_op, condition_value
+        )
+        for condition_op, condition_value in conditions.items()
+    )
+
+
+def filter_items(items: list[any], query: dict[str, dict[str, any]]) -> list[any]:
+    """
+    Filter items based on a complex query dictionary.
+
+    Parameters
+    ----------
+    items : list[any]
+        List of STAC items to filter.
+    query : dict[str, dict[str, any]]
+        Query filter with operations to apply to item properties.
+
+    Returns
+    -------
+    list[any]
+        Filtered list of items matching the query.
+
+    Examples
+    --------
+    >>> query = {
+    ...     'eo:cloud_cover': {'lt': [10], 'gt': [0]},
+    ...     'datetime': {'eq': '2023-01-01'}
+    ... }
+    >>> filtered_items = filter_items(catalog_items, query)
+    """
+    return [
+        item
+        for item in items
+        if all(
+            validate_property_condition(item, property_name, conditions)
+            for property_name, conditions in query.items()
+        )
+    ]
+
+
+def post_query_items(
+    items: list[any], query: dict[str, dict[str, any]]
+) -> ItemCollection:
+    """
+    Apply a query filter to items fetched from a STAC catalog and return an ItemCollection.
+
+    Parameters
+    ----------
+    items : list[any]
+        List of STAC items to filter.
+    query : dict[str, dict[str, any]]
+        Query filter with operations to apply to item properties.
+
+    Returns
+    -------
+    ItemCollection
+        Filtered collection of items matching the query.
+
+    Examples
+    --------
+    >>> query = {
+    ...     'eo:cloud_cover': {'lt': [10], 'gt': [0]},
+    ...     'datetime': {'eq': '2023-01-01'}
+    ... }
+    >>> filtered_items = post_query_items(catalog_items, query)
+    """
+    filtered_items = filter_items(items, query)
+    return ItemCollection(
+        filtered_items
+    )  # Assuming ItemCollection is imported/defined elsewhere
 
 
 def _select_last_common_occurrences(first, second):
@@ -132,30 +221,30 @@ def _cloud_path_to_http(cloud_path):
 
 
 def enhance_assets(
-    items,
-    alternate="download",
-    use_http_url=False,
-    add_default_scale_factor=False,
-):
-    """Enhance assets with extra fields.
+    items: ItemCollection,
+    alternate: str = "download",
+    use_http_url: bool = False,
+    add_default_scale_factor: bool = False,
+) -> ItemCollection:
+    """
+    Enhance STAC item assets with additional metadata and URL transformations.
 
     Parameters
     ----------
     items : ItemCollection
-        A PySTAC ItemCollection
-    alternate : str, optional
-        Alternate asset to use, by default "download"
+        Collection of STAC items to enhance
+    alternate : Optional[str], optional
+        Alternate asset href to use, by default "download"
     use_http_url : bool, optional
-        Use HTTP URL instead of cloud path, by default False
+        Convert cloud URLs to HTTP URLs, by default False
     add_default_scale_factor : bool, optional
-        Add default scale, offset, nodata factor to assets, by default False
+        Add default scale, offset, nodata to raster bands, by default False
 
     Returns
     -------
-    items : ItemCollection
-        Updated PySTAC ItemCollection
+    ItemCollection
+        Enhanced collection of STAC items
     """
-
     if any((alternate, use_http_url, add_default_scale_factor)):
         for idx, item in enumerate(items):
             keys = list(item.assets.keys())
@@ -941,13 +1030,17 @@ def datacube(
                     assets.append(sensor_mask)
                 elif isinstance(assets, dict):
                     assets[sensor_mask] = sensor_mask
+        bbox_query = None
+
+        if bbox is None and intersects is not None:
+            bbox_query = list(cube_utils.GeometryManager(intersects).to_bbox())
+        elif bbox is not None and intersects is None:
+            bbox_query = bbox
 
         # query the items
         items = self.search(
             collections=collections,
-            bbox=list(cube_utils.GeometryManager(intersects).to_bbox())
-            if bbox is None
-            else bbox,
+            bbox=bbox_query,
             datetime=datetime,
             assets=assets,
             prefer_alternate=prefer_alternate,