diff --git a/codecarbon/cli/main.py b/codecarbon/cli/main.py
index 93f627e5b..21d301edc 100644
--- a/codecarbon/cli/main.py
+++ b/codecarbon/cli/main.py
@@ -390,6 +390,15 @@ def monitor(
str,
typer.Option(help="Log level (critical, error, warning, info, debug)"),
] = "error",
+ ui: Annotated[
+ bool,
+ typer.Option(help="Serve a live dashboard in your browser"),
+ ] = False,
+ ui_port: Annotated[int, typer.Option(help="Port of the live dashboard")] = 8050,
+ ui_host: Annotated[
+ str,
+ typer.Option(help="Host to bind the live dashboard to"),
+ ] = "127.0.0.1",
):
"""Monitor your machine's carbon emissions."""
@@ -424,6 +433,20 @@ def monitor(
tracker_args = {**tracker_args, "save_to_api": api}
+ if ui:
+ from codecarbon.viz.live import LiveDashboardOutput
+
+ live_output = LiveDashboardOutput(port=ui_port, host=ui_host)
+ tracker_args.setdefault("output_handlers", []).append(live_output)
+ if live_output.is_serving:
+ print(f"Live dashboard: {live_output.url}")
+ else:
+ print(
+ f"WARNING: could not start the live dashboard on {ui_host}:{ui_port}, "
+ "monitoring continues without it.",
+ file=sys.stderr,
+ )
+
from codecarbon.emissions_tracker import EmissionsTracker, OfflineEmissionsTracker
# If extra args are provided (e.g. `codecarbon monitor -- my_script.py`), delegate to `run_and_monitor`
diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py
index 96ed00c91..c4766fb01 100644
--- a/codecarbon/emissions_tracker.py
+++ b/codecarbon/emissions_tracker.py
@@ -1275,6 +1275,27 @@ def _measure_power_and_energy(self) -> None:
self._do_measurements()
self._last_measured_time = time.perf_counter()
self._measure_occurrence += 1
+
+ # Handlers displaying data locally opt in to every measure by defining
+ # `on_measure`. They get the total only: computing the delta here would
+ # consume it for the periodic call below. The power fields of
+ # EmissionsData are averages since `start()`, which is not what a live
+ # view wants, so they are replaced by the last measured power.
+ on_measure_handlers = [
+ handler.on_measure
+ for handler in self._output_handlers
+ if hasattr(handler, "on_measure")
+ ]
+ if on_measure_handlers:
+ live = dataclasses.replace(
+ self._prepare_emissions_data(),
+ cpu_power=self._cpu_power.W,
+ gpu_power=self._gpu_power.W,
+ ram_power=self._ram_power.W,
+ )
+ for on_measure in on_measure_handlers:
+ on_measure(live)
+
# Special case: metrics and api calls are sent every `api_call_interval` measures
if (
self._api_call_interval != -1
diff --git a/codecarbon/viz/live.html b/codecarbon/viz/live.html
new file mode 100644
index 000000000..0322d140e
--- /dev/null
+++ b/codecarbon/viz/live.html
@@ -0,0 +1,142 @@
+
+
+
+
+
+CodeCarbon live
+
+
+
+
CodeCarbon waiting for the first measurement…
+
+
+
Power
–W
+
Emissions
–gCO₂eq
+
Energy
–kWh
+
Elapsed
–
+
+
+
+
+ CPU
+ GPU
+ RAM
+
+
+
+
Component
Model
Power (W)
Load
+
Task
Energy (kWh)
Emissions (g)
Duration (s)
+
+
+
+
+
diff --git a/codecarbon/viz/live.py b/codecarbon/viz/live.py
new file mode 100644
index 000000000..2ffdfb8de
--- /dev/null
+++ b/codecarbon/viz/live.py
@@ -0,0 +1,174 @@
+"""
+Live local dashboard.
+
+An output handler that keeps a bounded window of live measurements in memory and
+serves them, with a single self-contained HTML page, over a stdlib HTTP server.
+No dependency, no database, no network access: it is meant for watching a run on
+the machine that is being measured.
+"""
+
+import dataclasses
+import json
+import threading
+from collections import deque
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from importlib import resources
+from typing import List
+
+from codecarbon.external.logger import logger
+from codecarbon.output_methods.base_output import BaseOutput
+from codecarbon.output_methods.emissions_data import EmissionsData, TaskEmissionsData
+
+# Fields kept for every sample. The rest of EmissionsData is either static
+# (hardware, geography) or not useful on a live chart.
+SAMPLE_FIELDS = (
+ "timestamp",
+ "duration",
+ "cpu_power",
+ "gpu_power",
+ "ram_power",
+ "energy_consumed",
+ "cpu_utilization_percent",
+ "gpu_utilization_percent",
+ "ram_utilization_percent",
+)
+
+METADATA_FIELDS = (
+ "project_name",
+ "experiment_id",
+ "run_id",
+ "cpu_count",
+ "cpu_model",
+ "gpu_count",
+ "gpu_model",
+ "ram_total_size",
+ "country_name",
+ "country_iso_code",
+ "region",
+ "os",
+ "python_version",
+ "codecarbon_version",
+ "tracking_mode",
+)
+
+
+def _page() -> bytes:
+ return resources.files("codecarbon.viz").joinpath("live.html").read_bytes()
+
+
+class LiveDashboardOutput(BaseOutput):
+ """
+ Serve a live view of the current run on http://:.
+
+ Usage::
+
+ tracker = EmissionsTracker(output_handlers=[LiveDashboardOutput()])
+
+ The handler keeps at most ``history`` samples in memory, so it is safe to
+ leave running for days. If the port is already taken the handler logs an
+ error and stays inert: a busy port must never take down a measurement run.
+ """
+
+ def __init__(self, port: int = 8050, host: str = "127.0.0.1", history: int = 720):
+ self.port = port
+ self.host = host
+ self._history = deque(maxlen=history)
+ self._metadata = {}
+ self._tasks = []
+ self._lock = threading.Lock()
+ self._server = None
+ self._start_server()
+
+ def _start_server(self) -> None:
+ handler_self = self
+
+ class Handler(BaseHTTPRequestHandler):
+ def do_GET(self): # http.server API naming
+ if self.path.startswith("/data"):
+ self._respond(
+ 200, "application/json", handler_self._snapshot().encode()
+ )
+ elif self.path.startswith("/health"):
+ self._respond(200, "application/json", b'{"status": "ok"}')
+ elif self.path == "/":
+ self._respond(200, "text/html; charset=utf-8", _page())
+ else:
+ self._respond(404, "text/plain", b"not found")
+
+ def _respond(self, status, content_type, body):
+ self.send_response(status)
+ self.send_header("Content-Type", content_type)
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+ def log_message(self, *args):
+ """Silence the default stderr access log."""
+
+ try:
+ self._server = ThreadingHTTPServer((self.host, self.port), Handler)
+ except OSError as e:
+ logger.error(
+ f"Live dashboard could not bind {self.host}:{self.port} ({e}). "
+ "Continuing without the live dashboard."
+ )
+ return
+
+ # The OS assigns the port when 0 was requested, so report the real one.
+ self.port = self._server.server_address[1]
+ if self.host not in ("127.0.0.1", "localhost", "::1"):
+ logger.warning(
+ f"Live dashboard is listening on {self.host}:{self.port} and is "
+ "not authenticated. Prefer 127.0.0.1 with SSH port forwarding."
+ )
+ threading.Thread(
+ target=self._server.serve_forever, daemon=True, name="codecarbon-live-ui"
+ ).start()
+ logger.info(f"Live dashboard available on http://{self.host}:{self.port}")
+
+ @property
+ def is_serving(self) -> bool:
+ return self._server is not None
+
+ @property
+ def url(self) -> str:
+ return f"http://{self.host}:{self.port}"
+
+ def _snapshot(self) -> str:
+ with self._lock:
+ return json.dumps(
+ {
+ "samples": list(self._history),
+ "metadata": self._metadata,
+ "tasks": self._tasks,
+ }
+ )
+
+ def on_measure(self, total: EmissionsData):
+ """
+ Record one sample. Called by the tracker after every measurement, with
+ the power fields holding the last measured power rather than the
+ average since ``start()``.
+
+ Defining this method is what opts the handler into the per-measurement
+ cadence; `live_out` and `out` are deliberately left as no-ops so the
+ chart has a single feed of comparable samples.
+ """
+ values = total.values
+ sample = {k: values[k] for k in SAMPLE_FIELDS}
+ # Grams are what a human reads; kg is what the dataclass carries.
+ sample["emissions_g"] = total.emissions * 1000
+ with self._lock:
+ self._history.append(sample)
+ self._metadata = {k: values[k] for k in METADATA_FIELDS}
+
+ def task_out(self, data: List[TaskEmissionsData], experiment_name: str):
+ tasks = [dataclasses.asdict(task) for task in data]
+ with self._lock:
+ self._tasks = tasks
+
+ def exit(self):
+ if self._server is not None:
+ self._server.shutdown()
+ self._server.server_close()
+ self._server = None
diff --git a/docs/how-to/visualize.md b/docs/how-to/visualize.md
index f307d4b48..552804c7a 100644
--- a/docs/how-to/visualize.md
+++ b/docs/how-to/visualize.md
@@ -1,6 +1,42 @@
# Visualize
-CodeCarbon provides two ways to visualize your emissions data: a local Python dashboard for offline analysis, and an online web dashboard for cloud-based tracking and team collaboration.
+CodeCarbon provides three ways to visualize your emissions data: a live local dashboard for watching a run in progress, a local Python dashboard for offline analysis of finished runs, and an online web dashboard for cloud-based tracking and team collaboration.
+
+## Live Local Dashboard
+
+To watch power and emissions while a run is in progress, add `--ui` to the `monitor` command:
+
+``` bash
+codecarbon monitor --ui
+```
+
+CodeCarbon then serves a page on `http://127.0.0.1:8050` showing current power draw, cumulative emissions, energy, elapsed time, a chart of CPU / GPU / RAM power over the last couple of hours, and the hardware it detected. It is also a quick way to check that your GPU is being read.
+
+The dashboard uses only the Python standard library: no extra dependency, no database and no internet access are required.
+
+**Options:**
+
+- `--ui-port`: port to serve on (default `8050`)
+- `--ui-host`: interface to bind to (default `127.0.0.1`)
+
+The server is unauthenticated by design, so it listens on localhost only. To view it from another machine, use SSH port forwarding (`ssh -L 8050:127.0.0.1:8050 user@host`) rather than binding to a public interface.
+
+It also works with a wrapped command:
+
+``` bash
+codecarbon monitor --ui -- python train.py
+```
+
+From Python, the same view is available as an output handler:
+
+``` python
+from codecarbon import EmissionsTracker
+from codecarbon.viz.live import LiveDashboardOutput
+
+tracker = EmissionsTracker(output_handlers=[LiveDashboardOutput(port=8050)])
+```
+
+The live dashboard only shows the current run; for historical analysis across runs, use carbonboard below.
## Offline Visualization (carbonboard)
diff --git a/docs/reference/cli.md b/docs/reference/cli.md
index 71ba05a85..d0e9258e7 100644
--- a/docs/reference/cli.md
+++ b/docs/reference/cli.md
@@ -45,6 +45,9 @@ Displays real-time emissions data for all processes on your machine. Press `Ctrl
| `--offline` | flag | false | Run without internet access |
| `--country-iso-code` | string | - | ISO 3166-1 alpha-3 country code (required in offline mode) |
| `--log-level` | choice | ERROR | Log level: DEBUG, INFO, WARNING, ERROR |
+| `--ui` | flag | false | Serve a live dashboard in your browser |
+| `--ui-port` | int | 8050 | Port of the live dashboard |
+| `--ui-host` | string | 127.0.0.1 | Host to bind the live dashboard to |
**Examples:**
```bash
@@ -59,6 +62,9 @@ codecarbon monitor --offline --country-iso-code FRA
# Monitor with debug logging
codecarbon monitor --log-level DEBUG
+
+# Monitor with the live dashboard on http://127.0.0.1:8050
+codecarbon monitor --ui
```
### `codecarbon monitor -- `
diff --git a/pyproject.toml b/pyproject.toml
index 26a360338..c4fc49782 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -65,6 +65,7 @@ codecarbon = [
"data/private_infra/nordic_emissions.json",
"data/private_infra/2016/usa_emissions.json",
"data/private_infra/2023/canada_energy_mix.json",
+ "viz/live.html",
"viz/assets/car_icon.png",
"viz/assets/house_icon.png",
"viz/assets/tv_icon.png"
@@ -167,5 +168,10 @@ source = [
"codecarbon",
]
omit = [
- "codecarbon/viz/*",
+ # The legacy Dash dashboard is not unit tested. `live.py` is, so it stays in.
+ "codecarbon/viz/carbonboard.py",
+ "codecarbon/viz/carbonboard_on_api.py",
+ "codecarbon/viz/components.py",
+ "codecarbon/viz/data.py",
+ "codecarbon/viz/units.py",
]
diff --git a/tests/cli/test_cli_main.py b/tests/cli/test_cli_main.py
index 8bb4d66f4..36d16848a 100644
--- a/tests/cli/test_cli_main.py
+++ b/tests/cli/test_cli_main.py
@@ -386,6 +386,96 @@ def stop(self):
assert calls["kwargs"]["region"] == "IDF"
+def _invoke_offline_monitor(monkeypatch, extra_args):
+ """Run `codecarbon monitor --offline ...` against a tracker that does nothing."""
+ calls = {"kwargs": None}
+
+ class FakeOfflineTracker:
+ def __init__(self, **kwargs):
+ calls["kwargs"] = kwargs
+ # Breaks the CLI's infinite monitoring loop on the first iteration.
+ self._another_instance_already_running = True
+
+ def start(self):
+ pass
+
+ def stop(self):
+ return None
+
+ monkeypatch.setattr(
+ "codecarbon.emissions_tracker.OfflineEmissionsTracker", FakeOfflineTracker
+ )
+ monkeypatch.setattr(cli_main.signal, "signal", lambda *args, **kwargs: None)
+
+ result = CliRunner().invoke(
+ cli_main.codecarbon,
+ ["monitor", "--offline", "--country-iso-code", "FRA"] + extra_args,
+ )
+ return result, calls["kwargs"]
+
+
+def test_monitor_without_ui_registers_no_output_handler(monkeypatch):
+ result, kwargs = _invoke_offline_monitor(monkeypatch, [])
+
+ assert result.exit_code == 0
+ assert "output_handlers" not in kwargs
+
+
+def test_monitor_ui_registers_a_serving_live_dashboard(monkeypatch):
+ from codecarbon.viz.live import LiveDashboardOutput
+
+ # Port 0 lets the OS pick a free port, so the test never collides on CI.
+ result, kwargs = _invoke_offline_monitor(monkeypatch, ["--ui", "--ui-port", "0"])
+
+ (handler,) = kwargs["output_handlers"]
+ try:
+ assert isinstance(handler, LiveDashboardOutput)
+ assert handler.is_serving
+ assert handler.host == "127.0.0.1"
+ assert handler.port != 0
+ assert result.exit_code == 0
+ assert f"Live dashboard: http://127.0.0.1:{handler.port}" in result.output
+ finally:
+ handler.exit()
+
+
+def test_monitor_ui_host_is_passed_to_the_dashboard(monkeypatch):
+ _, kwargs = _invoke_offline_monitor(
+ monkeypatch, ["--ui", "--ui-port", "0", "--ui-host", "localhost"]
+ )
+
+ (handler,) = kwargs["output_handlers"]
+ try:
+ assert handler.host == "localhost"
+ assert handler.url == f"http://localhost:{handler.port}"
+ finally:
+ handler.exit()
+
+
+def test_monitor_ui_on_a_busy_port_warns_but_keeps_monitoring(monkeypatch):
+ import socket
+
+ with socket.socket() as sock:
+ sock.bind(("127.0.0.1", 0))
+ sock.listen(1)
+ busy_port = sock.getsockname()[1]
+
+ result, kwargs = _invoke_offline_monitor(
+ monkeypatch, ["--ui", "--ui-port", str(busy_port)]
+ )
+
+ (handler,) = kwargs["output_handlers"]
+ try:
+ # A busy port must never take the run down: the tracker still starts.
+ assert result.exit_code == 0
+ assert not handler.is_serving
+ assert f"could not start the live dashboard on 127.0.0.1:{busy_port}" in (
+ result.output
+ )
+ finally:
+ handler.exit()
+
+
def test_monitor_delegates_offline_flag_to_run_and_monitor(monkeypatch):
captured = {}
diff --git a/tests/test_live_dashboard.py b/tests/test_live_dashboard.py
new file mode 100644
index 000000000..b7868b9cc
--- /dev/null
+++ b/tests/test_live_dashboard.py
@@ -0,0 +1,330 @@
+import dataclasses
+import json
+import socket
+import threading
+import unittest
+import urllib.error
+import urllib.request
+
+from codecarbon.output_methods.base_output import BaseOutput
+from codecarbon.output_methods.emissions_data import EmissionsData, TaskEmissionsData
+from codecarbon.viz.live import LiveDashboardOutput
+
+
+def _make_emissions_data(**overrides) -> EmissionsData:
+ defaults = dict(
+ timestamp="2025-01-15T10:30:00",
+ project_name="test_project",
+ run_id="550e8400-e29b-41d4-a716-446655440000",
+ experiment_id="exp-001",
+ duration=3600.0,
+ emissions=0.042,
+ emissions_rate=1.17e-05,
+ cpu_power=12.5,
+ gpu_power=85.0,
+ ram_power=3.2,
+ cpu_energy=0.0125,
+ gpu_energy=0.085,
+ ram_energy=0.0032,
+ energy_consumed=0.1007,
+ water_consumed=0.0,
+ country_name="France",
+ country_iso_code="FRA",
+ region="Ile-de-France",
+ cloud_provider="",
+ cloud_region="",
+ os="Linux-5.15.0",
+ python_version="3.11.5",
+ codecarbon_version="2.5.0",
+ cpu_count=8,
+ cpu_model="Intel Core i7-12700",
+ gpu_count=1,
+ gpu_model="NVIDIA RTX 3090",
+ longitude=2.3522,
+ latitude=48.8566,
+ ram_total_size=32.0,
+ tracking_mode="machine",
+ )
+ defaults.update(overrides)
+ return EmissionsData(**defaults)
+
+
+def _make_task_emissions_data(**overrides) -> TaskEmissionsData:
+ fields = {f.name for f in dataclasses.fields(TaskEmissionsData)}
+ defaults = {
+ k: v
+ for k, v in dataclasses.asdict(_make_emissions_data()).items()
+ if k in fields
+ }
+ defaults["task_name"] = "a_task"
+ defaults.update(overrides)
+ return TaskEmissionsData(**defaults)
+
+
+def _get(url):
+ with urllib.request.urlopen(url, timeout=5) as response:
+ return response.status, response.read()
+
+
+class TestLiveDashboardOutput(unittest.TestCase):
+ def setUp(self):
+ # Port 0 lets the OS pick a free port, so tests never collide.
+ self.output = LiveDashboardOutput(port=0, history=5)
+ self.addCleanup(self.output.exit)
+
+ def test_on_measure_appends_and_respects_maxlen(self):
+ for i in range(15):
+ self.output.on_measure(_make_emissions_data(duration=float(i)))
+
+ samples = json.loads(self.output._snapshot())["samples"]
+ self.assertEqual(5, len(samples))
+ self.assertEqual(
+ [10.0, 11.0, 12.0, 13.0, 14.0], [s["duration"] for s in samples]
+ )
+
+ def test_data_endpoint_returns_expected_payload(self):
+ self.output.on_measure(_make_emissions_data())
+
+ status, body = _get(f"{self.output.url}/data")
+ payload = json.loads(body)
+
+ self.assertEqual(200, status)
+ self.assertEqual(1, len(payload["samples"]))
+ sample = payload["samples"][0]
+ self.assertEqual(12.5, sample["cpu_power"])
+ self.assertEqual(85.0, sample["gpu_power"])
+ self.assertAlmostEqual(42.0, sample["emissions_g"])
+ self.assertEqual("NVIDIA RTX 3090", payload["metadata"]["gpu_model"])
+ self.assertEqual([], payload["tasks"])
+
+ def test_index_serves_html_and_unknown_path_is_404(self):
+ status, body = _get(self.output.url + "/")
+ self.assertEqual(200, status)
+ self.assertIn(b"CodeCarbon", body)
+
+ with self.assertRaises(urllib.error.HTTPError) as context:
+ _get(f"{self.output.url}/nope")
+ self.assertEqual(404, context.exception.code)
+
+ def test_health_endpoint(self):
+ status, body = _get(f"{self.output.url}/health")
+ self.assertEqual(200, status)
+ self.assertEqual({"status": "ok"}, json.loads(body))
+
+ def test_concurrent_reads_and_writes_return_valid_json(self):
+ errors = []
+ stop = threading.Event()
+
+ def write():
+ while not stop.is_set():
+ self.output.on_measure(_make_emissions_data())
+
+ def read():
+ try:
+ for _ in range(20):
+ json.loads(_get(f"{self.output.url}/data")[1])
+ except Exception as e: # pragma: no cover - only on failure
+ errors.append(e)
+
+ writer = threading.Thread(target=write, daemon=True)
+ writer.start()
+ readers = [threading.Thread(target=read) for _ in range(4)]
+ for reader in readers:
+ reader.start()
+ for reader in readers:
+ reader.join()
+ stop.set()
+ writer.join()
+
+ self.assertEqual([], errors)
+
+ def test_exit_releases_the_port(self):
+ port = self.output.port
+ self.output.exit()
+
+ with socket.socket() as sock:
+ sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ sock.bind(("127.0.0.1", port))
+
+ def test_tracker_feeds_a_sample_on_every_measure(self):
+ from codecarbon.emissions_tracker import OfflineEmissionsTracker
+
+ tracker = OfflineEmissionsTracker(
+ country_iso_code="FRA",
+ measure_power_secs=1,
+ api_call_interval=30,
+ save_to_file=False,
+ output_handlers=[self.output],
+ allow_multiple_runs=True,
+ )
+ tracker.start()
+ try:
+ # Two measures, well below api_call_interval: without the
+ # `on_measure` opt-in nothing would arrive for minutes.
+ tracker._measure_power_and_energy()
+ tracker._measure_power_and_energy()
+ samples = json.loads(self.output._snapshot())["samples"]
+ finally:
+ tracker.stop()
+
+ self.assertGreaterEqual(len(samples), 2)
+
+ def test_busy_port_does_not_raise(self):
+ with socket.socket() as sock:
+ sock.bind(("127.0.0.1", 0))
+ sock.listen(1)
+ busy_port = sock.getsockname()[1]
+
+ output = LiveDashboardOutput(port=busy_port)
+ self.addCleanup(output.exit)
+
+ self.assertFalse(output.is_serving)
+ # Still usable as an output handler, it just serves nothing.
+ output.on_measure(_make_emissions_data())
+ self.assertEqual(1, len(json.loads(output._snapshot())["samples"]))
+
+ def test_only_on_measure_feeds_the_history(self):
+ # `out` and `live_out` carry power averaged since start(); mixing them
+ # into the chart would put incomparable points on the same line.
+ self.output.out(_make_emissions_data(duration=7.0), None)
+ self.output.live_out(_make_emissions_data(duration=8.0), None)
+
+ samples = json.loads(_get(f"{self.output.url}/data")[1])["samples"]
+ self.assertEqual([], samples)
+
+ def test_task_out_is_served_on_the_data_endpoint(self):
+ self.output.task_out(
+ [_make_task_emissions_data(task_name="train", emissions=0.5)],
+ "experiment",
+ )
+
+ tasks = json.loads(_get(f"{self.output.url}/data")[1])["tasks"]
+ self.assertEqual(1, len(tasks))
+ self.assertEqual("train", tasks[0]["task_name"])
+ self.assertEqual(0.5, tasks[0]["emissions"])
+
+ # A second call replaces the previous list rather than appending to it.
+ self.output.task_out(
+ [_make_task_emissions_data(task_name="eval")], "experiment"
+ )
+ tasks = json.loads(_get(f"{self.output.url}/data")[1])["tasks"]
+ self.assertEqual(["eval"], [t["task_name"] for t in tasks])
+
+ def test_binding_outside_loopback_warns_about_the_open_port(self):
+ with self.assertLogs("codecarbon", level="WARNING") as logs:
+ output = LiveDashboardOutput(port=0, host="0.0.0.0")
+ self.addCleanup(output.exit)
+
+ self.assertTrue(output.is_serving)
+ self.assertTrue(
+ any("not authenticated" in message for message in logs.output),
+ logs.output,
+ )
+
+ def test_exit_is_idempotent(self):
+ self.output.exit()
+ self.output.exit()
+ self.assertFalse(self.output.is_serving)
+
+ def test_data_is_served_after_the_os_assigned_the_port(self):
+ # port=0 means the OS picks: the handler must report the real port, not 0.
+ self.assertNotEqual(0, self.output.port)
+ self.assertEqual(f"http://127.0.0.1:{self.output.port}", self.output.url)
+ self.assertEqual(200, _get(f"{self.output.url}/health")[0])
+
+
+class _RecordingOutput(BaseOutput):
+ """Plain handler: fed only every `api_call_interval` measures."""
+
+ def __init__(self):
+ self.calls = []
+
+ def live_out(self, total, delta):
+ self.calls.append((total, delta))
+
+
+class _RecordingLiveOutput(_RecordingOutput):
+ """Opt-in handler: defining `on_measure` gets it fed on every measure."""
+
+ def __init__(self):
+ super().__init__()
+ self.measures = []
+
+ def on_measure(self, total):
+ self.measures.append(total)
+
+
+class TestLiveOutEveryMeasure(unittest.TestCase):
+ """The tracker must not feed an every-measure handler twice on an API tick."""
+
+ def _tracker(self, handlers, api_call_interval):
+ from codecarbon.emissions_tracker import OfflineEmissionsTracker
+
+ tracker = OfflineEmissionsTracker(
+ country_iso_code="FRA",
+ measure_power_secs=1,
+ api_call_interval=api_call_interval,
+ save_to_file=False,
+ output_handlers=handlers,
+ allow_multiple_runs=True,
+ )
+ tracker.start()
+ self.addCleanup(tracker.stop)
+ # `start()` takes a first measure: ignore it so the counts below are exact.
+ tracker._measure_occurrence = 0
+ for handler in handlers:
+ handler.calls.clear()
+ getattr(handler, "measures", []).clear()
+ return tracker
+
+ def test_on_measure_is_called_every_measure_and_live_out_stays_periodic(self):
+ live, plain = _RecordingLiveOutput(), _RecordingOutput()
+ tracker = self._tracker([live, plain], api_call_interval=3)
+
+ tracker._measure_power_and_energy()
+ tracker._measure_power_and_energy()
+
+ # Below api_call_interval: only the `on_measure` opt-in has been fed.
+ self.assertEqual(2, len(live.measures))
+ self.assertEqual([], live.calls)
+ self.assertEqual([], plain.calls)
+
+ tracker._measure_power_and_energy()
+
+ # The API tick feeds `live_out` on every handler, with a delta.
+ self.assertEqual(3, len(live.measures))
+ self.assertEqual(1, len(live.calls))
+ self.assertEqual(1, len(plain.calls))
+ self.assertIsNotNone(plain.calls[0][1])
+
+ def test_on_measure_receives_instantaneous_power_not_the_average(self):
+ from codecarbon.core.units import Power
+
+ live = _RecordingLiveOutput()
+ tracker = self._tracker([live], api_call_interval=-1)
+
+ # No real hardware in the loop: the measurement writes known values,
+ # and the running sums (the averages EmissionsData carries) are
+ # poisoned to unmistakably different ones.
+ def fake_measurements():
+ tracker._cpu_power = Power.from_watts(42.0)
+ tracker._gpu_power = Power.from_watts(7.0)
+ tracker._ram_power = Power.from_watts(3.0)
+
+ tracker._do_measurements = fake_measurements
+ tracker._cpu_power_sum = 10_000.0
+ tracker._gpu_power_sum = 20_000.0
+ tracker._ram_power_sum = 30_000.0
+ tracker._power_measurement_count = 100
+
+ tracker._measure_power_and_energy()
+
+ sample = live.measures[-1]
+ self.assertEqual(42.0, sample.cpu_power)
+ self.assertEqual(7.0, sample.gpu_power)
+ self.assertEqual(3.0, sample.ram_power)
+ # ... while the periodic path still reports the averages (sum / count).
+ averaged = tracker._prepare_emissions_data()
+ self.assertEqual(100.0, averaged.cpu_power)
+ self.assertEqual(200.0, averaged.gpu_power)
+ self.assertEqual(300.0, averaged.ram_power)