Implementation and tests for TPI stats endpoints

.vscode/settings.json

@@ -109,6 +109,7 @@
         "RDPS",
         "reproject",
         "rocketchat",
+        "rollup",
         "sessionmaker",
         "sfms",
         "sqlalchemy",

api/app/db/crud/auto_spatial_advisory.py

@@ -352,6 +352,23 @@ async def get_zonal_elevation_stats(session: AsyncSession, fire_zone_id: int, ru
     return await session.execute(stmt)
 
 
+async def get_zonal_tpi_stats(session: AsyncSession, fire_zone_id: int, run_type: RunType, run_datetime: datetime, for_date: date) -> AdvisoryTPIStats:
+    run_parameters_id = await get_run_parameters_id(session, run_type, run_datetime, for_date)
+    stmt = select(Shape.id).where(Shape.source_identifier == str(fire_zone_id))
+    result = await session.execute(stmt)
+    shape_id = result.scalar()
+
+    stmt = select(
+        AdvisoryTPIStats.advisory_shape_id,
+        AdvisoryTPIStats.valley_bottom,
+        AdvisoryTPIStats.mid_slope,
+        AdvisoryTPIStats.upper_slope,
+    ).where(AdvisoryTPIStats.advisory_shape_id == shape_id, AdvisoryTPIStats.run_parameters == run_parameters_id)
+
+    result = await session.execute(stmt)
+    return result.first()
+
+
 async def get_provincial_rollup(session: AsyncSession, run_type: RunTypeEnum, run_datetime: datetime, for_date: date) -> List[Row]:
     logger.info("gathering provincial rollup")
     run_parameter_id = await get_run_parameters_id(session, run_type, run_datetime, for_date)

api/app/routers/fba.py

@@ -1,7 +1,7 @@
-""" Routers for Auto Spatial Advisory
-"""
+"""Routers for Auto Spatial Advisory"""
 
 import logging
+import math
 from datetime import date, datetime
 from typing import List
 from fastapi import APIRouter, Depends
@@ -15,6 +15,7 @@
     get_provincial_rollup,
     get_run_datetimes,
     get_zonal_elevation_stats,
+    get_zonal_tpi_stats,
 )
 from app.db.models.auto_spatial_advisory import RunTypeEnum
 from app.schemas.fba import (
@@ -25,13 +26,14 @@
     FireZoneElevationStats,
     FireZoneElevationStatsByThreshold,
     FireZoneElevationStatsListResponse,
+    FireZoneTPIStats,
     SFMSFuelType,
     HfiThreshold,
     FireShapeAreaDetail,
     ProvincialSummaryResponse,
 )
 from app.auth import authentication_required, audit
-from app.wildfire_one.wfwx_api import (get_auth_header, get_fire_centers)
+from app.wildfire_one.wfwx_api import get_auth_header, get_fire_centers
 from app.auto_spatial_advisory.process_hfi import RunType
 
 logger = logging.getLogger(__name__)
@@ -42,38 +44,37 @@
 )
 
 
-@router.get('/fire-centers', response_model=FireCenterListResponse)
+@router.get("/fire-centers", response_model=FireCenterListResponse)
 async def get_all_fire_centers(_=Depends(authentication_required)):
-    """ Returns fire centers for all active stations. """
-    logger.info('/fba/fire-centers/')
+    """Returns fire centers for all active stations."""
+    logger.info("/fba/fire-centers/")
     async with ClientSession() as session:
         header = await get_auth_header(session)
         fire_centers = await get_fire_centers(session, header)
     return FireCenterListResponse(fire_centers=fire_centers)
 
 
-@router.get('/fire-shape-areas/{run_type}/{run_datetime}/{for_date}',
-            response_model=FireShapeAreaListResponse)
+@router.get("/fire-shape-areas/{run_type}/{run_datetime}/{for_date}", response_model=FireShapeAreaListResponse)
 async def get_shapes(run_type: RunType, run_datetime: datetime, for_date: date, _=Depends(authentication_required)):
-    """ Return area of each zone unit shape, and percentage of area of zone unit shape with high hfi. """
+    """Return area of each zone unit shape, and percentage of area of zone unit shape with high hfi."""
     async with get_async_read_session_scope() as session:
         shapes = []
 
-        rows = await get_hfi_area(session,
-                                  RunTypeEnum(run_type.value),
-                                  run_datetime,
-                                  for_date)
+        rows = await get_hfi_area(session, RunTypeEnum(run_type.value), run_datetime, for_date)
 
         # Fetch rows.
         for row in rows:
             combustible_area = row.combustible_area  # type: ignore
             hfi_area = row.hfi_area  # type: ignore
-            shapes.append(FireShapeArea(
-                fire_shape_id=row.source_identifier,  # type: ignore
-                threshold=row.threshold,  # type: ignore
-                combustible_area=row.combustible_area,  # type: ignore
-                elevated_hfi_area=row.hfi_area,  # type: ignore
-                elevated_hfi_percentage=hfi_area / combustible_area * 100))
+            shapes.append(
+                FireShapeArea(
+                    fire_shape_id=row.source_identifier,  # type: ignore
+                    threshold=row.threshold,  # type: ignore
+                    combustible_area=row.combustible_area,  # type: ignore
+                    elevated_hfi_area=row.hfi_area,  # type: ignore
+                    elevated_hfi_percentage=hfi_area / combustible_area * 100,
+                )
+            )
         return FireShapeAreaListResponse(shapes=shapes)
 
 
@@ -102,16 +103,12 @@
     return ProvincialSummaryResponse(provincial_summary=fire_shape_area_details)
 
 
-@router.get('/hfi-fuels/{run_type}/{for_date}/{run_datetime}/{zone_id}',
-            response_model=dict[int, List[ClassifiedHfiThresholdFuelTypeArea]])
-async def get_hfi_fuels_data_for_fire_zone(run_type: RunType,
-                                           for_date: date,
-                                           run_datetime: datetime,
-                                           zone_id: int):
+@router.get("/hfi-fuels/{run_type}/{for_date}/{run_datetime}/{zone_id}", response_model=dict[int, List[ClassifiedHfiThresholdFuelTypeArea]])
+async def get_hfi_fuels_data_for_fire_zone(run_type: RunType, for_date: date, run_datetime: datetime, zone_id: int):
     """
     Fetch rollup of fuel type/HFI threshold/area data for a specified fire zone.
     """
-    logger.info('hfi-fuels/%s/%s/%s/%s', run_type.value, for_date, run_datetime, zone_id)
+    logger.info("hfi-fuels/%s/%s/%s/%s", run_type.value, for_date, run_datetime, zone_id)
 
     async with get_async_read_session_scope() as session:
         # get thresholds data
@@ -120,11 +117,9 @@
         fuel_types = await get_all_sfms_fuel_types(session)
 
         # get HFI/fuels data for specific zone
-        hfi_fuel_type_ids_for_zone = await get_precomputed_high_hfi_fuel_type_areas_for_shape(session,
-                                                                            run_type=RunTypeEnum(run_type.value),
-                                                                            for_date=for_date,
-                                                                            run_datetime=run_datetime,
-                                                                            advisory_shape_id=zone_id)
+        hfi_fuel_type_ids_for_zone = await get_precomputed_high_hfi_fuel_type_areas_for_shape(
+            session, run_type=RunTypeEnum(run_type.value), for_date=for_date, run_datetime=run_datetime, advisory_shape_id=zone_id
+        )
         data = []
 
         for record in hfi_fuel_type_ids_for_zone:
@@ -135,25 +130,21 @@
             area = record[3] / 10000
             fuel_type_obj = next((ft for ft in fuel_types if ft.fuel_type_id == fuel_type_id), None)
             threshold_obj = next((th for th in thresholds if th.id == threshold_id), None)
-            data.append(ClassifiedHfiThresholdFuelTypeArea(
-                fuel_type=SFMSFuelType(
-                    fuel_type_id=fuel_type_obj.fuel_type_id,
-                    fuel_type_code=fuel_type_obj.fuel_type_code,
-                    description=fuel_type_obj.description
-                ),
-                threshold=HfiThreshold(
-                    id=threshold_obj.id,
-                    name=threshold_obj.name,
-                    description=threshold_obj.description),
-                area=area))
+            data.append(
+                ClassifiedHfiThresholdFuelTypeArea(
+                    fuel_type=SFMSFuelType(fuel_type_id=fuel_type_obj.fuel_type_id, fuel_type_code=fuel_type_obj.fuel_type_code, description=fuel_type_obj.description),
+                    threshold=HfiThreshold(id=threshold_obj.id, name=threshold_obj.name, description=threshold_obj.description),
+                    area=area,
+                )
+            )
 
         return {zone_id: data}
 
 
-@router.get('/sfms-run-datetimes/{run_type}/{for_date}', response_model=List[datetime])
+@router.get("/sfms-run-datetimes/{run_type}/{for_date}", response_model=List[datetime])
 async def get_run_datetimes_for_date_and_runtype(run_type: RunType, for_date: date, _=Depends(authentication_required)):
-    """ Return list of datetimes for which SFMS has run, given a specific for_date and run_type.
-    Datetimes should be ordered with most recent first. """
+    """Return list of datetimes for which SFMS has run, given a specific for_date and run_type.
+    Datetimes should be ordered with most recent first."""
     async with get_async_read_session_scope() as session:
         datetimes = []
 
@@ -165,21 +156,29 @@
         return datetimes
 
 
-@router.get('/fire-zone-elevation-info/{run_type}/{for_date}/{run_datetime}/{fire_zone_id}',
-            response_model=FireZoneElevationStatsListResponse)
-async def get_fire_zone_elevation_stats(fire_zone_id: int, run_type: RunType, run_datetime: datetime, for_date: date,
-                                        _=Depends(authentication_required)):
-    """ Return the elevation statistics for each advisory threshold """
+@router.get("/fire-zone-elevation-info/{run_type}/{for_date}/{run_datetime}/{fire_zone_id}", response_model=FireZoneElevationStatsListResponse)
+async def get_fire_zone_elevation_stats(fire_zone_id: int, run_type: RunType, run_datetime: datetime, for_date: date, _=Depends(authentication_required)):
+    """Return the elevation statistics for each advisory threshold"""
     async with get_async_read_session_scope() as session:
         data = []
         rows = await get_zonal_elevation_stats(session, fire_zone_id, run_type, run_datetime, for_date)
         for row in rows:
-            stats = FireZoneElevationStats(
-                minimum=row.minimum,
-                quartile_25=row.quartile_25,
-                median=row.median,
-                quartile_75=row.quartile_75,
-                maximum=row.maximum)
+            stats = FireZoneElevationStats(minimum=row.minimum, quartile_25=row.quartile_25, median=row.median, quartile_75=row.quartile_75, maximum=row.maximum)
             stats_by_threshold = FireZoneElevationStatsByThreshold(threshold=row.threshold, elevation_info=stats)
             data.append(stats_by_threshold)
         return FireZoneElevationStatsListResponse(hfi_elevation_info=data)
+
+
+@router.get("/fire-zone-tpi-stats/{run_type}/{for_date}/{run_datetime}/{fire_zone_id}", response_model=FireZoneTPIStats)
+async def get_fire_zone_tpi_stats(fire_zone_id: int, run_type: RunType, run_datetime: datetime, for_date: date, _=Depends(authentication_required)):
+    """Return the elevation TPI statistics for each advisory threshold"""
+    logger.info("/fba/fire-zone-tpi-stats/")
+    async with get_async_read_session_scope() as session:
+        stats = await get_zonal_tpi_stats(session, fire_zone_id, run_type, run_datetime, for_date)
+        square_metres = math.pow(stats.pixel_size_metres, 2)
+        return FireZoneTPIStats(
+            fire_zone_id=fire_zone_id,
+            valley_bottom=stats.valley_bottom * square_metres,
+            mid_slope=stats.mid_slope * square_metres,
+            upper_slope=stats.upper_slope * square_metres,
+        )

api/app/schemas/fba.py

@@ -1,31 +1,35 @@
-""" This module contains pydantic models related to the new formal/non-tinker fba. """
+"""This module contains pydantic models related to the new formal/non-tinker fba."""
 
 from typing import List, Optional
 from pydantic import BaseModel
 
 
 class FireCenterStation(BaseModel):
-    """ A fire weather station has a code, name and geographical coordinate. """
+    """A fire weather station has a code, name and geographical coordinate."""
+
     code: int
     name: str
     zone: Optional[str] = None
 
 
 class FireCentre(BaseModel):
-    """ The highest-level organizational unit for wildfire planning. Each fire centre
-    has 1 or more planning areas within it. """
+    """The highest-level organizational unit for wildfire planning. Each fire centre
+    has 1 or more planning areas within it."""
+
     id: str
     name: str
     stations: List[FireCenterStation]
 
 
 class FireCenterListResponse(BaseModel):
-    """ Response for all fire centers, in a list """
+    """Response for all fire centers, in a list"""
+
     fire_centers: List[FireCentre]
 
 
 class FireShapeArea(BaseModel):
-    """ A zone is a grouping of planning areas within a fire centre. """
+    """A zone is a grouping of planning areas within a fire centre."""
+
     fire_shape_id: int
     threshold: Optional[int] = None
     combustible_area: float
@@ -34,11 +38,14 @@ class FireShapeArea(BaseModel):
 
 
 class FireShapeAreaListResponse(BaseModel):
-    """ Response for all planning areas, in a list """
+    """Response for all planning areas, in a list"""
+
     shapes: List[FireShapeArea]
 
+
 class FireShapeAreaDetail(FireShapeArea):
-    """ Summary information about an advisory shape """
+    """Summary information about an advisory shape"""
+
     fire_shape_name: str
     fire_centre_name: str
 
@@ -48,62 +55,80 @@ class ProvincialSummaryResponse(BaseModel):
 
 
 class FireShapeHighHfiAreas(BaseModel):
-    """ A fire zone and the area exceeding HFI thresholds. """
+    """A fire zone and the area exceeding HFI thresholds."""
+
     fire_shape_id: int
     advisory_area: float
     warn_area: float
 
 
 class FireShapeHighHfiAreasListResponse(BaseModel):
-    """ Response for all fire zones and their areas exceeding high HFI thresholds. """
+    """Response for all fire zones and their areas exceeding high HFI thresholds."""
+
     zones: List[FireShapeHighHfiAreas]
 
 
 class HfiThresholdAreaByFuelType(BaseModel):
-    """ Total area in sq.m. within HFI threshold for a specific fuel type """
+    """Total area in sq.m. within HFI threshold for a specific fuel type"""
+
     fuel_type_id: int
     threshold: int
     area: float
 
 
 class HfiThreshold(BaseModel):
-    """ An HFI Classification threshold """
+    """An HFI Classification threshold"""
+
     id: int
     name: str
     description: str
 
 
 class SFMSFuelType(BaseModel):
-    """ Data for fuel types used by SFMS system to calculate HFI spatially. """
+    """Data for fuel types used by SFMS system to calculate HFI spatially."""
+
     fuel_type_id: int
     fuel_type_code: str
     description: str
 
 
 class ClassifiedHfiThresholdFuelTypeArea(BaseModel):
-    """ Collection of data objects recording the area within an advisory shape
+    """Collection of data objects recording the area within an advisory shape
     that meets a particular HfiThreshold for a specific SFMSFuelType
     """
+
     fuel_type: SFMSFuelType
     threshold: HfiThreshold
     area: float
 
 
 class FireZoneElevationStats(BaseModel):
-    """ Basic elevation statistics for a firezone """
+    """Basic elevation statistics for a firezone"""
+
     minimum: float
     quartile_25: float
     median: float
     quartile_75: float
     maximum: float
 
 
+class FireZoneTPIStats(BaseModel):
+    """Classified TPI areas of the fire zone contributing to the HFI >4k. Each area is in square metres."""
+
+    fire_zone_id: int
+    valley_bottom: int
+    mid_slope: int
+    upper_slope: int
+
+
 class FireZoneElevationStatsByThreshold(BaseModel):
-    """ Elevation statistics for a firezone by threshold"""
+    """Elevation statistics for a firezone by threshold"""
+
     threshold: int
     elevation_info: FireZoneElevationStats
 
 
 class FireZoneElevationStatsListResponse(BaseModel):
-    """ Response for a firezone that includes elevation statistics by threshold for the run parameters of interest """
+    """Response for a firezone that includes elevation statistics by threshold for the run parameters of interest"""
+
     hfi_elevation_info: List[FireZoneElevationStatsByThreshold]