Merge fa7b2de into 96e0a58

flexmeasures/data/models/reporting/cost.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from typing import Any, List, Dict
+
+
+from flexmeasures.data.models.reporting import Reporter
+from flexmeasures.data.schemas.reporting.cost import (
+    CostReporterConfigSchema,
+    CostReporterParametersSchema,
+)
+from flexmeasures.data.models.time_series import Sensor
+from flexmeasures.utils.time_utils import server_now
+from flexmeasures.data.queries.utils import simplify_index
+
+
+class CostReporter(Reporter):
+    """Compute the cashflow due to energy/power flow.
+
+    Given power/energy and price sensors, this reporter computes the cost/profit
+    of a power/energy flow under a certain tariff.
+
+    Sign convention
+    ----------------
+
+    Power flows:
+        (+) production
+        (-) consumption
+
+    Cash flows:
+        (+) to charge
+        (-) to pay
+
+    """
+
+    __version__ = "1"
+    __author__ = "Seita"
+
+    _config_schema = CostReporterConfigSchema()
+    _parameters_schema = CostReporterParametersSchema()
+
+    weights: dict
+    method: str
+
+    def _compute_report(
+        self,
+        start: datetime,
+        end: datetime,
+        input: List[Dict[str, Any]],
+        output: List[Dict[str, Any]],
+        resolution: timedelta | None = None,
+        belief_time: datetime | None = None,
+    ) -> List[Dict[str, Any]]:
+        """
+        :param start: start time of the report
+        :param end: end time of the report
+        :param input: power/energy sensor to consider
+        :param output: sensor where to save the report to. Specify multiple
+                       output sensors with different resolutions to save
+                       the results in multiple time frames (e.g. hourly, daily).
+        :param resolution: _description_, defaults to None
+        :param belief_time: time where the information is available.
+        """
+
+        production_price_sensor: Sensor = self._config.get("production_price_sensor")
+        consumption_price_sensor: Sensor = self._config.get("consumption_price_sensor")
+
+        input_sensor: Sensor = input[0]["sensor"]  # power or energy sensor
+        timezone = input_sensor.timezone
+
+        if belief_time is None:
+            belief_time = server_now()
+
+        # get prices
+        production_price = simplify_index(
+            production_price_sensor.search_beliefs(
+                event_starts_after=start,
+                event_ends_before=end,
+                beliefs_before=belief_time,
+                resolution=input_sensor.event_resolution,
+            )
+        )
+        production_price = production_price.tz_convert(timezone)
+        consumption_price = simplify_index(
+            consumption_price_sensor.search_beliefs(
+                event_starts_after=start,
+                event_ends_before=end,
+                beliefs_before=belief_time,
+                resolution=input_sensor.event_resolution,
+            )
+        )
+        consumption_price = consumption_price.tz_convert(timezone)
+
+        # get power/energy time series
+        power_energy_data = simplify_index(
+            input_sensor.search_beliefs(
+                event_starts_after=start,
+                event_ends_before=end,
+                resolution=resolution,
+                beliefs_before=belief_time,
+            )
+        )
+
+        # compute energy flow from power flow
+        if input_sensor.measures_power:
+            power_energy_data *= input_sensor.event_resolution / timedelta(hours=1)
+
+        # compute cashflow
+        # this step assumes that positive flows represent production and negative flows consumption
+        result = (
+            power_energy_data.clip(lower=0) * production_price
+            + power_energy_data.clip(upper=0) * consumption_price
+        )
+
+        results = []
+
+        for output_description in output:
+            output_sensor = output_description["sensor"]
+            _result = result.copy()
+
+            # resample result to the event_resolution of the output sensor
+            _result = _result.resample(output_sensor.event_resolution).sum()
+
+            # convert BeliefsSeries into a BeliefsDataFrame
+            _result["belief_time"] = belief_time
+            _result["cumulative_probability"] = 0.5
+            _result["source"] = self.data_source
+            _result.event_resolution = output_sensor.event_resolution
+
+            _result = _result.set_index(
+                ["belief_time", "source", "cumulative_probability"], append=True
+            )
+
+            results.append(
+                {
+                    "name": "cashflow",
+                    "column": "event_value",
+                    "sensor": output_description["sensor"],
+                    "data": _result,
+                }
+            )
+
+        return results

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

@@ -2,32 +2,126 @@
 from datetime import datetime, timedelta
 
 from pytz import utc
+import pandas as pd
 
+from flexmeasures.data.models.planning.utils import initialize_index
 from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType
 from flexmeasures.data.models.data_sources import DataSource
 from flexmeasures.data.models.time_series import Sensor, TimedBelief
 
 
 @pytest.fixture(scope="module")
-def setup_dummy_data(db, app):
+def generic_report(db, app):
+    report_asset_type = GenericAssetType(name="ReportAssetType")
+
+    db.session.add(report_asset_type)
+
+    generic_report = GenericAsset(
+        name="GenericReport", generic_asset_type=report_asset_type
+    )
+
+    db.session.add(generic_report)
+
+    return generic_report
+
+
+@pytest.fixture(scope="module")
+def cost_report(db, app, generic_report, add_market_prices, setup_sources):
+
+    device_type = GenericAssetType(name="Device")
+
+    db.session.add(device_type)
+
+    electricity_device = GenericAsset(
+        name="Electricity Consuming Device", generic_asset_type=device_type
+    )
+
+    db.session.add(electricity_device)
+
+    power_sensor = Sensor(
+        "power",
+        generic_asset=electricity_device,
+        event_resolution=timedelta(minutes=15),
+        unit="MW",
+        timezone="Europe/Amsterdam",
+    )
+
+    energy_sensor = Sensor(
+        "energy",
+        generic_asset=electricity_device,
+        event_resolution=timedelta(minutes=15),
+        unit="MWh",
+        timezone="Europe/Amsterdam",
+    )
+
+    cashflow_sensor_hourly = Sensor(
+        "cashflow hourly",
+        generic_asset=generic_report,
+        event_resolution=timedelta(hours=1),
+        unit="EUR",
+        timezone="Europe/Amsterdam",
+    )
+
+    cashflow_sensor_daily = Sensor(
+        "cashflow daily",
+        generic_asset=generic_report,
+        event_resolution=timedelta(hours=24),
+        unit="EUR",
+        timezone="Europe/Amsterdam",
+    )
+
+    db.session.add_all(
+        [cashflow_sensor_hourly, cashflow_sensor_daily, energy_sensor, power_sensor]
+    )
+
+    time_slots = initialize_index(
+        start=pd.Timestamp("2015-01-03").tz_localize("Europe/Amsterdam"),
+        end=pd.Timestamp("2015-01-04").tz_localize("Europe/Amsterdam"),
+        resolution="15min",
+    )
+
+    def save_values(sensor, values):
+        beliefs = [
+            TimedBelief(
+                event_start=dt,
+                belief_horizon=timedelta(hours=0),
+                event_value=val,
+                source=setup_sources["Seita"],
+                sensor=sensor,
+            )
+            for dt, val in zip(time_slots, values)
+        ]
+        db.session.add_all(beliefs)
+
+    # periodic pattern of producing 100kW for 4h and consuming 100kW for 4h:
+    # i.e. [0.1 0.1 0.1 0.1 ... -0.1 -0.1 -0.1 -0.1]
+    save_values(power_sensor, ([0.1] * 16 + [-0.1] * 16) * 3)
+
+    # creating the same pattern as above but with energy
+    # a flat consumption / production rate of 100kW is equivalent to consume / produce 25kWh
+    # every 15min block for 1h
+    save_values(energy_sensor, ([0.025] * 16 + [-0.025] * 16) * 3)
+
+    db.session.commit()
+
+    yield cashflow_sensor_hourly, cashflow_sensor_daily, power_sensor, energy_sensor
+
+
+@pytest.fixture(scope="module")
+def setup_dummy_data(db, app, generic_report):
     """
     Create 2 Sensors, 1 Asset and 1 AssetType
     """
 
     dummy_asset_type = GenericAssetType(name="DummyGenericAssetType")
-    report_asset_type = GenericAssetType(name="ReportAssetType")
 
-    db.session.add_all([dummy_asset_type, report_asset_type])
+    db.session.add(dummy_asset_type)
 
     dummy_asset = GenericAsset(
         name="DummyGenericAsset", generic_asset_type=dummy_asset_type
     )
 
-    pandas_report = GenericAsset(
-        name="PandasReport", generic_asset_type=report_asset_type
-    )
-
-    db.session.add_all([dummy_asset, pandas_report])
+    db.session.add(dummy_asset)
 
     sensor1 = Sensor("sensor 1", generic_asset=dummy_asset, event_resolution="1h")
     db.session.add(sensor1)
@@ -42,12 +136,12 @@ def setup_dummy_data(db, app):
     db.session.add(sensor3)
 
     report_sensor = Sensor(
-        "report sensor", generic_asset=pandas_report, event_resolution="1h"
+        "report sensor", generic_asset=generic_report, event_resolution="1h"
     )
     db.session.add(report_sensor)
     daily_report_sensor = Sensor(
         "daily report sensor",
-        generic_asset=pandas_report,
+        generic_asset=generic_report,
         event_resolution="1D",
         timezone="Europe/Amsterdam",
     )

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

@@ -0,0 +1,72 @@
+import pytest
+
+from datetime import timedelta
+from flexmeasures.data.models.reporting.cost import CostReporter
+from flexmeasures.data.models.time_series import Sensor
+from datetime import datetime
+from pytz import timezone
+
+
+@pytest.mark.parametrize(
+    "use_power_sensor",
+    [True, False],
+)
+def test_cost_reporter(app, db, cost_report, use_power_sensor):
+    (
+        cashflow_sensor_hourly,
+        cashflow_sensor_daily,
+        power_sensor,
+        energy_sensor,
+    ) = cost_report
+    output_sensor = energy_sensor
+
+    if use_power_sensor:
+        output_sensor = power_sensor
+
+    epex_da = Sensor.query.filter(Sensor.name == "epex_da").one_or_none()
+    epex_da_production = Sensor.query.filter(
+        Sensor.name == "epex_da_production"
+    ).one_or_none()
+
+    cost_reporter = CostReporter(
+        consumption_price_sensor=epex_da, production_price_sensor=epex_da_production
+    )
+
+    tz = timezone("Europe/Amsterdam")
+
+    result = cost_reporter.compute(
+        start=tz.localize(datetime(2015, 1, 3)),
+        end=tz.localize(datetime(2015, 1, 4)),
+        input=[dict(sensor=output_sensor)],
+        output=[
+            dict(sensor=cashflow_sensor_hourly),
+            dict(sensor=cashflow_sensor_daily),
+        ],
+    )
+
+    result_hourly = result[0]["data"]
+    result_daily = result[1]["data"]
+
+    assert result_hourly.event_resolution == timedelta(hours=1)
+
+    # period of negative prices
+
+    # in the period from 00:00 to 03:00
+    # the device produces 100kWh hourly at a -50 EUR/MWh price
+    assert (result_hourly[0:4] == -5).event_value.all()
+
+    # in the period from 04:00 to 08:00
+    # the device consumes 100kWh hourly at a 10 EUR/MWh price
+    assert (result_hourly[4:8] == 1).event_value.all()
+
+    # period of positive prices
+
+    # in the period from 08:00 to 12:00
+    # the device produces 100kWh hourly at a 60 EUR/MWh price
+    assert (result_hourly[8:12] == 6).event_value.all()
+
+    # in the period from 12:00 to 16:00
+    # the device produces 100kWh hourly at a 100 EUR/MWh price
+    assert (result_hourly[12:16] == -10).event_value.all()
+
+    assert result_daily.event_value.iloc[0] == result_hourly.sum().iloc[0]