feat(DataMart): add AdminAreaOfInterest to TCL by driver

Author: gtempus

app/crud/geostore.py

@@ -9,6 +9,7 @@
 from sqlalchemy.sql.elements import Label, TextClause
 
 from app.application import db
+from app.crud.tasks import create_or_update_task
 from app.errors import (
     BadAdminSourceException,
     BadAdminVersionException,
@@ -190,15 +191,15 @@ async def get_first_row(sql: Select):
     return row
 
 
-async def get_gadm_geostore(
+async def build_gadm_geostore(
     admin_provider: str,
     admin_version: str,
     adm_level: int,
     simplify: float | None,
     country_id: str,
     region_id: str | None = None,
     subregion_id: str | None = None,
-) -> AdminGeostoreResponse:
+) -> AdminGeostore:
     dv: Tuple[str, str] = await admin_params_to_dataset_version(
         admin_provider, admin_version
     )
@@ -281,7 +282,7 @@ async def get_gadm_geostore(
             "GeoJSON is None, try reducing or eliminating simplification."
         )
 
-    geostore: AdminGeostore = await form_admin_geostore(
+    return await form_admin_geostore(
         adm_level=adm_level,
         admin_version=admin_version,
         area=float(row.gfw_area__ha),
@@ -293,6 +294,26 @@ async def get_gadm_geostore(
         simplify=simplify,
     )
 
+
+async def get_gadm_geostore(
+    admin_provider: str,
+    admin_version: str,
+    adm_level: int,
+    simplify: float | None,
+    country_id: str,
+    region_id: str | None = None,
+    subregion_id: str | None = None,
+) -> AdminGeostoreResponse:
+    geostore: AdminGeostore = await build_gadm_geostore(
+        admin_provider=admin_provider,
+        admin_version=admin_version,
+        adm_level=adm_level,
+        simplify=simplify,
+    country_id=country_id,
+        region_id=region_id,
+        subregion_id=subregion_id,
+    )
+
     return AdminGeostoreResponse(data=geostore)
 
 

app/models/pydantic/datamart.py

@@ -8,21 +8,45 @@
 from app.models.pydantic.responses import Response
 
 from .base import StrictBaseModel
+from ...crud.geostore import build_gadm_geostore
+
 
 class AreaOfInterest(StrictBaseModel, ABC):
     @abstractmethod
-    def get_geostore_id(self) -> UUID:
+    async def get_geostore_id(self) -> UUID:
         """Return the unique identifier for the area of interest."""
         pass
 
 
 class GeostoreAreaOfInterest(AreaOfInterest):
     geostore_id:UUID = Field(..., title="Geostore ID")
 
-    def get_geostore_id(self) -> UUID:
+    async def get_geostore_id(self) -> UUID:
         return self.geostore_id
 
 
+class AdminAreaOfInterest(AreaOfInterest):
+    country: str = Field(..., title="ISO Country Code")
+    region: Optional[str] = Field(None, title="Region")
+    subregion: Optional[str] = Field(None, title="Subregion")
+    provider: str = Field('gadm', title="Administrative Boundary Provider")
+    version: str = Field('4.1', title="Administrative Boundary Version")
+
+
+    async def get_geostore_id(self) -> UUID:
+        admin_level = sum(1 for field in (self.country, self.region, self.subregion) if field is not None) - 1
+        geostore = await build_gadm_geostore(
+            admin_provider=self.provider,
+            admin_version=self.version,
+            adm_level=admin_level,
+            simplify=None,
+            country_id=self.country,
+            region_id=self.region,
+            subregion_id=self.subregion,
+        )
+        return UUID(geostore.id)
+
+
 class AnalysisStatus(str, Enum):
     saved = "saved"
     pending = "pending"
@@ -57,7 +81,7 @@ class DataMartResourceLinkResponse(Response):
 
 
 class TreeCoverLossByDriverIn(StrictBaseModel):
-    aoi: Union[GeostoreAreaOfInterest]
+    aoi: Union[GeostoreAreaOfInterest, AdminAreaOfInterest]
     canopy_cover: int = 30
     dataset_version: Dict[str, str] = {}
 

app/routes/datamart/land.py

@@ -28,7 +28,10 @@
     DataMartResourceLinkResponse,
     TreeCoverLossByDriver,
     TreeCoverLossByDriverIn,
-    TreeCoverLossByDriverResponse, AreaOfInterest, GeostoreAreaOfInterest,
+    TreeCoverLossByDriverResponse,
+    AreaOfInterest,
+    GeostoreAreaOfInterest,
+    AdminAreaOfInterest,
 )
 from app.settings.globals import API_URL
 from app.tasks.datamart.land import (
@@ -63,8 +66,22 @@ def _parse_dataset_versions(request: Request) -> Dict[str, str]:
 
 
 def _parse_area_of_interest(request: Request) -> AreaOfInterest:
-    aoi_info = request.query_params.getlist('aoi[geostore_id]');
-    return GeostoreAreaOfInterest(geostore_id=aoi_info[0])
+    if 'aoi[geostore_id]' in request.query_params:
+        return GeostoreAreaOfInterest(geostore_id=request.query_params['aoi[geostore_id]'])
+
+        # Otherwise, check if the request contains admin area information
+    if 'aoi[country]' in request.query_params:
+        return AdminAreaOfInterest(
+            country=request.query_params['aoi[country]'],
+            region=request.query_params.get('aoi[region]'),
+            subregion=request.query_params.get('aoi[subregion]'),
+            provider=request.query_params.get('aoi[provider]'),
+            version=request.query_params.get('aoi[version]'),
+        )
+
+    # If neither type is provided, raise an error
+    raise HTTPException(status_code=422, detail="Invalid Area of Interest parameters")
+
 
 @router.get(
     "/tree_cover_loss_by_driver",
@@ -87,7 +104,7 @@ def _parse_area_of_interest(request: Request) -> AreaOfInterest:
                 "schema": {
                     "oneOf": [
                         {"$ref": "#/components/schemas/GeostoreAreaOfInterest"},
-                        {"$ref": "#/components/schemas/WdpaAreaOfInterest"},
+                        {"$ref": "#/components/schemas/AdminAreaOfInterest"},
                     ]
                 }
             },
@@ -120,8 +137,9 @@ async def tree_cover_loss_by_driver_search(
     api_key: APIKey = Depends(get_api_key),
 ):
     """Search if a resource exists for a given geostore and canopy cover."""
+    geostore_id = await aoi.get_geostore_id()
     resource_id = _get_resource_id(
-        "tree_cover_loss_by_driver", aoi.get_geostore_id(), canopy_cover, dataset_versions
+        "tree_cover_loss_by_driver", geostore_id, canopy_cover, dataset_versions
     )
 
     # check if it exists
@@ -174,10 +192,11 @@ async def tree_cover_loss_by_driver_post(
 ):
     """Create new tree cover loss by drivers resource for a given geostore and
     canopy cover."""
+    geostore_id = await data.aoi.get_geostore_id()
 
     # check geostore is valid
     try:
-        await get_geostore(data.aoi.get_geostore_id(), GeostoreOrigin.rw)
+        await get_geostore(geostore_id, GeostoreOrigin.rw)
     except HTTPException:
         raise HTTPException(
             status_code=422,
@@ -190,7 +209,7 @@ async def tree_cover_loss_by_driver_post(
     dataset_version = DEFAULT_LAND_DATASET_VERSIONS | data.dataset_version
     resource_id = _get_resource_id(
         "tree_cover_loss_by_driver",
-        data.aoi.get_geostore_id(),
+        geostore_id,
         data.canopy_cover,
         dataset_version,
     )
@@ -200,7 +219,7 @@ async def tree_cover_loss_by_driver_post(
     background_tasks.add_task(
         compute_tree_cover_loss_by_driver,
         resource_id,
-        data.aoi.get_geostore_id(),
+        geostore_id,
         data.canopy_cover,
         dataset_version,
     )