Generate WMS legend values dynamically and expose API path for it

arpav_ppcv/operations.py

@@ -33,6 +33,7 @@
 from . import (
     config,
     database,
+    palette,
 )
 from .schemas import (
     base,
@@ -104,6 +105,58 @@ def get_climate_barometer_time_series(
     return result
 
 
+def apply_palette_to_coverage(
+    settings: config.ArpavPpcvSettings,
+    coverage: coverages.CoverageInternal,
+    temporal_instant: Optional[dt.datetime] = None,
+) -> list[tuple[float, str]]:
+    opendap_url = "/".join(
+        (
+            settings.thredds_server.base_url,
+            settings.thredds_server.opendap_service_url_fragment,
+            crawler.get_thredds_url_fragment(
+                coverage, settings.thredds_server.base_url
+            ),
+        )
+    )
+    if temporal_instant is not None:
+        ds = netCDF4.Dataset(opendap_url)
+        netcdf_variable_name = coverage.configuration.get_main_netcdf_variable_name(
+            coverage.identifier
+        )
+        time_var = ds["time"]
+        time_index = cftime.date2index(
+            temporal_instant, time_var, time_var.calendar, select="nearest"
+        )
+        found_instant = cftime.num2pydate(
+            time_index, units=time_var.units, calendar=time_var.calendar
+        )
+        logger.info(f"Found temporal instant {found_instant}")
+        data_max = np.nanmax(ds[netcdf_variable_name][time_index, :, :])
+        data_min = np.nanmin(ds[netcdf_variable_name][time_index, :, :])
+    else:
+        data_max = coverage.configuration.color_scale_max
+        data_min = coverage.configuration.color_scale_min
+    palette_colors = palette.parse_palette(
+        coverage.configuration.palette, settings.palettes_dir
+    )
+    applied_colors = []
+    if palette_colors is not None:
+        if abs(data_max - data_min) > 0.001:
+            applied_colors = palette.apply_palette(
+                palette_colors, data_min, data_max, num_stops=settings.palette_num_stops
+            )
+        else:
+            logger.warning(
+                f"Cannot calculate applied colors for coverage "
+                f"configuration {coverage.configuration.name!r} - check the "
+                f"colorscale min and max values"
+            )
+    else:
+        logger.warning(f"Unable to parse palette {coverage.configuration.palette!r}")
+    return applied_colors
+
+
 def _get_climate_barometer_data(
     settings: config.ArpavPpcvSettings,
     coverage: coverages.CoverageInternal,

arpav_ppcv/thredds/utils.py

@@ -58,6 +58,7 @@ def tweak_wms_get_map_request(
         num_color_bands = "2"
         query_params["NUMCOLORBANDS"] = num_color_bands
     else:
+        num_color_bands = "250"
         if "uncertainty_group" in layer_name:
             palette = ncwms_palette
         else:
@@ -75,4 +76,7 @@ def tweak_wms_get_map_request(
 
     query_params["styles"] = palette
     query_params["colorscalerange"] = color_scale_range
+    query_params["NUMCOLORBANDS"] = num_color_bands
+    query_params["ABOVEMAXCOLOR"] = "extend"
+    query_params["BELOWMINCOLOR"] = "extend"
     return query_params

arpav_ppcv/webapp/api_v2/routers/coverages.py

@@ -1,3 +1,4 @@
+import datetime as dt
 import logging
 import urllib.parse
 from operator import itemgetter
@@ -285,6 +286,34 @@ def get_coverage_identifier(
         raise HTTPException(400, detail=_INVALID_COVERAGE_IDENTIFIER_ERROR_DETAIL)
 
 
+@router.get(
+    "/wms-legend/{coverage_identifier}",
+    response_model=coverage_schemas.CoverageImageLegend,
+)
+def get_wms_legend(
+    request: Request,
+    db_session: Annotated[Session, Depends(dependencies.get_db_session)],
+    settings: Annotated[ArpavPpcvSettings, Depends(dependencies.get_settings)],
+    coverage_identifier: str,
+    datetime: Optional[dt.datetime] = None,
+):
+    """Get legend for WMS GetMap calls"""
+    if (coverage := db.get_coverage(db_session, coverage_identifier)) is not None:
+        applied_colors = operations.apply_palette_to_coverage(
+            settings, coverage, datetime
+        )
+        return coverage_schemas.CoverageImageLegend(
+            color_entries=[
+                coverage_schemas.ImageLegendColor(value=v, color=c)
+                for v, c in applied_colors
+            ]
+        )
+    else:
+        raise HTTPException(
+            status_code=400, detail=_INVALID_COVERAGE_IDENTIFIER_ERROR_DETAIL
+        )
+
+
 @router.get("/wms/{coverage_identifier}")
 async def wms_endpoint(
     request: Request,