Generate forecast menu dynamically

arpav_ppcv/operations.py

@@ -893,3 +893,108 @@ def _get_spatial_buffer(
         point_geom_projected, distance=distance_meters
     )
     return transform(inverse_coordinate_transformer, buffer_geom_projected)
+
+
+def get_forecast_variable_parameters(
+    session: sqlmodel.Session,
+) -> dict[str, coverages.ForecastVariableMenuTree]:
+    result = {}
+    all_cov_confs = database.collect_all_coverage_configurations(session)
+    for cov_conf in all_cov_confs:
+        variable_pvs = [
+            pv
+            for pv in cov_conf.possible_values
+            if pv.configuration_parameter_value.configuration_parameter.name
+            == "climatological_variable"
+        ]
+        if (num_pvs := len(variable_pvs)) >= 1:
+            variable_pv = variable_pvs[0]
+            if num_pvs > 1:
+                logger.info(
+                    f"coverage configuration {cov_conf.name!r} defines "
+                    f"{len(variable_pvs)} `climatological_variable`s, keeping only the "
+                    f"first one..."
+                )
+        else:
+            logger.info(
+                f"coverage configuration {cov_conf.name!r} does not have a "
+                f"`climatological_variable`, skipping..."
+            )
+            continue
+
+        aggreg_period_pvs = [
+            pv
+            for pv in cov_conf.possible_values
+            if pv.configuration_parameter_value.configuration_parameter.name
+            == "aggregation_period"
+        ]
+        if (num_aggreg_periods := len(aggreg_period_pvs)) >= 1:
+            aggreg_period_pv = aggreg_period_pvs[0]
+            if num_aggreg_periods > 1:
+                logger.info(
+                    f"coverage configuration {cov_conf.name!r} defines "
+                    f"{len(variable_pvs)} `aggregation_period`s, keeping only the "
+                    f"first one..."
+                )
+        else:
+            logger.info(
+                f"coverage configuration {cov_conf.name!r} does not have a "
+                f"`aggregation_period`, skipping..."
+            )
+            continue
+
+        measure_pvs = [
+            pv
+            for pv in cov_conf.possible_values
+            if pv.configuration_parameter_value.configuration_parameter.name
+            == "measure"
+        ]
+        if (num_measures := len(measure_pvs)) >= 1:
+            measure_pv = measure_pvs[0]
+            if num_measures > 1:
+                logger.info(
+                    f"coverage configuration {cov_conf.name!r} defines "
+                    f"{len(variable_pvs)} `measures`s, keeping only the "
+                    f"first one..."
+                )
+        else:
+            logger.info(
+                f"coverage configuration {cov_conf.name!r} does not have a "
+                f"`measure`, skipping..."
+            )
+            continue
+
+        result_key = "-".join(
+            (
+                variable_pv.configuration_parameter_value.name,
+                aggreg_period_pv.configuration_parameter_value.name,
+                measure_pv.configuration_parameter_value.name,
+            )
+        )
+        aggregated_variable = result.setdefault(result_key, {})
+        aggregated_variable[
+            "climatological_variable"
+        ] = variable_pv.configuration_parameter_value
+        aggregated_variable[
+            "aggregation_period"
+        ] = aggreg_period_pv.configuration_parameter_value
+        aggregated_variable["measure"] = measure_pv.configuration_parameter_value
+        combinations = aggregated_variable.setdefault("combinations", {})
+        params_to_ignore = (
+            "climatological_variable",
+            "aggregation_period",
+            "measure",
+            "uncertainty_type",
+        )
+        for pv in cov_conf.possible_values:
+            param_name = pv.configuration_parameter_value.configuration_parameter.name
+            if param_name not in params_to_ignore:
+                param_entry = combinations.setdefault(param_name, {})
+                param_entry[
+                    "configuration_parameter"
+                ] = pv.configuration_parameter_value.configuration_parameter
+                values = param_entry.setdefault("values", [])
+                existing_value_names = [cpv.name for cpv in values]
+                if pv.configuration_parameter_value.name not in existing_value_names:
+                    values.append(pv.configuration_parameter_value)
+    return result

arpav_ppcv/schemas/coverages.py

@@ -7,6 +7,7 @@
     Optional,
     Final,
     TYPE_CHECKING,
+    TypedDict,
 )
 
 import pydantic
@@ -525,3 +526,15 @@ class CoverageInternal:
 
     def __hash__(self):
         return hash(self.identifier)
+
+
+class ForecastVariableMenuTreeCombination(TypedDict):
+    configuration_parameter: ConfigurationParameter
+    values: list[ConfigurationParameterValue]
+
+
+class ForecastVariableMenuTree(TypedDict):
+    climatological_variable: ConfigurationParameterValue
+    aggregation_period: ConfigurationParameterValue
+    measure: ConfigurationParameterValue
+    combinations: dict[str, ForecastVariableMenuTreeCombination]

arpav_ppcv/webapp/api_v2/routers/coverages.py

@@ -552,3 +552,24 @@ def get_time_series(
             raise HTTPException(status_code=400, detail="Invalid coverage_identifier")
     else:
         raise HTTPException(status_code=400, detail="Invalid coverage_identifier")
+
+
+@router.get(
+    "/variable-combinations",
+    response_model=coverage_schemas.ForecastVariableCombinationsList,
+)
+def get_variable_combinations(
+    db_session: Annotated[Session, Depends(dependencies.get_db_session)],
+):
+    variable_combinations = operations.get_forecast_variable_parameters(db_session)
+    var_combinations = []
+    for var_name, var_menu in variable_combinations.items():
+        var_combinations.append(
+            coverage_schemas.VariableCombinations.from_items(var_menu)
+        )
+    return coverage_schemas.ForecastVariableCombinationsList(
+        combinations=var_combinations,
+        translations=coverage_schemas.ForecastMenuTranslations.from_items(
+            list(variable_combinations.values())
+        ),
+    )

arpav_ppcv/webapp/api_v2/schemas/coverages.py

@@ -4,7 +4,11 @@
 import pydantic
 from fastapi import Request
 
-from ....config import ArpavPpcvSettings
+from ....config import (
+    ArpavPpcvSettings,
+    LOCALE_EN,
+    LOCALE_IT,
+)
 from ....schemas import coverages as app_models
 from .base import WebResourceList
 
@@ -247,3 +251,108 @@ class ConfigurationParameterList(WebResourceList):
     items: list[ConfigurationParameterReadListItem]
     list_item_type = ConfigurationParameterReadListItem
     path_operation_name = "list_configuration_parameters"
+
+
+class ConfigurationParameterMenuTranslation(pydantic.BaseModel):
+    name: dict[str, str]
+    description: dict[str, str]
+
+
+class ForecastMenuTranslations(pydantic.BaseModel):
+    variable: dict[str, ConfigurationParameterMenuTranslation]
+    aggregation_period: dict[str, ConfigurationParameterMenuTranslation]
+    measure: dict[str, ConfigurationParameterMenuTranslation]
+    other_parameters: dict[str, dict[str, ConfigurationParameterMenuTranslation]]
+
+    @classmethod
+    def from_items(
+        cls, variable_menu_trees: typing.Sequence[app_models.ForecastVariableMenuTree]
+    ):
+        result = {}
+        for variable_menu_tree in variable_menu_trees:
+            variable_cp = variable_menu_tree["climatological_variable"]
+            aggregation_period_cp = variable_menu_tree["aggregation_period"]
+            measure_cp = variable_menu_tree["measure"]
+            vars = result.setdefault("variable", {})
+            vars[variable_cp.name] = ConfigurationParameterMenuTranslation(
+                name={
+                    LOCALE_EN.language: variable_cp.display_name_english,
+                    LOCALE_IT.language: variable_cp.display_name_english,
+                },
+                description={
+                    LOCALE_EN.language: variable_cp.description_english,
+                    LOCALE_IT.language: variable_cp.description_italian,
+                },
+            )
+            aggreg_periods = result.setdefault("aggregation_period", {})
+            aggreg_periods[
+                aggregation_period_cp.name
+            ] = ConfigurationParameterMenuTranslation(
+                name={
+                    LOCALE_EN.language: aggregation_period_cp.display_name_english,
+                    LOCALE_IT.language: aggregation_period_cp.display_name_english,
+                },
+                description={
+                    LOCALE_EN.language: aggregation_period_cp.description_english,
+                    LOCALE_IT.language: aggregation_period_cp.description_italian,
+                },
+            )
+            measures = result.setdefault("measure", {})
+            measures[measure_cp.name] = ConfigurationParameterMenuTranslation(
+                name={
+                    LOCALE_EN.language: measure_cp.display_name_english,
+                    LOCALE_IT.language: measure_cp.display_name_english,
+                },
+                description={
+                    LOCALE_EN.language: measure_cp.description_english,
+                    LOCALE_IT.language: measure_cp.description_italian,
+                },
+            )
+            others = result.setdefault("other_parameters", {})
+            for combination_info in variable_menu_tree["combinations"].values():
+                cp = combination_info["configuration_parameter"]
+                param_ = others.setdefault(cp.name, {})
+                for cpv in cp.allowed_values:
+                    param_[cpv.name] = ConfigurationParameterMenuTranslation(
+                        name={
+                            LOCALE_EN.language: cpv.display_name_english,
+                            LOCALE_IT.language: cpv.display_name_english,
+                        },
+                        description={
+                            LOCALE_EN.language: cpv.description_english,
+                            LOCALE_IT.language: cpv.description_italian,
+                        },
+                    )
+        return cls(
+            variable=result["variable"],
+            aggregation_period=result["aggregation_period"],
+            measure=result["measure"],
+            other_parameters=result["other_parameters"],
+        )
+
+
+class VariableCombinations(pydantic.BaseModel):
+    variable: str
+    aggregation_period: str
+    measure: str
+    other_parameters: dict[str, list[str]]
+
+    @classmethod
+    def from_items(cls, menu_tree: app_models.ForecastVariableMenuTree):
+        combinations = {}
+        for param_name, param_combinations in menu_tree["combinations"].items():
+            combinations[param_name] = []
+            for valid_value in param_combinations["values"]:
+                combinations[param_name].append(valid_value.name)
+
+        return cls(
+            variable=menu_tree["climatological_variable"].name,
+            aggregation_period=menu_tree["aggregation_period"].name,
+            measure=menu_tree["measure"].name,
+            other_parameters=combinations,
+        )
+
+
+class ForecastVariableCombinationsList(pydantic.BaseModel):
+    combinations: list[VariableCombinations]
+    translations: ForecastMenuTranslations