Merge branch 'develop' into gtc-3168/datamart-results-table

app/crud/geostore.py

@@ -1,16 +1,33 @@
-from typing import List
+import json
+from typing import Dict, List, Optional, Tuple
 from uuid import UUID
 
 from asyncpg.exceptions import UniqueViolationError
 from fastapi.logger import logger
-from sqlalchemy import Column, Table
-from sqlalchemy.sql import Select
-from sqlalchemy.sql.elements import TextClause
+from sqlalchemy import Column, Table, func
+from sqlalchemy.sql import Select, label
+from sqlalchemy.sql.elements import Label, TextClause
 
 from app.application import db
-from app.errors import RecordNotFoundError
+from app.errors import (
+    BadAdminSourceException,
+    BadAdminVersionException,
+    GeometryIsNullError,
+    RecordNotFoundError,
+)
 from app.models.orm.user_areas import UserArea as ORMUserArea
-from app.models.pydantic.geostore import Geometry, Geostore
+from app.models.pydantic.geostore import (
+    Adm0BoundaryInfo,
+    Adm1BoundaryInfo,
+    Adm2BoundaryInfo,
+    AdminGeostore,
+    AdminGeostoreResponse,
+    AdminListResponse,
+    Geometry,
+    Geostore,
+)
+from app.settings.globals import ENV, per_env_admin_boundary_versions
+from app.utils.gadm import extract_level_id, fix_id_pattern
 
 GEOSTORE_COLUMNS: List[Column] = [
     db.column("gfw_geostore_id"),
@@ -115,3 +132,253 @@ async def create_user_area(geometry: Geometry) -> Geostore:
         geostore = await get_gfw_geostore_from_any_dataset(geo_id)
 
     return geostore
+
+
+async def get_admin_boundary_list(
+    admin_provider: str, admin_version: str
+) -> AdminListResponse:
+    dv: Tuple[str, str] = await admin_params_to_dataset_version(
+        admin_provider, admin_version
+    )
+    dataset, version = dv
+
+    src_table: Table = db.table(version)
+    src_table.schema = dataset
+
+    # What exactly is served-up by RW? It looks like it INTENDS to just
+    # serve admin 0s, but the response contains much more
+    where_clause: TextClause = db.text("adm_level=:adm_level").bindparams(adm_level="0")
+
+    gadm_admin_list_columns: List[Column] = [
+        db.column("adm_level"),
+        db.column("gfw_geostore_id"),
+        db.column("gid_0"),
+        db.column("country"),
+    ]
+    sql: Select = (
+        db.select(gadm_admin_list_columns)
+        .select_from(src_table)
+        .where(where_clause)
+        .order_by("gid_0")
+    )
+
+    rows = await get_all_rows(sql)
+
+    return AdminListResponse.parse_obj(
+        {
+            "data": [
+                {
+                    "geostoreId": str(row.gfw_geostore_id),
+                    "iso": str(row.gid_0),
+                    "name": str(row.country),
+                }
+                for row in rows
+            ],
+        }
+    )
+
+
+async def get_all_rows(sql: Select):
+    rows = await db.all(sql)
+
+    return rows
+
+
+async def get_first_row(sql: Select):
+    row = await db.first(sql)
+
+    return row
+
+
+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:
+    dv: Tuple[str, str] = await admin_params_to_dataset_version(
+        admin_provider, admin_version
+    )
+    dataset, version = dv
+
+    src_table: Table = db.table(version)
+    src_table.schema = dataset
+
+    columns_etc: List[Column | Label] = [
+        db.column("adm_level"),
+        db.column("gfw_area__ha"),
+        db.column("gfw_bbox"),
+        db.column("gfw_geostore_id"),
+        label("level_id", db.column(f"gid_{adm_level}")),
+    ]
+
+    if adm_level == 0:
+        columns_etc.append(label("name", db.column("country")))
+    else:
+        columns_etc.append(label("name", db.column(f"name_{adm_level}")))
+
+    if simplify is None:
+        columns_etc.append(label("geojson", func.ST_AsGeoJSON(db.column("geom"))))
+    else:
+        columns_etc.append(
+            label(
+                "geojson",
+                func.ST_AsGeoJSON(func.ST_Simplify(db.column("geom"), simplify)),
+            )
+        )
+
+    where_clauses: List[TextClause] = [
+        db.text("adm_level=:adm_level").bindparams(adm_level=str(adm_level))
+    ]
+
+    # gid_0 is just a three-character value, but all more specific ids are
+    # followed by an underscore (which has to be escaped because normally in
+    # SQL an underscore is a wildcard) and a revision number (for which we
+    # use an UN-escaped underscore).
+    level_id_pattern: str = country_id
+
+    if adm_level == 0:  # Special-case to avoid slow LIKE
+        where_clauses.append(
+            db.text("gid_0=:level_id_pattern").bindparams(
+                level_id_pattern=level_id_pattern
+            )
+        )
+    else:
+        assert region_id is not None
+        level_id_pattern = ".".join((level_id_pattern, region_id))
+        if adm_level >= 2:
+            assert subregion_id is not None
+            level_id_pattern = ".".join((level_id_pattern, subregion_id))
+        level_id_pattern += r"\__"
+
+        # Adjust for any errata
+        level_id_pattern = fix_id_pattern(
+            adm_level, level_id_pattern, admin_provider, admin_version
+        )
+
+        where_clauses.append(
+            db.text(f"gid_{adm_level} LIKE :level_id_pattern").bindparams(
+                level_id_pattern=level_id_pattern
+            )
+        )
+
+    sql: Select = db.select(columns_etc).select_from(src_table)
+
+    for clause in where_clauses:
+        sql = sql.where(clause)
+
+    row = await get_first_row(sql)
+    if row is None:
+        raise RecordNotFoundError(
+            f"Admin boundary not found in {admin_provider} version {admin_version}"
+        )
+
+    if row.geojson is None:
+        raise GeometryIsNullError(
+            "GeoJSON is None, try reducing or eliminating simplification."
+        )
+
+    geostore: AdminGeostore = await form_admin_geostore(
+        adm_level=adm_level,
+        admin_version=admin_version,
+        area=float(row.gfw_area__ha),
+        bbox=[float(val) for val in row.gfw_bbox],
+        name=str(row.name),
+        geojson=json.loads(row.geojson),
+        geostore_id=str(row.gfw_geostore_id),
+        level_id=str(row.level_id),
+        simplify=simplify,
+    )
+
+    return AdminGeostoreResponse(data=geostore)
+
+
+async def admin_params_to_dataset_version(
+    source_provider: str, source_version: str
+) -> Tuple[str, str]:
+    admin_source_to_dataset: Dict[str, str] = {"GADM": "gadm_administrative_boundaries"}
+
+    try:
+        dataset: str = admin_source_to_dataset[source_provider.upper()]
+    except KeyError:
+        raise BadAdminSourceException(
+            (
+                "Invalid admin boundary source. Valid sources:"
+                f" {[source.lower() for source in admin_source_to_dataset.keys()]}"
+            )
+        )
+
+    try:
+        version: str = per_env_admin_boundary_versions[ENV][source_provider.upper()][
+            source_version
+        ]
+    except KeyError:
+        raise BadAdminVersionException(
+            (
+                "Invalid admin boundary version. Valid versions:"
+                f" {[v for v in per_env_admin_boundary_versions[ENV][source_provider.upper()].keys()]}"
+            )
+        )
+
+    return dataset, version
+
+
+async def form_admin_geostore(
+    adm_level: int,
+    bbox: List[float],
+    area: float,
+    geostore_id: str,
+    level_id: str,
+    simplify: Optional[float],
+    admin_version: str,
+    geojson: Dict,
+    name: str,
+) -> AdminGeostore:
+    info = Adm0BoundaryInfo.parse_obj(
+        {
+            "use": {},
+            "simplifyThresh": simplify,
+            "gadm": admin_version,
+            "name": name,
+            "iso": extract_level_id(0, level_id),
+        }
+    )
+    if adm_level >= 1:
+        info = Adm1BoundaryInfo(
+            **info.dict(),
+            id1=int(extract_level_id(1, level_id)),
+        )
+    if adm_level == 2:
+        info = Adm2BoundaryInfo(
+            **info.dict(),
+            id2=int(extract_level_id(2, level_id)),
+        )
+
+    return AdminGeostore.parse_obj(
+        {
+            "type": "geoStore",
+            "id": geostore_id,
+            "attributes": {
+                "geojson": {
+                    "crs": {},
+                    "type": "FeatureCollection",
+                    "features": [
+                        {
+                            "geometry": geojson,
+                            "properties": None,
+                            "type": "Feature",
+                        }
+                    ],
+                },
+                "hash": geostore_id,
+                "provider": {},
+                "areaHa": area,
+                "bbox": bbox,
+                "lock": False,
+                "info": info.dict(),
+            },
+        }
+    )

app/errors.py

@@ -51,3 +51,15 @@ def http_error_handler(exc: HTTPException) -> ORJSONResponse:
     return ORJSONResponse(
         status_code=exc.status_code, content={"status": status, "message": message}
     )
+
+
+class BadAdminSourceException(Exception):
+    pass
+
+
+class BadAdminVersionException(Exception):
+    pass
+
+
+class GeometryIsNullError(Exception):
+    pass

app/models/orm/queries/raster_assets.py

@@ -1,4 +1,4 @@
-latest_raster_tile_sets = """
+data_environment_raster_tile_sets = """
     SELECT
       assets.asset_id,
       assets.dataset,

app/models/pydantic/datamart.py

@@ -4,7 +4,6 @@
 from pydantic import Field
 
 from app.models.pydantic.responses import Response
-
 from .base import StrictBaseModel
 
 

app/models/pydantic/geostore.py

@@ -78,19 +78,17 @@ class CreateGeostoreResponseInfo(StrictBaseModel):
     use: Dict
 
 
-class RWAdminListItem(StrictBaseModel):
+class AdminListItem(StrictBaseModel):
     geostoreId: str
     iso: str
-    name: str
 
 
-class RWAdminListItemWithName(StrictBaseModel):
-    geostoreId: str
-    iso: str
+class AdminListItemWithName(AdminListItem):
+    name: str
 
 
-class RWAdminListResponse(StrictBaseModel):
-    data: List[RWAdminListItem | RWAdminListItemWithName]
+class AdminListResponse(Response):
+    data: List[AdminListItem | AdminListItemWithName]
 
 
 class WDPAInfo(StrictBaseModel):
@@ -108,7 +106,7 @@ class LandUseInfo(StrictBaseModel):
     simplify: bool
 
 
-class RWGeostoreAttributes(StrictBaseModel):
+class AdminGeostoreAttributes(StrictBaseModel):
     geojson: FeatureCollection
     hash: str
     provider: Dict
@@ -125,11 +123,11 @@ class RWGeostoreAttributes(StrictBaseModel):
     )
 
 
-class RWGeostore(StrictBaseModel):
+class AdminGeostore(StrictBaseModel):
     type: Literal["geoStore"]
     id: str
-    attributes: RWGeostoreAttributes
+    attributes: AdminGeostoreAttributes
 
 
-class RWGeostoreResponse(StrictBaseModel):
-    data: RWGeostore
+class AdminGeostoreResponse(Response):
+    data: AdminGeostore