Fix lint errors

Author: jacobbieker

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_megawatt_power: xr_data.isel(time_utc=0)["capacity_megawatt_power"].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/load/gsp.py

@@ -85,7 +85,9 @@ def __iter__(self) -> xr.DataArray:
             # 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_megawatt_power=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/pv.py

@@ -51,7 +51,11 @@ 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_watt_power, pv_system_row_number = _load_pv_power_watts_and_capacity_watt_power(
+    (
+        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)
@@ -108,7 +112,8 @@ def _load_pv_power_watts_and_capacity_watt_power(
     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_watt_power` 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.
@@ -134,7 +139,9 @@ def _load_pv_power_watts_and_capacity_watt_power(
 
     # Drop any PV systems whose PV capacity is too low:
     PV_CAPACITY_THRESHOLD_W = 100
-    pv_systems_to_drop = pv_capacity_watt_power.index[pv_capacity_watt_power <= 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"
@@ -164,53 +171,6 @@ def _load_pv_power_watts_and_capacity_watt_power(
     return pv_power_watts, pv_capacity_watt_power, 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_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"
-        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_watt_power = pv_capacity_watt_power.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."
-    )
-
-
-"""
-
-
 # Adapted from nowcasting_dataset.data_sources.pv.pv_data_source
 def _load_pv_metadata(filename: str) -> pd.DataFrame:
     """Return pd.DataFrame of PV metadata.

ocf_datapipes/load/pv/utils.py

@@ -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_watt_power: 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.
     """

ocf_datapipes/production/power_perceiver.py

@@ -111,7 +111,9 @@ def power_perceiver_production_datapipe(configuration_filename: Union[Path, str]
         )
         .fork(2)
     )
-    topo_datapipe = topo_datapipe.reproject_topography().normalize(calculate_mean_std_from_example=True)
+    topo_datapipe = topo_datapipe.reproject_topography().normalize(
+        calculate_mean_std_from_example=True
+    )
     sat_hrv_datapipe, sat_t0_datapipe = (
         sat_hrv_datapipe.convert_satellite_to_int8()
         .add_t0_idx_and_sample_period_duration(

ocf_datapipes/select/select_pv_systems_on_capacity.py

@@ -40,7 +40,8 @@ def __iter__(self) -> Union[xr.DataArray, xr.Dataset]:
 
 # Drop any PV systems whose PV capacity is too low:
     PV_CAPACITY_THRESHOLD_W = 100
-    pv_systems_to_drop = pv_capacity_watt_power.index[pv_capacity_watt_power <= 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"

tests/batch/test_merge_numpy_modalities.py

@@ -14,13 +14,17 @@ def test_merge_modalities(sat_hrv_datapipe, nwp_datapipe, gsp_datapipe, passiv_d
     batch_size = 4
 
     sat_hrv_datapipe = AddT0IdxAndSamplePeriodDuration(
-        sat_hrv_datapipe, sample_period_duration=timedelta(minutes=5), history_duration=timedelta(hours=1)
+        sat_hrv_datapipe,
+        sample_period_duration=timedelta(minutes=5),
+        history_duration=timedelta(hours=1),
     )
     sat_hrv_datapipe = ConvertSatelliteToNumpyBatch(sat_hrv_datapipe, is_hrv=True)
     sat_hrv_datapipe = MergeNumpyExamplesToBatch(sat_hrv_datapipe, n_examples_per_batch=batch_size)
 
     nwp_datapipe = AddT0IdxAndSamplePeriodDuration(
-        nwp_datapipe, sample_period_duration=timedelta(minutes=30), history_duration=timedelta(hours=1)
+        nwp_datapipe,
+        sample_period_duration=timedelta(minutes=30),
+        history_duration=timedelta(hours=1),
     )
     nwp_datapipe = ConvertNWPToNumpyBatch(nwp_datapipe)
     nwp_datapipe = MergeNumpyExamplesToBatch(nwp_datapipe, n_examples_per_batch=batch_size)
@@ -32,7 +36,9 @@ def test_merge_modalities(sat_hrv_datapipe, nwp_datapipe, gsp_datapipe, passiv_d
     gsp_datapipe = MergeNumpyExamplesToBatch(gsp_datapipe, n_examples_per_batch=batch_size)
 
     passiv_datapipe = AddT0IdxAndSamplePeriodDuration(
-        passiv_datapipe, sample_period_duration=timedelta(minutes=5), history_duration=timedelta(hours=1)
+        passiv_datapipe,
+        sample_period_duration=timedelta(minutes=5),
+        history_duration=timedelta(hours=1),
     )
     passiv_datapipe = ConvertPVToNumpyBatch(passiv_datapipe)
     passiv_datapipe = MergeNumpyExamplesToBatch(passiv_datapipe, n_examples_per_batch=batch_size)

tests/convert/test_gsp.py

@@ -7,7 +7,9 @@
 
 def test_convert_gsp_to_numpy_batch(gsp_datapipe):
     gsp_datapipe = AddT0IdxAndSamplePeriodDuration(
-        gsp_datapipe, sample_period_duration=timedelta(minutes=5), history_duration=timedelta(minutes=60)
+        gsp_datapipe,
+        sample_period_duration=timedelta(minutes=5),
+        history_duration=timedelta(minutes=60),
     )
     gsp_datapipe = ConvertGSPToNumpyBatch(gsp_datapipe)
     data = next(iter(gsp_datapipe))

tests/convert/test_nwp.py

@@ -8,7 +8,9 @@
 
 def test_convert_nwp_to_numpy_batch(nwp_datapipe):
     nwp_datapipe = AddT0IdxAndSamplePeriodDuration(
-        nwp_datapipe, sample_period_duration=timedelta(minutes=60), history_duration=timedelta(minutes=60)
+        nwp_datapipe,
+        sample_period_duration=timedelta(minutes=60),
+        history_duration=timedelta(minutes=60),
     )
     t0_datapipe = SelectLiveT0Time(nwp_datapipe, dim_name="init_time_utc")
 

tests/convert/test_satellite.py

@@ -8,7 +8,9 @@
 def test_convert_satellite_to_numpy_batch(sat_datapipe):
 
     sat_datapipe = AddT0IdxAndSamplePeriodDuration(
-        sat_datapipe, sample_period_duration=timedelta(minutes=5), history_duration=timedelta(minutes=60)
+        sat_datapipe,
+        sample_period_duration=timedelta(minutes=5),
+        history_duration=timedelta(minutes=60),
     )
     sat_datapipe = ConvertSatelliteToNumpyBatch(sat_datapipe, is_hrv=False)
     data = next(iter(sat_datapipe))
@@ -20,7 +22,9 @@ def test_convert_satellite_to_numpy_batch(sat_datapipe):
 
 def test_convert_hrvsatellite_to_numpy_batch(sat_datapipe):
     sat_datapipe = AddT0IdxAndSamplePeriodDuration(
-        sat_datapipe, sample_period_duration=timedelta(minutes=5), history_duration=timedelta(minutes=60)
+        sat_datapipe,
+        sample_period_duration=timedelta(minutes=5),
+        history_duration=timedelta(minutes=60),
     )
     sat_datapipe = ConvertSatelliteToNumpyBatch(sat_datapipe, is_hrv=True)
     data = next(iter(sat_datapipe))

tests/end2end/test_power_perceiver_production.py

@@ -46,7 +46,9 @@
 from ocf_datapipes.utils.consts import NWP_MEAN, NWP_STD, SAT_MEAN, SAT_STD, BatchKey
 
 
-def test_power_perceiver_production(sat_hrv_datapipe, passiv_datapipe, topo_datapipe, gsp_datapipe, nwp_datapipe):
+def test_power_perceiver_production(
+    sat_hrv_datapipe, passiv_datapipe, topo_datapipe, gsp_datapipe, nwp_datapipe
+):
     ####################################
     #
     # Equivalent to PP's loading and filtering methods
@@ -58,13 +60,19 @@ def test_power_perceiver_production(sat_hrv_datapipe, passiv_datapipe, topo_data
     topo_datapipe = ReprojectTopography(topo_datapipe)
     sat_datapipe = ConvertSatelliteToInt8(sat_hrv_datapipe)
     sat_datapipe = AddT0IdxAndSamplePeriodDuration(
-        sat_datapipe, sample_period_duration=timedelta(minutes=5), history_duration=timedelta(minutes=60)
+        sat_datapipe,
+        sample_period_duration=timedelta(minutes=5),
+        history_duration=timedelta(minutes=60),
     )
     pv_datapipe = AddT0IdxAndSamplePeriodDuration(
-        pv_datapipe, sample_period_duration=timedelta(minutes=5), history_duration=timedelta(minutes=60)
+        pv_datapipe,
+        sample_period_duration=timedelta(minutes=5),
+        history_duration=timedelta(minutes=60),
     )
     gsp_datapipe = AddT0IdxAndSamplePeriodDuration(
-        gsp_datapipe, sample_period_duration=timedelta(minutes=30), history_duration=timedelta(hours=2)
+        gsp_datapipe,
+        sample_period_duration=timedelta(minutes=30),
+        history_duration=timedelta(hours=2),
     )
     nwp_datapipe = AddT0IdxAndSamplePeriodDuration(
         nwp_datapipe, sample_period_duration=timedelta(hours=1), history_duration=timedelta(hours=2)
@@ -82,9 +90,14 @@ def test_power_perceiver_production(sat_hrv_datapipe, passiv_datapipe, topo_data
         3
     )  # Its in order then
     pv_datapipe = SelectSpatialSliceMeters(
-        pv_datapipe, location_datapipe=location_datapipe1, roi_width_meters=960_000, roi_height_meters=960_000
+        pv_datapipe,
+        location_datapipe=location_datapipe1,
+        roi_width_meters=960_000,
+        roi_height_meters=960_000,
     )  # Has to be large as test PV systems aren't in first 20 GSPs it seems
-    pv_datapipe, pv_t0_datapipe = EnsureNPVSystemsPerExample(pv_datapipe, n_pv_systems_per_example=8).fork(2)
+    pv_datapipe, pv_t0_datapipe = EnsureNPVSystemsPerExample(
+        pv_datapipe, n_pv_systems_per_example=8
+    ).fork(2)
     sat_datapipe, sat_t0_datapipe = SelectSpatialSlicePixels(
         sat_datapipe,
         location_datapipe=location_datapipe2,
@@ -166,7 +179,9 @@ def test_power_perceiver_production(sat_hrv_datapipe, passiv_datapipe, topo_data
     # Don't need to do NWP as it does go into the future
     nwp_datapipe = ConvertNWPToNumpyBatch(nwp_datapipe)
     nwp_datapipe = MergeNumpyExamplesToBatch(nwp_datapipe, n_examples_per_batch=4)
-    combined_datapipe = MergeNumpyModalities([gsp_datapipe, pv_datapipe, sat_datapipe, nwp_datapipe])
+    combined_datapipe = MergeNumpyModalities(
+        [gsp_datapipe, pv_datapipe, sat_datapipe, nwp_datapipe]
+    )
 
     combined_datapipe = AlignGSPto5Min(
         combined_datapipe, batch_key_for_5_min_datetimes=BatchKey.hrvsatellite_time_utc
@@ -197,7 +212,9 @@ def test_power_perceiver_production(sat_hrv_datapipe, passiv_datapipe, topo_data
     assert batch[BatchKey.hrvsatellite_surface_height].shape == (4, 128, 256)
 
 
-def test_power_perceiver_production_functional(sat_hrv_datapipe, passiv_datapipe, topo_datapipe, gsp_datapipe, nwp_datapipe):
+def test_power_perceiver_production_functional(
+    sat_hrv_datapipe, passiv_datapipe, topo_datapipe, gsp_datapipe, nwp_datapipe
+):
     ####################################
     #
     # Equivalent to PP's loading and filtering methods
@@ -220,12 +237,16 @@ def test_power_perceiver_production_functional(sat_hrv_datapipe, passiv_datapipe
             sample_period_duration=timedelta(minutes=5), history_duration=timedelta(minutes=60)
         )
         .select_spatial_slice_meters(
-            location_datapipe=location_datapipe1, roi_width_meters=960_000, roi_height_meters=960_000
+            location_datapipe=location_datapipe1,
+            roi_width_meters=960_000,
+            roi_height_meters=960_000,
         )
         .ensure_n_pv_systems_per_example(n_pv_systems_per_example=8)
         .fork(2)
     )
-    topo_datapipe = topo_datapipe.reproject_topography().normalize(calculate_mean_std_from_example=True)
+    topo_datapipe = topo_datapipe.reproject_topography().normalize(
+        calculate_mean_std_from_example=True
+    )
     sat_hrv_datapipe, sat_t0_datapipe = (
         sat_hrv_datapipe.convert_satellite_to_int8()
         .add_t0_idx_and_sample_period_duration(
@@ -262,7 +283,9 @@ def test_power_perceiver_production_functional(sat_hrv_datapipe, passiv_datapipe
     pv_t0_datapipe = pv_t0_datapipe.select_live_t0_time()
 
     gsp_datapipe = (
-        gsp_datapipe.select_live_time_slice(t0_datapipe=gsp_t0_datapipe, history_duration=timedelta(hours=2))
+        gsp_datapipe.select_live_time_slice(
+            t0_datapipe=gsp_t0_datapipe, history_duration=timedelta(hours=2)
+        )
         .gsp_iterator()
         .convert_gsp_to_numpy_batch()
         .extend_timesteps_to_future(

tests/select/test_location_picker.py

@@ -8,7 +8,9 @@
 
 def test_location_picker_single_location(gsp_datapipe):
     gsp_datapipe = AddT0IdxAndSamplePeriodDuration(
-        gsp_datapipe, sample_period_duration=timedelta(minutes=30), history_duration=timedelta(hours=1)
+        gsp_datapipe,
+        sample_period_duration=timedelta(minutes=30),
+        history_duration=timedelta(hours=1),
     )
     location_datapipe = LocationPicker(gsp_datapipe)
     data = next(iter(location_datapipe))
@@ -18,7 +20,9 @@ def test_location_picker_single_location(gsp_datapipe):
 def test_location_picker_all_locations(gsp_datapipe):
     dataset = next(iter(gsp_datapipe))
     gsp_datapipe = AddT0IdxAndSamplePeriodDuration(
-        gsp_datapipe, sample_period_duration=timedelta(minutes=30), history_duration=timedelta(hours=1)
+        gsp_datapipe,
+        sample_period_duration=timedelta(minutes=30),
+        history_duration=timedelta(hours=1),
     )
     location_datapipe = LocationPicker(gsp_datapipe, return_all_locations=True)
     loc_iterator = iter(location_datapipe)

tests/select/test_select_live_time_slice.py

@@ -18,7 +18,9 @@ def test_select_hrv(sat_hrv_datapipe):
 def test_select_gsp(gsp_datapipe):
     time_len = len(next(iter(gsp_datapipe)).time_utc.values)
     t0_datapipe = SelectLiveT0Time(gsp_datapipe, dim_name="time_utc")
-    gsp_datapipe = SelectLiveTimeSlice(gsp_datapipe, history_duration=timedelta(minutes=120), t0_datapipe=t0_datapipe)
+    gsp_datapipe = SelectLiveTimeSlice(
+        gsp_datapipe, history_duration=timedelta(minutes=120), t0_datapipe=t0_datapipe
+    )
     data = next(iter(gsp_datapipe))
     assert len(data.time_utc.values) == 5
     assert len(data.time_utc.values) < time_len

tests/select/test_select_spatial_slice.py

@@ -4,7 +4,10 @@
 def test_select_spatial_slice_meters_passiv(passiv_datapipe):
     loc_datapipe = LocationPicker(passiv_datapipe)
     passiv_datapipe = SelectSpatialSliceMeters(
-        passiv_datapipe, location_datapipe=loc_datapipe, roi_width_meters=96_000, roi_height_meters=96_000
+        passiv_datapipe,
+        location_datapipe=loc_datapipe,
+        roi_width_meters=96_000,
+        roi_height_meters=96_000,
     )
     data = next(iter(passiv_datapipe))
     assert len(data.pv_system_id) == 1