diff --git a/README.md b/README.md index e33dbd3..d76e4c4 100644 --- a/README.md +++ b/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| diff --git a/Tests/test_pvlive_api.py b/Tests/test_pvlive_api.py index 76758da..cff3ff1 100644 --- a/Tests/test_pvlive_api.py +++ b/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) diff --git a/pvlive_api/pvlive.py b/pvlive_api/pvlive.py index 1302432..562cc34 100644 --- a/pvlive_api/pvlive.py +++ b/pvlive_api/pvlive.py @@ -47,6 +47,14 @@ class PVLive: proxies : Optional[Dict] Optionally specify a Dict of proxies for http and https requests in the format: {"http": "
", "https": "
"} + 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, @@ -54,10 +62,11 @@ def __init__( 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,15 +176,15 @@ 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) @@ -181,6 +192,7 @@ def deployment(self, 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))}.") diff --git a/requirements_dev.txt b/requirements_dev.txt index 7eb2584..cd646c6 100644 --- a/requirements_dev.txt +++ b/requirements_dev.txt @@ -1,3 +1,4 @@ sphinx sphinx_rtd_theme numpydoc +pytest