feat: Adapting flex-config to SensorsToShowSchema

documentation/changelog.rst

@@ -20,12 +20,14 @@ New features
 * Give ability to edit sensor timezone from the UI [see `PR #1900 <https://www.github.com/FlexMeasures/flexmeasures/pull/1900>`_]
 * Support creating schedules with only information known prior to some time, now also via the CLI (the API already supported it) [see `PR #1871 <https://www.github.com/FlexMeasures/flexmeasures/pull/1871>`_].
 * Add ``fields`` param to the asset-listing endpoints, to save bandwidth in response data [see `PR #1884 <https://www.github.com/FlexMeasures/flexmeasures/pull/1884>`_]
+* Support for flex-config in the ``SensorsToShowSchema`` [see `PR #1904 <https://www.github.com/FlexMeasures/flexmeasures/pull/1904>`_]
+
 
   .. note:: For backwards-compatibility, the new ``fields`` parameter will only be fully active, i.e. also returning less fields per default, in v0.32. Set ``FLEXMEASURES_API_SUNSET_ACTIVE=True`` to test the full effect now.
 
 * Allow testing out the scheduling CLI without saving anything, using ``flexmeasures add schedule --dry-run`` [see `PR #1892 <https://www.github.com/FlexMeasures/flexmeasures/pull/1892>`_]
 * Allow unsupported ``flex-context`` or ``flex-model`` fields to be shown in the UI editors (they will be un-editable) [see `PR #1915 <https://www.github.com/FlexMeasures/flexmeasures/pull/1915>`_]
-* Add back save buttons to both ``flex-context`` and ``flex-model`` UI editors [see `PR #1916 <https://www.github.com/FlexMeasures/flexmeasures/pull/1916>`_]
+* Add back save buttons to both ``flex-context`` and ``flex-model`` UI editors [see `PR #1880 <https://www.github.com/FlexMeasures/flexmeasures/pull/1880>`_]
 
 Infrastructure / Support
 ----------------------

flexmeasures/data/models/charts/belief_charts.py

@@ -514,7 +514,14 @@ def chart_for_multiple_sensors(
         title = entry.get("title")
         if title == "Charge Point sessions":
             continue
-        sensors = entry.get("sensors")
+        plots = entry.get("plots", [])
+        sensors = []
+        for plot in plots:
+            if "sensors" in plot:
+                sensors.extend(plot.get("sensors"))
+            elif "sensor" in plot:
+                sensors.extend([plot.get("sensor")])
+
         # List the sensors that go into one row
         row_sensors: list["Sensor"] = sensors  # noqa F821
 

flexmeasures/data/models/generic_assets.py

@@ -287,6 +287,7 @@ def validate_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()
@@ -324,7 +325,17 @@ def validate_sensors_to_show(
         for entry in standardized_sensors_to_show:
 
             title = entry.get("title")
-            sensors = entry.get("sensors")
+            sensors = []
+            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)
 
             accessible_sensors = [
                 accessible_sensor_map.get(sid)
@@ -334,7 +345,9 @@ def validate_sensors_to_show(
             inaccessible = [sid for sid in sensors if sid not in accessible_sensor_map]
             missed_sensor_ids.extend(inaccessible)
             if accessible_sensors:
-                sensors_to_show.append({"title": title, "sensors": accessible_sensors})
+                sensors_to_show.append(
+                    {"title": title, "plots": [{"sensors": accessible_sensors}]}
+                )
 
         if missed_sensor_ids:
             current_app.logger.warning(

flexmeasures/data/schemas/generic_assets.py

@@ -19,6 +19,7 @@
 from flexmeasures.data.schemas.utils import (
     FMValidationError,
     MarshmallowClickMixin,
+    extract_sensors_from_flex_config,
 )
 from flexmeasures.auth.policy import user_has_admin_access
 from flexmeasures.cli import is_running as running_as_cli
@@ -80,45 +81,141 @@ def deserialize(self, value, **kwargs) -> list:
 
     def _standardize_item(self, item) -> dict:
         """
-        Standardize different input formats to a consistent dictionary format.
+        Normalize various input formats (int, list, or dict) into a standard plot dictionary.
         """
         if isinstance(item, int):
-            return {"title": None, "sensors": [item]}
+            return {"title": None, "plots": [{"sensor": item}]}
         elif isinstance(item, list):
             if not all(isinstance(sensor_id, int) for sensor_id in item):
                 raise ValidationError(
                     "All elements in a list within 'sensors_to_show' must be integers."
                 )
-            return {"title": None, "sensors": item}
+            return {"title": None, "plots": [{"sensors": item}]}
         elif isinstance(item, dict):
-            if "title" not in item:
-                raise ValidationError("Dictionary must contain a 'title' key.")
-            else:
-                title = item["title"]
-                if not isinstance(title, str) and title is not None:
-                    raise ValidationError("'title' value must be a string.")
-
-            if "sensor" in item:
-                sensor = item["sensor"]
-                if not isinstance(sensor, int):
-                    raise ValidationError("'sensor' value must be an integer.")
-                return {"title": title, "sensors": [sensor]}
-            elif "sensors" in item:
-                sensors = item["sensors"]
-                if not isinstance(sensors, list) or not all(
-                    isinstance(sensor_id, int) for sensor_id in sensors
-                ):
-                    raise ValidationError("'sensors' value must be a list of integers.")
-                return {"title": title, "sensors": sensors}
-            else:
-                raise ValidationError(
-                    "Dictionary must contain either 'sensor' or 'sensors' key."
-                )
+            return self._standardize_dict_item(item)
         else:
             raise ValidationError(
                 "Invalid item type in 'sensors_to_show'. Expected int, list, or dict."
             )
 
+    def _standardize_dict_item(self, item: dict) -> dict:
+        """
+        Transform a dictionary-based sensor configuration into a standardized 'plots' structure.
+        Ensures 'title' is a string and processes 'sensor', 'sensors', or direct 'plots' keys.
+        """
+        title = None
+
+        if "title" in item:
+            title = item["title"]
+            if not isinstance(title, str) and title is not None:
+                title = None
+
+        if "sensor" in item:
+            sensor = item["sensor"]
+            if not isinstance(sensor, int):
+                raise ValidationError("'sensor' value must be an integer.")
+            return {"title": title, "plots": [{"sensor": sensor}]}
+        elif "sensors" in item:
+            sensors = item["sensors"]
+            if not isinstance(sensors, list) or not all(
+                isinstance(sensor_id, int) for sensor_id in sensors
+            ):
+                raise ValidationError("'sensors' value must be a list of integers.")
+            return {"title": title, "plots": [{"sensors": sensors}]}
+        elif "plots" in item:
+            plots = item["plots"]
+            if not isinstance(plots, list):
+                raise ValidationError("'plots' must be a list or dictionary.")
+            for plot in plots:
+                self._validate_single_plot(plot)
+
+            return {"title": title, "plots": plots}
+        else:
+            raise ValidationError(
+                "Dictionary must contain either 'sensor' or 'sensors' key."
+            )
+
+    def _validate_single_plot(self, plot):
+        """
+        Perform structural validation on an individual plot dictionary.
+        Requires at least one of: 'sensor', 'sensors', or 'asset'.
+        """
+        if not isinstance(plot, dict):
+            raise ValidationError("Each plot in 'plots' must be a dictionary.")
+
+        if "sensor" not in plot and "sensors" not in plot and "asset" not in plot:
+            raise ValidationError(
+                "Each plot must contain either 'sensor', 'sensors' or an 'asset' key."
+            )
+
+        if "asset" in plot:
+            self._validate_asset_in_plot(plot)
+        if "sensor" in plot:
+            sensor = plot["sensor"]
+            if not isinstance(sensor, int):
+                raise ValidationError("'sensor' value must be an integer.")
+        if "sensors" in plot:
+            sensors = plot["sensors"]
+            if not isinstance(sensors, list) or not all(
+                isinstance(sensor_id, int) for sensor_id in sensors
+            ):
+                raise ValidationError("'sensors' value must be a list of integers.")
+
+    def _validate_asset_in_plot(self, plot):
+        """
+        Validate plots that reference a GenericAsset.
+        Ensures flex-config schemas are respected when an asset is provided.
+        """
+        from flexmeasures.data.schemas.scheduling import (
+            DBFlexContextSchema,
+        )
+        from flexmeasures.data.schemas.scheduling.storage import (
+            DBStorageFlexModelSchema,
+        )
+
+        if "flex-context" not in plot and "flex-model" not in plot:
+            raise ValidationError(
+                "When 'asset' is provided in a plot, 'flex-context' or 'flex-model' must also be provided."
+            )
+
+        self._validate_flex_config_field_is_valid_choice(
+            plot, "flex-context", DBFlexContextSchema.mapped_schema_keys.values()
+        )
+        self._validate_flex_config_field_is_valid_choice(
+            plot, "flex-model", DBStorageFlexModelSchema().mapped_schema_keys.values()
+        )
+
+    def _validate_flex_config_field_is_valid_choice(
+        self, plot_config, field_name, valid_collection
+    ):
+        """
+        Verify that the chosen flex-config field exists on the specific asset and matches
+        allowed schema keys.
+        """
+        if field_name in plot_config:
+            value = plot_config[field_name]
+            asset_id = plot_config.get("asset")
+            asset = GenericAssetIdField().deserialize(asset_id)
+
+            if asset is None:
+                raise ValidationError(f"Asset with ID {asset_id} does not exist.")
+
+            if value and not isinstance(value, str):
+                raise ValidationError(f"The value for '{field_name}' must be a string.")
+
+            if value not in valid_collection:
+                raise ValidationError(f"'{field_name}' value '{value}' is not valid.")
+
+            attr_to_check = (
+                "flex_model" if field_name == "flex-model" else "flex_context"
+            )
+            asset_flex_config = getattr(asset, attr_to_check, {})
+
+            if value not in asset_flex_config:
+                raise ValidationError(
+                    f"The asset with ID '{asset_id}' does not have a '{value}' set in its '{attr_to_check}'."
+                )
+
     @classmethod
     def flatten(cls, nested_list) -> list[int]:
         """
@@ -127,7 +224,7 @@ def flatten(cls, nested_list) -> list[int]:
         This method processes the following formats, for each of the entries of the nested list:
         - A list of sensor IDs: `[1, 2, 3]`
         - A list of dictionaries where each dictionary contains a `sensors` list or a `sensor` key:
-        `[{"title": "Temperature", "sensors": [1, 2]}, {"title": "Pressure", "sensor": 3}]`
+        `[{"title": "Temperature", "sensors": [1, 2]}, {"title": "Pressure", "sensor": 3},  {"title": "Pressure", "plots": [{"sensor": 4}, {"sensors": [5,6]}]}]`
         - Mixed formats: `[{"title": "Temperature", "sensors": [1, 2]}, {"title": "Pressure", "sensor": 3}, 4, 5, 1]`
 
         It extracts all sensor IDs, removes duplicates, and returns a flattened list of unique sensor IDs.
@@ -138,18 +235,21 @@ def flatten(cls, nested_list) -> list[int]:
         Returns:
             list: A unique list of sensor IDs.
         """
-
         all_objects = []
         for s in nested_list:
             if isinstance(s, list):
                 all_objects.extend(s)
             elif isinstance(s, dict):
-                if "sensors" in s:
-                    all_objects.extend(s["sensors"])
-                if "sensor" in s:
-                    all_objects.append(s["sensor"])
-            else:
-                all_objects.append(s)
+                if "plots" in s:
+                    for plot in s["plots"]:
+                        if "sensors" in plot:
+                            all_objects.extend(plot["sensors"])
+                        if "sensor" in plot:
+                            all_objects.append(plot["sensor"])
+                        if "asset" in plot:
+                            sensors = extract_sensors_from_flex_config(plot)
+                            all_objects.extend(sensors)
+
         return list(dict.fromkeys(all_objects).keys())
 
 

flexmeasures/data/schemas/utils.py

@@ -5,6 +5,7 @@
 from pint import DefinitionSyntaxError, DimensionalityError, UndefinedUnitError
 
 from flexmeasures.utils.unit_utils import to_preferred, ur
+from flexmeasures.data.models.time_series import Sensor
 
 
 class MarshmallowClickMixin(click.ParamType):
@@ -83,3 +84,40 @@ def convert_to_quantity(value: str, to_unit: str) -> ur.Quantity:
         raise FMValidationError(
             f"Cannot convert value '{value}' to a valid quantity. {e}"
         )
+
+
+def extract_sensors_from_flex_config(plot: dict) -> list[Sensor]:
+    """
+    Extracts a consolidated list of sensors from an asset based on
+    flex-context or flex-model definitions provided in a plot dictionary.
+    """
+    all_sensors = []
+
+    from flexmeasures.data.schemas.generic_assets import (
+        GenericAssetIdField,
+    )  # Import here to avoid circular imports
+
+    asset = GenericAssetIdField().deserialize(plot.get("asset"))
+
+    fields_to_check = {
+        "flex-context": asset.flex_context,
+        "flex-model": asset.flex_model,
+    }
+
+    for plot_key, flex_config in fields_to_check.items():
+        if plot_key in plot:
+            field_key = plot[plot_key]
+            data = flex_config or {}
+            field_value = data.get(field_key)
+
+            if isinstance(field_value, dict):
+                # Add multiple sensors if they exist as a list
+                sensors = field_value.get("sensors", [])
+                all_sensors.extend(sensors)
+
+                # Add a single sensor if it exists
+                sensor = field_value.get("sensor")
+                if sensor:
+                    all_sensors.append(sensor)
+
+    return all_sensors

flexmeasures/data/services/generic_assets.py

@@ -1,3 +1,4 @@
+from dictdiffer import diff
 from flask import current_app
 from sqlalchemy import delete
 
@@ -6,7 +7,7 @@
 from flexmeasures.data.models.audit_log import AssetAuditLog
 from flexmeasures.data.schemas.scheduling import DBFlexContextSchema
 from flexmeasures.data.schemas.scheduling.storage import DBStorageFlexModelSchema
-from dictdiffer import diff
+from flexmeasures.data.schemas.generic_assets import SensorsToShowSchema
 
 """Services for managing assets"""
 
@@ -120,6 +121,7 @@ def patch_asset(db_asset: GenericAsset, asset_data: dict) -> GenericAsset:
     schema_map = dict(
         flex_context=DBFlexContextSchema,
         flex_model=DBStorageFlexModelSchema,
+        sensors_to_show=SensorsToShowSchema,
     )
 
     for k, v in asset_data.items():
@@ -135,7 +137,10 @@ def patch_asset(db_asset: GenericAsset, asset_data: dict) -> GenericAsset:
             continue
         if k in schema_map:
             # Validate the JSON field against the given schema
-            schema_map[k]().load(v)
+            if k != "sensors_to_show":
+                schema_map[k]().load(v)
+            else:
+                schema_map[k]().deserialize(v)
 
         if k.lower() in {"sensors_to_show", "flex_context", "flex_model"}:
             audit_log_data.append(format_json_field_change(k, getattr(db_asset, k), v))