add 5/15 sat logic

pvnet_app/app.py

@@ -41,7 +41,9 @@
 from pvnet_app.utils import (
     worker_init_fn, populate_data_config_sources, convert_dataarray_to_forecasts, preds_to_dataarray
 )
-from pvnet_app.data import regrid_nwp_data, download_sat_data, download_nwp_data
+from pvnet_app.data import (
+    download_sat_data, download_nwp_data, preprocess_sat_data, regrid_nwp_data,
+)
 
 # ---------------------------------------------------------------------------
 # GLOBAL SETTINGS
@@ -177,6 +179,9 @@ def app(
     logger.info("Downloading satellite data")
     download_sat_data()
 
+    # Process the 5/15 minutely satellite data
+    preprocess_sat_data(t0)
+
     # Download NWP data
     logger.info("Downloading NWP data")
     download_nwp_data()

pvnet_app/consts.py

@@ -1,3 +1,2 @@
 sat_path = "sat.zarr"
-sat_15_path = "sat_15.zarr"
 nwp_path = "nwp.zarr"

pvnet_app/data.py

@@ -3,28 +3,56 @@
 import logging
 import os
 import fsspec
+from datetime import timedelta
+import ocf_blosc2
 
-from pvnet_app.consts import sat_path, sat_15_path, nwp_path
+from pvnet_app.consts import sat_path, nwp_path
 
 logger = logging.getLogger(__name__)
 
 this_dir = os.path.dirname(os.path.abspath(__file__))
 
+sat_5_path = "sat_5_min.zarr"
+sat_15_path = "sat_15_min.zarr"
+
+
 def download_sat_data():
     """Download the sat data"""
     fs = fsspec.open(os.environ["SATELLITE_ZARR_PATH"]).fs
-    fs.get(os.environ["SATELLITE_ZARR_PATH"], "sat.zarr.zip")
-    os.system(f"rm -r {sat_path}")
-    os.system(f"unzip sat.zarr.zip -d {sat_path}")
+    fs.get(os.environ["SATELLITE_ZARR_PATH"], "sat_5_min.zarr.zip")
+    os.system(f"rm -r {sat_5_path} {sat_15_path} {sat_path}")
+    os.system(f"unzip sat_5_min.zarr.zip -d {sat_5_path}")
 
     # Also download 15-minute satellite if it exists
     sat_latest_15 = os.environ["SATELLITE_ZARR_PATH"].replace("sat.zarr", "sat_15.zarr")
     if fs.exists(sat_latest_15):
         logger.info("Downloading 15-minute satellite data")
-        fs.get(sat_latest_15, "sat_15.zarr")
-        os.system(f"unzip sat_15.zarr.zip -d {sat_15_path}")
+        fs.get(sat_latest_15, sat_15_path)
+        os.system(f"unzip sat_15_min.zarr.zip -d {sat_15_path}")
 
 
+def preprocess_sat_data(t0):
+    """Select and/or combine the 5 and 15-minutely satellite data"""
+
+    ds_sat_5 = xr.open_zarr(sat_5_path)
+    latest_time_5 = pd.to_datetime(ds_sat_5.time.max().values)
+    sat_delay_5 = t0 - latest_time_5
+    logger.info(f"Latest 5-minute timestamp is {latest_time_5} for t0 time {t0}.")
+
+    if sat_delay_5 < timedelta(minutes=60):
+        logger.info(f"5-min satellite delay is only {sat_delay_5} - Using 5-minutely data.")
+        os.system(f"mv {sat_5_path} {sat_path}")
+    else:
+        logger.info(f"5-min satellite delay is {sat_delay} - Switching to 15-minutely data.")
+
+        ds_sat_15 = xr.open_zarr(sat_15_path)
+        latest_time_15 = pd.to_datetime(ds_sat_15.time.max().values)        
+        logger.info(f"Latest 15-minute timestamp is {latest_time_15} for t0 time {t0}.")
+
+        logger.debug("Resampling 15 minute data to 5 mins")
+        ds_sat_15.resample(time="5T").interpolate("linear").to_zarr(sat_path)
+
+
 def download_nwp_data():
     """Download the NWP data"""
     fs = fsspec.open(os.environ["NWP_ZARR_PATH"]).fs

requirements.txt

@@ -1,7 +1,7 @@
 torch[cpu]>=2.1.1
 PVNet-summation>=0.0.9
-pvnet>=2.4.0
-ocf_datapipes>=2.2.2
+pvnet==2.4.0
+ocf_datapipes==2.2.14
 nowcasting_datamodel>=1.5.22
 fsspec[s3]
 xarray