Merge branch 'develop' into bump_crypto_for_security

Author: dmannarino

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,7 +1,4 @@
-from ....application import db
-from ...enum.pixetl import Grid
-
-sql = """
+data_environment_raster_tile_sets = """
     SELECT
       assets.asset_id,
       assets.dataset,
@@ -18,11 +15,7 @@
         AND versions.version = assets.version
       LEFT JOIN raster_band_metadata rb
         ON rb.asset_metadata_id = am.id
-      WHERE versions.is_latest = true
-      AND assets.asset_type = 'Raster tile set'
+      WHERE assets.asset_type = 'Raster tile set'
       AND assets.creation_options->>'pixel_meaning' NOT LIKE '%tcd%'
       AND assets.creation_options->>'grid' = :grid
     """
-
-
-latest_raster_tile_sets = db.text(sql)

app/models/pydantic/datamart.py

@@ -1,8 +1,49 @@
+from typing import Dict, Optional
 from uuid import UUID
 
+from app.models.pydantic.responses import Response
+
 from .base import StrictBaseModel
 
 
+class DataMartSource(StrictBaseModel):
+    dataset: str
+    version: str
+
+
+class DataMartMetadata(StrictBaseModel):
+    geostore_id: UUID
+    sources: list[DataMartSource]
+
+
+class DataMartResource(StrictBaseModel):
+    status: str = "saved"
+    details: Optional[str] = None
+    metadata: Optional[DataMartMetadata] = None
+
+
+class DataMartResourceLink(StrictBaseModel):
+    link: str
+
+
+class DataMartResourceLinkResponse(Response):
+    data: DataMartResourceLink
+
+
 class TreeCoverLossByDriverIn(StrictBaseModel):
     geostore_id: UUID
     canopy_cover: int = 30
+    dataset_version: Dict[str, str] = {}
+
+
+class TreeCoverLossByDriverMetadata(DataMartMetadata):
+    canopy_cover: int
+
+
+class TreeCoverLossByDriver(DataMartResource):
+    tree_cover_loss_by_driver: Optional[Dict[str, float]] = None
+    metadata: Optional[TreeCoverLossByDriverMetadata] = None
+
+
+class TreeCoverLossByDriverResponse(Response):
+    data: TreeCoverLossByDriver