Creation of Reporter class

flexmeasures/data/models/reporting/__init__.py

@@ -117,13 +117,13 @@ def get_data_source_info(cls: type) -> dict:
             source_info["version"] = str(cls.__version__)
         else:
             current_app.logger.warning(
-                f"Scheduler {cls.__name__} loaded, but has no __version__ attribute."
+                f"Reporter {cls.__name__} loaded, but has no __version__ attribute."
             )
         if hasattr(cls, "__author__"):
             source_info["name"] = str(cls.__author__)
         else:
             current_app.logger.warning(
-                f"Scheduler {cls.__name__} has no __author__ attribute."
+                f"Reporter {cls.__name__} has no __author__ attribute."
             )
         return source_info
 
@@ -132,7 +132,7 @@ def deserialize_config(self):
         Check all configurations we have, throwing either ValidationErrors or ValueErrors.
         Other code can decide if/how to handle those.
         """
-        self.deserialize_report_config()
+        self.deserialize_reporter_config()
         self.deserialize_timing_config()
 
     def deserialize_timing_config(self):
@@ -149,7 +149,7 @@ def deserialize_timing_config(self):
             if end < start:
                 raise ValueError(f"Start {start} cannot be after end {end}.")
 
-    def deserialize_report_config(self):
+    def deserialize_reporter_config(self):
         """
         Validate the report config against a Marshmallow Schema.
         Ideas:

flexmeasures/data/models/reporting/pandas_reporter.py

@@ -1,4 +1,7 @@
 import pandas as pd
+
+from flask import current_app
+
 from flexmeasures.data.models.reporting import Reporter
 from flexmeasures.data.schemas.reporting.pandas_reporter import (
     PandasReporterConfigSchema,
@@ -12,20 +15,25 @@ class PandasReporter(Reporter):
     __author__ = None
     schema = PandasReporterConfigSchema()
 
-    def deserialize_report_config(self):
-        # call super class deserialize_report_config
-        super().deserialize_report_config()
+    def deserialize_reporter_config(self):
+        # call super class deserialize_reporter_config
+        super().deserialize_reporter_config()
 
         # extract PandasReporter specific fields
         self.transformations = self.reporter_config.get("transformations")
         self.final_df_output = self.reporter_config.get("final_df_output")
 
     def _compute(self) -> pd.Series:
-        """"""
-        # apply pandas transformations to the dataframes in `self.data``
-        self.apply_transformations()
+        """
+        This method applies the transformations and outputs the dataframe
+        defined in `final_df_output` field of the report_config.
+        """
+
+        # apply pandas transformations to the dataframes in `self.data`
+        self._apply_transformations()
 
         final_output = self.data[self.final_df_output]
+
         return final_output
 
     def get_object_or_literal(self, value, method):
@@ -40,10 +48,14 @@ def get_object_or_literal(self, value, method):
 
         Examples
         >> self.get_object_or_literal(["@sensor_1", "@sensor_2"], "sum")
-        [[                    ...n: 0:00:00,                     ...n: 0:00:00]]
+        [[ <BeliefsDataFrame sensor: VAT>, <BeliefsDataFrame sensor=PV>]]
         """
 
         if method in ["eval", "query"]:
+            if isinstance(value, str) and value.startswith("@"):
+                current_app.logger.warning(
+                    "Cannot reference objects in self.data using the method eval or query. That is because these methods use the symbol `@` to make reference to local variables."
+                )
             return value
 
         if isinstance(value, str) and value.startswith("@"):
@@ -55,31 +67,31 @@ def get_object_or_literal(self, value, method):
 
         return value
 
-    def process_pandas_args(self, args, method):
+    def _process_pandas_args(self, args, method):
         """This method applies the function get_object_or_literal to all the arguments
         to detect where to replace a string "@<object-name>" with the actual object stored in `self.data["<object-name>"]`.
         """
         for i in range(len(args)):
             args[i] = self.get_object_or_literal(args[i], method)
         return args
 
-    def process_pandas_kwargs(self, kwargs, method):
+    def _process_pandas_kwargs(self, kwargs, method):
         """This method applies the function get_object_or_literal to all the keyword arguments
         to detect where to replace a string "@<object-name>" with the actual object stored in `self.data["<object-name>"]`.
         """
         for k, v in kwargs.items():
             kwargs[k] = self.get_object_or_literal(v, method)
         return kwargs
 
-    def apply_transformations(self) -> pd.Series:
+    def _apply_transformations(self) -> pd.Series:
         """Convert the series using the given list of transformation specs, which is called in the order given.
 
         Each transformation specs should include a 'method' key specifying a method name of a Pandas DataFrame.
 
         Optionally, 'args' and 'kwargs' keys can be specified to pass on arguments or keyword arguments to the given method.
 
-        All data exchange is made through the dictionary `self.data`. The superclass Reporter already fetches and saves BeliefDataFrames
-        of the input sensors in the fields `sensor_<sensor_id>`. In case you need to perform complex operations on dataframes, you can
+        All data exchange is made through the dictionary `self.data`. The superclass Reporter already fetches BeliefsDataFrames of
+        the sensors and saves them in the self.data dictionary fields  `sensor_<sensor_id>`. In case you need to perform complex operations on dataframes, you can
         split the operations in several steps and saving the intermediate results using the parameters `df_input` and `df_output` for the
         input and output dataframes, respectively.
 
@@ -104,8 +116,8 @@ def apply_transformations(self) -> pd.Series:
             )  # default is OUTPUT = INPUT.method()
 
             method = transformation.get("method")
-            args = self.process_pandas_args(transformation.get("args", []), method)
-            kwargs = self.process_pandas_kwargs(
+            args = self._process_pandas_args(transformation.get("args", []), method)
+            kwargs = self._process_pandas_kwargs(
                 transformation.get("kwargs", {}), method
             )
 

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

@@ -30,7 +30,7 @@
     -50.07,
     -50.00,
     -0.90,
-    08.10,
+    108.10,
     128.10,
     151.00,
     155.20,
@@ -64,18 +64,18 @@
     35.8,
     33.6,
     32.0,
-]  # EUR/MWh
+]  # cents/kWh
 
 
 class TibberReporter(PandasReporter):
     def __init__(self, start: datetime, end: datetime) -> None:
         """This class calculates the price of energy of a tariff indexed to the Day Ahead prices.
-        Energy Price = (1 + VAT) x ( EB + Tiber + DA Prices)
+        Energy Price = (1 + VAT) x ( EnergyTax + Tiber + DA Prices)
         """
 
         # search the sensors
-        EB = Sensor.query.filter(Sensor.name == "EB").one_or_none()
-        BWV = Sensor.query.filter(Sensor.name == "BWV").one_or_none()
+        EnergyTax = Sensor.query.filter(Sensor.name == "EnergyTax").one_or_none()
+        VAT = Sensor.query.filter(Sensor.name == "VAT").one_or_none()
         tibber_tariff = Sensor.query.filter(
             Sensor.name == "Tibber Tariff"
         ).one_or_none()
@@ -84,31 +84,29 @@ def __init__(self, start: datetime, end: datetime) -> None:
 
         tb_query_config_extra = dict(
             resolution=3600,  # 1h = 3600s
-            event_starts_after=str(start),
-            event_ends_before=str(end),
         )
 
         # creating the PandasReporter reporter config
-        reporter_config_raw = dict(
+        reporter_config = dict(
             start=str(start),
             end=str(end),
             tb_query_config=[
-                dict(sensor=EB.id, **tb_query_config_extra),
-                dict(sensor=BWV.id, **tb_query_config_extra),
+                dict(sensor=EnergyTax.id, **tb_query_config_extra),
+                dict(sensor=VAT.id, **tb_query_config_extra),
                 dict(sensor=tibber_tariff.id, **tb_query_config_extra),
                 dict(sensor=da_prices.id, **tb_query_config_extra),
             ],
             transformations=[
                 dict(
                     df_input="sensor_1",
-                    df_output="BWV",
+                    df_output="VAT",
                     method="droplevel",
                     args=[[1, 2, 3]],
                 ),
-                dict(method="add", args=[1]),  # this is to get 1 + BWV
+                dict(method="add", args=[1]),  # this is to get 1 + VAT
                 dict(
                     df_input="sensor_2",
-                    df_output="EB",
+                    df_output="EnergyTax",
                     method="droplevel",
                     args=[[1, 2, 3]],
                 ),
@@ -125,22 +123,21 @@ def __init__(self, start: datetime, end: datetime) -> None:
                     args=[[1, 2, 3]],
                 ),
                 dict(
-                    method="multiply",
-                    args=[1 / 1000],
-                ),
+                    method="add", args=["@tibber_tariff"]
+                ),  # da_prices = da_prices + tibber_tariff
                 dict(
-                    df_output="energy_price",
-                    df_input="EB",
-                    method="add",
-                    args=["@tibber_tariff"],
-                ),
-                dict(method="add", args=["@da_prices"]),
-                dict(method="multiply", args=["@BWV"]),
+                    method="add", args=["@EnergyTax"]
+                ),  # da_prices = da_prices + EnergyTax
+                dict(
+                    method="multiply", args=["@VAT"]
+                ),  # da_prices = da_price * VAT, VAT
+                dict(method="round", args=[2]),  # round 2 decimals
+                dict(method="round", args=[1]),  # round 1 decimal
             ],
-            final_df_output="energy_price",
+            final_df_output="da_prices",
         )
 
-        super().__init__(reporter_config_raw)
+        super().__init__(reporter_config)
 
 
 def beliefs_from_timeseries(index, values, sensor, source):
@@ -172,36 +169,45 @@ def tibber_test_data(fresh_db, app):
 
     electricity_price = GenericAsset(name="Electricity Price", generic_asset_type=price)
 
-    VAT = GenericAsset(name="VAT", generic_asset_type=tax)
+    VAT_asset = GenericAsset(name="VAT", generic_asset_type=tax)
 
     electricity_tax = GenericAsset(name="Energy Tax", generic_asset_type=tax)
 
-    db.session.add_all([electricity_price, VAT, electricity_tax])
+    db.session.add_all([electricity_price, VAT_asset, electricity_tax])
 
     # Taxes
-    BWV = Sensor("BWV", generic_asset=VAT, event_resolution=timedelta(days=365))
-    EB = Sensor(
-        "EB", generic_asset=electricity_tax, event_resolution=timedelta(days=365)
+    VAT = Sensor(
+        "VAT",
+        generic_asset=VAT_asset,
+        event_resolution=timedelta(days=365),
+        unit="unit range",
+    )
+    EnergyTax = Sensor(
+        "EnergyTax",
+        generic_asset=electricity_tax,
+        event_resolution=timedelta(days=365),
+        unit="EUR/MWh",
     )
 
     # Tibber Tariff
     tibber_tariff = Sensor(
         "Tibber Tariff",
         generic_asset=electricity_price,
         event_resolution=timedelta(days=365),
+        unit="EUR/MWh",
     )
 
-    db.session.add_all([BWV, EB, tibber_tariff])
+    db.session.add_all([VAT, EnergyTax, tibber_tariff])
 
     """
         Saving TimeBeliefs to the DB
     """
 
-    # Adding EB, BWV and Tibber Tarriff beliefs to the DB
+    # Adding EnergyTax, VAT and Tibber Tarriff beliefs to the DB
     for sensor, source_name, value in [
-        (BWV, "Belastingdienst", 0.21),
-        (EB, "Belastingdienst", 0.12599),
-        (tibber_tariff, "Tibber", 0.018),
+        (VAT, "Tax Authority", 0.21),  # unit interval
+        (EnergyTax, "Tax Authority", 125.99),  # EUR / MWh
+        (tibber_tariff, "Tibber", 18.0),  # EUR /MWh
     ]:
         belief = TimedBelief(
             sensor=sensor,
@@ -243,10 +249,16 @@ def test_tibber_reporter(tibber_test_data):
     assert len(result) == 24
 
     tibber_app_price_df = (
-        pd.DataFrame(tibber_app_price, index=index, columns=["event_value"]) / 100
+        pd.DataFrame(tibber_app_price, index=index, columns=["event_value"])
+        * 10  # converting cents/kWh to EUR/MWh
     )
 
-    # checking that (EPEX+EB + Tibber Tariff)*(1+BWV) = Tibber App Price
-    assert (
-        abs(result - tibber_app_price_df).mean().iloc[0] < 0.01
-    )  # difference of less than 1 cent / kWh
+    error = abs(result - tibber_app_price_df)
+
+    # checking that (EPEX + EnergyTax + Tibber Tariff)*(1 + VAT) = Tibber App Price
+
+    # mean error is low enough, i.e 1 EUR/MWh = 0.1 cent/kWh
+    assert error.mean().iloc[0] < 1
+
+    # max error is low enough, i.e 1 EUR/MWh = 0.1 cent/kWh
+    assert error.max().iloc[0] < 1