Merge branch 'v2.1' of github.com:simvue-io/python-api into v2.1

Author: kzscisoft

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

@@ -47,12 +47,12 @@ class CO2Monitor(pydantic.BaseModel):
     local_data_directory: pydantic.DirectoryPath
     intensity_refresh_interval: int | None | str
     co2_intensity: float | None
-    co2_signal_api_token: str | None
+    co2_signal_api_token: pydantic.SecretStr | None
     offline: bool = False
 
     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 (
@@ -252,7 +252,6 @@ def estimate_co2_emissions(
             )
             _current_co2_intensity = self._current_co2_data.data.carbon_intensity
             _co2_units = self._current_co2_data.carbon_intensity_units
-
         _process.gpu_percentage = gpu_percent
         _process.cpu_percentage = cpu_percent
         _previous_energy: float = _process.total_energy
@@ -271,8 +270,6 @@ def estimate_co2_emissions(
         # Measured value is in g/kWh, convert to kg/kWs
         _carbon_intensity_kgpws: float = _current_co2_intensity / (60 * 60 * 1e3)
 
-        _previous_emission: float = _process.co2_emission
-
         _process.co2_delta = (
             _process.power_usage * _carbon_intensity_kgpws * measure_interval
         )

simvue/metrics.py

@@ -139,7 +139,6 @@ def __init__(
         self,
         processes: list[psutil.Process],
         interval: float | None,
-        cpu_only: bool = False,
     ) -> None:
         """Perform a measurement of system resource consumption.
 
@@ -149,14 +148,10 @@ def __init__(
             processes to measure across.
         interval: float | None
             interval to measure, if None previous measure time used for interval.
-        cpu_only: bool, optional
-            only record CPU information, default False
         """
         self.cpu_percent: float | None = get_process_cpu(processes, interval=interval)
         self.cpu_memory: float | None = get_process_memory(processes)
-        self.gpus: list[dict[str, float]] = (
-            None if cpu_only else get_gpu_metrics(processes)
-        )
+        self.gpus: list[dict[str, float]] = get_gpu_metrics(processes)
 
     def to_dict(self) -> dict[str, float]:
         """Create metrics dictionary for sending to a Simvue server."""

simvue/run.py

@@ -334,10 +334,7 @@ def _terminate_run(
 
     def _get_internal_metrics(
         self,
-        system_metrics_step: int | None,
-        emission_metrics_step: int | None,
-        res_measure_interval: int | None = None,
-        ems_measure_interval: int | None = None,
+        system_metrics_step: int,
     ) -> None:
         """Refresh resource and emissions metrics.
 
@@ -346,55 +343,49 @@ def _get_internal_metrics(
 
         Parameters
         ----------
-        system_metrics_step: int | None
-            the current step for this resource metric record,
-            None if skipping resource metrics.
-        emission_metrics_step: int | None
-            the current step for this emission metrics record,
-            None if skipping emission metrics.
-        res_measure_interval: int | None, optional
-            the interval for resource metric gathering, default is None
-        ems_measure_interval: int | None, optional
-            the interval for emission metric gathering, default is None
+        system_metrics_step: int
+            The current step for this system metric record
 
         Return
         ------
         tuple[float, float]
             new resource metric measure time
             new emissions metric measure time
         """
+
+        # In order to get a resource metric reading at t=0
+        # because there is no previous CPU reading yet we cannot
+        # use the default of None for the interval here, so we measure
+        # at an interval of 1s.
         _current_system_measure = SystemResourceMeasurement(
             self.processes,
-            interval=res_measure_interval,
-            cpu_only=not system_metrics_step,
+            interval=1 if system_metrics_step == 0 else None,
         )
 
-        if system_metrics_step is not None:
-            # Set join on fail to false as if an error is thrown
-            # join would be called on this thread and a thread cannot
-            # join itself!
-            self._add_metrics_to_dispatch(
-                _current_system_measure.to_dict(),
-                join_on_fail=False,
-                step=system_metrics_step,
-            )
+        # Set join on fail to false as if an error is thrown
+        # join would be called on this thread and a thread cannot
+        # join itself!
+        self._add_metrics_to_dispatch(
+            _current_system_measure.to_dict(),
+            join_on_fail=False,
+            step=system_metrics_step,
+        )
 
-        if (
-            self._emissions_monitor
-            and emission_metrics_step is not None
-            and ems_measure_interval is not None
-            and _current_system_measure.cpu_percent is not None
-        ):
+        # For the first emissions metrics reading, the time interval to use
+        # Is the time since the run started, otherwise just use the time between readings
+        if self._emissions_monitor:
             self._emissions_monitor.estimate_co2_emissions(
                 process_id=f"{self._name}",
                 cpu_percent=_current_system_measure.cpu_percent,
-                measure_interval=ems_measure_interval,
+                measure_interval=(time.time() - self._start_time)
+                if system_metrics_step == 0
+                else self._system_metrics_interval,
                 gpu_percent=_current_system_measure.gpu_percent,
             )
             self._add_metrics_to_dispatch(
                 self._emissions_monitor.simvue_metrics(),
                 join_on_fail=False,
-                step=emission_metrics_step,
+                step=system_metrics_step,
             )
 
     def _create_heartbeat_callback(
@@ -416,61 +407,29 @@ def _heartbeat(
                 raise RuntimeError("Expected initialisation of heartbeat")
 
             last_heartbeat: float = 0
-            last_res_metric_call: float = 0
-            last_co2_metric_call: float = 0
-
-            co2_step: int = 0
-            res_step: int = 0
+            last_sys_metric_call: float = 0
 
-            initial_ems_metrics_interval: float = time.time() - self._start_time
+            sys_step: int = 0
 
             while not heartbeat_trigger.is_set():
                 with self._configuration_lock:
                     _current_time: float = time.time()
+
                     _update_system_metrics: bool = (
                         self._system_metrics_interval is not None
-                        and _current_time - last_res_metric_call
-                        > self._system_metrics_interval
-                        and self._status == "running"
-                    )
-                    _update_emissions_metrics: bool = (
-                        self._system_metrics_interval is not None
-                        and self._emissions_monitor
-                        and _current_time - last_co2_metric_call
+                        and _current_time - last_sys_metric_call
                         > self._system_metrics_interval
                         and self._status == "running"
                     )
 
-                    # In order to get a resource metric reading at t=0
-                    # because there is no previous CPU reading yet we cannot
-                    # use the default of None for the interval here, so we measure
-                    # at an interval of 1s. For emissions metrics the first step
-                    # is time since run start
-                    self._get_internal_metrics(
-                        emission_metrics_step=co2_step
-                        if _update_emissions_metrics
-                        else None,
-                        system_metrics_step=res_step
-                        if _update_system_metrics
-                        else None,
-                        res_measure_interval=1 if res_step == 0 else None,
-                        ems_measure_interval=initial_ems_metrics_interval
-                        if co2_step == 0
-                        else self._system_metrics_interval,
-                    )
+                    if _update_system_metrics:
+                        self._get_internal_metrics(system_metrics_step=sys_step)
+                        sys_step += 1
 
-                    res_step += 1
-                    co2_step += 1
-
-                    last_res_metric_call = (
+                    last_sys_metric_call = (
                         _current_time
                         if _update_system_metrics
-                        else last_res_metric_call
-                    )
-                    last_co2_metric_call = (
-                        _current_time
-                        if _update_emissions_metrics
-                        else last_co2_metric_call
+                        else last_sys_metric_call
                     )
 
                 if time.time() - last_heartbeat < self._heartbeat_interval:
@@ -1055,7 +1014,7 @@ def config(
         queue_blocking: bool | None = None,
         system_metrics_interval: pydantic.PositiveInt | None = None,
         enable_emission_metrics: bool | None = None,
-        disable_system_metrics: bool | None = None,
+        disable_resources_metrics: bool | None = None,
         storage_id: str | None = None,
         abort_on_alert: typing.Literal["run", "all", "ignore"] | bool | None = None,
     ) -> bool:
@@ -1069,10 +1028,10 @@ def config(
         queue_blocking : bool, optional
             block thread queues during metric/event recording
         system_metrics_interval : int, optional
-            frequency at which to collect resource metrics
+            frequency at which to collect resource and emissions metrics, if enabled
         enable_emission_metrics : bool, optional
             enable monitoring of emission metrics
-        disable_system_metrics : bool, optional
+        disable_resources_metrics : bool, optional
             disable monitoring of resource metrics
         storage_id : str, optional
             identifier of storage to use, by default None
@@ -1095,17 +1054,30 @@ def config(
             if queue_blocking is not None:
                 self._queue_blocking = queue_blocking
 
-            if system_metrics_interval and disable_system_metrics:
+            if system_metrics_interval and disable_resources_metrics:
                 self._error(
                     "Setting of resource metric interval and disabling resource metrics is ambiguous"
                 )
                 return False
 
-            if disable_system_metrics:
+            if system_metrics_interval:
+                self._system_metrics_interval = system_metrics_interval
+
+            if disable_resources_metrics:
+                if self._emissions_monitor:
+                    self._error(
+                        "Emissions metrics require resource metrics collection."
+                    )
+                    return False
                 self._pid = None
                 self._system_metrics_interval = None
 
             if enable_emission_metrics:
+                if not self._system_metrics_interval:
+                    self._error(
+                        "Emissions metrics require resource metrics collection - make sure resource metrics are enabled!"
+                    )
+                    return False
                 if self._user_config.run.mode == "offline":
                     # Create an emissions monitor with no API calls
                     self._emissions_monitor = CO2Monitor(
@@ -1130,9 +1102,6 @@ def config(
             elif enable_emission_metrics is False and self._emissions_monitor:
                 self._error("Cannot disable emissions monitor once it has been started")
 
-            if system_metrics_interval:
-                self._system_metrics_interval = system_metrics_interval
-
             if abort_on_alert is not None:
                 if isinstance(abort_on_alert, bool):
                     warnings.warn(

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: