feat(cli): expose tracker options on codecarbon monitor

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)

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

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