From 5f110a883af6bbb3e7324192a9258ec366eaebdc Mon Sep 17 00:00:00 2001 From: Harsh Raj Singhania <40535627+HarshRajSinghania@users.noreply.github.com> Date: Mon, 21 Sep 2026 14:10:51 +0530 Subject: [PATCH] feat(cli): expose tracker options on codecarbon monitor Generate additional `codecarbon monitor` flags from BaseEmissionsTracker.__init__ so options such as --pue and --wue stay in sync with the package API. Fixes #1273 --- codecarbon/cli/main.py | 20 ++++- codecarbon/cli/tracker_options.py | 118 ++++++++++++++++++++++++++++++ tests/cli/test_tracker_options.py | 71 ++++++++++++++++++ 3 files changed, 208 insertions(+), 1 deletion(-) create mode 100644 codecarbon/cli/tracker_options.py create mode 100644 tests/cli/test_tracker_options.py diff --git a/codecarbon/cli/main.py b/codecarbon/cli/main.py index f81fb302a..14f5dd1b9 100644 --- a/codecarbon/cli/main.py +++ b/codecarbon/cli/main.py @@ -18,6 +18,10 @@ get_existing_exp_id, overwrite_local_config, ) +from codecarbon.cli.tracker_options import ( + extend_monitor_signature, + extra_tracker_param_names, +) API_URL = os.environ.get("API_URL", "https://dashboard.codecarbon.io/api") @@ -418,8 +422,14 @@ def monitor( str, typer.Option(help="Log level (critical, error, warning, info, debug)"), ] = "error", + **tracker_kwargs, ): - """Monitor your machine's carbon emissions.""" + """Monitor your machine's carbon emissions. + + Additional flags are generated from ``EmissionsTracker`` constructor + arguments (for example ``--pue`` and ``--wue``) so the CLI stays in + sync with the package API. + """ external_conf = _external_config() @@ -434,6 +444,12 @@ def monitor( tracker_args = { name: value for name, value in cli_defaults if _cli_provided(ctx, name) } + for name in extra_tracker_param_names(): + value = tracker_kwargs.get(name) + if value is None and ctx is not None: + value = ctx.params.get(name) + if value is not None: + tracker_args[name] = value for name, value in cli_defaults: if name not in tracker_args and name not in external_conf: # Nothing configures it: keep the defaults advertised by `--help` @@ -546,3 +562,5 @@ def questionary_prompt(prompt, list_options, default): if __name__ == "__main__": main() + +extend_monitor_signature(monitor) diff --git a/codecarbon/cli/tracker_options.py b/codecarbon/cli/tracker_options.py new file mode 100644 index 000000000..8a62e34b2 --- /dev/null +++ b/codecarbon/cli/tracker_options.py @@ -0,0 +1,118 @@ +"""Derive extra `codecarbon monitor` flags from the tracker constructor. + +The option list is generated from ``BaseEmissionsTracker.__init__`` so it +stays aligned with the package API (issue #1273). +""" + +from __future__ import annotations + +import inspect +import re +from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union, get_args, get_origin + + +_SKIP_PARAMS = { + "self", + "logging_logger", + "output_handlers", + "output_methods", +} + +_DECLARED_ON_MONITOR = { + "api_call_interval", + "country_iso_code", + "log_level", + "measure_power_secs", + "region", + "save_to_api", +} + + +def _unwrap_optional(annotation: Any) -> Any: + origin = get_origin(annotation) + if origin is Union: + non_none = [arg for arg in get_args(annotation) if arg is not type(None)] + if len(non_none) == 1: + return non_none[0] + return annotation + + +def _eval_annotation(raw: str) -> Any: + """Best-effort eval of a postponed annotation string.""" + namespace = { + "Optional": Optional, + "Union": Union, + "List": list, + "str": str, + "int": int, + "float": float, + "bool": bool, + } + try: + return eval(raw, {"__builtins__": {}}, namespace) + except Exception: + return raw + + +def _cli_type(annotation: Any) -> Optional[type]: + inner = _unwrap_optional(annotation) + if inner in (str, int, float, bool): + return inner + return None + + +def _param_help_from_docstring(doc: Optional[str]) -> Dict[str, str]: + if not doc: + return {} + help_map: Dict[str, str] = {} + pattern = re.compile( + r":param\s+(\w+)\s*:\s*(.+?)(?=\n\s*:param|\n\s*:return|\Z)", + re.S, + ) + for name, text in pattern.findall(doc): + help_map[name] = " ".join(text.split())[:200] + return help_map + + +def iter_tracker_cli_params() -> Iterable[Tuple[str, type, str]]: + """Yield ``(name, python_type, help)`` for constructor args mapped to flags.""" + from codecarbon.emissions_tracker import BaseEmissionsTracker + + signature = inspect.signature(BaseEmissionsTracker.__init__) + help_map = _param_help_from_docstring(BaseEmissionsTracker.__init__.__doc__) + for name, param in signature.parameters.items(): + if name in _SKIP_PARAMS or name in _DECLARED_ON_MONITOR: + continue + annotation = param.annotation + if isinstance(annotation, str): + annotation = _eval_annotation(annotation) + cli_type = _cli_type(annotation) + if cli_type is None: + continue + yield name, cli_type, help_map.get(name, name.replace("_", " ")) + + +def extra_tracker_param_names() -> List[str]: + return [name for name, _, _ in iter_tracker_cli_params()] + + +def extend_monitor_signature(func: Callable) -> Callable: + """Add generated keyword-only parameters so Typer exposes matching flags.""" + sig = inspect.signature(func) + params = list(sig.parameters.values()) + existing = {param.name for param in params} + extra = [] + for name, cli_type, _help_text in iter_tracker_cli_params(): + if name in existing: + continue + extra.append( + inspect.Parameter( + name, + kind=inspect.Parameter.KEYWORD_ONLY, + default=None, + annotation=Optional[cli_type], + ) + ) + head = [p for p in params if p.kind != inspect.Parameter.VAR_KEYWORD] + func.__signature__ = sig.replace(parameters=head + extra) + return func diff --git a/tests/cli/test_tracker_options.py b/tests/cli/test_tracker_options.py new file mode 100644 index 000000000..a26bb8bdf --- /dev/null +++ b/tests/cli/test_tracker_options.py @@ -0,0 +1,71 @@ +from codecarbon.cli.tracker_options import extra_tracker_param_names, iter_tracker_cli_params + + +def test_generated_options_include_pue_and_wue(): + names = extra_tracker_param_names() + assert "pue" in names + assert "wue" in names + assert "measure_power_secs" not in names + assert "logging_logger" not in names + + +def test_generated_option_types(): + types = {name: cli_type for name, cli_type, _ in iter_tracker_cli_params()} + assert types["pue"] is float + assert types["wue"] is float + assert types["project_name"] is str + assert types["allow_multiple_runs"] is bool + + +def test_monitor_forwards_generated_tracker_options(monkeypatch): + import codecarbon.cli.main as cli_main + from typer.testing import CliRunner + + calls = {"kwargs": None} + + class FakeOfflineTracker: + def __init__(self, **kwargs): + calls["kwargs"] = kwargs + self._another_instance_already_running = True + + def start(self): + return None + + 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", + "--pue", + "1.4", + "--wue", + "0.5", + "--project-name", + "cli-monitor", + ], + ) + assert result.exit_code == 0, result.output + assert calls["kwargs"]["pue"] == 1.4 + assert calls["kwargs"]["wue"] == 0.5 + assert calls["kwargs"]["project_name"] == "cli-monitor" + + +def test_monitor_help_lists_generated_package_options(): + import codecarbon.cli.main as cli_main + from typer.testing import CliRunner + + result = CliRunner().invoke(cli_main.codecarbon, ["monitor", "--help"]) + assert result.exit_code == 0 + assert "--pue" in result.output + assert "--wue" in result.output + assert "--force-cpu-power" in result.output