[Feat] CarbonIntensityProvider and ElectricityMaps implementation (#…

…129)

Co-authored-by: Jae-Won Chung <[email protected]>

tests/test_carbon.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import json
+import pytest
+import requests
+
+from unittest.mock import patch
+
+from zeus.carbon import (
+    ElectrictyMapsClient,
+    get_ip_lat_long,
+    ZeusCarbonIntensityNotFoundError,
+)
+
+
+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 = r'{"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()
+
+
+def test_get_current_carbon_intensity(mock_requests):
+    latlong = get_ip_lat_long()
+    assert latlong == (pytest.approx(42.2776), pytest.approx(-83.7409))
+    provider = ElectrictyMapsClient(
+        latlong, estimate=True, emission_factor_type="lifecycle"
+    )
+    assert provider.get_current_carbon_intensity() == 466
+
+    provider.emission_factor_type = "direct"
+    assert provider.get_current_carbon_intensity() == 506
+
+
+def test_get_current_carbon_intensity_no_response(mock_requests):
+    latlong = get_ip_lat_long()
+    assert latlong == (pytest.approx(42.2776), pytest.approx(-83.7409))
+    provider = ElectrictyMapsClient(latlong)
+
+    with pytest.raises(ZeusCarbonIntensityNotFoundError):
+        provider.get_current_carbon_intensity()

zeus/carbon.py

@@ -0,0 +1,101 @@
+"""Carbon intensity providers used for carbon-aware optimizers."""
+
+from __future__ import annotations
+
+import abc
+import requests
+from typing import Literal
+
+from zeus.exception import ZeusBaseError
+from zeus.utils.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+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(","))
+        logger.info("Retrieved latitude and longitude: %s, %s", lat, long)
+        return lat, long
+    except requests.exceptions.RequestException as e:
+        logger.exception(
+            "Failed to retrieve current latitude and longitude of IP: %s", e
+        )
+        raise
+
+
+class ZeusCarbonIntensityNotFoundError(ZeusBaseError):
+    """Exception when carbon intensity measurement could not be retrieved."""
+
+    def __init__(self, message: str) -> None:
+        """Initialize carbon not found exception."""
+        super().__init__(message)
+
+
+class CarbonIntensityProvider(abc.ABC):
+    """Abstract class for implementing ways to fetch carbon intensity."""
+
+    @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
+
+
+class ElectrictyMapsClient(CarbonIntensityProvider):
+    """Carbon Intensity Provider with ElectricityMaps API.
+
+    Reference:
+
+    1. [ElectricityMaps](https://www.electricitymaps.com/)
+    2. [ElectricityMaps API](https://static.electricitymaps.com/api/docs/index.html)
+    3. [ElectricityMaps GitHub](https://github.com/electricitymaps/electricitymaps-contrib)
+    """
+
+    def __init__(
+        self,
+        location: tuple[float, float],
+        estimate: bool = False,
+        emission_factor_type: Literal["direct", "lifecycle"] = "direct",
+    ) -> None:
+        """Iniitializes ElectricityMaps Carbon Provider.
+
+        Args:
+            location: tuple of latitude and longitude (latitude, longitude)
+            estimate: bool to toggle whether carbon intensity is estimated or not
+            emission_factor_type: emission factor to be measured (`direct` or `lifestyle`)
+        """
+        self.lat, self.long = location
+        self.estimate = estimate
+        self.emission_factor_type = emission_factor_type
+
+    def get_current_carbon_intensity(self) -> float:
+        """Fetches current carbon intensity of the location of the class.
+
+        !!! Note
+            In some locations, there is no recent carbon intensity data. `self.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 self.estimate}&emissionFactorType={self.emission_factor_type}"
+            )
+            resp = requests.get(url)
+        except requests.exceptions.RequestException as e:
+            logger.exception(
+                "Failed to retrieve recent carbon intensnity measurement: %s", e
+            )
+            raise
+
+        try:
+            return resp.json()["carbonIntensity"]
+        except KeyError as e:
+            # Raise exception when carbonIntensity does not exist in response
+            raise ZeusCarbonIntensityNotFoundError(
+                f"Recent carbon intensity measurement not found at `({self.lat}, {self.long})` "
+                f"with estimate set to `{self.estimate}` and emission_factor_type set to `{self.emission_factor_type}`\n"
+                f"JSON Response: {resp.text}"
+            ) from e