From 7b9d9737b4daea0f99fcf01853f28eef171bf9f8 Mon Sep 17 00:00:00 2001 From: sam-watttime Date: Fri, 22 Dec 2023 14:18:36 -0700 Subject: [PATCH 1/4] 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/4] 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/4] 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 79644b54481398307c190131f52468421ba67791 Mon Sep 17 00:00:00 2001 From: sam-watttime Date: Thu, 1 Feb 2024 13:57:35 -0700 Subject: [PATCH 4/4] 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()