GTC-2825: True color integrated alerts

app/routes/gfw_integrated_alerts/raster_tiles.py

@@ -0,0 +1,64 @@
+from typing import Optional, Tuple
+
+from aenum import Enum, extend_enum
+from fastapi import APIRouter, BackgroundTasks, Depends, Path, Query, Response
+
+from ...crud.sync_db.tile_cache_assets import get_versions
+from ...models.enumerators.tile_caches import TileCacheType
+from .. import DATE_REGEX, optional_implementation_dependency, raster_xyz
+from ..dynamic_deforestation_alerts_tile import get_dynamic_deforestation_alert_tile
+
+router = APIRouter()
+
+dataset = "gfw_integrated_alerts"
+
+
+class GFWIntegratedAlertsVersions(str, Enum):
+    """GFW Integrated Alerts versions. When using `latest` call will be redirected (307) to version tagged as latest."""
+
+    latest = "latest"
+
+
+_versions = get_versions(dataset, TileCacheType.raster_tile_cache)
+for _version in _versions:
+    extend_enum(GFWIntegratedAlertsVersions, _version, _version)
+
+
+@router.get(
+    f"/{dataset}/{{version}}/dynamic/{{z}}/{{x}}/{{y}}.png",
+    response_class=Response,
+    tags=["Raster Tiles"],
+    response_description="PNG Raster Tile",
+)
+async def gfw_integrated_alerts_raster_tile(
+    *,
+    version: GFWIntegratedAlertsVersions = Path(..., description=GFWIntegratedAlertsVersions.__doc__),
+    xyz: Tuple[int, int, int] = Depends(raster_xyz),
+    start_date: Optional[str] = Query(
+        None,
+        regex=DATE_REGEX,
+        description="Only show alerts for given date and after",
+    ),
+    end_date: Optional[str] = Query(
+        None, regex=DATE_REGEX, description="Only show alerts until given date."
+    ),
+    confirmed_only: Optional[bool] = Query(
+        None, description="Only show confirmed alerts"
+    ),
+    implementation: str = Depends(optional_implementation_dependency),
+    background_tasks: BackgroundTasks,
+) -> Response:
+    """
+    GFW Integrated alerts raster tiles.
+    """
+
+    return await get_dynamic_deforestation_alert_tile(
+        dataset,
+        version,
+        implementation,
+        xyz,
+        start_date,
+        end_date,
+        confirmed_only,
+        background_tasks,
+    )

lambdas/raster_tiler/lambda_function.py

@@ -172,6 +172,58 @@ def get_alpha_band(
     return alpha
 
 
+def get_integrated_alerts_alpha_band(
+    rgba: ndarray,
+    start_date: Optional[int],
+    end_date: Optional[int],
+    confirmed_only: Optional[bool],
+) -> ndarray:
+    """Compute alpha value based on RGBA encoding and applied filters.
+
+    Expecting 3D array.
+    """
+
+    logger.debug("Get Deforestation Alert Alpha Band")
+
+    # encode false color tiles
+    red, green, blue, alpha = rgba
+
+    # date and intensity must be Unit16 to stay in value range
+    days = (red.astype("uint16") * 255 + green).astype("uint16")
+    intensity = (blue % 100).astype("uint16")
+
+    # build masks
+    date_mask: Union[bool, ndarray] = (start_date is None or start_date <= days) * (
+        end_date is None or days <= end_date
+    )
+
+    # From the GFW repo:
+    #  First 6 bits Alpha channel used to individual alert confidence
+    #  First two bits (leftmost) are GLAD-L
+    #  Next, 3rd and 4th bits are GLAD-S2
+    #  Finally, 5th and 6th bits are RADD
+    #  Bits are either: 00 (0, no alerts), 01 (1, low conf), or 10 (2, high conf)
+    #  e.g. 00 10 01 00 --> no GLAD-L, high conf GLAD-S2, low conf RADD
+    conf_1 = ((alpha >> 6) & 3)
+    conf_2 = ((alpha >> 4) & 3)
+    conf_3 = ((alpha >> 2) & 3)
+    combined_conf = conf_1 + conf_2 + conf_3
+
+    if confirmed_only:
+        confidence_mask: Union[bool, ndarray] = combined_conf >= 4
+    else:
+        confidence_mask = True
+
+    no_data_mask: ndarray = red + green + blue > 0
+
+    # compute new alpha value
+    new_alpha: ndarray = (
+        np.minimum(255, intensity * 50) * date_mask * confidence_mask * no_data_mask
+    ).astype("uint8")
+
+    return new_alpha
+
+
 def apply_deforestation_filter(
     data: ndarray,
     start_date: Optional[str],
@@ -206,6 +258,40 @@ def apply_deforestation_filter(
     return np.array([red, green, blue, alpha])
 
 
+def apply_integrated_alerts_filter(
+    data: ndarray,
+    start_date: Optional[str],
+    end_date: Optional[str],
+    confirmed_only: Optional[bool],
+    **kwargs,
+) -> ndarray:
+    """Decode using Pink alert color and filtering out unwanted alerts."""
+
+    logger.debug("Apply Deforestation Filter")
+
+    start_day = (
+        days_since_bog(datetime.strptime(start_date, "%Y-%m-%d").date())
+        if start_date
+        else None
+    )
+    end_day = (
+        days_since_bog(datetime.strptime(end_date, "%Y-%m-%d").date())
+        if end_date
+        else None
+    )
+
+    # Create an all pink image with varying opacity
+    red = np.ones(data[0].shape).astype("uint8") * 228
+    green = np.ones(data[1].shape).astype("uint8") * 102
+    blue = np.ones(data[2].shape).astype("uint8") * 153
+
+    # Compute alpha value
+    alpha = get_integrated_alerts_alpha_band(data, start_day, end_day, confirmed_only)
+
+    # stack bands and return
+    return np.array([red, green, blue, alpha])
+
+
 ############################
 # Data Lake Reader
 ############################
@@ -427,6 +513,7 @@ def handler(event: Dict[str, Any], _: Dict[str, Any]) -> Dict[str, str]:
     filter_constructor = {
         "annual_loss": apply_annual_loss_filter,
         "deforestation_alerts": apply_deforestation_filter,
+        "integrated_alerts": apply_integrated_alerts_filter
     }
 
     source: str = event.get("source", "datalake")