options for inteperolate live pv data

Author: peterdudfield

nowcasting_dataset/config/model.py

@@ -231,6 +231,16 @@ class PV(DataSourceMixin, StartEndDatetimeMixin):
         False, description="Option if to use live data from the nowcasting pv database"
     )
 
+    live_interpolate_minutes: int = Field(
+        30, description="The number of minutes we allow PV data to interpolate"
+    )
+    live_load_extra_minutes: int = Field(
+        0,
+        description="The number of extra minutes in the past we should load. Then the recent "
+        "values can be interpolated, and the extra minutes removed. This is "
+        "because some live data takes ~1 hour to come in.",
+    )
+
     @classmethod
     def model_validation(cls, v):
         """Move old way of storing filenames to new way"""

nowcasting_dataset/data_sources/pv/live.py

@@ -45,23 +45,40 @@ def get_metadata_from_database() -> pd.DataFrame:
     return pv_systems_df
 
 
-def get_pv_power_from_database(history_duration: timedelta) -> pd.DataFrame:
+def get_pv_power_from_database(
+    history_duration: timedelta, interpolate_minutes: int = 30, load_extra_minutes: int = 60
+) -> pd.DataFrame:
     """
     Get pv power from database
 
+    :param history_duration: a timedelta of how many minutes to load in the past
+    :param interpolate_minutes: how many minutes we should interpolate the data froward for
+    :param load_extra_minutes: the extra minutes we should load, in order to load more data.
+        This is because some data from a site lags significantly behind 'now'
+
     Returns: pandas data frame with the following columns
     - pv systems indexes
     The index is the datetime
 
     """
 
+    logger.info("Loading PV data from database")
+    logger.debug(f"{history_duration=} {interpolate_minutes=} {load_extra_minutes=}")
+
+    extra_duration = timedelta(minutes=load_extra_minutes)
+    now = datetime.now(tz=timezone.utc)
+    start_utc = now - history_duration
+    start_utc_extra = start_utc - extra_duration
+
+    # create empty dataframe with 5 mins periods
+    empty_df = pd.DataFrame(index=pd.date_range(start=start_utc_extra, end=now, freq="5T"))
+
     # make database connection
     url = os.getenv("DB_URL_PV")
     db_connection = DatabaseConnection(url=url, base=Base_PV)
 
     with db_connection.get_session() as session:
-        start_utc = datetime.now(tz=timezone.utc) - history_duration
-        pv_yields: List[PVYieldSQL] = get_pv_yield(session=session, start_utc=start_utc)
+        pv_yields: List[PVYieldSQL] = get_pv_yield(session=session, start_utc=start_utc_extra)
 
         pv_yields_df = pd.DataFrame(
             [(PVYield.from_orm(pv_yield)).__dict__ for pv_yield in pv_yields]
@@ -72,9 +89,10 @@ def get_pv_power_from_database(history_duration: timedelta) -> pd.DataFrame:
     else:
         logger.debug(f"Found {len(pv_yields_df)} pv yields")
 
-    # get the system id from 'pv_system_id=xxxx provider=.....'
-    print(pv_yields_df.columns)
-    print(pv_yields_df["pv_system"])
+    if len(pv_yields_df) == 0:
+        return pv_yields_df
+
+        # get the system id from 'pv_system_id=xxxx provider=.....'
     pv_yields_df["pv_system_id"] = (
         pv_yields_df["pv_system"].astype(str).str.split(" ").str[0].str.split("=").str[-1]
     )
@@ -90,7 +108,14 @@ def get_pv_power_from_database(history_duration: timedelta) -> pd.DataFrame:
 
     pv_yields_df.columns = encode_label(pv_yields_df.columns, label="pvoutput")
 
-    # interpolate in between, maximum 30 mins
-    pv_yields_df.interpolate(limit=3, limit_area="inside", inplace=True)
+    # interpolate in between, maximum 'live_interpolate_minutes' mins
+    # note data is in 5 minutes chunks
+    pv_yields_df = empty_df.join(pv_yields_df)
+    limit = int(interpolate_minutes / 5)
+    if limit > 0:
+        pv_yields_df.interpolate(limit=limit, inplace=True)
+
+    # filter out the extra minutes loaded
+    pv_yields_df = pv_yields_df[pv_yields_df.index >= start_utc]
 
     return pv_yields_df

nowcasting_dataset/data_sources/pv/pv_data_source.py

@@ -50,6 +50,8 @@ class PVDataSource(ImageDataSource):
     load_from_gcs: bool = True  # option to load data from gcs, or local file
     get_center: bool = True
     is_live: bool = False
+    live_interpolate_minutes: int = 30
+    live_load_extra_minutes: int = 60
 
     def __post_init__(
         self, image_size_pixels_height: int, image_size_pixels_width: int, meters_per_pixel: int
@@ -172,7 +174,11 @@ def _load_pv_power(self):
             pv_power = pd.concat(pv_power_all, axis="columns")
 
         else:
-            pv_power = get_pv_power_from_database(history_duration=self.history_duration)
+            pv_power = get_pv_power_from_database(
+                history_duration=self.history_duration,
+                interpolate_minutes=self.live_interpolate_minutes,
+                load_extra_minutes=self.live_load_extra_minutes,
+            )
             logger.debug(f"Found {len(pv_power)} pv power datetimes from database ")
             logger.debug(f"Found {len(pv_power.columns)} pv power pv system ids from database")
 

tests/data_sources/pv/conftest.py

@@ -52,8 +52,8 @@ def pv_yields_and_systems(db_session_pv):
 
     Pv systems: Two systems
     PV yields:
-        FOr system 1, pv yields from 4 to 10 at 5 minutes
-        For system 2: 1 pv yield at 16.00
+        For system 1, pv yields from 4 to 10 at 5 minutes. Last one at 09.55
+        For system 2: 1 pv yield at 04.00
     """
 
     pv_system_sql_1: PVSystemSQL = PVSystem(

tests/data_sources/pv/test_pv_live.py

@@ -33,16 +33,20 @@ def test_get_pv_power_from_database(pv_yields_and_systems):
     )
 
 
-@freeze_time("2022-01-01 17:00")
+@freeze_time("2022-01-01 10:55:00")
 def test_get_pv_power_from_database_interpolate(pv_yields_and_systems):
     """Get pv power from database, test out get extra minutes and interpolate"""
-    pv_power = get_pv_power_from_database(history_duration=timedelta(hours=1))
-    assert len(pv_power) == 0  # last data point is at 16:00
 
     pv_power = get_pv_power_from_database(
-        history_duration=timedelta(hours=1), load_extra_minutes=60, interpolate_minutes=60
+        history_duration=timedelta(hours=0.5), load_extra_minutes=0, interpolate_minutes=0
+    )
+    assert len(pv_power) == 0  # last data point is at 09:55
+
+    pv_power = get_pv_power_from_database(
+        history_duration=timedelta(hours=1), load_extra_minutes=60, interpolate_minutes=30
     )
-    assert len(pv_power) == 12  # 1 hours at 5 mins = 12
+    assert len(pv_power) == 13  # 1 hours at 5 mins = 12
+    assert pv_power.isna().sum().sum() == 6  # the last 30 mins is still nans
 
 
 @freeze_time("2022-01-01")