update BL to store predictions in influx for auditing

Author: Nick van der Burgt

src/application/generate_events.py

@@ -140,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.

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")