Merge pull request #166 from wri/titiler/encoded-alert-tiles

Titiler/encoded alert tiles

app/models/enumerators/alerts_confidence.py → app/models/enumerators/titiler.py

@@ -5,3 +5,8 @@ class AlertConfidence(str, Enum):
     low = "low"
     high = "high"
     highest = "highest"
+
+
+class RenderType(str, Enum):
+    true_color = "true_color"
+    encoded = "encoded"

app/routes/titiler/algorithms/alerts.py

@@ -3,11 +3,12 @@
 
 import numpy as np
 from dateutil.relativedelta import relativedelta
+from fastapi.logger import logger
 from pydantic import Field
 from rio_tiler.models import ImageData
 from titiler.core.algorithm import BaseAlgorithm
 
-from app.models.enumerators.alerts_confidence import AlertConfidence
+from app.models.enumerators.titiler import AlertConfidence, RenderType
 
 Colors: namedtuple = namedtuple("Colors", ["red", "green", "blue"])
 AlertConfig: namedtuple = namedtuple("AlertConfig", ["confidence", "colors"])
@@ -53,31 +54,57 @@ class Alerts(BaseAlgorithm):
         AlertConfidence.low, description="Alert confidence"
     )
 
+    render_type: RenderType = Field(
+        RenderType.true_color,
+        description="Render true color or encoded pixels",
+    )
+
     # metadata
     input_nbands: int = 2
     output_nbands: int = 4
     output_dtype: str = "uint8"
 
     def __call__(self, img: ImageData) -> ImageData:
-        """Decode deforestation and land disturbance alert raster data to
-        RGBA."""
+        """Process the input image and decode deforestation or land disturbance
+        alert raster data into RGBA format.
+
+        Args:
+            img (ImageData): Input image data with alert date/confidence and intensity
+            (zoom-level visibility) layers.
+
+        Returns:
+            ImageData: Processed image with RGBA channels either with true colors ready for
+            visualization or encoding date and confidence for front-end processing.
+        """
         date_conf_data = img.data[0]
 
         self.intensity = img.data[1]
         self.no_data = img.array.mask[0]
         self.data_alert_confidence = date_conf_data // 10000
         self.alert_date = date_conf_data % 10000
 
-        mask = self.create_mask()
-        rgb = self.create_rgb()
-        alpha = self.create_alpha(mask)
+        self.mask = self.create_mask()
+
+        if self.render_type == RenderType.true_color:
+            rgb = self.create_true_color_rgb()
+            alpha = self.create_true_color_alpha()
+        else:  # encoded
+            rgb = self.create_encoded_rgb()
+            alpha = self.create_encoded_alpha()
 
         data = np.vstack([rgb, alpha[np.newaxis, ...]]).astype(self.output_dtype)
         data = np.ma.MaskedArray(data, mask=False)
 
         return ImageData(data, assets=img.assets, crs=img.crs, bounds=img.bounds)
 
     def create_mask(self):
+        """Generate a mask for pixel visibility based on date and confidence
+        filters, and no data values.
+
+        Returns:
+            np.ndarray: A mask array pixels with no alert or alerts not meeting filter
+            condition are masked.
+        """
         start_mask = self.alert_date >= (
             np.datetime64(self.start_date) - np.datetime64(self.record_start_date)
         )
@@ -93,10 +120,13 @@ def create_mask(self):
 
         return mask
 
-    def create_rgb(self):
-        r = np.zeros_like(self.data_alert_confidence, dtype=np.uint8)
-        g = np.zeros_like(self.data_alert_confidence, dtype=np.uint8)
-        b = np.zeros_like(self.data_alert_confidence, dtype=np.uint8)
+    def create_true_color_rgb(self):
+        """Map alert confidence levels to RGB values for visualization.
+
+        Returns:
+            np.ndarray: A 3D array with RGB channels.
+        """
+        r, g, b = self._rgb_zeros_array()
 
         for properties in self.conf_colors.values():
             confidence = properties.confidence
@@ -108,6 +138,50 @@ def create_rgb(self):
 
         return np.stack([r, g, b], axis=0)
 
-    def create_alpha(self, mask):
-        alpha = np.where(mask, self.intensity * 150, 0)
+    def create_encoded_rgb(self):
+        """Encode the alert date and confidence into the RGB channels, allowing
+        interactive date filtering and color control on Flagship.
+
+        Returns:
+            np.ndarray: A 3D array with encoded RGB values.
+        """
+        r, g, b = self._rgb_zeros_array()
+        r = self.alert_date // 255
+        g = self.alert_date % 255
+        b = (self.data_alert_confidence // 3 + 1) * 100 + self.intensity
+
+        return np.stack([r, g, b], axis=0)
+
+    def create_true_color_alpha(self):
+        """Set the transparency (alpha) channel for alert pixels based on date,
+        confidence filters, and intensity input. The intensity multiplier is
+        used to control how isolated alerts fade out at low zoom levels,
+        matching the rendering behavior in Flagship.
+
+        Returns:
+            np.ndarray: Array representing the alpha (transparency) channel, where pixel
+            visibility is adjusted by intensity.
+        """
+        alpha = np.where(self.mask, self.intensity * 150, 0)
         return np.minimum(255, alpha)
+
+    def create_encoded_alpha(self):
+        """Generate the alpha channel for encoded alerts. The default
+        implementation sets pixel visibility based on date/confidence filters
+        and intensity input. Can be overridden for specific alert types.
+
+        Returns:
+            np.ndarray: An array representing the alpha channel.
+        """
+        logger.info(
+            """Encoded alpha not provided, returning alpha
+                    from input layer and date/confidence mask."""
+        )
+        return self.create_true_color_alpha()
+
+    def _rgb_zeros_array(self):
+        r = np.zeros_like(self.data_alert_confidence, dtype=np.uint8)
+        g = np.zeros_like(self.data_alert_confidence, dtype=np.uint8)
+        b = np.zeros_like(self.data_alert_confidence, dtype=np.uint8)
+
+        return r, g, b

app/routes/titiler/algorithms/dist_alerts.py

@@ -1,6 +1,6 @@
 from collections import OrderedDict
 
-from app.models.enumerators.alerts_confidence import AlertConfidence
+from app.models.enumerators.titiler import AlertConfidence
 
 from .alerts import AlertConfig, Alerts, Colors
 

app/routes/titiler/algorithms/integrated_alerts.py

@@ -1,6 +1,8 @@
 from collections import OrderedDict
 
-from app.models.enumerators.alerts_confidence import AlertConfidence
+import numpy as np
+
+from app.models.enumerators.titiler import AlertConfidence
 
 from .alerts import AlertConfig, Alerts, Colors
 
@@ -24,3 +26,27 @@ class IntegratedAlerts(Alerts):
     )
 
     record_start_date: str = "2014-12-31"
+
+    def create_encoded_alpha(self):
+        # Using the same method used in Data API to provide the confidence encoding
+        # GFW Flagship expects in the alpha channel where the three alerts' confidence
+        # levels are packed in the alpha channel with 2 bits for each alert (starting at 3rd).
+        # The values used are follows (bit layout and decimal value):
+        # low confidence
+        # GLADL  GLADS2  RADD  Unused
+        #   00     00     01     00     = 4 (decimal)
+        # high confidence
+        # GLADL  GLADS2  RADD  Unused
+        #   00     00     10     00     = 8 (decimal)
+        # highest confidence
+        # GLADL  GLADS2  RADD  Unused
+        #   00     01     10     00     = 24 (decimal)
+        #  More explanation: https://github.com/wri/gfw-data-api/blob/master/app/tasks/raster_tile_cache_assets/symbology.py#L92
+
+        alpha = np.zeros_like(self.data_alert_confidence, dtype=np.uint8)
+
+        alpha[self.data_alert_confidence == self.conf_colors["low"].confidence] = 4
+        alpha[self.data_alert_confidence == self.conf_colors["high"].confidence] = 8
+        alpha[self.data_alert_confidence == self.conf_colors["highest"].confidence] = 24
+
+        return alpha

docker-compose.dev.yml

@@ -19,8 +19,8 @@ services:
       - DB_USER_RO=${GFW_DB_USER_RO}
       - DB_PASSWORD_RO=${GFW_DB_PASSWORD_RO}
       - DB_PORT_RO=5432
-      - AWS_DEFAULT_PROFILE=gfw-staging
-      - DATA_LAKE_BUCKET=gfw-data-lake-staging
+      - AWS_DEFAULT_PROFILE=${AWS_PROFILE}
+      - DATA_LAKE_BUCKET=${DATA_LAKE_BUCKET}
       - LOG_LEVEL=debug
       - RASTER_TILER_LAMBDA_NAME=test
       - ENV=dev

tests/titiler/test_alerts_algo.py

@@ -5,7 +5,7 @@
 from dateutil.relativedelta import relativedelta
 from rio_tiler.models import ImageData
 
-from app.models.enumerators.alerts_confidence import AlertConfidence
+from app.models.enumerators.titiler import AlertConfidence, RenderType
 from app.routes.titiler.algorithms.integrated_alerts import IntegratedAlerts
 from tests.conftest import DATE_CONF_TIF, INTENSITY_TIF
 
@@ -77,7 +77,7 @@ def test_mask_logic_with_nodata():
     assert output.data[3, 0, 0] == 0  # Alpha should be 0 for no-data pixel
 
 
-def test_rgb():
+def test_true_color_rgb():
     """Test that the right pink pixels are used."""
     alerts = IntegratedAlerts(start_date="2022-01-01")
 
@@ -93,3 +93,20 @@ def test_rgb():
     np.testing.assert_array_equal(
         rgba.array[:, 154, 71], np.array([220, 102, 153, 255])
     )
+
+
+def test_encoded_rgba():
+    """Test encoding used for tiles served to Flagship."""
+    alerts = IntegratedAlerts(start_date="2022-01-01", render_type=RenderType.encoded)
+
+    img = get_tile_data()
+    rgba = alerts(img)
+
+    # test date encoding in red and green channels
+    np.testing.assert_array_equal(rgba.array[:2, 122, 109], np.array([12, 154]))
+
+    # test highest confidence in alpha channel
+    assert rgba.array[3, 122, 109] == 24
+
+    # test high confidence in alpha channel
+    assert rgba.array[3, 154, 71] == 8