Merge pull request #667 from openclimatefix/issue/load-more-pv-data

pv live interpolation options

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,42 @@ 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, load_extra_minutes: int
+) -> pd.DataFrame:
     """
     Get pv power from database
 
-    Returns: pandas data frame with the following columns
-    - pv systems indexes
+    Args:
+        history_duration: a timedelta of how many minutes to load in the past
+        interpolate_minutes: how many minutes we should interpolate the data froward for
+        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)
+
+        logger.debug(f"Found {len(pv_yields)} PV yields from the database")
 
         pv_yields_df = pd.DataFrame(
             [(PVYield.from_orm(pv_yield)).__dict__ for pv_yield in pv_yields]
@@ -72,9 +91,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 +110,16 @@ 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
+    logger.debug(f"{len(pv_yields_df)} of datetimes before filter on {start_utc}")
+    pv_yields_df = pv_yields_df[pv_yields_df.index >= start_utc]
+    logger.debug(f"{len(pv_yields_df)} of datetimes after filter on {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

@@ -19,12 +19,14 @@ def test_get_metadata_from_database(pv_yields_and_systems):
     assert len(meteadata) == 2
 
 
-@freeze_time("2022-01-01")
+@freeze_time("2022-01-01 05:00")
 def test_get_pv_power_from_database(pv_yields_and_systems):
     """Get pv power from database"""
-    pv_power = get_pv_power_from_database(history_duration=timedelta(hours=1))
+    pv_power = get_pv_power_from_database(
+        history_duration=timedelta(hours=1), load_extra_minutes=30, interpolate_minutes=30
+    )
 
-    assert len(pv_power) == 72  # 6 hours at 5 mins = 6*12
+    assert len(pv_power) == 13  # 1 hours at 5 mins = 6*12
     assert len(pv_power.columns) == 2
     assert pv_power.columns[0] == "11"
     assert (
@@ -33,7 +35,23 @@ def test_get_pv_power_from_database(pv_yields_and_systems):
     )
 
 
-@freeze_time("2022-01-01")
+@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=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) == 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 05:00")
 def test_get_example_and_batch(pv_yields_and_systems):
     """Test PVDataSource with data source from database"""
 
@@ -61,6 +79,6 @@ def test_get_example_and_batch(pv_yields_and_systems):
     assert len(pv_data_source.pv_metadata) > 0
 
     locations = pv_data_source.get_locations(pv_data_source.pv_power.index)
-    assert len(locations) == 72  # 6 hours at 5 mins
+    assert len(locations) == 7  # 30 minutes at 5 mins, inclusive
 
     _ = pv_data_source.get_example(location=locations[0])