Add /maps and /region-from-loc endpoints

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')
+```

tests/test_sdk.py

@@ -9,6 +9,7 @@
     WattTimeHistorical,
     WattTimeMyAccess,
     WattTimeForecast,
+    WattTimeMaps,
 )
 from pathlib import Path
 
@@ -303,5 +304,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()

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
@@ -74,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,
@@ -88,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(
@@ -384,3 +420,30 @@ def get_historical_forecast_pandas(
                 _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()