diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 813fb1f7c1..b2ecd736bb 100644 --- a/documentation/changelog.rst +++ b/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 `_] * 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 `_] * The statistics table on a sensor page now shows all data sources together by default, as the graph does [see `PR #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 `_] Infrastructure / Support ------------------------- diff --git a/documentation/features/reporting.rst b/documentation/features/reporting.rst index 5262af925c..e5f1e7e912 100644 --- a/documentation/features/reporting.rst +++ b/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 --------------------------- diff --git a/flexmeasures/data/models/reporting/aggregator.py b/flexmeasures/data/models/reporting/aggregator.py index 63f0db702e..492fa406e9 100644 --- a/flexmeasures/data/models/reporting/aggregator.py +++ b/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, } ] diff --git a/flexmeasures/data/models/reporting/tests/conftest.py b/flexmeasures/data/models/reporting/tests/conftest.py index d6e82798fc..7add753c31 100644 --- a/flexmeasures/data/models/reporting/tests/conftest.py +++ b/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 diff --git a/flexmeasures/data/models/reporting/tests/test_aggregator.py b/flexmeasures/data/models/reporting/tests/test_aggregator.py index e32f980fa2..e84a6170f9 100644 --- a/flexmeasures/data/models/reporting/tests/test_aggregator.py +++ b/flexmeasures/data/models/reporting/tests/test_aggregator.py @@ -1,4 +1,5 @@ import pytest +from marshmallow import ValidationError from flexmeasures.data.models.reporting.aggregator import AggregatorReporter from flexmeasures.data.models.data_sources import DataSource @@ -286,3 +287,157 @@ def test_source_transition(setup_dummy_data, db): assert len(result) == 6 assert (result[:5] == -1).all().event_value # beliefs from the older version assert (result[5:] == 3).all().event_value # belief from the latest version + + +def test_aggregator_over_asset(setup_site_data, db): + """Aggregate every PV sensor below the site, whatever asset it sits on. + + The roof records 100 kW in quarter-hourly events and the carport records 0.2 MW in hourly events, + so reporting onto an hourly sensor in MW needs both a unit conversion and a resampling step to arrive at 0.3 MW. + """ + site, site_power_sensor, roof_pv_sensor, carport_pv_sensor, _ = setup_site_data + + agg_reporter = AggregatorReporter( + config=dict(method="sum", asset=site.id, sensor_name_pattern="PV") + ) + + assert sorted(sensor.id for sensor in agg_reporter.input_sensors) == sorted( + [roof_pv_sensor.id, carport_pv_sensor.id] + ) + + result = agg_reporter.compute( + output=[dict(sensor=site_power_sensor)], + start=datetime(2023, 5, 10, tzinfo=utc), + end=datetime(2023, 5, 11, tzinfo=utc), + belief_time=datetime(2023, 12, 1, tzinfo=utc), + )[0]["data"] + + assert len(result) == 24 + assert result["event_value"].values == pytest.approx(0.3) + + +def test_aggregator_over_asset_leaves_out_output_sensor(setup_site_data, db): + """The sensor a report is recorded on sits below the site, but must not be aggregated into itself. + + It already holds a previous report of 99 MW, which would show up in the aggregate if it were read along with the rest. + """ + site, site_power_sensor, _, _, _ = setup_site_data + + agg_reporter = AggregatorReporter( + config=dict(method="sum", asset=site.id, sensor_units=["MW"]) + ) + + result = agg_reporter.compute( + output=[dict(sensor=site_power_sensor)], + start=datetime(2023, 5, 10, tzinfo=utc), + end=datetime(2023, 5, 11, tzinfo=utc), + belief_time=datetime(2023, 12, 1, tzinfo=utc), + )[0]["data"] + + assert len(result) == 24 + assert result["event_value"].values == pytest.approx(0.3) + + +def test_aggregator_over_listed_sensors(setup_site_data, db): + """Select the sensors to aggregate by ID, and weigh one of them by its generated name.""" + site, site_power_sensor, roof_pv_sensor, carport_pv_sensor, _ = setup_site_data + + agg_reporter = AggregatorReporter( + config=dict( + method="sum", + sensors=[roof_pv_sensor.id, carport_pv_sensor.id], + weights={f"sensor_{carport_pv_sensor.id}": -1.0}, + ) + ) + + result = agg_reporter.compute( + output=[dict(sensor=site_power_sensor)], + start=datetime(2023, 5, 10, tzinfo=utc), + end=datetime(2023, 5, 11, tzinfo=utc), + belief_time=datetime(2023, 12, 1, tzinfo=utc), + )[0]["data"] + + assert len(result) == 24 + assert result["event_value"].values == pytest.approx(-0.1) + + +def test_aggregator_without_unit_conversion(setup_site_data, db): + """Without unit conversion, the reporter adds up what the sensors record, however they record it.""" + site, site_power_sensor, _, _, _ = setup_site_data + + agg_reporter = AggregatorReporter( + config=dict( + method="sum", + asset=site.id, + sensor_name_pattern="PV", + convert_units=False, + ) + ) + + result = agg_reporter.compute( + output=[dict(sensor=site_power_sensor)], + start=datetime(2023, 5, 10, tzinfo=utc), + end=datetime(2023, 5, 11, tzinfo=utc), + belief_time=datetime(2023, 12, 1, tzinfo=utc), + )[0]["data"] + + assert len(result) == 24 + assert result["event_value"].values == pytest.approx(100.2) + + +def test_aggregator_refuses_incompatible_units(setup_site_data, db): + """Aggregating a temperature onto a power sensor says so, rather than silently adding up degrees and megawatts.""" + site, site_power_sensor, _, _, temperature_sensor = setup_site_data + + agg_reporter = AggregatorReporter(config=dict(method="sum", asset=site.id)) + + with pytest.raises(ValueError, match="°C"): + agg_reporter.compute( + output=[dict(sensor=site_power_sensor)], + start=datetime(2023, 5, 10, tzinfo=utc), + end=datetime(2023, 5, 11, tzinfo=utc), + belief_time=datetime(2023, 12, 1, tzinfo=utc), + ) + + +def test_aggregator_without_sensors(setup_site_data, db): + """A reporter that selects no sensor at all says what to do about it.""" + site, site_power_sensor, _, _, _ = setup_site_data + + agg_reporter = AggregatorReporter( + config=dict(method="sum", asset=site.id, sensor_name_pattern="no such sensor") + ) + + with pytest.raises(ValueError, match="no sensors to aggregate"): + agg_reporter.compute( + output=[dict(sensor=site_power_sensor)], + start=datetime(2023, 5, 10, tzinfo=utc), + end=datetime(2023, 5, 11, tzinfo=utc), + belief_time=datetime(2023, 12, 1, tzinfo=utc), + ) + + +def test_aggregator_invalid_sensor_name_pattern(setup_site_data, db): + """An unparsable regular expression is caught where it is configured, not where it is used.""" + with pytest.raises(ValidationError, match="not a valid regular expression"): + AggregatorReporter(config=dict(method="sum", sensor_name_pattern="PV(")) + + +def test_aggregator_data_source_records_sensor_selection(setup_site_data, db): + """The data source of the report records how its sensors were selected, so the report can be traced back to it.""" + site, site_power_sensor, _, _, _ = setup_site_data + + agg_reporter = AggregatorReporter( + config=dict(method="sum", asset=site.id, sensor_name_pattern="PV") + ) + + agg_reporter.compute( + output=[dict(sensor=site_power_sensor)], + start=datetime(2023, 5, 10, tzinfo=utc), + end=datetime(2023, 5, 11, tzinfo=utc), + belief_time=datetime(2023, 12, 1, tzinfo=utc), + ) + + config = agg_reporter.data_source.attributes["data_generator"]["config"] + assert config["asset"] == site.id + assert config["sensor_name_pattern"] == "PV" diff --git a/flexmeasures/data/schemas/reporting/aggregation.py b/flexmeasures/data/schemas/reporting/aggregation.py index 5ac6bc6cd1..7f5228a35d 100644 --- a/flexmeasures/data/schemas/reporting/aggregation.py +++ b/flexmeasures/data/schemas/reporting/aggregation.py @@ -1,16 +1,24 @@ -from marshmallow import fields, validate +import re + +from marshmallow import fields, validate, validates, ValidationError from flexmeasures.data.schemas.reporting import ( ReporterConfigSchema, ReporterParametersSchema, ) -from flexmeasures.data.schemas.io import Output +from flexmeasures.data.schemas.generic_assets import GenericAssetIdField +from flexmeasures.data.schemas.io import Input, Output +from flexmeasures.data.schemas.sensors import SensorIdField class AggregatorConfigSchema(ReporterConfigSchema): """Schema for the AggregatorReporter configuration + Besides the aggregation method and the weights, this schema describes which sensors to aggregate. + Sensors can be selected by asset, so that everything below a site asset is aggregated, and by an explicit list of sensor IDs. + Both selections can be narrowed down by a regular expression on the sensor name, and by a list of units. + Example: .. code-block:: json { @@ -20,15 +28,42 @@ class AggregatorConfigSchema(ReporterConfigSchema): "consumption" : -1.0 } } + + Example, aggregating the power of every PV sensor below asset 3: + .. code-block:: json + { + "method" : "sum", + "asset" : 3, + "sensor_name_pattern" : "(?i)pv", + "sensor_units" : ["MW"] + } """ method = fields.Str(required=False, dump_default="sum", load_default="sum") weights = fields.Dict(fields.Str(), fields.Float(), required=False) + asset = GenericAssetIdField(required=False) + sensors = fields.List(SensorIdField(), required=False) + sensor_name_pattern = fields.Str(required=False) + sensor_units = fields.List(fields.Str(), required=False) + + convert_units = fields.Bool(required=False, dump_default=True, load_default=True) + + @validates("sensor_name_pattern") + def validate_sensor_name_pattern(self, pattern: str, **kwargs): + try: + re.compile(pattern) + except re.error as e: + raise ValidationError( + f"'{pattern}' is not a valid regular expression: {e}. Sensor names are matched with Python's `re` module." + ) + class AggregatorParametersSchema(ReporterParametersSchema): """Schema for the AggregatorReporter parameters + The `input` field is optional here, unlike in the base schema, because the sensors to aggregate can also be selected in the reporter's configuration. + Example: .. code-block:: json { @@ -54,6 +89,13 @@ class AggregatorParametersSchema(ReporterParametersSchema): } """ + # redefining input, because the sensors to aggregate can also come from the reporter's config + input = fields.List( + fields.Nested(Input()), + required=False, + load_default=list, + ) + # redefining output to restrict the output length to 1 output = fields.List( fields.Nested(Output()),