Merge pull request #6 from ElaadNL/audit-predictions

Audit predictions

Author: nicburgt

src/application/generate_events.py

@@ -16,7 +16,7 @@
 from src.models.predicted_load import PredictedGridAssetLoad
 
 
-class PredictionActionsBase[ReadOnlySession](ABC):
+class PredictionActionsBase[ReadOnlySession, WriteSession](ABC):
     """Abstract class which contains methods used by this workflow.
 
     These methods are implemented on a higher level and provided to the functions of this
@@ -27,6 +27,10 @@ class PredictionActionsBase[ReadOnlySession](ABC):
     def get_query_api(self) -> ReadOnlySession:
         """Retrieve a read-only session to the database."""
 
+    @abstractmethod
+    def get_write_api(self) -> WriteSession:
+        """Retrieve a write session to the database."""
+
     @abstractmethod
     async def get_predicted_grid_asset_load(
         self, query_api: ReadOnlySession, from_date: datetime, to_date: datetime
@@ -42,6 +46,19 @@ async def get_predicted_grid_asset_load(
             list[PredictedGridAssetLoad]: The list of predicted grid asset loads.
         """
 
+    @abstractmethod
+    async def audit_predicted_grid_asset_loads(
+        self,
+        write_api: WriteSession,
+        predicted_grid_asset_loads: list[PredictedGridAssetLoad],
+    ) -> None:
+        """Audit predicted grid asset loads by storing them in the database.
+
+        Args:
+            write_api (WriteSession): The write connection to the database.
+            predicted_grid_asset_loads (list[PredictedGridAssetLoad]): The list of predicted grid asset loads to audit.
+        """
+
 
 def _generate_capacity_limitation_intervals(
     interval_id: int,
@@ -123,7 +140,7 @@ async def get_capacity_limitation_event(
     Args:
         actions (PredictionActionsBase): The actions to use.
         from_date (datetime): The start time (inclusive) from which to fetch OpenADR events.
-        to_date (datetime): The end time (exclusive) from which to fetch OpenADR events.
+        to_date (datetime): The end time (inclusive) from which to fetch OpenADR events.
 
     Returns:
         Event | None: The OpenADR3 capacity limitation event. None if no data to base the event on could be retrieved.
@@ -140,4 +157,9 @@ async def get_capacity_limitation_event(
         )
         return None
 
+    write_api = actions.get_write_api()
+    await actions.audit_predicted_grid_asset_loads(
+        write_api, predicted_grid_asset_loads
+    )
+
     return _generate_capacity_limitation_event(predicted_grid_asset_loads, MAX_CAPACITY)

src/config.py

@@ -28,7 +28,9 @@
 STANDARD_PROFILES_BUCKET_NAME = config(
     "STANDARD_PROFILES_BUCKET_NAME", default="ditm_standard_profiles"
 )
-DALIDATA_BUCKET_NAME = config("DALIDATA_BUCKET_NAME", default="dalidata")
+DALIDATA_BUCKET_NAME = config(
+    "DALIDATA_BUCKET_NAME", default="ditm-dali-data-processed"
+)
 
 # External services URLs
 WEATHER_FORECAST_API_URL = config("WEATHER_FORECAST_API_URL")

src/infrastructure/azureml/feature_generation.py

@@ -144,28 +144,28 @@ async def _get_lag_features_for_dates(
     )
 
     predict_datetimes_df["lag_1_year"] = dalidata_df.reindex(lag_1_year_dt)[
-        "value"
+        "WAARDE"
     ].values
     predict_datetimes_df["lag_1_days"] = dalidata_df.reindex(lag_1_day_dt)[
-        "value"
+        "WAARDE"
     ].values
     predict_datetimes_df["lag_2_days"] = dalidata_df.reindex(lag_2_day_dt)[
-        "value"
+        "WAARDE"
     ].values
     predict_datetimes_df["lag_3_days"] = dalidata_df.reindex(lag_3_day_dt)[
-        "value"
+        "WAARDE"
     ].values
     predict_datetimes_df["lag_4_days"] = dalidata_df.reindex(lag_4_day_dt)[
-        "value"
+        "WAARDE"
     ].values
     predict_datetimes_df["lag_5_days"] = dalidata_df.reindex(lag_5_day_dt)[
-        "value"
+        "WAARDE"
     ].values
     predict_datetimes_df["lag_6_days"] = dalidata_df.reindex(lag_6_day_dt)[
-        "value"
+        "WAARDE"
     ].values
     predict_datetimes_df["lag_7_days"] = dalidata_df.reindex(lag_7_day_dt)[
-        "value"
+        "WAARDE"
     ].values
 
     return predict_datetimes_df

src/infrastructure/influxdb/dalidata/query_dali_data.py

@@ -11,24 +11,22 @@ async def retrieve_dali_data_between(
     start_date_inclusive: datetime,
     end_date_inclusive: datetime,
 ) -> pd.DataFrame:
-    """Retrieve standard profiles from InfluxDB between the given dates.
+    """Retrieve dalidata from InfluxDB between the given dates.
 
     Args:
         start_date_inclusive (datetime): The start date (inclusive)
         end_date_inclusive (datetime): The end date (inclusive)
 
     Returns:
-        pd.DataFrame: The dataframe containing standard profiles for the date range.
+        pd.DataFrame: The dataframe containing dalidata for the date range.
     """
     start_date_str = start_date_inclusive.strftime(format="%Y-%m-%dT%H:%M:%SZ")
     end_date_str = end_date_inclusive.strftime(format="%Y-%m-%dT%H:%M:%SZ")
 
     query = f"""from(bucket: "{DALIDATA_BUCKET_NAME}")
             |> range(start: {start_date_str}, stop: {end_date_str})
-            |> filter(fn: (r) => r["_measurement"] == "kinesis_data")
-            |> filter(fn: (r) => r["_field"] == "value")
-            |> filter(fn: (r) => r["boxid"] == "CVD.091030-1")
-            |> filter(fn: (r) => r["description"] == "Trafometing vermogen som 3 fases - gem. 15 min." or r["description"] == "Trafometing_vermogen_som_3_fases_-_gem._15_min.")
+            |> filter(fn: (r) => r["_measurement"] == "WAARDE")
+            |> filter(fn: (r) => r["_field"] == "WAARDE")
             |> group(columns: [])
             |> sort(columns: ["_time"])
             |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")

src/infrastructure/prediction_actions_impl.py

@@ -4,14 +4,16 @@
 
 from influxdb_client.client.influxdb_client_async import InfluxDBClientAsync
 from influxdb_client.client.query_api_async import QueryApiAsync
+from influxdb_client.client.write_api_async import WriteApiAsync
 
 from src.application.generate_events import PredictionActionsBase
 from src.infrastructure.azureml.feature_generation import get_features_between_dates
 from src.infrastructure.azureml.predictions import get_predictions_for_features
 from src.models.predicted_load import PredictedGridAssetLoad
+from src.infrastructure.influxdb.trafo_load_audit import store_predictions_for_audit
 
 
-class PredictionActionsInfluxDB(PredictionActionsBase[QueryApiAsync]):
+class PredictionActionsInfluxDB(PredictionActionsBase[QueryApiAsync, WriteApiAsync]):
     """Implementation of the prediction actions using influxDB.
 
     This is implemented seperately in the infrastructure layer to promote decoupling of business
@@ -34,6 +36,10 @@ def get_query_api(self) -> QueryApiAsync:
         """Retrieve a read-only connection for the database."""
         return self.client.query_api()
 
+    def get_write_api(self) -> WriteApiAsync:
+        """Retrieve a write connection for the database."""
+        return self.client.write_api()
+
     async def get_predicted_grid_asset_load(
         self, query_api: QueryApiAsync, from_date: datetime, to_date: datetime
     ) -> list[PredictedGridAssetLoad]:
@@ -53,3 +59,19 @@ async def get_predicted_grid_asset_load(
             end_date_inclusive=to_date,
         )
         return get_predictions_for_features(features=features_for_time_range)
+
+    async def audit_predicted_grid_asset_loads(
+        self,
+        write_api: WriteApiAsync,
+        predicted_grid_asset_loads: list[PredictedGridAssetLoad],
+    ) -> None:
+        """Audit predicted grid asset loads by storing them in the database.
+
+        Args:
+            write_api (WriteApi): The write connection to the database.
+            predicted_grid_asset_loads (list[PredictedGridAssetLoad]): The list of predicted grid asset loads to audit.
+        """
+        await store_predictions_for_audit(
+            write_api=write_api,
+            predicted_loads=predicted_grid_asset_loads,
+        )

src/infrastructure/predictions_actions_stub_impl.py

@@ -9,7 +9,7 @@
 from src.models.predicted_load import PredictedGridAssetLoad
 
 
-class PredictionActionsStub(PredictionActionsBase[None]):
+class PredictionActionsStub(PredictionActionsBase[None, None]):
     """Stub implementation of the prediction actions, generates predicted grid asset loads when called.
 
     Args:
@@ -24,6 +24,10 @@ def get_query_api(self) -> None:
         """Retrieve a read-only connection for the database."""
         return None
 
+    def get_write_api(self) -> None:
+        """Retrieve a write connection for the database."""
+        return None
+
     async def get_predicted_grid_asset_load(
         self, query_api: None, from_date: datetime, to_date: datetime
     ) -> list[PredictedGridAssetLoad]:
@@ -54,3 +58,17 @@ async def get_predicted_grid_asset_load(
             current_time += step
 
         return predicted_grid_asset_loads
+
+    async def audit_predicted_grid_asset_loads(
+        self,
+        write_api: None,
+        predicted_grid_asset_loads: list[PredictedGridAssetLoad],
+    ) -> None:
+        """Stub implementation of auditing predicted grid asset loads.
+
+        Args:
+            write_api (None): The write connection.
+            predicted_grid_asset_loads (list[PredictedGridAssetLoad]): The list of predicted grid asset loads to audit.
+        """
+        # In this stub implementation, we do nothing.
+        pass