From 7b9d9737b4daea0f99fcf01853f28eef171bf9f8 Mon Sep 17 00:00:00 2001 From: sam-watttime Date: Fri, 22 Dec 2023 14:18:36 -0700 Subject: [PATCH 1/6] README updates for /register --- README.md | 13 +++++++++++-- tests/test_sdk.py | 41 +++++++++++++++++++++++++++++++++++++++-- watttime/api.py | 13 +++++++++++++ 3 files changed, 63 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index b61a0ba8..d6e516eb 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # About This SDK is meant to help users with basic queries to WattTime’s API (version 3), and to get data returned in specific formats (e.g., JSON, pandas, csv). -Users must first [register for access to the WattTime API here](https://watttime.org/docs-dev/data-plans/). +Users may register for access to the WattTime API through this client, however the basic user scoping given will only allow newly registered users to access data for the `CAISO_NORTH` region. Additionally, data may not be available for all signal types for newly registered users. Full documentation of WattTime's API, along with response samples and information about [available endpoints is also available](https://docs.watttime.org/). @@ -11,7 +11,16 @@ The SDK can be installed as a python package from the PyPi repository, we recomm pip install watttime ``` -Once registered for the WattTime API, you may set your credentials as environment variables to avoid passing these during class initialization: +If you are not registered for the WattTime API, you can do so using the SDK: +``` +from watttime import WattTimeMyAccess + +wt = WattTimeMyAccess(username=, password=) +wt.register(email=, organization=) + +``` + +If you are already registered for the WattTime API, you may set your credentials as environment variables to avoid passing these during class initialization: ``` # linux or mac export WATTTIME_USER= diff --git a/tests/test_sdk.py b/tests/test_sdk.py index eb596836..3017a8f0 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -1,7 +1,9 @@ import unittest +import unittest.mock as mock from datetime import datetime, timedelta from dateutil.parser import parse from pytz import timezone, UTC +import os from watttime import ( WattTimeBase, WattTimeHistorical, @@ -16,6 +18,30 @@ REGION = "CAISO_NORTH" +def mocked_register(*args, **kwargs): + url = args[0] + + class MockResponse: + def __init__(self, json_data, status_code): + self.json_data = json_data + self.status_code = status_code + + def json(self): + return self.json_data + + if ( + (url == "https://api.watttime.org/register") + & (kwargs["json"]["email"] == os.getenv("WATTTIME_EMAIL")) + & (kwargs["json"]["username"] == os.getenv("WATTTIME_USER")) + & (kwargs["json"]["password"] == os.getenv("WATTTIME_PASSWORD")) + ): + return MockResponse( + {"ok": "User created", "user": kwargs["json"]["username"]}, 200 + ) + else: + raise MockResponse({"error": "Failed to create user"}, 400) + + class TestWattTimeBase(unittest.TestCase): def setUp(self): self.base = WattTimeBase() @@ -91,6 +117,12 @@ def test_parse_dates_with_datetime(self): self.assertIsInstance(parsed_end, datetime) self.assertEqual(parsed_end.tzinfo, UTC) + @mock.patch("requests.post", side_effect=mocked_register) + def test_mock_register(self, mock_post): + resp = self.base.register(email=os.getenv("WATTTIME_EMAIL")) + self.assertEqual(resp.status_code, 200) + self.assertEqual(len(mock_post.call_args_list), 1) + class TestWattTimeHistorical(unittest.TestCase): def setUp(self): @@ -154,16 +186,21 @@ def test_get_historical_pandas_meta(self): self.assertIn("point_time", df.columns) self.assertIn("value", df.columns) self.assertIn("meta", df.columns) - + def test_get_historical_csv(self): start = parse("2022-01-01 00:00Z") end = parse("2022-01-02 00:00Z") self.historical.get_historical_csv(start, end, REGION) - fp = Path.home() / "watttime_historical_csvs" / f"{REGION}_co2_moer_{start.date()}_{end.date()}.csv" + fp = ( + Path.home() + / "watttime_historical_csvs" + / f"{REGION}_co2_moer_{start.date()}_{end.date()}.csv" + ) assert fp.exists() fp.unlink() + class TestWattTimeMyAccess(unittest.TestCase): def setUp(self): self.access = WattTimeMyAccess() diff --git a/watttime/api.py b/watttime/api.py index b1f7a53c..6cbe7236 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -70,6 +70,19 @@ def _get_chunks( # API response is inclusive, avoid overlap in chunks chunks = [(s, e - timedelta(minutes=5)) for s, e in chunks[0:-1]] + [chunks[-1]] return chunks + + def register(self, email: str, organization: Optional[str] = None) -> requests.Response: + """Register for the WattTime API, if you do not already have an account.""" + url = f"{self.url_base}/register" + params = { + 'username': self.username, + 'password': self.password, + 'email': email, + 'org': organization, + } + + rsp = requests.post(url, json=params, timeout=20) + return rsp class WattTimeHistorical(WattTimeBase): From 00a0cff03ad18aca6e1eb22d775b55a4005218ab Mon Sep 17 00:00:00 2001 From: sam-watttime Date: Fri, 22 Dec 2023 15:22:31 -0700 Subject: [PATCH 2/6] add maps and region-from-loc methods --- README.md | 29 +++++++++++++++- tests/test_sdk.py | 39 ++++++++++++++++++++++ watttime/api.py | 84 ++++++++++++++++++++++++++++++++++++++++------- 3 files changed, 140 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index d6e516eb..2dfbfa22 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,9 @@ wt_myaccess.get_access_json() wt_myaccess.get_access_pandas() ``` -Once you confirm your access, you may wish to request data for a particular balancing authority: +### Accessing Historical Data + +Once you confirm your access, you may wish to request data for a particular region: ```python from watttime import WattTimeHistorical @@ -95,6 +97,7 @@ for region in moer_regions: moers = pd.concat([moers, region_df], axis='rows') ``` +### Accessing Real-Time and Historical Forecasts You can also use the SDK to request a current forecast for some signal types, such as co2_moer and health_damage: ```python @@ -107,6 +110,7 @@ forecast = wt_forecast.get_forecast_json( ) ``` +We recommend using the `WattTimeForecast` class to access data for real-time optimization. The first item of the response from this call is always guaranteed to be an estimate of the signal_type for the current five minute period, and forecasts extend at least 24 hours at a five minute granularity, which is useful for scheduling utilization during optimal times. Methods also exist to request historical forecasts, however these responses may be slower as the volume of data can be significant: ```python @@ -117,3 +121,26 @@ hist_forecasts = wt_forecast.get_historical_forecast_json( signal_type = 'health_damage' ) ``` + +### Accessing Location Data +We provide two methods to access location data: + +1) The `region_from_loc()` method allows users to provide a latitude and longitude coordinates in order to receive the valid region for a given signal type. + +2) the `WattTimeMaps` class provides a `get_maps_json()` method which returns a [GeoJSON](https://en.wikipedia.org/wiki/GeoJSON) object with complete boundaries for all regions available for a given signal type. Note that access to this endpoint is only available for Pro and Analyst subscribers. + +```python +from watttime import WattTimeMaps + +wt = WattTimeMaps() + +# get BA region for a given location +wt.region_from_loc( + latitude=39.7522, + longitude=-105.0, + signal_type='co2_moer' +) + +# get shape files for all regions of a signal type +wt.get_maps_json('co2_moer') +``` \ No newline at end of file diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 3017a8f0..0e6b833f 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -9,6 +9,7 @@ WattTimeHistorical, WattTimeMyAccess, WattTimeForecast, + WattTimeMaps, ) from pathlib import Path @@ -301,5 +302,43 @@ def test_historical_forecast_pandas(self): self.assertIn("generated_at", df.columns) +class TestWattTimeMaps(unittest.TestCase): + def setUp(self): + self.maps = WattTimeMaps() + + def test_get_maps_json_moer(self): + moer = self.maps.get_maps_json(signal_type="co2_moer") + self.assertEqual(moer["type"], "FeatureCollection") + self.assertEqual(moer["meta"]["signal_type"], "co2_moer") + self.assertGreater( + parse(moer["meta"]["last_updated"]), parse("2023-01-01 00:00Z") + ) + self.assertGreater(len(moer["features"]), 100) # 172 as of 2023-12-01 + + def test_get_maps_json_aoer(self): + aoer = self.maps.get_maps_json(signal_type="co2_aoer") + self.assertEqual(aoer["type"], "FeatureCollection") + self.assertEqual(aoer["meta"]["signal_type"], "co2_aoer") + self.assertGreater( + parse(aoer["meta"]["last_updated"]), parse("2023-01-01 00:00Z") + ) + self.assertGreater(len(aoer["features"]), 50) # 87 as of 2023-12-01 + + def test_get_maps_json_health(self): + health = self.maps.get_maps_json(signal_type="health_damage") + self.assertEqual(health["type"], "FeatureCollection") + self.assertEqual(health["meta"]["signal_type"], "health_damage") + self.assertGreater( + parse(health["meta"]["last_updated"]), parse("2022-01-01 00:00Z") + ) + self.assertGreater(len(health["features"]), 100) # 114 as of 2023-12-01 + + def test_region_from_loc(self): + region = self.maps.region_from_loc(latitude=39.7522, longitude=-105.0, signal_type='co2_moer') + self.assertEqual(region["region"], "PSCO") + self.assertEqual(region["region_full_name"], "Public Service Co of Colorado") + self.assertEqual(region["signal_type"], "co2_moer") + + if __name__ == "__main__": unittest.main() diff --git a/watttime/api.py b/watttime/api.py index 6cbe7236..0ace6a67 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -5,6 +5,7 @@ import os import time from pathlib import Path +from functools import cache import requests import pandas as pd @@ -70,20 +71,52 @@ def _get_chunks( # API response is inclusive, avoid overlap in chunks chunks = [(s, e - timedelta(minutes=5)) for s, e in chunks[0:-1]] + [chunks[-1]] return chunks - - def register(self, email: str, organization: Optional[str] = None) -> requests.Response: + + def register( + self, email: str, organization: Optional[str] = None + ) -> requests.Response: """Register for the WattTime API, if you do not already have an account.""" url = f"{self.url_base}/register" params = { - 'username': self.username, - 'password': self.password, - 'email': email, - 'org': organization, + "username": self.username, + "password": self.password, + "email": email, + "org": organization, } - + rsp = requests.post(url, json=params, timeout=20) return rsp + @cache + def region_from_loc( + self, + latitude: Union[str, float], + longitude: Union[str, float], + signal_type: Optional[ + Literal["co2_moer", "co2_aoer", "health_damage"] + ] = "co2_moer" + ) -> Dict[str, str]: + """ + Retrieve the region information based on the given latitude and longitude. + + Args: + latitude (Union[str, float]): The latitude of the location. + longitude (Union[str, float]): The longitude of the location. + signal_type (Optional[Literal["co2_moer", "co2_aoer", "health_damage"]], optional): + The type of signal to be used for the region classification. + Defaults to "co2_moer". + + Returns: + Dict[str, str]: A dictionary containing the region information with keys "region" and "region_full_name". + """ + if not self._is_token_valid(): + self._login() + url = f"{self.url_base}/v3/region-from-loc" + headers = {"Authorization": "Bearer " + self.token} + params = {"latitude": str(latitude), "longitude": str(longitude), "signal_type":signal_type} + rsp = requests.get(url, headers=headers, params=params) + return rsp.json() + class WattTimeHistorical(WattTimeBase): def get_historical_jsons( @@ -196,10 +229,10 @@ def get_historical_csv( None """ df = self.get_historical_pandas(start, end, region, signal_type, model_date) - + out_dir = Path.home() / "watttime_historical_csvs" out_dir.mkdir(exist_ok=True) - + start, end = self._parse_dates(start, end) fp = out_dir / f"{region}_{signal_type}_{start.date()}_{end.date()}.csv" df.to_csv(fp, index=False) @@ -317,7 +350,9 @@ def get_forecast_pandas( pd.DataFrame: _description_ """ j = self.get_forecast_json(region, signal_type, model_date) - return pd.json_normalize(j, record_path="data", meta=["meta"] if include_meta else []) + return pd.json_normalize( + j, record_path="data", meta=["meta"] if include_meta else [] + ) def get_historical_forecast_json( self, @@ -373,8 +408,35 @@ def get_historical_forecast_pandas( ) out = pd.DataFrame() for json in json_list: - for entry in json['data']: + for entry in json["data"]: _df = pd.json_normalize(entry, record_path=["forecast"]) _df = _df.assign(generated_at=pd.to_datetime(entry["generated_at"])) out = pd.concat([out, _df]) return out + + +class WattTimeMaps(WattTimeBase): + def get_maps_json( + self, + signal_type: Optional[ + Literal["co2_moer", "co2_aoer", "health_damage"] + ] = "co2_moer", + ): + """ + Retrieves JSON data for the maps API. + + Args: + signal_type (Optional[str]): The type of signal to retrieve data for. + Valid options are "co2_moer", "co2_aoer", and "health_damage". + Defaults to "co2_moer". + + Returns: + dict: The JSON response from the API. + """ + if not self._is_token_valid(): + self._login() + url = "{}/v3/maps".format(self.url_base) + headers = {"Authorization": "Bearer " + self.token} + params = {"signal_type": signal_type} + rsp = requests.get(url, headers=headers, params=params) + return rsp.json() From 7f5b3bb65617d4ceef87a0bad8db8afa8bb38e79 Mon Sep 17 00:00:00 2001 From: sam-watttime Date: Fri, 22 Dec 2023 15:28:42 -0700 Subject: [PATCH 3/6] no response from register() needed --- tests/test_sdk.py | 4 +++- watttime/api.py | 32 +++++++++++++++++++------------- 2 files changed, 22 insertions(+), 14 deletions(-) diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 3017a8f0..49e70c20 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -28,6 +28,9 @@ def __init__(self, json_data, status_code): def json(self): return self.json_data + + def raise_for_status(self): + assert self.status_code == 200 if ( (url == "https://api.watttime.org/register") @@ -120,7 +123,6 @@ def test_parse_dates_with_datetime(self): @mock.patch("requests.post", side_effect=mocked_register) def test_mock_register(self, mock_post): resp = self.base.register(email=os.getenv("WATTTIME_EMAIL")) - self.assertEqual(resp.status_code, 200) self.assertEqual(len(mock_post.call_args_list), 1) diff --git a/watttime/api.py b/watttime/api.py index 6cbe7236..08f18629 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -70,19 +70,23 @@ def _get_chunks( # API response is inclusive, avoid overlap in chunks chunks = [(s, e - timedelta(minutes=5)) for s, e in chunks[0:-1]] + [chunks[-1]] return chunks - - def register(self, email: str, organization: Optional[str] = None) -> requests.Response: - """Register for the WattTime API, if you do not already have an account.""" + + def register( + self, email: str, organization: Optional[str] = None + ) -> requests.Response: url = f"{self.url_base}/register" params = { - 'username': self.username, - 'password': self.password, - 'email': email, - 'org': organization, + "username": self.username, + "password": self.password, + "email": email, + "org": organization, } - + rsp = requests.post(url, json=params, timeout=20) - return rsp + rsp.raise_for_status() + print( + f"Successfully registered {self.username}, please check {email} for a verification email" + ) class WattTimeHistorical(WattTimeBase): @@ -196,10 +200,10 @@ def get_historical_csv( None """ df = self.get_historical_pandas(start, end, region, signal_type, model_date) - + out_dir = Path.home() / "watttime_historical_csvs" out_dir.mkdir(exist_ok=True) - + start, end = self._parse_dates(start, end) fp = out_dir / f"{region}_{signal_type}_{start.date()}_{end.date()}.csv" df.to_csv(fp, index=False) @@ -317,7 +321,9 @@ def get_forecast_pandas( pd.DataFrame: _description_ """ j = self.get_forecast_json(region, signal_type, model_date) - return pd.json_normalize(j, record_path="data", meta=["meta"] if include_meta else []) + return pd.json_normalize( + j, record_path="data", meta=["meta"] if include_meta else [] + ) def get_historical_forecast_json( self, @@ -373,7 +379,7 @@ def get_historical_forecast_pandas( ) out = pd.DataFrame() for json in json_list: - for entry in json['data']: + for entry in json["data"]: _df = pd.json_normalize(entry, record_path=["forecast"]) _df = _df.assign(generated_at=pd.to_datetime(entry["generated_at"])) out = pd.concat([out, _df]) From 952fd120dedc05762bf996fbc43b7d6a1727926d Mon Sep 17 00:00:00 2001 From: sam-watttime Date: Fri, 22 Dec 2023 15:34:58 -0700 Subject: [PATCH 4/6] add missing docstrings --- watttime/api.py | 171 +++++++++++++++++++++++++++++++++++------------- 1 file changed, 124 insertions(+), 47 deletions(-) diff --git a/watttime/api.py b/watttime/api.py index 0ace6a67..7621da04 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -15,13 +15,26 @@ class WattTimeBase: url_base = "https://api.watttime.org" def __init__(self, username: Optional[str] = None, password: Optional[str] = None): + """ + Initializes a new instance of the class. + + Parameters: + username (Optional[str]): The username to use for authentication. If not provided, the value will be retrieved from the environment variable "WATTTIME_USER". + password (Optional[str]): The password to use for authentication. If not provided, the value will be retrieved from the environment variable "WATTTIME_PASSWORD". + """ self.username = username or os.getenv("WATTTIME_USER") self.password = password or os.getenv("WATTTIME_PASSWORD") self.token = None self.token_valid_until = None def _login(self): - """Login to the WattTime API, which provides a JWT valid for 30 minutes.""" + """ + Login to the WattTime API, which provides a JWT valid for 30 minutes + + Raises: + Exception: If the login fails and the credentials are incorrect. + """ + url = f"{self.url_base}/login" rsp = requests.get( url, @@ -41,6 +54,16 @@ def _is_token_valid(self) -> bool: def _parse_dates( self, start: Union[str, datetime], end: Union[str, datetime] ) -> Tuple[datetime, datetime]: + """ + Parse the given start and end dates. + + Args: + start (Union[str, datetime]): The start date to parse. It can be either a string or a datetime object. + end (Union[str, datetime]): The end date to parse. It can be either a string or a datetime object. + + Returns: + Tuple[datetime, datetime]: A tuple containing the parsed start and end dates as datetime objects. + """ if isinstance(start, str): start = parse(start) if isinstance(end, str): @@ -61,7 +84,17 @@ def _parse_dates( def _get_chunks( self, start: datetime, end: datetime, chunk_size: timedelta = timedelta(days=30) ) -> List[Tuple[datetime, datetime]]: - """Internal function turns a start and end into 30-day chunks""" + """ + Generate a list of tuples representing chunks of time within a given time range. + + Args: + start (datetime): The start datetime of the time range. + end (datetime): The end datetime of the time range. + chunk_size (timedelta, optional): The size of each chunk. Defaults to timedelta(days=30). + + Returns: + List[Tuple[datetime, datetime]]: A list of tuples representing the chunks of time. + """ chunks = [] while start < end: chunk_end = min(end, start + chunk_size) @@ -72,10 +105,18 @@ def _get_chunks( chunks = [(s, e - timedelta(minutes=5)) for s, e in chunks[0:-1]] + [chunks[-1]] return chunks - def register( - self, email: str, organization: Optional[str] = None - ) -> requests.Response: - """Register for the WattTime API, if you do not already have an account.""" + def register(self, email: str, organization: Optional[str] = None) -> None: + """ + Register a user with the given email and organization. + + Parameters: + email (str): The email of the user. + organization (Optional[str], optional): The organization the user belongs to. Defaults to None. + + Returns: + None: An error will be raised if registration was unsuccessful. + """ + url = f"{self.url_base}/register" params = { "username": self.username, @@ -94,18 +135,18 @@ def region_from_loc( longitude: Union[str, float], signal_type: Optional[ Literal["co2_moer", "co2_aoer", "health_damage"] - ] = "co2_moer" + ] = "co2_moer", ) -> Dict[str, str]: """ Retrieve the region information based on the given latitude and longitude. - + Args: latitude (Union[str, float]): The latitude of the location. longitude (Union[str, float]): The longitude of the location. signal_type (Optional[Literal["co2_moer", "co2_aoer", "health_damage"]], optional): The type of signal to be used for the region classification. Defaults to "co2_moer". - + Returns: Dict[str, str]: A dictionary containing the region information with keys "region" and "region_full_name". """ @@ -113,7 +154,11 @@ def region_from_loc( self._login() url = f"{self.url_base}/v3/region-from-loc" headers = {"Authorization": "Bearer " + self.token} - params = {"latitude": str(latitude), "longitude": str(longitude), "signal_type":signal_type} + params = { + "latitude": str(latitude), + "longitude": str(longitude), + "signal_type": signal_type, + } rsp = requests.get(url, headers=headers, params=params) return rsp.json() @@ -185,7 +230,8 @@ def get_historical_pandas( model_date: Optional[Union[str, date]] = None, include_meta: bool = False, ): - """Return a pd.DataFrame with point_time, and values. + """ + Return a pd.DataFrame with point_time, and values. Args: See .get_hist_jsons() for shared arguments. @@ -215,18 +261,18 @@ def get_historical_csv( model_date: Optional[Union[str, date]] = None, ): """ - Retrieves historical data from a specified start date to an end date and saves it as a CSV file. - CSV naming scheme is like "CAISO_NORTH_co2_moer_2022-01-01_2022-01-07.csv" - + - Args: - start (Union[str, datetime]): The start date for retrieving historical data. It can be a string in the format "YYYY-MM-DD" or a datetime object. - end (Union[str, datetime]): The end date for retrieving historical data. It can be a string in the format "YYYY-MM-DD" or a datetime object. - region (str): The region for which historical data is requested. - signal_type (Optional[Literal["co2_moer", "co2_aoer", "health_damage"]]): The type of signal for which historical data is requested. Default is "co2_moer". - model_date (Optional[Union[str, date]]): The date of the model for which historical data is requested. It can be a string in the format "YYYY-MM-DD" or a date object. Default is None. + Retrieves historical data from a specified start date to an end date and saves it as a CSV file. + CSV naming scheme is like "CAISO_NORTH_co2_moer_2022-01-01_2022-01-07.csv" + + Args: + start (Union[str, datetime]): The start date for retrieving historical data. It can be a string in the format "YYYY-MM-DD" or a datetime object. + end (Union[str, datetime]): The end date for retrieving historical data. It can be a string in the format "YYYY-MM-DD" or a datetime object. + region (str): The region for which historical data is requested. + signal_type (Optional[Literal["co2_moer", "co2_aoer", "health_damage"]]): The type of signal for which historical data is requested. Default is "co2_moer". + model_date (Optional[Union[str, date]]): The date of the model for which historical data is requested. It can be a string in the format "YYYY-MM-DD" or a date object. Default is None. - Returns: - None + Returns: + None, results are saved to a csv file in the user's home directory. """ df = self.get_historical_pandas(start, end, region, signal_type, model_date) @@ -338,7 +384,8 @@ def get_forecast_pandas( model_date: Optional[Union[str, date]] = None, include_meta: bool = False, ) -> pd.DataFrame: - """Return a pd.DataFrame with point_time, and values. + """ + Return a pd.DataFrame with point_time, and values. Args: See .get_forecast_json() for shared arguments. @@ -364,6 +411,22 @@ def get_historical_forecast_json( ] = "co2_moer", model_date: Optional[Union[str, date]] = None, ) -> List[Dict[str, Any]]: + """ + Retrieves the historical forecast data from the API as a list of dictionaries. + + Args: + start (Union[str, datetime]): The start date of the historical forecast. Can be a string or a datetime object. + end (Union[str, datetime]): The end date of the historical forecast. Can be a string or a datetime object. + region (str): The region for which to retrieve the forecast data. + signal_type (Optional[Literal["co2_moer", "co2_aoer", "health_damage"]]): The type of signal to retrieve. Defaults to "co2_moer". + model_date (Optional[Union[str, date]]): The date of the model version to use. Defaults to None. + + Returns: + List[Dict[str, Any]]: A list of dictionaries representing the forecast data. + + Raises: + Exception: If there is an API response error. + """ if not self._is_token_valid(): self._login() url = "{}/v3/forecast/historical".format(self.url_base) @@ -403,6 +466,20 @@ def get_historical_forecast_pandas( ] = "co2_moer", model_date: Optional[Union[str, date]] = None, ) -> pd.DataFrame: + """ + Retrieves the historical forecast data as a pandas DataFrame. + + Args: + start (Union[str, datetime]): The start date or datetime for the historical forecast. + end (Union[str, datetime]): The end date or datetime for the historical forecast. + region (str): The region for which the historical forecast data is retrieved. + signal_type (Optional[Literal["co2_moer", "co2_aoer", "health_damage"]], optional): + The type of signal for the historical forecast data. Defaults to "co2_moer". + model_date (Optional[Union[str, date]], optional): The model date for the historical forecast data. Defaults to None. + + Returns: + pd.DataFrame: A pandas DataFrame containing the historical forecast data. + """ json_list = self.get_historical_forecast_json( start, end, region, signal_type, model_date ) @@ -416,27 +493,27 @@ def get_historical_forecast_pandas( class WattTimeMaps(WattTimeBase): - def get_maps_json( - self, - signal_type: Optional[ - Literal["co2_moer", "co2_aoer", "health_damage"] - ] = "co2_moer", - ): - """ - Retrieves JSON data for the maps API. - - Args: - signal_type (Optional[str]): The type of signal to retrieve data for. - Valid options are "co2_moer", "co2_aoer", and "health_damage". - Defaults to "co2_moer". - - Returns: - dict: The JSON response from the API. - """ - if not self._is_token_valid(): - self._login() - url = "{}/v3/maps".format(self.url_base) - headers = {"Authorization": "Bearer " + self.token} - params = {"signal_type": signal_type} - rsp = requests.get(url, headers=headers, params=params) - return rsp.json() + def get_maps_json( + self, + signal_type: Optional[ + Literal["co2_moer", "co2_aoer", "health_damage"] + ] = "co2_moer", + ): + """ + Retrieves JSON data for the maps API. + + Args: + signal_type (Optional[str]): The type of signal to retrieve data for. + Valid options are "co2_moer", "co2_aoer", and "health_damage". + Defaults to "co2_moer". + + Returns: + dict: The JSON response from the API. + """ + if not self._is_token_valid(): + self._login() + url = "{}/v3/maps".format(self.url_base) + headers = {"Authorization": "Bearer " + self.token} + params = {"signal_type": signal_type} + rsp = requests.get(url, headers=headers, params=params) + return rsp.json() From 1974e4d8be675009889c3c008c2a4281387ec088 Mon Sep 17 00:00:00 2001 From: sam-watttime Date: Fri, 22 Dec 2023 15:37:15 -0700 Subject: [PATCH 5/6] add rsp.raise_for_status() throughout --- watttime/api.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/watttime/api.py b/watttime/api.py index 7621da04..4fcfcfad 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -41,6 +41,7 @@ def _login(self): auth=requests.auth.HTTPBasicAuth(self.username, self.password), timeout=20, ) + rsp.raise_for_status() self.token = rsp.json().get("token", None) self.token_valid_until = datetime.now() + timedelta(minutes=30) if not self.token: @@ -160,6 +161,7 @@ def region_from_loc( "signal_type": signal_type, } rsp = requests.get(url, headers=headers, params=params) + rsp.raise_for_status() return rsp.json() @@ -208,10 +210,11 @@ def get_historical_jsons( for c in chunks: params["start"], params["end"] = c rsp = requests.get(url, headers=headers, params=params) - if rsp.status_code == 200: + try: + rsp.raise_for_status() j = rsp.json() responses.append(j) - else: + except Exception as e: raise Exception(f"\nAPI Response Error: {rsp.status_code}, {rsp.text}") if len(j["meta"]["warnings"]): @@ -302,6 +305,7 @@ def get_access_json(self) -> Dict: url = "{}/v3/my-access".format(self.url_base) headers = {"Authorization": "Bearer " + self.token} rsp = requests.get(url, headers=headers) + rsp.raise_for_status() return rsp.json() def get_access_pandas(self) -> pd.DataFrame: @@ -373,6 +377,7 @@ def get_forecast_json( url = "{}/v3/forecast".format(self.url_base) headers = {"Authorization": "Bearer " + self.token} rsp = requests.get(url, headers=headers, params=params) + rsp.raise_for_status() return rsp.json() def get_forecast_pandas( @@ -444,10 +449,11 @@ def get_historical_forecast_json( for c in chunks: params["start"], params["end"] = c rsp = requests.get(url, headers=headers, params=params) - if rsp.status_code == 200: + try: + rsp.raise_for_status() j = rsp.json() responses.append(j) - else: + except Exception as e: raise Exception(f"\nAPI Response Error: {rsp.status_code}, {rsp.text}") if len(j["meta"]["warnings"]): @@ -516,4 +522,5 @@ def get_maps_json( headers = {"Authorization": "Bearer " + self.token} params = {"signal_type": signal_type} rsp = requests.get(url, headers=headers, params=params) + rsp.raise_for_status() return rsp.json() From 79644b54481398307c190131f52468421ba67791 Mon Sep 17 00:00:00 2001 From: sam-watttime Date: Thu, 1 Feb 2024 13:57:35 -0700 Subject: [PATCH 6/6] merge fix and lint --- watttime/api.py | 83 +++++++++++++++++++++++++++++++++++-------------- 1 file changed, 59 insertions(+), 24 deletions(-) diff --git a/watttime/api.py b/watttime/api.py index 5f4a56ff..01d0fcf8 100644 --- a/watttime/api.py +++ b/watttime/api.py @@ -75,6 +75,7 @@ def _get_chunks( def register( self, email: str, organization: Optional[str] = None ) -> requests.Response: + """Register for the WattTime API, if you do not already have an account.""" url = f"{self.url_base}/register" params = { "username": self.username, @@ -89,6 +90,40 @@ def register( f"Successfully registered {self.username}, please check {email} for a verification email" ) + @cache + def region_from_loc( + self, + latitude: Union[str, float], + longitude: Union[str, float], + signal_type: Optional[ + Literal["co2_moer", "co2_aoer", "health_damage"] + ] = "co2_moer", + ) -> Dict[str, str]: + """ + Retrieve the region information based on the given latitude and longitude. + + Args: + latitude (Union[str, float]): The latitude of the location. + longitude (Union[str, float]): The longitude of the location. + signal_type (Optional[Literal["co2_moer", "co2_aoer", "health_damage"]], optional): + The type of signal to be used for the region classification. + Defaults to "co2_moer". + + Returns: + Dict[str, str]: A dictionary containing the region information with keys "region" and "region_full_name". + """ + if not self._is_token_valid(): + self._login() + url = f"{self.url_base}/v3/region-from-loc" + headers = {"Authorization": "Bearer " + self.token} + params = { + "latitude": str(latitude), + "longitude": str(longitude), + "signal_type": signal_type, + } + rsp = requests.get(url, headers=headers, params=params) + return rsp.json() + class WattTimeHistorical(WattTimeBase): def get_historical_jsons( @@ -388,27 +423,27 @@ def get_historical_forecast_pandas( class WattTimeMaps(WattTimeBase): - def get_maps_json( - self, - signal_type: Optional[ - Literal["co2_moer", "co2_aoer", "health_damage"] - ] = "co2_moer", - ): - """ - Retrieves JSON data for the maps API. - - Args: - signal_type (Optional[str]): The type of signal to retrieve data for. - Valid options are "co2_moer", "co2_aoer", and "health_damage". - Defaults to "co2_moer". - - Returns: - dict: The JSON response from the API. - """ - if not self._is_token_valid(): - self._login() - url = "{}/v3/maps".format(self.url_base) - headers = {"Authorization": "Bearer " + self.token} - params = {"signal_type": signal_type} - rsp = requests.get(url, headers=headers, params=params) - return rsp.json() + def get_maps_json( + self, + signal_type: Optional[ + Literal["co2_moer", "co2_aoer", "health_damage"] + ] = "co2_moer", + ): + """ + Retrieves JSON data for the maps API. + + Args: + signal_type (Optional[str]): The type of signal to retrieve data for. + Valid options are "co2_moer", "co2_aoer", and "health_damage". + Defaults to "co2_moer". + + Returns: + dict: The JSON response from the API. + """ + if not self._is_token_valid(): + self._login() + url = "{}/v3/maps".format(self.url_base) + headers = {"Authorization": "Bearer " + self.token} + params = {"signal_type": signal_type} + rsp = requests.get(url, headers=headers, params=params) + return rsp.json()