[Feat] CarbonIntensityProvider and ElectricityMaps implementation

tests/carbon/test_electricity_maps_carbon_intensity_provider.py

@@ -0,0 +1,121 @@
+import requests
+import pytest
+import json
+
+from unittest.mock import patch
+
+from zeus.carbon import get_ip_lat_long
+from zeus.carbon.electricity_maps_carbon_intensity_provider import (
+    ElectricityMapsCarbonIntensityProvider,
+)
+
+
+class MockHttpResponse:
+    def __init__(self, text):
+        self.text = text
+        self.json_obj = json.loads(text)
+
+    def json(self):
+        return self.json_obj
+
+
+@pytest.fixture
+def mock_requests():
+    IP_INFO_RESPONSE = """{
+		"ip": "35.3.237.23",
+		"hostname": "0587459863.wireless.umich.net",
+		"city": "Ann Arbor",
+		"region": "Michigan",
+		"country": "US",
+		"loc": "42.2776,-83.7409",
+		"org": "AS36375 University of Michigan",
+		"postal": "48109",
+		"timezone": "America/Detroit",
+		"readme": "https://ipinfo.io/missingauth"
+	}"""
+
+    NO_MEASUREMENT_RESPONSE = '{"error":"No recent data for zone "US-MIDW-MISO""}'
+
+    ELECTRICITY_MAPS_RESPONSE_LIFECYCLE = (
+        '{"zone":"US-MIDW-MISO","carbonIntensity":466,"datetime":"2024-09-24T03:00:00.000Z",'
+        '"updatedAt":"2024-09-24T02:47:02.408Z","createdAt":"2024-09-21T03:45:20.860Z",'
+        '"emissionFactorType":"lifecycle","isEstimated":true,"estimationMethod":"TIME_SLICER_AVERAGE"}'
+    )
+
+    ELECTRICITY_MAPS_RESPONSE_DIRECT = (
+        '{"zone":"US-MIDW-MISO","carbonIntensity":506,"datetime":"2024-09-27T00:00:00.000Z",'
+        '"updatedAt":"2024-09-27T00:43:50.277Z","createdAt":"2024-09-24T00:46:38.741Z",'
+        '"emissionFactorType":"direct","isEstimated":true,"estimationMethod":"TIME_SLICER_AVERAGE"}'
+    )
+
+    real_requests_get = requests.get
+
+    def mock_requests_get(url, *args, **kwargs):
+        if url == "http://ipinfo.io/json":
+            return MockHttpResponse(IP_INFO_RESPONSE)
+        elif (
+            url
+            == "https://api.electricitymap.org/v3/carbon-intensity/latest?lat=42.2776&lon=-83.7409&disableEstimations=True&emissionFactorType=direct"
+        ):
+            return MockHttpResponse(NO_MEASUREMENT_RESPONSE)
+        elif (
+            url
+            == "https://api.electricitymap.org/v3/carbon-intensity/latest?lat=42.2776&lon=-83.7409&disableEstimations=False&emissionFactorType=direct"
+        ):
+            return MockHttpResponse(ELECTRICITY_MAPS_RESPONSE_DIRECT)
+        elif (
+            url
+            == "https://api.electricitymap.org/v3/carbon-intensity/latest?lat=42.2776&lon=-83.7409&disableEstimations=False&emissionFactorType=lifecycle"
+        ):
+            return MockHttpResponse(ELECTRICITY_MAPS_RESPONSE_LIFECYCLE)
+        else:
+            return real_requests_get(url, *args, **kwargs)
+
+    patch_request_get = patch("requests.get", side_effect=mock_requests_get)
+
+    patch_request_get.start()
+
+    yield
+
+    patch_request_get.stop()
+
+
+@pytest.fixture
+def mock_exception_ip():
+    def mock_requests_get(url):
+        raise ConnectionError
+
+    patch_request_get = patch("requests.get", side_effect=mock_requests_get)
+
+    patch_request_get.start()
+
+    yield
+
+    patch_request_get.stop()
+
+
+def test_get_current_carbon_intensity(mock_requests):
+    latlong = get_ip_lat_long()
+    assert latlong == (42.2776, -83.7409)
+    provider = ElectricityMapsCarbonIntensityProvider(latlong)
+    assert (
+        provider.get_current_carbon_intensity(
+            estimate=True, emission_factor_type="lifecycle"
+        )
+        == 466
+    )
+    assert provider.get_current_carbon_intensity(estimate=True) == 506
+
+
+def test_get_current_carbon_intensity_no_response(mock_requests):
+    latlong = get_ip_lat_long()
+    assert latlong == (42.2776, -83.7409)
+    provider = ElectricityMapsCarbonIntensityProvider(latlong)
+
+    with pytest.raises(Exception):
+        provider.get_current_carbon_intensity()
+
+
+def test_get_lat_long_excpetion(mock_exception_ip):
+    with pytest.raises(ConnectionError):
+        get_ip_lat_long()

zeus/carbon/__init__.py

@@ -0,0 +1,21 @@
+"""Carbon intensity providers used for carbon-aware optimizers."""
+
+from zeus.carbon.electricity_maps_carbon_intensity_provider import (
+    ElectricityMapsCarbonIntensityProvider,
+)
+
+import requests
+
+
+def get_ip_lat_long() -> tuple[float, float]:
+    """Retrieve the latitude and longitude of the current IP position."""
+    try:
+        ip_url = "http://ipinfo.io/json"
+        resp = requests.get(ip_url)
+        loc = resp.json()["loc"]
+        lat, long = map(float, loc.split(","))
+        print(f"Retrieve latitude and longitude: {lat}, {long}")
+        return lat, long
+    except Exception as e:
+        print(f"Failed to Retrieve Current IP's Latitude and Longitude: {e}")
+        raise (e)

zeus/carbon/carbon_intensity_provider.py

@@ -0,0 +1,19 @@
+"""Abstract Carbon Intensity Provider Class."""
+import abc
+
+
+class CarbonIntensityProvider(abc.ABC):
+    """Abstract class for implementing ways to fetch carbon intensity."""
+
+    def __init__(self, location: tuple[float, float]) -> None:
+        """Initializes carbon intensity provider location to the latitude and longitude of the input `location`.
+
+        Location is a tuple of floats where latitude is the first float and longitude is the second float.
+        """
+        self.lat = location[0]
+        self.long = location[1]
+
+    @abc.abstractmethod
+    def get_current_carbon_intensity(self) -> float:
+        """Abstract method for fetching the current carbon intensity of the set location of the class."""
+        pass

zeus/carbon/electricity_maps_carbon_intensity_provider.py

@@ -0,0 +1,30 @@
+"""Carbon Intensity Provider using ElectrictyMaps API."""
+from zeus.carbon.carbon_intensity_provider import CarbonIntensityProvider
+import requests
+
+
+class ElectricityMapsCarbonIntensityProvider(CarbonIntensityProvider):
+    """Carbon Intensity Provider with ElectricityMaps API."""
+
+    def get_current_carbon_intensity(
+        self, estimate: bool = False, emission_factor_type: str = "direct"
+    ) -> float:
+        """Fetches current carbon intensity of the location of the class.
+
+        Args:
+            estimate: bool to toggle whether carbon intensity is estimated or not
+            emission_factor_type: emission factor to be measured (`direct` or `lifestyle`)
+
+        !!! Note
+            In some locations, there is no recent carbon intensity data. `estimate` can be used to approximate the carbon intensity in such cases.
+        """
+        try:
+            url = (
+                f"https://api.electricitymap.org/v3/carbon-intensity/latest?lat={self.lat}&lon={self.long}"
+                + f"&disableEstimations={not estimate}&emissionFactorType={emission_factor_type}"
+            )
+            resp = requests.get(url)
+            return resp.json()["carbonIntensity"]
+        except Exception as e:
+            print(f"Failed to retrieve live carbon intensity data: {e}")
+            raise (e)