diff --git a/planet/subscription_request.py b/planet/subscription_request.py index 648c93e6..cf299143 100644 --- a/planet/subscription_request.py +++ b/planet/subscription_request.py @@ -13,7 +13,8 @@ # the License. """Functionality for preparing subscription requests.""" from datetime import datetime -from typing import Any, Dict, Optional, List, Mapping, Sequence +from dataclasses import dataclass, asdict +from typing import Any, Dict, Optional, List, Mapping, Sequence, Union from typing_extensions import Literal @@ -651,3 +652,63 @@ def toar_tool(scale_factor: int = 10000) -> dict: reflectances not fitting in 16bit integers. """ return _tool('toar', {'scale_factor': scale_factor}) + + +@dataclass +class FilterValue: + """Represents a filter value with optional greater than or equal to (gte) + and less than or equal to (lte) constraints. + + Attributes: + gte (Optional[float]): The minimum threshold value for the filter. + lte (Optional[float]): The maximum threshold value for the filter. + """ + + gte: Optional[float] = None + lte: Optional[float] = None + + +def cloud_filter_tool( + clear_percent: Optional[FilterValue] = None, + cloud_percent: Optional[FilterValue] = None, + shadow_percent: Optional[FilterValue] = None, + heavy_haze_percent: Optional[FilterValue] = None, + light_haze_percent: Optional[FilterValue] = None, + snow_ice_percent: Optional[FilterValue] = None, +) -> Dict[str, Dict[str, Union[float, int]]]: + """Specify a subscriptions API cloud_filter tool. + + The cloud_filter tool filters imagery after the clip tool has run and certain + metadata values have been updated to pertain to the clip AOI. This tool offers + a more detailed filtering of cloudy imagery than what can be achieved using + only catalog source filters. For instance, you might want to receive only images + that, after clipping, have a cloud_percent value of less than or equal to 25%. + + Parameters: + clear_percent: Filters for images based on the percentage of clear sky. + cloud_percent: Filters for images based on the percentage of cloud cover. + shadow_percent: Filters for images based on the percentage of shadow cover. + heavy_haze_percent: Filters for images based on the percentage of heavy haze cover. + light_haze_percent: Filters for images based on the percentage of light haze cover. + snow_ice_percent: Filters for images based on the percentage of snow or ice cover. + """ + filters = { + "clear_percent": clear_percent, + "cloud_percent": cloud_percent, + "shadow_percent": shadow_percent, + "heavy_haze_percent": heavy_haze_percent, + "light_haze_percent": light_haze_percent, + "snow_ice_percent": snow_ice_percent, + } + + result = {} + + for key, value in filters.items(): + if value: + inner_dict = asdict(value) + result[key] = { + k: v + for k, v in inner_dict.items() if v is not None + } + + return _tool("cloud_filter", result) diff --git a/tests/unit/test_subscription_request.py b/tests/unit/test_subscription_request.py index 1b09afaf..bd93a3ac 100644 --- a/tests/unit/test_subscription_request.py +++ b/tests/unit/test_subscription_request.py @@ -407,3 +407,22 @@ def test_catalog_source_time_range_type_acquired(geom_geojson): ) assert source["parameters"]["time_range_type"] == "acquired" + + +def test_cloud_filter_tool_success(): + res = subscription_request.cloud_filter_tool( + clear_percent=subscription_request.FilterValue(gte=90), + cloud_percent=subscription_request.FilterValue(lte=10, gte=5)) + expected = { + "type": "cloud_filter", + "parameters": { + "clear_percent": { + "gte": 90 + }, "cloud_percent": { + "lte": 10, + "gte": 5 + } + } + } + + assert res == expected