From a7ace931876ba7c64c21334056d836d4c69a382c Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 17:55:06 +0200 Subject: [PATCH 1/4] feat(cli): live local dashboard for monitor Add `codecarbon monitor --ui`, serving a single self-contained page over a stdlib HTTP server so a run can be watched while it happens, without the carbonboard extra, carbonserver or any network access. Output handlers can now opt into `live_out` on every measurement via `live_out_every_measure`, instead of once per `api_call_interval`. Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/cli/main.py | 23 +++ codecarbon/emissions_tracker.py | 17 ++- codecarbon/output_methods/base_output.py | 5 + codecarbon/viz/live.html | 142 +++++++++++++++++++ codecarbon/viz/live.py | 170 +++++++++++++++++++++++ docs/how-to/visualize.md | 38 ++++- docs/reference/cli.md | 6 + pyproject.toml | 1 + tests/test_live_dashboard.py | 170 +++++++++++++++++++++++ 9 files changed, 570 insertions(+), 2 deletions(-) create mode 100644 codecarbon/viz/live.html create mode 100644 codecarbon/viz/live.py create mode 100644 tests/test_live_dashboard.py diff --git a/codecarbon/cli/main.py b/codecarbon/cli/main.py index 93f627e5b..a70277c0f 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["output_handlers"] = [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..7af33df6e 100644 --- a/codecarbon/emissions_tracker.py +++ b/codecarbon/emissions_tracker.py @@ -1275,6 +1275,20 @@ 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 want every measure, not one every + # `api_call_interval`. They get the total only: computing the delta here + # would consume it for the periodic call below. + every_measure_handlers = [ + handler + for handler in self._output_handlers + if getattr(handler, "live_out_every_measure", False) + ] + if every_measure_handlers: + total = self._prepare_emissions_data() + for handler in every_measure_handlers: + handler.live_out(total, None) + # Special case: metrics and api calls are sent every `api_call_interval` measures if ( self._api_call_interval != -1 @@ -1288,7 +1302,8 @@ def _measure_power_and_energy(self) -> None: + f"{emissions_delta.emissions_rate * 3600 * 24 * 365:,} kg.CO2eq/year" ) for handler in self._output_handlers: - handler.live_out(emissions, emissions_delta) + if not getattr(handler, "live_out_every_measure", False): + handler.live_out(emissions, emissions_delta) self._measure_occurrence = 0 logger.debug(f"last_duration={last_duration}\n------------------------") diff --git a/codecarbon/output_methods/base_output.py b/codecarbon/output_methods/base_output.py index 373d23edd..dfba2cd11 100644 --- a/codecarbon/output_methods/base_output.py +++ b/codecarbon/output_methods/base_output.py @@ -41,6 +41,11 @@ class BaseOutput: emissions segregated by task """ + #: When True, `live_out` is called on every measurement instead of every + #: `api_call_interval` measurements, and `delta` is passed as None. + #: Only useful for handlers that display data locally. + live_out_every_measure = False + def out(self, total: EmissionsData, delta: EmissionsData): pass 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 + +
+ +
ComponentModelPower (W)Load
+ +

+ + + + diff --git a/codecarbon/viz/live.py b/codecarbon/viz/live.py new file mode 100644 index 000000000..e37ab8295 --- /dev/null +++ b/codecarbon/viz/live.py @@ -0,0 +1,170 @@ +""" +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. + """ + + live_out_every_measure = True + + 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 live_out(self, total: EmissionsData, delta: EmissionsData): + 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 out(self, total: EmissionsData, delta: EmissionsData): + self.live_out(total, delta) + + 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..bee3f2ab9 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" diff --git a/tests/test_live_dashboard.py b/tests/test_live_dashboard.py new file mode 100644 index 000000000..01573c304 --- /dev/null +++ b/tests/test_live_dashboard.py @@ -0,0 +1,170 @@ +import json +import socket +import threading +import unittest +import urllib.error +import urllib.request + +from codecarbon.output_methods.emissions_data import EmissionsData +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 _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_live_out_appends_and_respects_maxlen(self): + for i in range(15): + self.output.live_out(_make_emissions_data(duration=float(i)), None) + + 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.live_out(_make_emissions_data(), None) + + 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.live_out(_make_emissions_data(), None) + + 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 + # `live_out_every_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.live_out(_make_emissions_data(), None) + self.assertEqual(1, len(json.loads(output._snapshot())["samples"])) From 60bd90777aa8261d745f3ad75006abe9e90accf2 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 19:26:44 +0200 Subject: [PATCH 2/4] test: cover the live dashboard CLI wiring Measure codecarbon/viz/live.py instead of omitting it with the legacy Dash dashboard, and cover the paths the new feature added: the --ui/--ui-port/--ui-host wiring (serving and busy-port branches), task_out, out, the non-loopback bind warning, idempotent exit, and the live_out_every_measure handling in the tracker. Co-Authored-By: Claude Opus 5 (1M context) --- pyproject.toml | 7 +- tests/cli/test_cli_main.py | 90 +++++++++++++++++++++++++ tests/test_live_dashboard.py | 126 ++++++++++++++++++++++++++++++++++- 3 files changed, 221 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index bee3f2ab9..c4fc49782 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -168,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 index 01573c304..7de5d3331 100644 --- a/tests/test_live_dashboard.py +++ b/tests/test_live_dashboard.py @@ -1,3 +1,4 @@ +import dataclasses import json import socket import threading @@ -5,7 +6,8 @@ import urllib.error import urllib.request -from codecarbon.output_methods.emissions_data import EmissionsData +from codecarbon.output_methods.base_output import BaseOutput +from codecarbon.output_methods.emissions_data import EmissionsData, TaskEmissionsData from codecarbon.viz.live import LiveDashboardOutput @@ -47,6 +49,18 @@ def _make_emissions_data(**overrides) -> EmissionsData: 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() @@ -168,3 +182,113 @@ def test_busy_port_does_not_raise(self): # Still usable as an output handler, it just serves nothing. output.live_out(_make_emissions_data(), None) self.assertEqual(1, len(json.loads(output._snapshot())["samples"])) + + def test_out_feeds_the_same_history_as_live_out(self): + self.output.out(_make_emissions_data(duration=7.0), None) + + samples = json.loads(_get(f"{self.output.url}/data")[1])["samples"] + self.assertEqual([7.0], [s["duration"] for s in 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: fed on every measure, with no delta.""" + + live_out_every_measure = True + + +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() + return tracker + + def test_every_measure_handler_is_not_called_again_on_the_api_tick(self): + live, plain = _RecordingLiveOutput(), _RecordingOutput() + tracker = self._tracker([live, plain], api_call_interval=1) + + tracker._measure_power_and_energy() + tracker._measure_power_and_energy() + + # api_call_interval=1 makes every measure an API tick: the plain handler + # gets both, the live one still gets exactly one call per measure. + self.assertEqual(2, len(live.calls)) + self.assertEqual(2, len(plain.calls)) + # The every-measure path passes no delta, the periodic one does. + self.assertEqual([None, None], [delta for _, delta in live.calls]) + self.assertTrue(all(delta is not None for _, delta in plain.calls)) + + def test_plain_handler_waits_for_the_api_interval(self): + live, plain = _RecordingLiveOutput(), _RecordingOutput() + tracker = self._tracker([live, plain], api_call_interval=3) + + tracker._measure_power_and_energy() + tracker._measure_power_and_energy() + + self.assertEqual(2, len(live.calls)) + self.assertEqual([], plain.calls) From bb9a00e3bd48ae65e0a5836e539e76512695d624 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 19:50:30 +0200 Subject: [PATCH 3/4] fix(live-ui): plot instantaneous power, drop the BaseOutput flag The live chart read cpu/gpu/ram_power off EmissionsData, which holds the average since start() (the _*_power_sum accumulators are never reset), so the 'live power' line was a running mean that ramped up and then barely decayed: measured 8.5 -> 35.7 -> 10.4 W of real power against a monotone 8.5 -> 22.6 -> 20.4 W average over the same idle/busy/idle run. The tracker now hands the handler the last measured power, and handlers opt into the per-measure cadence by defining `on_measure` instead of the `live_out_every_measure` class attribute that widened BaseOutput for one consumer. LiveDashboardOutput leaves `out`/`live_out` as no-ops so the chart has a single feed of comparable samples. Also append the live handler to output_handlers rather than replacing the list. Co-Authored-By: Claude Opus 5 (1M context) --- codecarbon/cli/main.py | 2 +- codecarbon/emissions_tracker.py | 30 +++++---- codecarbon/output_methods/base_output.py | 5 -- codecarbon/viz/live.py | 16 +++-- tests/test_live_dashboard.py | 78 ++++++++++++++++-------- 5 files changed, 82 insertions(+), 49 deletions(-) diff --git a/codecarbon/cli/main.py b/codecarbon/cli/main.py index a70277c0f..21d301edc 100644 --- a/codecarbon/cli/main.py +++ b/codecarbon/cli/main.py @@ -437,7 +437,7 @@ def monitor( from codecarbon.viz.live import LiveDashboardOutput live_output = LiveDashboardOutput(port=ui_port, host=ui_host) - tracker_args["output_handlers"] = [live_output] + tracker_args.setdefault("output_handlers", []).append(live_output) if live_output.is_serving: print(f"Live dashboard: {live_output.url}") else: diff --git a/codecarbon/emissions_tracker.py b/codecarbon/emissions_tracker.py index 7af33df6e..c4766fb01 100644 --- a/codecarbon/emissions_tracker.py +++ b/codecarbon/emissions_tracker.py @@ -1276,18 +1276,25 @@ def _measure_power_and_energy(self) -> None: self._last_measured_time = time.perf_counter() self._measure_occurrence += 1 - # Handlers displaying data locally want every measure, not one every - # `api_call_interval`. They get the total only: computing the delta here - # would consume it for the periodic call below. - every_measure_handlers = [ - handler + # 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 getattr(handler, "live_out_every_measure", False) + if hasattr(handler, "on_measure") ] - if every_measure_handlers: - total = self._prepare_emissions_data() - for handler in every_measure_handlers: - handler.live_out(total, None) + 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 ( @@ -1302,8 +1309,7 @@ def _measure_power_and_energy(self) -> None: + f"{emissions_delta.emissions_rate * 3600 * 24 * 365:,} kg.CO2eq/year" ) for handler in self._output_handlers: - if not getattr(handler, "live_out_every_measure", False): - handler.live_out(emissions, emissions_delta) + handler.live_out(emissions, emissions_delta) self._measure_occurrence = 0 logger.debug(f"last_duration={last_duration}\n------------------------") diff --git a/codecarbon/output_methods/base_output.py b/codecarbon/output_methods/base_output.py index dfba2cd11..373d23edd 100644 --- a/codecarbon/output_methods/base_output.py +++ b/codecarbon/output_methods/base_output.py @@ -41,11 +41,6 @@ class BaseOutput: emissions segregated by task """ - #: When True, `live_out` is called on every measurement instead of every - #: `api_call_interval` measurements, and `delta` is passed as None. - #: Only useful for handlers that display data locally. - live_out_every_measure = False - def out(self, total: EmissionsData, delta: EmissionsData): pass diff --git a/codecarbon/viz/live.py b/codecarbon/viz/live.py index e37ab8295..2ffdfb8de 100644 --- a/codecarbon/viz/live.py +++ b/codecarbon/viz/live.py @@ -69,8 +69,6 @@ class LiveDashboardOutput(BaseOutput): error and stays inert: a busy port must never take down a measurement run. """ - live_out_every_measure = True - def __init__(self, port: int = 8050, host: str = "127.0.0.1", history: int = 720): self.port = port self.host = host @@ -146,7 +144,16 @@ def _snapshot(self) -> str: } ) - def live_out(self, total: EmissionsData, delta: EmissionsData): + 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. @@ -155,9 +162,6 @@ def live_out(self, total: EmissionsData, delta: EmissionsData): self._history.append(sample) self._metadata = {k: values[k] for k in METADATA_FIELDS} - def out(self, total: EmissionsData, delta: EmissionsData): - self.live_out(total, delta) - def task_out(self, data: List[TaskEmissionsData], experiment_name: str): tasks = [dataclasses.asdict(task) for task in data] with self._lock: diff --git a/tests/test_live_dashboard.py b/tests/test_live_dashboard.py index 7de5d3331..a7eec3d67 100644 --- a/tests/test_live_dashboard.py +++ b/tests/test_live_dashboard.py @@ -72,9 +72,9 @@ def setUp(self): self.output = LiveDashboardOutput(port=0, history=5) self.addCleanup(self.output.exit) - def test_live_out_appends_and_respects_maxlen(self): + def test_on_measure_appends_and_respects_maxlen(self): for i in range(15): - self.output.live_out(_make_emissions_data(duration=float(i)), None) + self.output.on_measure(_make_emissions_data(duration=float(i))) samples = json.loads(self.output._snapshot())["samples"] self.assertEqual(5, len(samples)) @@ -83,7 +83,7 @@ def test_live_out_appends_and_respects_maxlen(self): ) def test_data_endpoint_returns_expected_payload(self): - self.output.live_out(_make_emissions_data(), None) + self.output.on_measure(_make_emissions_data()) status, body = _get(f"{self.output.url}/data") payload = json.loads(body) @@ -117,7 +117,7 @@ def test_concurrent_reads_and_writes_return_valid_json(self): def write(): while not stop.is_set(): - self.output.live_out(_make_emissions_data(), None) + self.output.on_measure(_make_emissions_data()) def read(): try: @@ -160,7 +160,7 @@ def test_tracker_feeds_a_sample_on_every_measure(self): tracker.start() try: # Two measures, well below api_call_interval: without the - # `live_out_every_measure` opt-in nothing would arrive for minutes. + # `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"] @@ -180,14 +180,17 @@ def test_busy_port_does_not_raise(self): self.assertFalse(output.is_serving) # Still usable as an output handler, it just serves nothing. - output.live_out(_make_emissions_data(), None) + output.on_measure(_make_emissions_data()) self.assertEqual(1, len(json.loads(output._snapshot())["samples"])) - def test_out_feeds_the_same_history_as_live_out(self): + 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([7.0], [s["duration"] for s in samples]) + self.assertEqual([], samples) def test_task_out_is_served_on_the_data_endpoint(self): self.output.task_out( @@ -241,9 +244,14 @@ def live_out(self, total, delta): class _RecordingLiveOutput(_RecordingOutput): - """Opt-in handler: fed on every measure, with no delta.""" + """Opt-in handler: defining `on_measure` gets it fed on every measure.""" - live_out_every_measure = True + def __init__(self): + super().__init__() + self.measures = [] + + def on_measure(self, total): + self.measures.append(total) class TestLiveOutEveryMeasure(unittest.TestCase): @@ -266,29 +274,49 @@ def _tracker(self, handlers, api_call_interval): tracker._measure_occurrence = 0 for handler in handlers: handler.calls.clear() + getattr(handler, "measures", []).clear() return tracker - def test_every_measure_handler_is_not_called_again_on_the_api_tick(self): + 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=1) + tracker = self._tracker([live, plain], api_call_interval=3) tracker._measure_power_and_energy() tracker._measure_power_and_energy() - # api_call_interval=1 makes every measure an API tick: the plain handler - # gets both, the live one still gets exactly one call per measure. - self.assertEqual(2, len(live.calls)) - self.assertEqual(2, len(plain.calls)) - # The every-measure path passes no delta, the periodic one does. - self.assertEqual([None, None], [delta for _, delta in live.calls]) - self.assertTrue(all(delta is not None for _, delta in plain.calls)) - - def test_plain_handler_waits_for_the_api_interval(self): - live, plain = _RecordingLiveOutput(), _RecordingOutput() - tracker = self._tracker([live, plain], api_call_interval=3) + # 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): + live = _RecordingLiveOutput() + tracker = self._tracker([live], api_call_interval=-1) + + # The tracker keeps running sums to compute the averages that + # EmissionsData carries. Poison them so an average is unmistakably + # different from the last measured power. + 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() - self.assertEqual(2, len(live.calls)) - self.assertEqual([], plain.calls) + sample = live.measures[-1] + self.assertEqual(tracker._cpu_power.W, sample.cpu_power) + self.assertEqual(tracker._gpu_power.W, sample.gpu_power) + self.assertEqual(tracker._ram_power.W, sample.ram_power) + # ... and the averages the periodic path would have reported are the + # poisoned ones, so the two are genuinely distinguishable here. + averaged = tracker._prepare_emissions_data() + self.assertGreater(averaged.cpu_power, 90.0) + self.assertLess(sample.cpu_power, 90.0) From 61a04726bae5ca85f43e3655919dd5af130439b5 Mon Sep 17 00:00:00 2001 From: David Berenstein Date: Wed, 12 Aug 2026 21:10:52 +0200 Subject: [PATCH 4/4] test: prove instantaneous power with injected values, not real hardware --- tests/test_live_dashboard.py | 28 ++++++++++++++++++---------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/tests/test_live_dashboard.py b/tests/test_live_dashboard.py index a7eec3d67..b7868b9cc 100644 --- a/tests/test_live_dashboard.py +++ b/tests/test_live_dashboard.py @@ -298,12 +298,20 @@ def test_on_measure_is_called_every_measure_and_live_out_stays_periodic(self): 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) - # The tracker keeps running sums to compute the averages that - # EmissionsData carries. Poison them so an average is unmistakably - # different from the last measured power. + # 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 @@ -312,11 +320,11 @@ def test_on_measure_receives_instantaneous_power_not_the_average(self): tracker._measure_power_and_energy() sample = live.measures[-1] - self.assertEqual(tracker._cpu_power.W, sample.cpu_power) - self.assertEqual(tracker._gpu_power.W, sample.gpu_power) - self.assertEqual(tracker._ram_power.W, sample.ram_power) - # ... and the averages the periodic path would have reported are the - # poisoned ones, so the two are genuinely distinguishable here. + 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.assertGreater(averaged.cpu_power, 90.0) - self.assertLess(sample.cpu_power, 90.0) + self.assertEqual(100.0, averaged.cpu_power) + self.assertEqual(200.0, averaged.gpu_power) + self.assertEqual(300.0, averaged.ram_power)