Aggregation reporter: select sensors by asset, name and unit

documentation/changelog.rst

@@ -32,6 +32,7 @@ New features
 * Both tabs of an asset's status page now name the asset each row belongs to, and the jobs tab also lists the jobs of the asset's sub-assets, so a site asset shows what happened anywhere below it, which you can switch off per session [see `PR #2500 <https://www.github.com/FlexMeasures/flexmeasures/pull/2500>`_]
 * Changing the selected time range on an asset or sensor chart now only loads the data that is actually new, instead of reloading the whole range, which makes stepping through or extending a long period much faster; reloading the page, or leaving it open for five minutes, still fetches everything afresh [see `PR #2433 <https://www.github.com/FlexMeasures/flexmeasures/pull/2433>`_]
 * The statistics table on a sensor page now shows all data sources together by default, as the graph does [see `PR #2462 <https://www.github.com/FlexMeasures/flexmeasures/pull/2462>`_]
+* Report an aggregated signal for a whole site, by telling the ``AggregatorReporter`` which asset to aggregate below, rather than naming every sensor by hand, optionally narrowed down by a pattern on the sensor name and by the units the sensors record in, with values converted to the unit of the sensor the report is recorded on and read at its resolution, so sensors recording in different units and at different resolutions can be aggregated [see `PR #2525 <https://www.github.com/FlexMeasures/flexmeasures/pull/2525>`_]
 
 Infrastructure / Support
 -------------------------

documentation/features/reporting.rst

@@ -85,6 +85,48 @@ This parameterizes the computation (from which sensors does data come from, whic
     These correspond to the same filters available on ``Sensor.search_beliefs``.
 
 
+Aggregating everything below an asset
+--------------------------------------
+
+Naming every sensor one by one is fine for two of them, but not for a site with a dozen PV inverters, let alone one where inverters get added over time.
+The ``AggregatorReporter`` can therefore also be told *which* sensors to aggregate in its configuration, rather than in its parameters:
+
+- ``asset``: aggregate the sensors of this asset and of all of its offspring, so pointing at a site asset covers everything below it.
+- ``sensors``: aggregate these sensors, listed by ID.
+- ``sensor_name_pattern``: keep only the sensors whose name matches this regular expression.
+- ``sensor_units``: keep only the sensors that record in one of these units, or in a unit measuring the same quantity, so ``["MW"]`` also keeps a sensor recording in ``kW``, but not one recording in ``MWh``.
+
+For example, to report the total PV power of a site, whose asset has ID 3:
+
+.. code-block:: json
+
+    {
+        "method" : "sum",
+        "asset" : 3,
+        "sensor_name_pattern" : "(?i)pv",
+        "sensor_units" : ["MW"]
+    }
+
+.. code-block:: json
+
+    {
+        "output": [
+            {
+                "sensor": 10
+            }
+        ],
+        "start" : "2023-01-01T00:00:00+00:00",
+        "end" : "2023-01-03T00:00:00+00:00"
+    }
+
+Values are converted to the unit of the output sensor, and read at its resolution, so sensors recording in different units and at different resolutions can be aggregated onto one sensor.
+Set ``convert_units`` to ``false`` to aggregate the values as they are recorded, and pass a ``resolution`` parameter to read at another resolution than the output sensor's.
+A sensor without a unit is never converted, because an empty unit says nothing about what its values mean, and a sensor recording a quantity that the output sensor cannot express (a temperature onto a power sensor, say) is reported as an error rather than silently added up.
+
+The output sensor itself is left out of the aggregation, so a report can be recorded on a sensor that sits below the very asset being aggregated.
+Sensors named in the ``input`` parameters are read as described there, and the selected sensors are added to them.
+
+
 Example: Profits & losses
 ---------------------------
 

flexmeasures/data/models/reporting/aggregator.py

@@ -1,10 +1,14 @@
 from __future__ import annotations
 
+import re
 from datetime import datetime, timedelta
 from typing import Any
 
 import pandas as pd
+import pint
+from flask import current_app
 
+from flexmeasures.data.models.generic_assets import GenericAsset
 from flexmeasures.data.models.reporting import Reporter
 from flexmeasures.data.models.time_series import Sensor
 from flexmeasures.data.schemas.reporting.aggregation import (
@@ -13,12 +17,35 @@
 )
 
 from flexmeasures.utils.time_utils import server_now
+from flexmeasures.utils.unit_utils import convert_units, units_are_convertible
+
+
+def units_match(unit: str, units: list[str]) -> bool:
+    """Tell whether a sensor unit is one of the units to filter on.
+
+    A unit matches when it is spelled exactly like one of them, or when it measures the same quantity,
+    so that filtering on "MW" also finds sensors recording in "kW", but not sensors recording in "MWh".
+    """
+    for other_unit in units:
+        if unit == other_unit:
+            return True
+        if units_are_convertible(unit, other_unit, duration_known=False):
+            return True
+    return False
 
 
 class AggregatorReporter(Reporter):
-    """This reporter applies an aggregation function to multiple sensors"""
+    """This reporter applies an aggregation function to multiple sensors.
+
+    The sensors to aggregate can be listed one by one, as `input` parameters, but they can also be selected in the reporter's configuration,
+    which is what makes this reporter useful for a whole site: name an asset and every sensor below it is aggregated,
+    optionally narrowed down by a pattern on the sensor name and by the units the sensors record in.
+
+    Values are converted to the unit of the output sensor, and resampled to its resolution,
+    so that sensors recording in different units and at different resolutions can be aggregated.
+    """
 
-    __version__ = "1"
+    __version__ = "2"
     __author__ = "Seita"
 
     _config_schema = AggregatorConfigSchema()
@@ -27,12 +54,105 @@ class AggregatorReporter(Reporter):
     weights: dict
     method: str
 
+    @property
+    def input_sensors(self) -> list:
+        """Return the sensors read by this reporter, including the ones selected in its config."""
+        return self._resolve_sensors(super().input_sensors, self._find_sensors())
+
+    def _find_sensors(self) -> list[Sensor]:
+        """Find the sensors that the reporter's configuration selects.
+
+        The pool of candidates holds the sensors of the asset named in the `asset` field and of its offspring, together with the sensors listed in the `sensors` field.
+        The `sensor_name_pattern` and `sensor_units` fields then narrow that pool down.
+        Sensors are returned ordered by ID, so that an aggregation over a site does not depend on the order in which its sensors happen to be loaded.
+        """
+        asset: GenericAsset | None = self._config.get("asset")
+        listed_sensors: list[Sensor] = self._config.get("sensors") or []
+        name_pattern: str | None = self._config.get("sensor_name_pattern")
+        units: list[str] | None = self._config.get("sensor_units")
+
+        candidates: dict[int, Sensor] = {}
+        if asset is not None:
+            for sub_asset in [asset] + asset.offspring:
+                for sensor in sub_asset.sensors:
+                    candidates[sensor.id] = sensor
+        for sensor in listed_sensors:
+            candidates[sensor.id] = sensor
+
+        sensors = list(candidates.values())
+
+        if name_pattern is not None:
+            pattern = re.compile(name_pattern)
+            sensors = [sensor for sensor in sensors if pattern.search(sensor.name)]
+
+        if units:
+            sensors = [sensor for sensor in sensors if units_match(sensor.unit, units)]
+
+        return sorted(sensors, key=lambda sensor: sensor.id)
+
+    def _collect_input_descriptions(
+        self, input: list[dict[str, Any]], output_sensor: Sensor
+    ) -> list[dict[str, Any]]:
+        """List what to read, combining the `input` parameters with the sensors selected in the config.
+
+        The input descriptions are copied, so that reading them does not consume the parameters the reporter was given.
+        A selected sensor that is already described as an input is left to that description, which is the more specific of the two.
+        The output sensor is never aggregated into itself, which it otherwise would be when it sits below the configured asset.
+        """
+        input_descriptions = [dict(input_description) for input_description in input]
+        described_sensor_ids = {
+            input_description["sensor"].id for input_description in input_descriptions
+        }
+
+        for sensor in self._find_sensors():
+            if sensor.id in described_sensor_ids or sensor.id == output_sensor.id:
+                continue
+            input_descriptions.append({"sensor": sensor})
+
+        return input_descriptions
+
+    def _convert_to_output_unit(
+        self,
+        df: pd.DataFrame,
+        sensor: Sensor,
+        output_sensor: Sensor,
+        resolution: timedelta,
+    ) -> pd.DataFrame:
+        """Convert the values read from one input sensor to the unit of the output sensor.
+
+        A sensor without a unit is left alone, with a warning, because an empty unit says nothing about what its values mean.
+        """
+        if sensor.unit == output_sensor.unit:
+            return df
+
+        if sensor.unit == "" or output_sensor.unit == "":
+            current_app.logger.warning(
+                f"Not converting the values of sensor {sensor.id} ({sensor.name}) from '{sensor.unit}' to '{output_sensor.unit}', because one of these units is empty."
+                f" Set a unit on both sensors, or set the reporter's `convert_units` config field to False to aggregate raw values on purpose."
+            )
+            return df
+
+        try:
+            df["event_value"] = convert_units(
+                df["event_value"],
+                from_unit=sensor.unit,
+                to_unit=output_sensor.unit,
+                event_resolution=resolution,
+            )
+        except (pint.errors.PintError, ValueError) as e:
+            raise ValueError(
+                f"Cannot aggregate sensor {sensor.id} ({sensor.name}), which records in '{sensor.unit}', onto sensor {output_sensor.id} ({output_sensor.name}), which records in '{output_sensor.unit}': {e}"
+                f" Either aggregate sensors that record a comparable quantity, or set the reporter's `convert_units` config field to False to aggregate raw values."
+            )
+
+        return df
+
     def _compute_report(
         self,
         start: datetime,
         end: datetime,
-        input: list[dict[str, Any]],
         output: list[dict[str, Any]],
+        input: list[dict[str, Any]] | None = None,
         resolution: timedelta | None = None,
         belief_time: datetime | None = None,
         belief_horizon: timedelta | None = None,
@@ -43,15 +163,31 @@ def _compute_report(
         columns.
         """
 
-        method: str = self._config.get("method")
+        method: str = self._config.get("method", "sum")
         weights: dict = self._config.get("weights", {})
+        convert_to_output_unit: bool = self._config.get("convert_units", True)
+
+        output_sensor: Sensor = output[0]["sensor"]
+
+        # Read and resample to the resolution of the output sensor, unless the caller asked for another resolution.
+        if resolution is None:
+            resolution = output_sensor.event_resolution
+
+        input_descriptions = self._collect_input_descriptions(
+            input or [], output_sensor=output_sensor
+        )
+        if len(input_descriptions) == 0:
+            raise ValueError(
+                "The AggregatorReporter has no sensors to aggregate."
+                " Name them in the `input` parameters, or select them in the reporter's config with the `asset`, `sensors`, `sensor_name_pattern` and `sensor_units` fields."
+            )
 
         dataframes = []
 
         if belief_time is None and belief_horizon is None:
             belief_time = server_now()
 
-        for input_description in input:
+        for input_description in input_descriptions:
             sensor: Sensor = input_description.pop("sensor")
             # if name is not in belief_search_config, using the Sensor id instead
             column_name = input_description.pop("name", f"sensor_{sensor.id}")
@@ -107,6 +243,12 @@ def _compute_report(
             # drop all indexes but event_start
             df = df.droplevel([1, 2, 3])
 
+            # express the values in the unit of the output sensor
+            if convert_to_output_unit and not df.empty:
+                df = self._convert_to_output_unit(
+                    df, sensor, output_sensor, resolution=resolution
+                )
+
             # apply weight
             if column_name in weights:
                 df *= weights[column_name]
@@ -128,8 +270,8 @@ def _compute_report(
             output_df[belief_col] = belief_horizon
         output_df["cumulative_probability"] = 0.5
         output_df["source"] = self.data_source
-        output_df.sensor = output[0]["sensor"]
-        output_df.event_resolution = output[0]["sensor"].event_resolution
+        output_df.sensor = output_sensor
+        output_df.event_resolution = output_sensor.event_resolution
 
         output_df = output_df.set_index(
             [belief_col, "source", "cumulative_probability"], append=True
@@ -139,7 +281,7 @@ def _compute_report(
             {
                 "name": "aggregate",
                 "column": "event_value",
-                "sensor": output[0]["sensor"],
+                "sensor": output_sensor,
                 "data": output_df,
             }
         ]

flexmeasures/data/models/reporting/tests/conftest.py

@@ -301,3 +301,90 @@ def setup_dummy_data(db, app, generic_report):
     db.session.commit()
 
     yield sensor1, sensor2, sensor3, sensor4, report_sensor, daily_report_sensor
+
+
+@pytest.fixture(scope="module")
+def setup_site_data(db, app, setup_dummy_data):
+    """Create a site asset with sensors spread over its offspring, to aggregate over.
+
+    The site holds the sensor the aggregate is reported on, so that a reporter aggregating everything below the site must leave its own output out.
+    Its PV sensors record in different units and at different resolutions, so that aggregating them needs both a unit conversion and a resampling step.
+    """
+
+    site_type = GenericAssetType(name="AggregationSiteType")
+    db.session.add(site_type)
+
+    site = GenericAsset(name="Aggregation Site", generic_asset_type=site_type)
+    db.session.add(site)
+
+    building = GenericAsset(
+        name="Building", generic_asset_type=site_type, parent_asset=site
+    )
+    db.session.add(building)
+
+    carport = GenericAsset(
+        name="Carport", generic_asset_type=site_type, parent_asset=building
+    )
+    db.session.add(carport)
+
+    site_power_sensor = Sensor(
+        "site power",
+        generic_asset=site,
+        event_resolution=timedelta(hours=1),
+        unit="MW",
+        timezone="UTC",
+    )
+    roof_pv_sensor = Sensor(
+        "roof PV power",
+        generic_asset=building,
+        event_resolution=timedelta(minutes=15),
+        unit="kW",
+        timezone="UTC",
+    )
+    carport_pv_sensor = Sensor(
+        "carport PV power",
+        generic_asset=carport,
+        event_resolution=timedelta(hours=1),
+        unit="MW",
+        timezone="UTC",
+    )
+    temperature_sensor = Sensor(
+        "temperature",
+        generic_asset=building,
+        event_resolution=timedelta(hours=1),
+        unit="°C",
+        timezone="UTC",
+    )
+    db.session.add_all(
+        [site_power_sensor, roof_pv_sensor, carport_pv_sensor, temperature_sensor]
+    )
+
+    site_source = DataSource("site source", type="A")
+    db.session.add(site_source)
+
+    start = datetime(2023, 5, 10, tzinfo=utc)
+
+    def save_values(sensor, value, n_events):
+        db.session.add_all(
+            [
+                TimedBelief(
+                    event_start=start + event * sensor.event_resolution,
+                    belief_horizon=timedelta(hours=24),
+                    event_value=value,
+                    sensor=sensor,
+                    source=site_source,
+                )
+                for event in range(n_events)
+            ]
+        )
+
+    save_values(roof_pv_sensor, 100, 24 * 4)  # 100 kW, in quarter-hourly events
+    save_values(carport_pv_sensor, 0.2, 24)  # 0.2 MW, in hourly events
+    save_values(temperature_sensor, 20, 24)  # 20 °C, in hourly events
+
+    # a previous report on the output sensor, which aggregating over the site should not pick up again
+    save_values(site_power_sensor, 99, 24)
+
+    db.session.commit()
+
+    yield site, site_power_sensor, roof_pv_sensor, carport_pv_sensor, temperature_sensor