Reverted previous implementation of dynamic WMS legend generation

This is not needed, as the underlying THREDDS WMS server does not perform dynamic contrast enhancement

arpav_ppcv/operations.py

@@ -33,7 +33,6 @@
 from . import (
     config,
     database,
-    palette,
 )
 from .schemas import (
     base,
@@ -105,58 +104,6 @@ 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/webapp/api_v2/routers/coverages.py

@@ -1,4 +1,3 @@
-import datetime as dt
 import logging
 import urllib.parse
 from operator import itemgetter
@@ -30,6 +29,7 @@
     datadownloads,
     exceptions,
     operations,
+    palette,
 )
 from ....config import ArpavPpcvSettings
 from ....thredds import (
@@ -177,6 +177,7 @@ def list_coverage_configurations(
 )
 def get_coverage_configuration(
     request: Request,
+    settings: Annotated[ArpavPpcvSettings, Depends(dependencies.get_settings)],
     db_session: Annotated[Session, Depends(dependencies.get_db_session)],
     coverage_configuration_id: pydantic.UUID4,
 ):
@@ -186,11 +187,33 @@ def get_coverage_configuration(
     allowed_coverage_identifiers = db.generate_coverage_identifiers(
         coverage_configuration=db_coverage_configuration
     )
+    palette_colors = palette.parse_palette(
+        db_coverage_configuration.palette, settings.palettes_dir
+    )
+    applied_colors = []
+    if palette_colors is not None:
+        minimum = db_coverage_configuration.color_scale_min
+        maximum = db_coverage_configuration.color_scale_max
+        if abs(maximum - minimum) > 0.001:
+            applied_colors = palette.apply_palette(
+                palette_colors, minimum, maximum, num_stops=settings.palette_num_stops
+            )
+        else:
+            logger.warning(
+                f"Cannot calculate applied colors for coverage "
+                f"configuration {db_coverage_configuration.name!r} - check the "
+                f"colorscale min and max values"
+            )
+    else:
+        logger.warning(f"Unable to parse palette {db_coverage_configuration.palette!r}")
     return coverage_schemas.CoverageConfigurationReadDetail.from_db_instance(
-        db_coverage_configuration, allowed_coverage_identifiers, request
+        db_coverage_configuration, allowed_coverage_identifiers, applied_colors, request
     )
 
 
+# PossibleValue: pydantic.StringConstraints(pattern="^[\w-_]+:[\w-_]+$")
+
+
 @router.get(
     "/coverage-identifiers",
     response_model=coverage_schemas.CoverageIdentifierList,
@@ -262,34 +285,6 @@ 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,

arpav_ppcv/webapp/api_v2/schemas/coverages.py

@@ -135,13 +135,15 @@ class CoverageConfigurationReadDetail(CoverageConfigurationReadListItem):
     allowed_coverage_identifiers: list[str]
     description_english: str | None
     description_italian: str | None
+    legend: CoverageImageLegend
     data_precision: int
 
     @classmethod
     def from_db_instance(
         cls,
         instance: app_models.CoverageConfiguration,
         allowed_coverage_identifiers: list[str],
+        legend_colors: list[tuple[float, str]],
         request: Request,
     ) -> "CoverageConfigurationReadDetail":
         url = request.url_for(
@@ -166,6 +168,11 @@ def from_db_instance(
                 for pv in instance.possible_values
             ],
             allowed_coverage_identifiers=allowed_coverage_identifiers,
+            legend=CoverageImageLegend(
+                color_entries=[
+                    ImageLegendColor(value=v, color=c) for v, c in legend_colors
+                ]
+            ),
         )