Key results for detail panel

.jshintrc

@@ -36,7 +36,8 @@
     "subscribeToEvents": false,
     "createChart": false,
     "clearChart": false,
-    "hidePotentialLayers": false
+    "hidePotentialLayers": false,
+    "URLSearchParams": false
   },
   "strict": "implied"
 }

Makefile

@@ -89,4 +89,8 @@ update_vendor_assets:
 	cp node_modules/shepherd.js/dist/js/shepherd.* digiplan/static/vendors/shepherd/
 	cp node_modules/shepherd.js/dist/css/shepherd.css digiplan/static/vendors/shepherd/
 
+	# HTMX https://htmx.org/
+	rm -r digiplan/static/vendors/htmx/js/*
+	cp node_modules/htmx.org/dist/htmx.min.js digiplan/static/vendors/htmx/js/
+
 	# Done

config/settings/base.py

@@ -90,6 +90,7 @@
 THIRD_PARTY_APPS = [
     "rest_framework",
     "django_distill",
+    "template_partials",
 ]
 
 LOCAL_APPS = ["digiplan.map.apps.MapConfig", "django_oemof", "django_mapengine"]

digiplan/map/datapackage.py

@@ -1,17 +1,36 @@
 """Read functionality for digipipe datapackage."""
 import csv
 import json
-from collections import defaultdict
+from collections import defaultdict, namedtuple
 from pathlib import Path
-from typing import Optional
+from typing import Optional, Union
 
 import pandas as pd
+from cache_memoize import cache_memoize
 from django.conf import settings
 from django_oemof.settings import OEMOF_DIR
 
-from config.settings.base import DATA_DIR
+from config.settings.base import DIGIPIPE_DIR
 from digiplan.map import config, models
 
+Source = namedtuple("Source", ["csv_file", "column"])
+
+
+def get_data_from_sources(sources: Union[Source, list[Source]]) -> pd.DataFrame:
+    """Extract data from single or multiple sources and merge into dataframe."""
+    source_files = defaultdict(list)
+    if isinstance(sources, Source):
+        source_files[sources.csv_file].append(sources.column)
+    else:
+        for source in sources:
+            source_files[source.csv_file].append(source.column)
+
+    dfs = []
+    for source_file, columns in source_files.items():
+        source_path = Path(DIGIPIPE_DIR, "scalars", source_file)
+        dfs.append(pd.read_csv(source_path, usecols=columns))
+    return pd.concat(dfs)
+
 
 def get_employment() -> pd.DataFrame:
     """Return employment data."""
@@ -144,42 +163,17 @@ def get_thermal_efficiency(component: str) -> float:
     return pd.read_csv(sequence_filename, sep=";").iloc[:, 1]
 
 
-def get_potential_values(*, per_municipality: bool = False) -> dict:
+@cache_memoize(timeout=None)
+def get_potential_values() -> dict:
     """
     Calculate max_values for sliders.
 
-    Parameters
-    ----------
-    per_municipality: bool
-        If set to True, potentials are not aggregated, but given per municipality
-
     Returns
     -------
     dict
         dictionary with each slider / switch and respective max_value
     """
-    scalars = {
-        "wind": "potentialarea_wind_area_stats_muns.csv",
-        "pv_ground": "potentialarea_pv_ground_area_stats_muns.csv",
-        "pv_roof": "potentialarea_pv_roof_area_stats_muns.csv",
-    }
-
-    areas = {
-        "wind": {
-            "wind_2018": "stp_2018_eg",
-            "wind_2024": "stp_2024_vr",
-            "wind_2027": area
-            if (area := models.Municipality.objects.all().values("area").aggregate(models.Sum("area"))["area__sum"])
-            else 0,  # to prevent None if regions are empty
-        },
-        "pv_ground": {
-            "pv_soil_quality_low": "soil_quality_low_region",
-            "pv_soil_quality_medium": "soil_quality_medium_region",
-            "pv_permanent_crops": "permanent_crops_region",
-        },
-        "pv_roof": {"pv_roof": "installable_power"},
-    }
-
+    areas = get_potential_areas()
     pv_density = {
         "pv_soil_quality_low": "pv_ground",
         "pv_soil_quality_medium": "pv_ground_vertical_bifacial",
@@ -189,25 +183,64 @@ def get_potential_values(*, per_municipality: bool = False) -> dict:
     power_density = json.load(Path.open(Path(settings.DIGIPIPE_DIR, "scalars/technology_data.json")))["power_density"]
 
     potentials = {}
-    for profile in areas:
-        path = Path(DATA_DIR, "digipipe/scalars", scalars[profile])
-        reader = pd.read_csv(path)
-        for key, value in areas[profile].items():
-            if key == "wind_2027":
-                # Value is already calculated from region area (see above)
-                potentials[key] = value
-            else:
-                if per_municipality:  # noqa: PLR5501
-                    potentials[key] = reader[value]
-                else:
-                    potentials[key] = reader[value].sum()
-            if profile == "wind":
-                potentials[key] = potentials[key] * power_density["wind"]
-            if profile == "pv_ground":
-                potentials[key] = potentials[key] * power_density[pv_density[key]]
+    for key in areas:
+        if key.startswith("wind"):
+            potentials[key] = areas[key] * power_density["wind"]
+        if key == "pv_roof":
+            potentials[key] = areas[key]
+            continue
+        if key.startswith("pv"):
+            potentials[key] = areas[key] * power_density[pv_density[key]]
     return potentials
 
 
+@cache_memoize(timeout=None)
+def get_potential_areas(technology: Optional[str] = None) -> dict:
+    """
+    Return potential areas.
+
+    Parameters
+    ----------
+    technology: str
+        If given, potential area only for this technology is returned
+
+    Returns
+    -------
+    dict
+        Potential areas of all technologies or specified one
+    """
+    sources = {
+        "wind_2018": Source("potentialarea_wind_area_stats_muns.csv", "stp_2018_eg"),
+        "wind_2024": Source("potentialarea_wind_area_stats_muns.csv", "stp_2024_vr"),
+        "pv_soil_quality_low": Source("potentialarea_pv_ground_area_stats_muns.csv", "soil_quality_low_region"),
+        "pv_soil_quality_medium": Source("potentialarea_pv_ground_area_stats_muns.csv", "soil_quality_medium_region"),
+        "pv_permanent_crops": Source("potentialarea_pv_ground_area_stats_muns.csv", "permanent_crops_region"),
+        "pv_roof": Source("potentialarea_pv_roof_area_stats_muns.csv", "installable_power"),
+    }
+
+    # Add wind for 2027 directly from model data, as it is not included in datapackage
+    areas = {
+        "wind_2027": area
+        if (area := models.Municipality.objects.all().values("area").aggregate(models.Sum("area"))["area__sum"])
+        else 0,
+    }
+    if technology is not None:
+        if technology == "wind_2027":
+            return areas["wind_2027"]
+        sources = {technology: sources[technology]}
+
+    data = get_data_from_sources(sources.values())
+    data = data.sum()
+    for index in data.index:
+        # Add extracted data to areas and map source columns to keys accordingly
+        areas[next(key for key, source in sources.items() if source.column == index)] = data[index]
+
+    if technology is not None:
+        return areas[technology]
+    return areas
+
+
+@cache_memoize(timeout=None)
 def get_full_load_hours(year: int) -> pd.Series:
     """Return full load hours for given year."""
     full_load_hours = pd.Series(
@@ -217,6 +250,7 @@ def get_full_load_hours(year: int) -> pd.Series:
     return full_load_hours
 
 
+@cache_memoize(timeout=None)
 def get_capacities_from_datapackage() -> pd.DataFrame:
     """Return renewable capacities for given year from datapackage."""
     capacities = pd.concat(
@@ -256,6 +290,7 @@ def get_capacities_from_sliders(year: int) -> pd.Series:
     return slider_settings
 
 
+@cache_memoize(timeout=None)
 def get_power_density(technology: Optional[str] = None) -> dict:
     """Return power density for technology."""
     if technology:

digiplan/map/forms.py

@@ -1,20 +1,28 @@
-from itertools import count  # noqa: D100
+"""Module containing django forms."""
+from __future__ import annotations
+
+from itertools import count
+from typing import TYPE_CHECKING
 
 from django.forms import BooleanField, Form, IntegerField, TextInput, renderers
+from django.shortcuts import reverse
 from django.utils.safestring import mark_safe
-from django_mapengine import legend
 
-from . import charts
+from . import charts, menu
 from .widgets import SwitchWidget
 
+if TYPE_CHECKING:
+    from django_mapengine import legend
+
 
 class TemplateForm(Form):  # noqa: D101
     template_name = None
+    extra_content = {}
 
     def __str__(self) -> str:  # noqa: D105
         if self.template_name:
             renderer = renderers.get_default_renderer()
-            return mark_safe(renderer.render(self.template_name, {"form": self}))  # noqa: S308
+            return mark_safe(renderer.render(self.template_name, {"form": self, **self.extra_content}))  # noqa: S308
         return super().__str__()
 
 
@@ -54,33 +62,36 @@ def __init__(self, parameters, additional_parameters=None, **kwargs) -> None:  #
         super().__init__(**kwargs)
         self.fields = {item["name"]: item["field"] for item in self.generate_fields(parameters, additional_parameters)}
 
-    @staticmethod
-    def generate_fields(parameters, additional_parameters=None):  # noqa: ANN001, ANN205, D102
+    def get_field_attrs(self, name: str, parameters: dict) -> dict:  # noqa: ARG002
+        """Set up field attributes from parameters."""
+        attrs = {
+            "class": parameters["class"],
+            "data-min": parameters["min"],
+            "data-max": parameters["max"],
+            "data-from": parameters["start"],
+            "data-grid": "true" if "grid" in parameters and parameters["grid"] else "false",
+            "data-has-sidepanel": "true" if "sidepanel" in parameters else "false",
+            "data-color": parameters["color"] if "color" in parameters else "",
+        }
+        if "to" in parameters:
+            attrs["data-to"] = parameters["to"]
+        if "step" in parameters:
+            attrs["data-step"] = parameters["step"]
+        if "from-min" in parameters:
+            attrs["data-from-min"] = parameters["from-min"]
+        if "from-max" in parameters:
+            attrs["data-from-max"] = parameters["from-max"]
+        return attrs
+
+    def generate_fields(self, parameters: dict, additional_parameters: dict | None = None) -> dict:
+        """Create fields from config parameters."""
         if additional_parameters is not None:
             charts.merge_dicts(parameters, additional_parameters)
         for name, item in parameters.items():
             if item["type"] == "slider":
-                attrs = {
-                    "class": item["class"],
-                    "data-min": item["min"],
-                    "data-max": item["max"],
-                    "data-from": item["start"],
-                    "data-grid": "true" if "grid" in item and item["grid"] else "false",
-                    "data-has-sidepanel": "true" if "sidepanel" in item else "false",
-                    "data-color": item["color"] if "color" in item else "",
-                }
-                if "to" in item:
-                    attrs["data-to"] = item["to"]
-                if "step" in item:
-                    attrs["data-step"] = item["step"]
-                if "from-min" in item:
-                    attrs["data-from-min"] = item["from-min"]
-                if "from-max" in item:
-                    attrs["data-from-max"] = item["from-max"]
-
                 field = IntegerField(
                     label=item["label"],
-                    widget=TextInput(attrs=attrs),
+                    widget=TextInput(attrs=self.get_field_attrs(name, item)),
                     help_text=item["tooltip"],
                     required=item.get("required", True),
                 )
@@ -103,6 +114,32 @@ def generate_fields(parameters, additional_parameters=None):  # noqa: ANN001, AN
 class EnergyPanelForm(PanelForm):  # noqa: D101
     template_name = "forms/panel_energy.html"
 
+    def __init__(self, parameters, additional_parameters=None, **kwargs) -> None:  # noqa: ANN001
+        """Overwrite init function to add initial key results for detail panels."""
+        super().__init__(parameters, additional_parameters, **kwargs)
+        for technology in ("wind_2018", "wind_2024", "wind_2027", "pv_ground", "pv_roof"):
+            # get initial slider values for wind and pv:
+            key_results = menu.detail_key_results(
+                technology,
+                id_s_w_6=parameters["s_w_6"]["start"],
+                id_s_w_7=parameters["s_w_7"]["start"],
+                id_s_pv_ff_3=parameters["s_pv_ff_3"]["start"],
+                id_s_pv_ff_4=parameters["s_pv_ff_4"]["start"],
+                id_s_pv_ff_5=parameters["s_pv_ff_5"]["start"],
+                id_s_pv_d_3=parameters["s_pv_d_3"]["start"],
+            )
+            for key, value in key_results.items():
+                self.extra_content[f"{technology}_key_result_{key}"] = value
+
+    def get_field_attrs(self, name: str, parameters: dict) -> dict:
+        """Add HTMX attributes to wind and pv detail sliders."""
+        detail_slider_targets = {"s_w_6": "wind_key_results_2024", "s_w_7": "wind_key_results_2027"}
+        attrs = super().get_field_attrs(name, parameters)
+        if name in detail_slider_targets:
+            attrs["hx-get"] = reverse("map:detail_key_results")
+            attrs["hx-target"] = detail_slider_targets[name]
+        return attrs
+
 
 class HeatPanelForm(PanelForm):  # noqa: D101
     template_name = "forms/panel_heat.html"

digiplan/map/menu.py

@@ -0,0 +1,49 @@
+"""Add calculations for menu items."""
+
+from . import config, datapackage
+
+
+def detail_key_results(technology: str, **kwargs: dict) -> dict:
+    """Calculate detail key results for given technology."""
+    areas = datapackage.get_potential_areas()
+    potential_capacities = datapackage.get_potential_values()  # in MW
+    full_load_hours = datapackage.get_full_load_hours(2045)
+    nominal_power_per_unit = config.TECHNOLOGY_DATA["nominal_power_per_unit"]["wind"]
+
+    if technology.startswith("wind"):
+        percentage = 1
+        if technology == "wind_2024":
+            percentage = float(kwargs["id_s_w_6"]) / 100
+        if technology == "wind_2027":
+            percentage = float(kwargs["id_s_w_7"]) / 100
+        return {
+            "area": areas[technology] / 100 * percentage,
+            "turbines": potential_capacities[technology] / nominal_power_per_unit * percentage,
+            "energy": potential_capacities[technology] * full_load_hours["wind"] * percentage * 1e-6,
+        }
+    if technology == "pv_ground":
+        percentages = {
+            "pv_soil_quality_low": int(kwargs["id_s_pv_ff_3"]) / 100,
+            "pv_soil_quality_medium": int(kwargs["id_s_pv_ff_4"]) / 100,
+            "pv_permanent_crops": int(kwargs["id_s_pv_ff_5"]) / 100,
+        }
+        flh_mapping = {
+            "pv_soil_quality_low": "pv_ground",
+            "pv_soil_quality_medium": "pv_ground_elevated",
+            "pv_permanent_crops": "pv_ground_vertical_bifacial",
+        }
+        return {
+            "area": sum(areas[pv_type] * percentages[pv_type] for pv_type in percentages),
+            "energy": sum(
+                potential_capacities[pv_type] * full_load_hours[flh_mapping[pv_type]] * percentages[pv_type]
+                for pv_type in percentages
+            )
+            * 1e-6,
+        }
+    if technology == "pv_roof":
+        percentage = int(kwargs["id_s_pv_d_3"]) / 100
+        return {
+            "area": areas[technology] * percentage,
+            "energy": potential_capacities[technology] * full_load_hours[technology] * percentage * 1e-6,
+        }
+    raise KeyError(f"Unknown technology '{technology}'.")

digiplan/map/urls.py

@@ -12,4 +12,5 @@
     path("choropleth/<str:lookup>/<str:layer_id>", views.get_choropleth, name="choropleth"),
     path("popup/<str:lookup>/<int:region>", views.get_popup, name="popup"),
     path("charts", views.get_charts, name="charts"),
+    path("detail_key_results", views.DetailKeyResultsView.as_view(), name="detail_key_results"),
 ]