Refactor to use new electricitymaps API endpoints

Author: wk9874

simvue/eco/api_client.py

@@ -19,38 +19,33 @@
 import geocoder.location
 import typing
 
-CO2_SIGNAL_API_ENDPOINT: str = "https://api.co2signal.com/v1/latest"
+CO2_SIGNAL_API_ENDPOINT: str = (
+    "https://api.electricitymap.org/v3/carbon-intensity/latest"
+)
 
 
 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"].replace("Z", "+00:00")
+                json_response["datetime"].replace("Z", "+00:00")
             ),
-            carbon_intensity=_data["carbonIntensity"],
-            fossil_fuel_percentage=_data["fossilFuelPercentage"],
+            carbon_intensity=json_response["carbonIntensity"],
         )
         return cls(
-            disclaimer=json_response["_disclaimer"],
-            country_code=json_response["countryCode"],
-            status=json_response["status"],
+            country_code=json_response["zone"],
             data=_co2_signal_data,
-            carbon_intensity_units=json_response["units"]["carbonIntensity"],
+            carbon_intensity_units="gCO2e/kWh",
         )
 
 
@@ -109,16 +104,14 @@ def _get_user_location_info(self) -> None:
     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,
+            "zone": 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)
+        _response = requests.get(f"{self.co2_api_endpoint}", headers=_params)
 
         if _response.status_code != http.HTTPStatus.OK:
             try:

simvue/eco/emissions_monitor.py

@@ -52,7 +52,7 @@ class CO2Monitor(pydantic.BaseModel):
 
     def now(self) -> str:
         """Return data file timestamp for the current time"""
-        _now: datetime.datetime = datetime.datetime.now(datetime.UTC)
+        _now: datetime.datetime = datetime.datetime.now(datetime.timezone.utc)
         return _now.strftime(TIME_FORMAT)
 
     @property
@@ -187,7 +187,7 @@ def check_refresh(self) -> bool:
             with self._data_file_path.open("r") as in_f:
                 self._local_data = json.load(in_f)
 
-        if not self._client or not self._local_data:
+        if not self._client:
             return False
 
         if (

tests/conftest.py

@@ -53,15 +53,14 @@ def clear_out_files() -> None:
 @pytest.fixture
 def mock_co2_signal(monkeypatch: monkeypatch.MonkeyPatch) -> dict[str, dict | str]:
     _mock_data = {
-        "data": {
-            "datetime": datetime.datetime.now().isoformat(),
-            "carbonIntensity": 40,
-            "fossilFuelPercentage": 39,
-        },
-        "_disclaimer": "test disclaimer",
-        "countryCode": "GB",
-        "status": "unknown",
-        "units": {"carbonIntensity": "eqCO2kg/kwh"}
+        "zone":"GB",
+        "carbonIntensity":85,
+        "datetime":"2025-04-04T12:00:00.000Z",
+        "updatedAt":"2025-04-04T11:41:12.947Z",
+        "createdAt":"2025-04-01T12:43:58.056Z",
+        "emissionFactorType":"lifecycle",
+        "isEstimated":True,
+        "estimationMethod":"TIME_SLICER_AVERAGE"
     }
     class MockCo2SignalAPIResponse:
         def json(*_, **__) -> dict:

tests/unit/test_ecoclient.py

@@ -18,10 +18,9 @@ def test_api_client_get_loc_info(mock_co2_signal) -> None:
 def test_api_client_query(mock_co2_signal: dict[str, dict | str]) -> None:
     _client = sv_eco_api.APIClient()
     _response: sv_eco_api.CO2SignalResponse = _client.get()
-    assert _response.carbon_intensity_units == mock_co2_signal["units"]["carbonIntensity"]
-    assert _response.country_code == mock_co2_signal["countryCode"]
-    assert _response.data.carbon_intensity == mock_co2_signal["data"]["carbonIntensity"]
-    assert _response.data.fossil_fuel_percentage == mock_co2_signal["data"]["fossilFuelPercentage"]
+    assert _response.carbon_intensity_units == "gCO2e/kWh"
+    assert _response.country_code == mock_co2_signal["zone"]
+    assert _response.data.carbon_intensity == mock_co2_signal["carbonIntensity"]
 
 
 @pytest.mark.eco
@@ -41,7 +40,7 @@ def test_outdated_data_check(
             thermal_design_power_per_cpu=80,
             thermal_design_power_per_gpu=130,
             local_data_directory=tempd,
-            intensity_refresh_interval=1 if refresh else 2,
+            intensity_refresh_interval=1 if refresh else None,
             co2_intensity=None,
             co2_signal_api_token=None
         )