Dev

app/main.py

@@ -44,6 +44,7 @@
 from .routes.titiler import routes as titiler_routes
 from .routes.titiler.gfw_integrated_alerts import router as integrated_alerts_router
 from .routes.titiler.umd_glad_dist_alerts import router as dist_alerts_router
+from .routes.titiler.gfw_forest_carbon_gross_emissions import router as emissions_router
 
 gunicorn_logger = logging.getLogger("gunicorn.error")
 logger.handlers = gunicorn_logger.handlers
@@ -59,6 +60,7 @@
     vector_tiles.router,
     integrated_alerts_router,
     dist_alerts_router,
+    emissions_router,
     umd_tree_cover_loss_raster_tiles.router,
     umd_glad_landsat_alerts_raster_tiles.router,
     umd_glad_sentinel2_alerts_raster_tiles.router,

app/models/enumerators/titiler.py

@@ -15,3 +15,9 @@ class IntegratedAlertConfidence(str, Enum):
 class RenderType(str, Enum):
     true_color = "true_color"
     encoded = "encoded"
+
+
+class TreeCoverDensityThreshold(str, Enum):
+    tcd_30 = 30
+    tcd_50 = 50
+    tcd_75 = 75

app/routes/titiler/algorithms/carbon_gross_emissions.py

@@ -0,0 +1,101 @@
+from collections import OrderedDict, namedtuple
+
+import numpy as np
+from pydantic import ConfigDict
+from rio_tiler.models import ImageData
+from titiler.core.algorithm import BaseAlgorithm
+
+Colors: namedtuple = namedtuple("Colors", ["red", "green", "blue"])
+
+
+class CarbonGrossEmissions(BaseAlgorithm):
+    """Visualize carbon gross emissions."""
+
+    title: str = "Carbon gross emissions"
+    description: str = "Visualize carbon gross emissions"
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    conf_colors: OrderedDict[float, tuple] = OrderedDict(
+        {
+            0.0: Colors(254, 246, 249),
+            1.0: Colors(245, 237, 242),
+            6.0: Colors(236, 228, 236),
+            14.0: Colors(227, 220, 231),
+            24.0: Colors(217, 210, 228),
+            39.0: Colors(209, 201, 227),
+            56.0: Colors(202, 192, 228),
+            76.0: Colors(194, 183, 229),
+            99.0: Colors(187, 173, 231),
+            126.0: Colors(182, 164, 232),
+            156.0: Colors(177, 154, 231),
+            188.0: Colors(173, 145, 229),
+            224.0: Colors(169, 134, 225),
+            263.0: Colors(165, 126, 218),
+            305.0: Colors(162, 117, 211),
+            351.0: Colors(158, 109, 202),
+            399.0: Colors(153, 100, 191),
+            451.0: Colors(149, 93, 181),
+            505.0: Colors(144, 86, 171),
+            563.0: Colors(140, 79, 160),
+            624.0: Colors(134, 71, 148),
+            688.0: Colors(128, 65, 138),
+            755.0: Colors(123, 59, 127),
+            825.0: Colors(116, 53, 117),
+            899.0: Colors(109, 47, 105),
+            975.0: Colors(102, 42, 95),
+            1055.0: Colors(95, 37, 85),
+            1137.0: Colors(88, 32, 76),
+            1223.0: Colors(80, 26, 66),
+            1312.0: Colors(72, 21, 57),
+            1404.0: Colors(64, 15, 50),
+            1500.0: Colors(57, 8, 42),
+        }
+    )
+
+    # metadata
+    input_nbands: int = 2
+    output_nbands: int = 4
+    output_dtype: str = "uint8"
+
+    def __call__(self, img: ImageData) -> ImageData:
+
+        self.emissions = img.data[0]
+        self.intensity = img.data[1]
+        self.no_data = img.array.mask[0]
+
+        rgb = self.create_true_color_rgb()
+        alpha = self.create_true_color_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_true_color_rgb(self):
+        r, g, b = self._rgb_zeros_array()
+
+        for k, colors in self.conf_colors.items():
+            r[self.emissions >= k] = colors.red
+            g[self.emissions >= k] = colors.green
+            b[self.emissions >= k] = colors.blue
+
+        return np.stack([r, g, b], axis=0)
+
+    def create_true_color_alpha(self):
+        """Set the transparency (alpha) channel based on intensity input. The
+        intensity multiplier is used to control how isolated pixels 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.no_data, self.intensity, 0)
+        return np.minimum(255, alpha)
+
+    def _rgb_zeros_array(self):
+        r = np.zeros_like(self.emissions, dtype=np.uint8)
+        g = np.zeros_like(self.emissions, dtype=np.uint8)
+        b = np.zeros_like(self.emissions, dtype=np.uint8)
+
+        return r, g, b

app/routes/titiler/gfw_forest_carbon_gross_emissions.py

@@ -0,0 +1,75 @@
+import os
+from typing import Optional, Tuple
+
+from aenum import Enum, extend_enum
+from fastapi import APIRouter, Depends, Query, Response
+from titiler.core.resources.enums import ImageType
+from titiler.core.utils import render_image
+
+from ...crud.sync_db.tile_cache_assets import get_versions
+from ...models.enumerators.tile_caches import TileCacheType
+from ...models.enumerators.titiler import TreeCoverDensityThreshold
+from .. import raster_xyz
+from .algorithms.carbon_gross_emissions import CarbonGrossEmissions
+from .readers import AlertsReader
+
+DATA_LAKE_BUCKET = os.environ.get("DATA_LAKE_BUCKET")
+
+router = APIRouter()
+
+dataset = "gfw_forest_carbon_gross_emissions"
+
+
+class GfwForestCarbonGrossEmissions(str, Enum):
+    latest = "v20240402"
+
+
+_versions = get_versions(dataset, TileCacheType.cog)
+for _version in _versions:
+    extend_enum(GfwForestCarbonGrossEmissions, _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 global_forest_carbon_gross_emissions_raster_tile(
+    *,
+    version: GfwForestCarbonGrossEmissions,
+    xyz: Tuple[int, int, int] = Depends(raster_xyz),
+    scale: int = Query(
+        1, ge=1, lt=4, description="Tile size scale. 1=256x256, 2=512x512..."
+    ),
+    tree_cover_density_threshold: Optional[TreeCoverDensityThreshold] = Query(
+        TreeCoverDensityThreshold.tcd_30,
+        description="Show alerts in pixels with tree cover density (in percent) greater than or equal to this threshold. `umd_tree_cover_density_2000` is used for this masking.",
+    ),
+) -> Response:
+    """Forest Carbon Gross Emissions raster tiles."""
+
+    tile_x, tile_y, zoom = xyz
+    bands = [
+        f"emission_tcd_{tree_cover_density_threshold}",
+        f"intensity_tcd_{tree_cover_density_threshold}",
+    ]
+    folder: str = f"s3://{DATA_LAKE_BUCKET}/{dataset}/{version}/raster/epsg-4326/cog"
+    with AlertsReader(
+        input=folder, default_band=f"emission_tcd_{tree_cover_density_threshold}"
+    ) as reader:
+        # NOTE: the bands in the output `image_data` array will be in the order of
+        # the input `bands` list
+        image_data = reader.tile(
+            tile_x, tile_y, zoom, bands=bands, tilesize=scale * 256
+        )
+
+    processed_image = CarbonGrossEmissions()(image_data)
+
+    content, media_type = render_image(
+        processed_image,
+        output_format=ImageType("png"),
+        add_mask=False,
+    )
+
+    return Response(content, media_type=media_type)

app/routes/titiler/readers.py

@@ -9,6 +9,7 @@
 class AlertsReader(MultiBandReader):
 
     input: str = attr.ib()
+    default_band: str = attr.ib(default="default")
     tms: morecantile.TileMatrixSet = attr.ib(
         default=morecantile.tms.get("WebMercatorQuad")
     )
@@ -29,7 +30,7 @@ def _maxzoom(self):
 
     def __attrs_post_init__(self):
         """Get grid bounds."""
-        band_url: str = self._get_band_url("default")
+        band_url: str = self._get_band_url(self.default_band)
         with self.reader(band_url) as cog:
             self.bounds = cog.bounds
             self.crs = cog.crs