Feat: Graph modal upgrade - flex-config integration

flexmeasures/data/models/charts/__init__.py

@@ -2,6 +2,7 @@
 
 from . import belief_charts
 from .defaults import apply_chart_defaults
+from .belief_charts import chart_for_flex_config_reference
 
 
 def chart_type_to_chart_specs(chart_type: str, **kwargs) -> dict:
@@ -19,6 +20,11 @@ def chart_type_to_chart_specs(chart_type: str, **kwargs) -> dict:
         for chart_type, chart_specs in getmembers(belief_charts)
         if isfunction(chart_specs) or isinstance(chart_specs, dict)
     }
+
+    belief_charts_mapping["chart_for_flex_config_reference"] = apply_chart_defaults(
+        chart_for_flex_config_reference
+    )
+
     # Create chart specs
     chart_specs_or_fnc = belief_charts_mapping[chart_type]
     if isfunction(chart_specs_or_fnc):

flexmeasures/data/models/charts/belief_charts.py

@@ -1258,3 +1258,99 @@ def chart_for_chargepoint_sessions(
     ]
     chart_specs["vconcat"].insert(0, cp_chart)
     return chart_specs
+
+
+def chart_for_flex_config_reference(
+    sensors_to_show: list["Sensor" | list["Sensor"] | dict[str, "Sensor"]],  # noqa F821
+    event_starts_after: datetime | None = None,
+    event_ends_before: datetime | None = None,
+    combine_legend: bool = True,
+    **override_chart_specs: dict,
+):
+    """
+    Generate chart specifications for comparing sensors against reference Flex Config values.
+
+    This function renders a vertically concatenated view of line charts for each provided sensor.
+    It features specific logic for temporary sensors (id < 0), treating them as configuration references:
+    instead of querying the database, it generates a constant horizontal line based on the
+    'graph_value' attribute found in the sensor's metadata.
+    """
+    all_sensors = flatten_unique(sensors_to_show)
+
+    charts = []
+
+    # Determine legend position based on config
+    # If combine_legend is True (default in asset view), we usually want a shared legend or
+    # to let the frontend defaults handle it. Here we force 'bottom' if combined, else 'right'.
+    legend_orient = "bottom" if combine_legend else "right"
+
+    for sensor in all_sensors:
+
+        # Define the chart
+        sensor_chart = {
+            "width": "container",
+            "height": 150,
+            "mark": {"type": "line", "point": True, "tooltip": True},
+            "encoding": {
+                "x": {
+                    "field": "event_start",
+                    "type": "temporal",
+                    "scale": {
+                        "domain": (
+                            [
+                                event_starts_after.timestamp() * 1000,
+                                event_ends_before.timestamp() * 1000,
+                            ]
+                            if event_starts_after and event_ends_before
+                            else None
+                        )
+                    },
+                    "title": None,
+                    "axis": {"labelOverlap": True},
+                },
+                "y": {
+                    "field": "event_value",
+                    "type": "quantitative",
+                    "title": f"{sensor.name} ({sensor.unit})",
+                },
+                "color": {
+                    "field": "sensor.name",
+                    "type": "nominal",
+                    "legend": {"title": "Sensor", "orient": legend_orient},
+                },
+            },
+            "transform": [{"filter": {"field": "sensor.id", "equal": sensor.id}}],
+        }
+
+        # SPECIAL HANDLING FOR TEMPORARY SENSORS
+        if sensor.id < 0:
+            custom_value = sensor.get_attribute("graph_value", 10)
+            start_ts = event_starts_after.timestamp() * 1000
+            end_ts = event_ends_before.timestamp() * 1000
+            manual_data = []
+            # loop and add 10 points between start and end
+            for i in range(11):
+                manual_data.append(
+                    {
+                        "event_start": start_ts + i * (end_ts - start_ts) / 10,
+                        "event_value": custom_value,
+                        "sensor": {"id": sensor.id, "name": sensor.name},
+                        "source": {"name": "Simulation"},
+                    }
+                )
+
+            sensor_chart["data"] = {"values": manual_data}
+            del sensor_chart["transform"]
+
+        charts.append(sensor_chart)
+
+    chart_specs = {
+        "description": "Custom Sensor View",
+        "vconcat": charts,
+        "resolve": {"scale": {"color": "shared", "x": "shared"}},
+    }
+
+    for k, v in override_chart_specs.items():
+        chart_specs[k] = v
+
+    return chart_specs

flexmeasures/data/models/generic_assets.py

@@ -2,6 +2,8 @@
 
 from datetime import datetime, timedelta
 from typing import Any
+import math
+import random
 import json
 
 from flask import current_app
@@ -275,19 +277,10 @@ def validate_sensors_to_show(
         """
         # If not set, use defaults (show first 2 sensors)
         if not self.sensors_to_show and suggest_default_sensors:
-            sensors_to_show = self.sensors[:2]
-            if (
-                len(sensors_to_show) == 2
-                and sensors_to_show[0].unit == sensors_to_show[1].unit
-            ):
-                # Sensors are shown together (e.g. they can share the same y-axis)
-                return [{"title": None, "sensors": sensors_to_show}]
-            # Otherwise, show separately
-            return [{"title": None, "sensors": [sensor]} for sensor in sensors_to_show]
+            return self._suggest_default_sensors_to_show()
 
         sensor_ids_to_show = self.sensors_to_show
         # Import the schema for validation
-        from flexmeasures.data.schemas.utils import extract_sensors_from_flex_config
         from flexmeasures.data.schemas.generic_assets import SensorsToShowSchema
 
         sensors_to_show_schema = SensorsToShowSchema()
@@ -299,51 +292,41 @@ def validate_sensors_to_show(
 
         sensor_id_allowlist = SensorsToShowSchema.flatten(standardized_sensors_to_show)
 
-        # Only allow showing sensors from assets owned by the user's organization,
-        # except in play mode, where any sensor may be shown
-        accounts = [self.owner] if self.owner is not None else None
-        if current_app.config.get("FLEXMEASURES_MODE") == "play":
-            from flexmeasures.data.models.user import Account
-
-            accounts = db.session.scalars(select(Account)).all()
-
-        from flexmeasures.data.services.sensors import get_sensors
-
-        accessible_sensor_map = {
-            sensor.id: sensor
-            for sensor in get_sensors(
-                account=accounts,
-                include_public_assets=True,
-                sensor_id_allowlist=sensor_id_allowlist,
-            )
-        }
+        accessible_sensor_map = self._get_accessible_sensor_map(sensor_id_allowlist)
 
         # Build list of sensor objects that are accessible
         sensors_to_show = []
         missed_sensor_ids = []
+        asset_refs: list[dict] = []
 
-        for entry in standardized_sensors_to_show:
+        from flexmeasures.data.schemas.utils import extract_sensors_from_flex_config
+        from flexmeasures.data.models.time_series import Sensor
 
+        for entry in standardized_sensors_to_show:
             title = entry.get("title")
-            sensors = []
+            sensors_for_entry: list[int] = []
             plots = entry.get("plots", [])
-            if len(plots) > 0:
-                for plot in plots:
-                    if "sensor" in plot:
-                        sensors.append(plot["sensor"])
-                    if "sensors" in plot:
-                        sensors.extend(plot["sensors"])
-                    if "asset" in plot:
-                        extracted_sensors = extract_sensors_from_flex_config(plot)
-                        sensors.extend(extracted_sensors)
+
+            for plot in plots:
+                self._process_plot_entry(
+                    plot,
+                    sensors_for_entry,
+                    asset_refs,
+                    extract_sensors_from_flex_config,
+                    Sensor,
+                )
 
             accessible_sensors = [
                 accessible_sensor_map.get(sid)
-                for sid in sensors
+                for sid in sensors_for_entry
                 if sid in accessible_sensor_map
             ]
-            inaccessible = [sid for sid in sensors if sid not in accessible_sensor_map]
+
+            inaccessible = [
+                sid for sid in sensors_for_entry if sid not in accessible_sensor_map
+            ]
             missed_sensor_ids.extend(inaccessible)
+
             if accessible_sensors:
                 sensors_to_show.append(
                     {"title": title, "plots": [{"sensors": accessible_sensors}]}
@@ -353,8 +336,84 @@ def validate_sensors_to_show(
             current_app.logger.warning(
                 f"Cannot include sensor(s) {missed_sensor_ids} in sensors_to_show on asset {self}, as it is not accessible to user {current_user}."
             )
+
+        if asset_refs:
+            self._add_temporary_asset_sensors(asset_refs, sensors_to_show, Sensor)
+
         return sensors_to_show
 
+    def _suggest_default_sensors_to_show(self):
+        """Helper to return default sensors if none are configured."""
+        sensors_to_show = self.sensors[:2]
+        if (
+            len(sensors_to_show) == 2
+            and sensors_to_show[0].unit == sensors_to_show[1].unit
+        ):
+            # Sensors are shown together (e.g. they can share the same y-axis)
+            return [{"title": None, "sensors": sensors_to_show}]
+        # Otherwise, show separately
+        return [{"title": None, "sensors": [sensor]} for sensor in sensors_to_show]
+
+    def _get_accessible_sensor_map(self, sensor_id_allowlist: list[int]) -> dict:
+        """Helper to fetch and map accessible sensors."""
+        from flexmeasures.data.services.sensors import get_sensors
+
+        # Only allow showing sensors from assets owned by the user's organization,
+        # except in play mode, where any sensor may be shown
+        accounts = [self.owner] if self.owner is not None else None
+        if current_app.config.get("FLEXMEASURES_MODE") == "play":
+            from flexmeasures.data.models.user import Account
+
+            accounts = db.session.scalars(select(Account)).all()
+
+        return {
+            sensor.id: sensor
+            for sensor in get_sensors(
+                account=accounts,
+                include_public_assets=True,
+                sensor_id_allowlist=sensor_id_allowlist,
+            )
+        }
+
+    def _process_plot_entry(
+        self, plot, sensors_list, asset_refs_list, extract_utils, SensorClass
+    ):
+        """Helper to extract sensors and asset refs from a single plot configuration."""
+        if "sensor" in plot:
+            sensors_list.append(plot["sensor"])
+        if "sensors" in plot:
+            sensors_list.extend(plot["sensors"])
+        if "asset" in plot:
+            extracted_sensors, refs = extract_utils(plot)
+            for sensor_id in extracted_sensors:
+                sensor = db.session.get(SensorClass, sensor_id)
+                flex_config_key = plot.get("flex-context") or plot.get("flex-model")
+
+                # temporarily update sensor name for display context
+                sensor_name = f"{sensor.name} ({flex_config_key} for ({sensor.generic_asset.name}))"
+                sensor.name = sensor_name
+
+            sensors_list.extend(extracted_sensors)
+            asset_refs_list.extend(refs)
+
+    def _add_temporary_asset_sensors(self, asset_refs, sensors_to_show, SensorClass):
+        """Helper to create temporary sensor objects for asset references."""
+        for ref in asset_refs:
+            parent_asset = db.session.get(GenericAsset, ref["id"])
+            sensor_name = f"Temporary Sensor ({ref['field']} for ({parent_asset.name}))"
+            # create temporary sensor with negative ID
+            temporary = SensorClass(
+                name=sensor_name,
+                unit=ref["unit"],
+                generic_asset=parent_asset,
+                attributes={"graph_value": ref["value"]},
+            )
+            # random negative number between -1 and -10000 to avoid conflicts with real sensor ids
+            temporary.id = -1 * math.ceil(random.random() * 10000)
+            sensors_to_show.append(
+                {"title": "Temporary Sensor", "plots": [{"sensor": temporary}]}
+            )
+
     @property
     def asset_type(self) -> GenericAssetType:
         """This property prepares for dropping the "generic" prefix later"""
@@ -1044,6 +1103,15 @@ def set_inflexible_sensors(self, inflexible_sensor_ids: list[int]) -> None:
             ).all()
         db.session.add(self)
 
+    def find_site_asset(self) -> GenericAsset | None:
+        """Find the site asset for this asset, if it exists.
+
+        The site asset is the highest ancestor asset without a parent.
+        """
+        if self.parent_asset is None:
+            return self
+        return self.parent_asset.find_site_asset()
+
 
 def create_generic_asset(generic_asset_type: str, **kwargs) -> GenericAsset:
     """Create a GenericAsset and assigns it an id.

flexmeasures/data/schemas/generic_assets.py

@@ -243,7 +243,7 @@ def flatten(cls, nested_list) -> list[int]:
                         if "sensor" in plot:
                             all_objects.append(plot["sensor"])
                         if "asset" in plot:
-                            sensors = extract_sensors_from_flex_config(plot)
+                            sensors, _ = extract_sensors_from_flex_config(plot)
                             all_objects.extend(sensors)
                 elif "sensors" in s:
                     all_objects.extend(s["sensors"])

flexmeasures/data/schemas/utils.py

@@ -4,7 +4,7 @@
 from flask.cli import with_appcontext as with_cli_appcontext
 from pint import DefinitionSyntaxError, DimensionalityError, UndefinedUnitError
 
-from flexmeasures.utils.unit_utils import to_preferred, ur
+from flexmeasures.utils.unit_utils import to_preferred, ur, extract_unit_from_string
 from flexmeasures.data.models.time_series import Sensor
 
 
@@ -86,19 +86,23 @@ def convert_to_quantity(value: str, to_unit: str) -> ur.Quantity:
         )
 
 
-def extract_sensors_from_flex_config(plot: dict) -> list[Sensor]:
+def extract_sensors_from_flex_config(plot: dict) -> tuple[list[Sensor], list[dict]]:
     """
     Extracts a consolidated list of sensors from an asset based on
     flex-context or flex-model definitions provided in a plot dictionary.
     """
     all_sensors = []
+    asset_refs = []
 
     from flexmeasures.data.schemas.generic_assets import (
         GenericAssetIdField,
     )  # Import here to avoid circular imports
 
     asset = GenericAssetIdField().deserialize(plot.get("asset"))
 
+    if asset is None:
+        raise FMValidationError("Asset not found for the provided plot configuration.")
+
     fields_to_check = {
         "flex-context": asset.flex_context,
         "flex-model": asset.flex_model,
@@ -115,5 +119,22 @@ def extract_sensors_from_flex_config(plot: dict) -> list[Sensor]:
                 sensor = field_value.get("sensor")
                 if sensor:
                     all_sensors.append(sensor)
-
-    return all_sensors
+            elif isinstance(field_value, str):
+                unit = None
+                # extract unit from the string value and add a dummy sensor with that unit
+                value, unit = extract_unit_from_string(field_value)
+                if unit is not None:
+                    asset_refs.append(
+                        {
+                            "id": asset.id,
+                            "field": field_key,
+                            "value": value,
+                            "unit": unit,
+                        }
+                    )
+                else:
+                    raise FMValidationError(
+                        f"Value '{field_value}' for field '{field_key}' in '{plot_key}' is not a valid quantity string."
+                    )
+
+    return all_sensors, asset_refs