Ecoclient Simvue Python API Extension

pyproject.toml

@@ -45,7 +45,6 @@ dependencies = [
   "gitpython (>=3.1.44,<4.0.0)",
   "humanfriendly (>=10.0,<11.0)",
   "randomname (>=0.2.1,<0.3.0)",
-  "codecarbon (>=2.8.3,<3.0.0)",
   "numpy (>=2.0.0,<3.0.0)",
   "flatdict (>=4.0.1,<5.0.0)",
   "semver (>=3.0.4,<4.0.0)",
@@ -54,6 +53,7 @@ dependencies = [
   "tenacity (>=9.0.0,<10.0.0)",
   "typing-extensions (>=4.12.2,<5.0.0) ; python_version < \"3.11\"",
   "deepmerge (>=2.0,<3.0)",
+  "geocoder (>=1.38.1,<2.0.0)",
 ]
 
 [project.urls]

simvue/api/objects/folder.py

@@ -10,8 +10,8 @@
 import pathlib
 import typing
 import datetime
+import json
 
-from codecarbon.output_methods.emissions_data import json
 import pydantic
 
 from simvue.exception import ObjectNotFoundError

simvue/api/request.py

@@ -10,9 +10,9 @@
 import copy
 import json as json_module
 import typing
+import logging
 import http
 
-from codecarbon.external.logger import logging
 import requests
 from tenacity import (
     retry,

simvue/config/parameters.py

@@ -48,8 +48,6 @@ def check_token(cls, v: typing.Any) -> str | None:
 
 class OfflineSpecifications(pydantic.BaseModel):
     cache: pathlib.Path | None = None
-    country_iso_code: str | None = None
-
 
 class MetricsSpecifications(pydantic.BaseModel):
     resources_metrics_interval: pydantic.PositiveInt | None = -1

simvue/config/user.py

@@ -33,6 +33,7 @@
 from simvue.version import __version__
 from simvue.api.request import get as sv_get
 from simvue.api.url import URL
+from simvue.eco.config import EcoConfig
 
 logger = logging.getLogger(__name__)
 
@@ -42,14 +43,15 @@
 
 class SimvueConfiguration(pydantic.BaseModel):
     # Hide values as they contain token and URL
-    model_config = pydantic.ConfigDict(hide_input_in_errors=True)
+    model_config = pydantic.ConfigDict(hide_input_in_errors=True, revalidate_instances="always")
     client: ClientGeneralOptions = ClientGeneralOptions()
     server: ServerSpecifications = pydantic.Field(
         ..., description="Specifications for Simvue server"
     )
     run: DefaultRunSpecifications = DefaultRunSpecifications()
     offline: OfflineSpecifications = OfflineSpecifications()
     metrics: MetricsSpecifications = MetricsSpecifications()
+    eco: EcoConfig = EcoConfig()
 
     @classmethod
     def _load_pyproject_configs(cls) -> dict | None:
@@ -135,14 +137,14 @@ def write(self, out_directory: pydantic.DirectoryPath) -> None:
 
     @pydantic.model_validator(mode="after")
     @classmethod
-    def check_valid_server(cls, values: "SimvueConfiguration") -> bool:
+    def check_valid_server(cls, values: "SimvueConfiguration") -> "SimvueConfiguration":
         if os.environ.get("SIMVUE_NO_SERVER_CHECK"):
             return values
 
         cls._check_server(values.server.token, values.server.url, values.run.mode)
 
         return values
-
+    
     @classmethod
     @sv_util.prettify_pydantic
     def fetch(

simvue/eco/__init__.py

@@ -0,0 +1,14 @@
+"""
+Simvue Eco
+==========
+
+Contains functionality for green IT, monitoring emissions etc.
+NOTE: The metrics calculated by these methods should be used for relative
+comparisons only. Any values returned should not be taken as absolute.
+
+"""
+__date__ = "2025-03-06"
+
+from .emissions_monitor import CO2Monitor as CO2Monitor
+
+__all__ = ["CO2Monitor"]

simvue/eco/api_client.py

@@ -0,0 +1,141 @@
+"""
+CO2 Signal API Client
+=====================
+
+Provides inteface to the CO2 Signal API,
+which provides real-time data on the carbon intensity of
+electricity generation in different countries.
+"""
+
+__date__ = "2025-02-27"
+
+import requests
+import pydantic
+import functools
+import http
+import logging
+import datetime
+import geocoder
+import geocoder.location
+import typing
+
+
+class CO2SignalData(pydantic.BaseModel):
+    datetime: datetime.datetime
+    carbon_intensity: float
+    fossil_fuel_percentage: float
+
+
+class CO2SignalResponse(pydantic.BaseModel):
+    disclaimer: str
+    country_code: str
+    status: str
+    data: CO2SignalData
+    carbon_intensity_units: str
+
+    @classmethod
+    def from_json_response(cls, json_response: dict) -> "CO2SignalResponse":
+        _data: dict[str, typing.Any] = json_response["data"]
+        _co2_signal_data = CO2SignalData(
+            datetime=datetime.datetime.fromisoformat(_data["datetime"]),
+            carbon_intensity=_data["carbonIntensity"],
+            fossil_fuel_percentage=_data["fossilFuelPercentage"],
+        )
+        return cls(
+            disclaimer=json_response["_disclaimer"],
+            country_code=json_response["countryCode"],
+            status=json_response["status"],
+            data=_co2_signal_data,
+            carbon_intensity_units=json_response["units"]["carbonIntensity"],
+        )
+
+
+@functools.lru_cache()
+def _call_geocoder_query() -> typing.Any:
+    """Call GeoCoder API for IP location
+
+    Cached so this API is only called once per session as required.
+    """
+    return geocoder.ip("me")
+
+
+class APIClient(pydantic.BaseModel):
+    """
+    CO2 Signal API Client
+
+    Provides an interface to the Electricity Maps API.
+    """
+
+    co2_api_endpoint: pydantic.HttpUrl = pydantic.HttpUrl(
+        "https://api.co2signal.com/v1/latest"
+    )
+    co2_api_token: pydantic.SecretStr | None = None
+    timeout: pydantic.PositiveInt = 10
+
+    def __init__(self, *args, **kwargs) -> None:
+        """Initialise the CO2 Signal API client.
+
+        Parameters
+        ----------
+        co2_api_endpoint : str
+            endpoint for CO2 signal API
+        co2_api_token: str
+            RECOMMENDED. The API token for the CO2 Signal API, default is None.
+        timeout : int
+            timeout for API
+        """
+        super().__init__(*args, **kwargs)
+        self._logger = logging.getLogger(self.__class__.__name__)
+
+        if not self.co2_api_token:
+            self._logger.warning(
+                "⚠️  No API token provided for CO2 Signal, it is recommended "
+            )
+
+        self._get_user_location_info()
+
+    def _get_user_location_info(self) -> None:
+        """Retrieve location information for the current user."""
+        self._logger.info("📍 Determining current user location.")
+        _current_user_loc_data: geocoder.location.BBox = _call_geocoder_query()
+        self._latitude: float
+        self._longitude: float
+        self._latitude, self._longitude = _current_user_loc_data.latlng
+        self._two_letter_country_code: str = _current_user_loc_data.country  # type: ignore
+
+    def get(self) -> CO2SignalResponse:
+        """Get the current data"""
+        _params: dict[str, float | str] = {
+            "lat": self._latitude,
+            "lon": self._longitude,
+            "countryCode": self._two_letter_country_code,
+        }
+
+        if self.co2_api_token:
+            _params["auth-token"] = self.co2_api_token.get_secret_value()
+
+        self._logger.debug(f"🍃 Retrieving carbon intensity data for: {_params}")
+        _response = requests.get(f"{self.co2_api_endpoint}", params=_params)
+
+        if not _response.status_code == http.HTTPStatus.OK:
+            raise RuntimeError(
+                "Failed to retrieve current CO2 signal data for"
+                f" country '{self._two_letter_country_code}': {_response.text}"
+            )
+
+        return CO2SignalResponse.from_json_response(_response.json())
+
+    @property
+    def country_code(self) -> str:
+        """Returns the country code"""
+        return self._two_letter_country_code
+
+    @property
+    def latitude(self) -> float:
+        """Returns current latitude"""
+        return self._latitude
+
+    @property
+    def longitude(self) -> float:
+        """Returns current longitude"""
+        return self._longitude