Merge pull request #58 from SheffieldSolar/56-update-capacity-methods-to-use-the-new-capacity-api-endpoints

Update capacity methods to use the new capacity endpoints form the PV_Live API

Author: JamieTaylor-TUOS

README.md

@@ -18,6 +18,8 @@ A Python implementation of the PV_Live web API. See https://www.solar.sheffield.
 
 ## Usage
 
+As of 2025-07-07, the production PV_Live API is hosted on Google Cloud Platform (GCP) at https://api.pvlive.uk. There is a also non-prod test/fix-on-fail (FOF) environment hosted on TUOS IT: https://api.solar.sheffield.ac.uk. To support switching between the two, the `pvlive-api` package exposes a parameter `domain_url`, which can be set to one of `["api.pvlive.uk", "api.solar.sheffield.ac.uk"]` but defaults to `api.pvlive.uk`. 
+
 There are three methods for extracting raw data from the PV_Live API:
 
 |Method|Description|Docs Link|

Tests/test_pvlive_api.py

@@ -133,25 +133,6 @@ def test_latest(self):
         )
         self.check_df_columns(data)
         self.check_df_dtypes(data)
-        data = self.api.latest(
-            entity_type="pes",
-            entity_id=0,
-            period=5
-        )
-        self.check_pes_tuple(data)
-        self.check_pes_tuple_dtypes(data)
-        data = self.api.latest(entity_type="pes", entity_id=0, period=5, dataframe=True)
-        self.check_df_columns(data)
-        self.check_df_dtypes(data)
-        data = self.api.latest(
-            entity_type="pes",
-            entity_id=0,
-            extra_fields="ucl_mw,lcl_mw,installedcapacity_mwp,stats_error",
-            period=5,
-            dataframe=True
-        )
-        self.check_df_columns(data)
-        self.check_df_dtypes(data)
         data = self.api.latest(entity_type="gsp", entity_id=103)
         self.check_gsp_tuple(data)
         self.check_gsp_tuple_dtypes(data)
@@ -175,27 +156,6 @@ def test_day_peak(self):
             entity_id=0,
             dataframe=True
         )
-        data = self.api.day_peak(d=test_date, entity_type="pes", entity_id=0, period=5)
-        self.check_pes_tuple(data)
-        self.check_pes_tuple_dtypes(data)
-        data = self.api.day_peak(
-            d=test_date,
-            entity_type="pes",
-            entity_id=0,
-            period=5,
-            dataframe=True
-        )
-        self.check_df_columns(data)
-        self.check_df_dtypes(data)
-        data = self.api.day_peak(
-            d=test_date,
-            extra_fields="ucl_mw,lcl_mw,installedcapacity_mwp,stats_error",
-            entity_type="pes",
-            entity_id=0,
-            period=5,
-            dataframe=True
-        )
-        self.check_df_dtypes(data)
         data = self.api.day_peak(d=test_date, entity_type="gsp", entity_id=54)
         self.check_gsp_tuple(data)
         self.check_gsp_tuple_dtypes(data)
@@ -235,25 +195,6 @@ def test_between(self):
         )
         self.check_df_columns(data)
         self.check_df_dtypes(data)
-        data = self.api.between(
-            start=get_test_time(0, 0),
-            end=get_test_time(14, 0),
-            entity_type="pes",
-            entity_id=0,
-            period=5
-        )
-        with self.subTest():
-            assert isinstance(data, list)
-        data = self.api.between(
-            start=get_test_time(12, 20),
-            end=get_test_time(14, 0),
-            entity_type="pes",
-            entity_id=0,
-            period=5,
-            dataframe=True
-        )
-        self.check_df_columns(data)
-        self.check_df_dtypes(data)
 
     def test_at_time(self):
         """Test the at_time function."""
@@ -272,21 +213,6 @@ def test_at_time(self):
             dataframe=True
         )
         self.check_df_dtypes(data)
-        data = self.api.at_time(dt=test_time, entity_type="pes", entity_id=0, period=5)
-        self.check_pes_tuple(data)
-        self.check_pes_tuple_dtypes(data)
-        data = self.api.at_time(test_time, entity_type="pes", entity_id=0, period=5, dataframe=True)
-        self.check_df_columns(data)
-        self.check_df_dtypes(data)
-        data = self.api.at_time(
-            test_time,
-            entity_type="pes",
-            entity_id=0,
-            extra_fields="ucl_mw,lcl_mw,installedcapacity_mwp,stats_error",
-            period=5,
-            dataframe=True
-        )
-        self.check_df_dtypes(data)
         data = self.api.at_time(dt=test_time, entity_type="gsp", entity_id=26)
         self.check_gsp_tuple(data)
         self.check_gsp_tuple_dtypes(data)

pvlive_api/pvlive.py

@@ -47,17 +47,26 @@ class PVLive:
     proxies : Optional[Dict]
         Optionally specify a Dict of proxies for http and https requests in the format:
         {"http": "<address>", "https": "<address>"}
+    ssl_verify : Bool
+        Set to False to disable SSL cert checks when querying the API. Useful if e.g. your company
+        network/VPN has issues propagating SSL certs. Defaults to True. Not recommended to set to
+        False!
+    domain_url : Literal["api.pvlive.uk", "api.solar.sheffield.ac.uk", "api0.solar.sheffield.ac.uk"]
+        Used to switch between the production PV_Live API on GCP and the test/FOF server on-prem.
+    gsp_boundaries_version : str
+        Some capacity datasets (by GSP) are encoded with the version of the GSP boundaries used.
     """
     def __init__(
         self,
         retries: int = 3,
         proxies: Optional[Dict] = None,
         ssl_verify: bool = True,
         domain_url: Literal[
-            "api0.solar.sheffield.ac.uk",
+            "api.pvlive.uk",
             "api.solar.sheffield.ac.uk",
-            "api.pvlive.uk"
-        ] = "api.solar.sheffield.ac.uk"
+            "api0.solar.sheffield.ac.uk",
+        ] = "api.pvlive.uk",
+        gsp_boundaries_version: str = "20250109",
     ):
         valid_domain_urls = [
             "api0.solar.sheffield.ac.uk",
@@ -79,7 +88,9 @@ def __init__(
         self.pes_list = self._get_pes_list()
         self.gsp_ids = self.gsp_list.gsp_id.dropna().astype(int64).unique()
         self.pes_ids = self.pes_list.pes_id.dropna().astype(int64).unique()
+        self.deployment_datasets = None
         self.deployment_releases = None
+        self.gsp_boundaries_version = gsp_boundaries_version
 
     def _get_gsp_list(self):
         """Fetch the GSP list from the API and convert to Pandas DataFrame."""
@@ -94,15 +105,15 @@ def _get_pes_list(self):
         return pd.DataFrame(response["data"], columns=response["meta"])
 
     def _get_deployment_releases(self):
-        """Get a list of deployment releases as datestamps (YYYYMMDD)."""
-        if self.deployment_releases is None:
+        """
+        Get the available deployment releases as a list of datestamps (YYYYMMDD) alongside a dict of
+        files available in each release.
+        """
+        if self.deployment_datasets is None:
             url = f"{self.domain_url}/capacity/"
-            response = self._fetch_url(url, parse_json=False)
-            soup = BeautifulSoup(response.content, "html.parser")
-            releases = [r["href"].strip("/") for r in soup.find_all("a", href=True)
-                        if re.match(r"[0-9]{8}/", r["href"])]
-            self.deployment_releases = sorted(releases, reverse=True)
-        return self.deployment_releases
+            self.deployment_datasets = self._fetch_url(url, parse_json=True)
+            self.deployment_releases = sorted(list(self.deployment_datasets.keys()), reverse=True)
+        return self.deployment_datasets, self.deployment_releases
 
     def _get_deployment_filenames(self, release):
         """Get a list of filenames for a given release."""
@@ -115,7 +126,7 @@ def _get_deployment_filenames(self, release):
 
     def _validate_deployment_inputs(self, region, include_history, by_system_size, release):
         """Validate input parameters to `deployment()`."""
-        releases = self._get_deployment_releases()
+        deployment_datasets, releases = self._get_deployment_releases()
         if not isinstance(region, str):
             raise TypeError("`region` must be a string.")
         supported_regions = ["gsp", "llsoa"]
@@ -165,22 +176,23 @@ def deployment(self,
             GSPs/llsoa and dc_capacity_mwp.
         """
         self._validate_deployment_inputs(region, include_history, by_system_size, release)
-        releases = self._get_deployment_releases()
+        deployment_datasets, releases = self._get_deployment_releases()
         release = releases[release] if isinstance(release, int) else release
-        filenames = self._get_deployment_filenames(release)
-        region_ = "20220314_GSP" if region == "gsp" else region
+        filenames = list(deployment_datasets[release].keys())
+        region_ = f"{self.gsp_boundaries_version}_GSP" if region == "gsp" else region
         history_ = "_and_month" if include_history else ""
         system_size_ = "_and_system_size" if by_system_size else ""
         filename_ending = f"_capacity_by_{region_}{history_}{system_size_}.csv.gz"
         filename = [f for f in filenames if f.endswith(filename_ending)][0]
-        url = f"{self.domain_url}/capacity/{release}/{filename}"
+        url = deployment_datasets[release][filename]
         kwargs = dict(parse_dates=["install_month"]) if include_history else {}
         response = self._fetch_url(url, parse_json=False)
         mock_file = BytesIO(response.content)
         deployment_data = pd.read_csv(mock_file, compression={"method": "gzip"}, **kwargs)
         deployment_data.insert(0, "release", release)
         deployment_data.rename(columns={"dc_capacity_MWp": "dc_capacity_mwp"}, inplace=True)
         deployment_data.system_count = deployment_data.system_count.astype("Int64")
+        deployment_data.dropna(how="any", inplace=True)
         return deployment_data
 
     def latest(self,
@@ -532,7 +544,7 @@ def _validate_inputs(self, entity_type="gsp", entity_id=0, extra_fields="", peri
         elif entity_type == "gsp":
             if entity_id not in self.gsp_ids:
                 raise PVLiveException(f"The gsp_id {entity_id} was not found.")
-        periods = [5, 30]
+        periods = [30]
         if period not in periods:
             raise ValueError("The period parameter must be one of: "
                              f"{', '.join(map(str, periods))}.")

requirements_dev.txt

@@ -1,3 +1,4 @@
 sphinx
 sphinx_rtd_theme
 numpydoc
+pytest