Merge commit '4e2e1448252da9bdc6ff0560157ea150bc394edf' into issue/gsp-database

# Conflicts:
#	ocf_datapipes/load/gsp/gsp.py
#	ocf_datapipes/production/power_perceiver.py

Author: peterdudfield

.bumpversion.cfg

@@ -1,7 +1,7 @@
 [bumpversion]
 commit = True
 tag = True
-current_version = 0.2.9
+current_version = 0.2.13
 
 [bumpversion:file:setup.py]
 search = version="{current_version}"

environment.yml

@@ -12,9 +12,18 @@ dependencies:
   - xarray
   - fsspec
   - zarr
+  - cartopy
+  - dask
+  - pyproj
+  - pyresample
+  - geopandas
+  - h5netcdf
 pip:
   - einops
   - pathy
   - git+https://github.com/SheffieldSolar/PV_Live-API
   - pyaml_env
   - nowcasting_datamodel
+  - gitpython
+  - tqdm
+  - bottleneck

ocf_datapipes/convert/gsp.py

@@ -27,7 +27,9 @@ def __iter__(self) -> NumpyBatch:
                 BatchKey.gsp: xr_data.values,
                 BatchKey.gsp_t0_idx: xr_data.attrs["t0_idx"],
                 BatchKey.gsp_id: xr_data.gsp_id.values,
-                BatchKey.gsp_capacity_mwp: xr_data.isel(time_utc=0)["capacity_mwp"].values,
+                BatchKey.gsp_capacity_megawatt_power: xr_data.isel(time_utc=0)[
+                    "capacity_megawatt_power"
+                ].values,
                 BatchKey.gsp_time_utc: datetime64_to_float(xr_data["time_utc"].values),
             }
 

ocf_datapipes/convert/pv.py

@@ -29,7 +29,7 @@ def __iter__(self) -> NumpyBatch:
                 BatchKey.pv_t0_idx: xr_data.attrs["t0_idx"],
                 BatchKey.pv_system_row_number: xr_data["pv_system_row_number"].values,
                 BatchKey.pv_id: xr_data["pv_system_id"].values.astype(np.float32),
-                BatchKey.pv_capacity_wp: xr_data["capacity_wp"].values,
+                BatchKey.pv_capacity_watt_power: xr_data["capacity_watt_power"].values,
                 BatchKey.pv_time_utc: datetime64_to_float(xr_data["time_utc"].values),
                 BatchKey.pv_x_osgb: xr_data["x_osgb"].values,
                 BatchKey.pv_y_osgb: xr_data["y_osgb"].values,

ocf_datapipes/load/gsp/gsp.py

@@ -78,15 +78,16 @@ def __iter__(self) -> xr.DataArray:
 
         # Ensure the centroids have the same GSP ID index as the GSP PV power:
         gsp_id_to_shape = gsp_id_to_shape.loc[gsp_pv_power_mw_ds.gsp_id]
-
         data_array = put_gsp_data_into_an_xr_dataarray(
             gsp_pv_power_mw=gsp_pv_power_mw_ds.generation_mw.data.astype(np.float32),
             time_utc=gsp_pv_power_mw_ds.datetime_gmt.data,
             gsp_id=gsp_pv_power_mw_ds.gsp_id.data,
             # TODO: Try using `gsp_id_to_shape.geometry.envelope.centroid`. See issue #76.
             x_osgb=gsp_id_to_shape.geometry.centroid.x.astype(np.float32),
             y_osgb=gsp_id_to_shape.geometry.centroid.y.astype(np.float32),
-            capacity_mwp=gsp_pv_power_mw_ds.installedcapacity_mwp.data.astype(np.float32),
+            capacity_megawatt_power=gsp_pv_power_mw_ds.installedcapacity_mwp.data.astype(  # noqa
+                np.float32
+            ),
         )
 
         del gsp_id_to_shape, gsp_pv_power_mw_ds

ocf_datapipes/load/pv/live.py

@@ -85,7 +85,7 @@ def __iter__(self):
             pv_power_watts=pv_power,  # TODO check this is watts
             y_osgb=pv_metadata.y_osgb.astype(np.float32),
             x_osgb=pv_metadata.x_osgb.astype(np.float32),
-            capacity_wp=pv_metadata.capacity_wp,
+            capacity_watt_power=pv_metadata.capacity_watt_power,
             pv_system_row_number=pv_system_row_number,
         )
 
@@ -148,7 +148,7 @@ def get_metadata_from_database(providers: List[str] = None) -> pd.DataFrame:
     pv_system_all_df["y_osgb"] = y_osgb
 
     pv_system_all_df["capacity_kw"] = pv_system_all_df["installed_capacity_kw"]
-    pv_system_all_df["capacity_wp"] = pv_system_all_df["capacity_kw"] * 1000
+    pv_system_all_df["capacity_watt_power"] = pv_system_all_df["capacity_kw"] * 1000
 
     # sort index
     pv_system_all_df = pv_system_all_df.sort_index()

ocf_datapipes/load/pv/pv.py

@@ -51,21 +51,25 @@ def __iter__(self):
 def load_everything_into_ram(pv_power_filename, pv_metadata_filename) -> xr.DataArray:
     """Open AND load PV data into RAM."""
     # Load pd.DataFrame of power and pd.Series of capacities:
-    pv_power_watts, pv_capacity_wp, pv_system_row_number = _load_pv_power_watts_and_capacity_wp(
+    (
+        pv_power_watts,
+        pv_capacity_watt_power,
+        pv_system_row_number,
+    ) = _load_pv_power_watts_and_capacity_watt_power(
         pv_power_filename,
     )
     pv_metadata = _load_pv_metadata(pv_metadata_filename)
-    # Ensure pv_metadata, pv_power_watts, and pv_capacity_wp all have the same set of
+    # Ensure pv_metadata, pv_power_watts, and pv_capacity_watt_power all have the same set of
     # PV system IDs, in the same order:
     pv_metadata, pv_power_watts = intersection_of_pv_system_ids(pv_metadata, pv_power_watts)
-    pv_capacity_wp = pv_capacity_wp.loc[pv_power_watts.columns]
+    pv_capacity_watt_power = pv_capacity_watt_power.loc[pv_power_watts.columns]
     pv_system_row_number = pv_system_row_number.loc[pv_power_watts.columns]
 
     data_in_ram = put_pv_data_into_an_xr_dataarray(
         pv_power_watts=pv_power_watts,
         y_osgb=pv_metadata.y_osgb.astype(np.float32),
         x_osgb=pv_metadata.x_osgb.astype(np.float32),
-        capacity_wp=pv_capacity_wp,
+        capacity_watt_power=pv_capacity_watt_power,
         pv_system_row_number=pv_system_row_number,
     )
 
@@ -77,12 +81,12 @@ def load_everything_into_ram(pv_power_filename, pv_metadata_filename) -> xr.Data
     return data_in_ram
 
 
-def _load_pv_power_watts_and_capacity_wp(
+def _load_pv_power_watts_and_capacity_watt_power(
     filename: Union[str, Path],
     start_date: Optional[datetime.datetime] = None,
     end_date: Optional[datetime.datetime] = None,
 ) -> tuple[pd.DataFrame, pd.Series, pd.Series]:
-    """Return pv_power_watts, pv_capacity_wp, pv_system_row_number.
+    """Return pv_power_watts, pv_capacity_watt_power, pv_system_row_number.
 
     The capacities and pv_system_row_number are computed across the *entire* dataset,
     and so is independent of the `start_date` and `end_date`. This ensures the PV system
@@ -94,7 +98,7 @@ def _load_pv_power_watts_and_capacity_wp(
     # Load data in a way that will work in the cloud and locally:
     with fsspec.open(filename, mode="rb") as file:
         pv_power_ds = xr.open_dataset(file, engine="h5netcdf")
-        pv_capacity_wp = pv_power_ds.max().to_pandas().astype(np.float32)
+        pv_capacity_watt_power = pv_power_ds.max().to_pandas().astype(np.float32)
         pv_power_watts = pv_power_ds.sel(datetime=slice(start_date, end_date)).to_dataframe()
         pv_power_watts = pv_power_watts.astype(np.float32)
         del pv_power_ds
@@ -105,14 +109,15 @@ def _load_pv_power_watts_and_capacity_wp(
             pv_power_watts.tz_localize("Europe/London").tz_convert("UTC").tz_convert(None)
         )
 
-    pv_capacity_wp.index = [np.int32(col) for col in pv_capacity_wp.index]
+    pv_capacity_watt_power.index = [np.int32(col) for col in pv_capacity_watt_power.index]
     pv_power_watts.columns = pv_power_watts.columns.astype(np.int64)
 
-    # Create pv_system_row_number. We use the index of `pv_capacity_wp` because that includes
+    # Create pv_system_row_number. We use the index of
+    # `pv_capacity_watt_power` because that includes
     # the PV system IDs for the entire dataset (independent of `start_date` and `end_date`).
     # We use `float32` for the ID because we use NaN to indicate a missing PV system,
     # or that this whole example doesn't include PV.
-    all_pv_system_ids = pv_capacity_wp.index
+    all_pv_system_ids = pv_capacity_watt_power.index
     pv_system_row_number = np.arange(start=0, stop=len(all_pv_system_ids), dtype=np.float32)
     pv_system_row_number = pd.Series(pv_system_row_number, index=all_pv_system_ids)
 
@@ -134,7 +139,9 @@ def _load_pv_power_watts_and_capacity_wp(
 
     # Drop any PV systems whose PV capacity is too low:
     PV_CAPACITY_THRESHOLD_W = 100
-    pv_systems_to_drop = pv_capacity_wp.index[pv_capacity_wp <= PV_CAPACITY_THRESHOLD_W]
+    pv_systems_to_drop = pv_capacity_watt_power.index[
+        pv_capacity_watt_power <= PV_CAPACITY_THRESHOLD_W
+    ]
     pv_systems_to_drop = pv_systems_to_drop.intersection(pv_power_watts.columns)
     _log.info(
         f"Dropping {len(pv_systems_to_drop)} PV systems because their max power is less than"
@@ -144,7 +151,7 @@ def _load_pv_power_watts_and_capacity_wp(
 
     # Ensure that capacity and pv_system_row_num use the same PV system IDs as the power DF:
     pv_system_ids = pv_power_watts.columns
-    pv_capacity_wp = pv_capacity_wp.loc[pv_system_ids]
+    pv_capacity_watt_power = pv_capacity_watt_power.loc[pv_system_ids]
     pv_system_row_number = pv_system_row_number.loc[pv_system_ids]
 
     _log.info(
@@ -157,58 +164,11 @@ def _load_pv_power_watts_and_capacity_wp(
     # Sanity checks:
     assert not pv_power_watts.columns.duplicated().any()
     assert not pv_power_watts.index.duplicated().any()
-    assert np.isfinite(pv_capacity_wp).all()
-    assert (pv_capacity_wp >= 0).all()
+    assert np.isfinite(pv_capacity_watt_power).all()
+    assert (pv_capacity_watt_power >= 0).all()
     assert np.isfinite(pv_system_row_number).all()
-    assert np.array_equal(pv_power_watts.columns, pv_capacity_wp.index)
-    return pv_power_watts, pv_capacity_wp, pv_system_row_number
-
-
-"""Filtering to be added in a different IterDataPipe
-
-    pv_power_watts = pv_power_watts.clip(lower=0, upper=5e7)
-    # Convert the pv_system_id column names from strings to ints:
-    pv_power_watts.columns = [np.int32(col) for col in pv_power_watts.columns]
-
-    if "passiv" not in filename:
-        _log.warning("Converting timezone. ARE YOU SURE THAT'S WHAT YOU WANT TO DO?")
-        pv_power_watts = (
-            pv_power_watts.tz_localize("Europe/London").tz_convert("UTC").tz_convert(None)
-        )
-
-    pv_power_watts = _drop_pv_systems_which_produce_overnight(pv_power_watts)
-
-    # Resample to 5-minutely and interpolate up to 15 minutes ahead.
-    # TODO: Issue #74: Give users the option to NOT resample (because Perceiver IO
-    # doesn't need all the data to be perfectly aligned).
-    pv_power_watts = pv_power_watts.resample("5T").interpolate(method="time", limit=3)
-    pv_power_watts.dropna(axis="index", how="all", inplace=True)
-    pv_power_watts.dropna(axis="columns", how="all", inplace=True)
-
-    # Drop any PV systems whose PV capacity is too low:
-    PV_CAPACITY_THRESHOLD_W = 100
-    pv_systems_to_drop = pv_capacity_wp.index[pv_capacity_wp <= PV_CAPACITY_THRESHOLD_W]
-    pv_systems_to_drop = pv_systems_to_drop.intersection(pv_power_watts.columns)
-    _log.info(
-        f"Dropping {len(pv_systems_to_drop)} PV systems because their max power is less than"
-        f" {PV_CAPACITY_THRESHOLD_W}"
-    )
-    pv_power_watts.drop(columns=pv_systems_to_drop, inplace=True)
-
-    # Ensure that capacity and pv_system_row_num use the same PV system IDs as the power DF:
-    pv_system_ids = pv_power_watts.columns
-    pv_capacity_wp = pv_capacity_wp.loc[pv_system_ids]
-    pv_system_row_number = pv_system_row_number.loc[pv_system_ids]
-
-    _log.info(
-        "After filtering & resampling to 5 minutes:"
-        f" pv_power = {pv_power_watts.values.nbytes / 1e6:,.1f} MBytes."
-        f" {len(pv_power_watts)} PV power datetimes."
-        f" {len(pv_power_watts.columns)} PV power PV system IDs."
-    )
-
-
-"""
+    assert np.array_equal(pv_power_watts.columns, pv_capacity_watt_power.index)
+    return pv_power_watts, pv_capacity_watt_power, pv_system_row_number
 
 
 # Adapted from nowcasting_dataset.data_sources.pv.pv_data_source

ocf_datapipes/load/pv/utils.py

@@ -25,7 +25,7 @@ def put_pv_data_into_an_xr_dataarray(
     pv_power_watts: pd.DataFrame,
     y_osgb: pd.Series,
     x_osgb: pd.Series,
-    capacity_wp: pd.Series,
+    capacity_watt_power: pd.Series,
     pv_system_row_number: pd.Series,
 ) -> xr.DataArray:
     """Convert to an xarray DataArray.
@@ -35,7 +35,8 @@ def put_pv_data_into_an_xr_dataarray(
             ints), and the index is UTC datetime.
         x_osgb: The x location. Index = PV system ID ints.
         y_osgb: The y location. Index = PV system ID ints.
-        capacity_wp: The max power output of each PV system in Watts. Index = PV system ID ints.
+        capacity_watt_power: The max power output of each PV system in Watts.
+         Index = PV system ID ints.
         pv_system_row_number: The integer position of the PV system in the metadata.
             Used to create the PV system ID embedding.
     """
@@ -44,7 +45,7 @@ def put_pv_data_into_an_xr_dataarray(
     for name, series in (
         ("x_osgb", x_osgb),
         ("y_osgb", y_osgb),
-        ("capacity_wp", capacity_wp),
+        ("capacity_watt_power", capacity_watt_power),
         ("pv_system_row_number", pv_system_row_number),
     ):
         logger.debug(f"Checking {name}")
@@ -64,7 +65,7 @@ def put_pv_data_into_an_xr_dataarray(
     data_array = data_array.assign_coords(
         x_osgb=("pv_system_id", x_osgb),
         y_osgb=("pv_system_id", y_osgb),
-        capacity_wp=("pv_system_id", capacity_wp),
+        capacity_watt_power=("pv_system_id", capacity_watt_power),
         pv_system_row_number=("pv_system_id", pv_system_row_number),
     )
     # Sample period duration is required so PVDownsample transform knows by how much