API Reference

This page provides the complete technical documentation for all public modules in EcoTrace.

---

Core Module

The EcoTrace class is the main entry point for all monitoring operations.

ecotrace.core

EcoTrace

High-precision carbon tracking engine for production Python.

Monitors CPU and GPU energy consumption at function-level granularity using continuous 50 ms sampling, TDP-based energy estimation, and region-specific carbon intensity factors.

Energy formula

energy (Wh) = TDP × (utilization% / 100) × duration / 3600 gCO2 = (Wh / 1000) × carbon_intensity

Parameters:

Name	Type	Description	Default
region_code		ISO 3166-1 alpha-2 country code for grid carbon intensity lookup. Falls back to DEFAULT_REGION if the code is not recognized.	'GLOBAL'
carbon_limit		Optional carbon budget threshold in gCO2. Reserved for future budget alert functionality.	None
gpu_index		Zero-based index selecting which GPU to monitor when multiple devices are present. Validated to be a non-negative integer.	0
api_key		Optional Google Gemini API key. If not provided, it will check the GEMINI_API_KEY environment variable.	None
grid_api_key		Optional Electricity Maps API key for real-time carbon intensity data. If not provided, checks the ECOTRACE_GRID_API_KEY environment variable. Falls back to static data if unavailable.	None
check_updates		If True (default), checks PyPI for newer versions at startup and prompts the user interactively. Set to False in CI/CD or non-interactive environments.	True

Raises:

Type	Description
TypeError	If gpu_index is not an integer or carbon_limit is not numeric.

Source code in ecotrace/core.py

class EcoTrace:
    """High-precision carbon tracking engine for production Python.

    Monitors CPU and GPU energy consumption at function-level granularity using
    continuous 50 ms sampling, TDP-based energy estimation, and region-specific
    carbon intensity factors.

    Energy formula:
        energy (Wh) = TDP × (utilization% / 100) × duration / 3600
        gCO2 = (Wh / 1000) × carbon_intensity

    Args:
        region_code: ISO 3166-1 alpha-2 country code for grid carbon intensity
            lookup. Falls back to DEFAULT_REGION if the code is not recognized.
        carbon_limit: Optional carbon budget threshold in gCO2. Reserved for
            future budget alert functionality.
        gpu_index: Zero-based index selecting which GPU to monitor when multiple
            devices are present. Validated to be a non-negative integer.
        api_key: Optional Google Gemini API key. If not provided, it will
            check the GEMINI_API_KEY environment variable.
        grid_api_key: Optional Electricity Maps API key for real-time carbon
            intensity data. If not provided, checks the ECOTRACE_GRID_API_KEY
            environment variable. Falls back to static data if unavailable.
        check_updates: If True (default), checks PyPI for newer versions at
            startup and prompts the user interactively. Set to False in CI/CD
            or non-interactive environments.

    Raises:
        TypeError: If gpu_index is not an integer or carbon_limit is not numeric.
    """

    FULL_UTILIZATION_PERCENT = 100.0
    MONITOR_INTERVAL_S = 0.05  # 50 ms
    SAMPLE_BUFFER_SIZE = 10000  # 8+ mins at 50ms buffer
    MONITOR_JOIN_TIMEOUT_S = 1.0
    BASELINE_MEASUREMENT_MS = 100  # 100ms idle baseline measurement

    _instances = weakref.WeakSet()

    SECONDS_PER_HOUR = 3600
    WATTS_PER_KILOWATT = 1000

    def __init__(self, region_code="GLOBAL", carbon_limit=None, gpu_index=0,
                 api_key=None, grid_api_key=None, check_updates=True, quiet=False,
                 on_budget_exceeded=None, session_summary=True, run_label=None):
        # Fail-safe auto-update check
        if check_updates:
            try:
                from .updater import check_for_updates
                from . import __version__
                check_for_updates(__version__)
            except Exception:
                pass

        if not isinstance(gpu_index, int) or gpu_index < 0:
            logger.warning(f"Invalid gpu_index={gpu_index!r}, defaulting to 0.")
            gpu_index = 0

        if carbon_limit is not None:
            if not isinstance(carbon_limit, (int, float)) or carbon_limit <= 0:
                logger.warning(f"Invalid carbon_limit={carbon_limit!r}, disabling limit.")
                carbon_limit = None

        self.carbon_limit = carbon_limit
        self.total_carbon = 0.0
        self.total_energy_kwh = 0.0
        self.gpu_index = gpu_index
        self.api_key = api_key or os.environ.get("GEMINI_API_KEY")
        self.grid_api_key = grid_api_key or os.environ.get("ECOTRACE_GRID_API_KEY")
        self.quiet = quiet

        self._run_id = uuid.uuid4().hex[:12]
        self._run_label = run_label or ""

        self._on_budget_exceeded = on_budget_exceeded
        self._budget_warning_fired = False   # 80% threshold
        self._budget_exceeded_fired = False  # 100% threshold
        self._tracked_functions_count = 0
        self._exporters = []
        self._exporter_pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="EcoTrace-Exporter")

        cli_cfg = load_cli_config()
        cloud_key = api_key if (isinstance(api_key, str) and api_key.startswith("eco_usr_")) else (
            os.environ.get("ECOTRACE_CLOUD_KEY") or cli_cfg.get("api_key")
        )
        self.cloud_key = cloud_key
        if self.cloud_key and isinstance(self.cloud_key, str) and self.cloud_key.startswith("eco_usr_"):
            try:
                from .exporters.cloud import CloudExporter
                cloud_exp = CloudExporter(api_key=self.cloud_key, endpoint=cli_cfg.get("endpoint"))
                self.add_exporter(cloud_exp)
            except Exception as e:
                logger.debug(f"Auto CloudExporter registration failed: {e}")

        self.base_dir = os.path.dirname(os.path.abspath(__file__))
        self.json_path = os.path.join(self.base_dir, "constants.json")
        self.csv_path = os.path.join(self.base_dir, "cpu_data.csv")

        self._constants_data = load_constants(self.json_path)
        self.tdp_db = load_tdp_database(self.csv_path)

        final_region = region_code
        if region_code == "GLOBAL":
            detected = identify_user_region()
            if detected:
                final_region = detected
                logger.debug(f"Detected region: {final_region}")
            else:
                logger.info(f"Using default region: {DEFAULT_REGION}")

        self.region_code = validate_region_code(final_region, self._constants_data)

        self._grid_cache_timestamp = 0.0
        self._grid_cached_intensity = None
        self._intensity_source = "static"

        self.carbon_intensity = self._resolve_intensity_with_live_fallback()

        self.gpu_tdp_defaults = load_gpu_tdp_defaults(self._constants_data)
        self.cpu_info = get_cpu_info(self.tdp_db, self._constants_data)
        self.gpu_info = get_gpu_info(self.gpu_index, self.gpu_tdp_defaults)
        self.ram_info = get_ram_info()
        self.hardware = HardwareMonitor()

        self._carbon_lock = threading.Lock()
        self._gpu_monitor_active = False
        self._gpu_monitor_thread = None
        self._gpu_samples = deque(maxlen=self.SAMPLE_BUFFER_SIZE)
        self._gpu_sample_lock = threading.Lock()
        self._cpu_monitor_active = False
        self._cpu_monitor_thread = None
        self._cpu_samples = deque(maxlen=self.SAMPLE_BUFFER_SIZE)
        self._cpu_sample_lock = threading.Lock()
        self._cpu_monitor_ref_count = 0
        self._gpu_monitor_ref_count = 0

        # 50ms (0.05s) optimal frequency balancing precision and low CPU overhead
        self._monitor_interval = self.MONITOR_INTERVAL_S
        self._current_process = psutil.Process()
        self._paused = False
        self._paused_at = None
        self._total_paused_duration = 0.0

        if not self.quiet:
            intensity_metadata = f"{self.carbon_intensity} gCO2/kWh"
            source_label = "LIVE" if self._intensity_source == "live" else "STATIC"

            logger.info(f"[INFO] EcoTrace instrumentation session initialized ({source_label}).")
            logger.info("-" * 53)
            logger.info(f"Run ID        : {self._run_id}" + (f" [{self._run_label}]" if self._run_label else ""))
            logger.info(f"Region        : {self.region_code} ({intensity_metadata})")
            cpu_brand = self.cpu_info.get('brand', 'Unknown') if isinstance(self.cpu_info, dict) else 'Unknown'
            cpu_cores = self.cpu_info.get('cores', 1) if isinstance(self.cpu_info, dict) else 1
            cpu_tdp = self.cpu_info.get('tdp', 65.0) if isinstance(self.cpu_info, dict) else 65.0
            logger.info(f"Hardware Logic: {cpu_brand}")
            logger.info(f"Specifications: {cpu_cores} Cores | {cpu_tdp}W TDP")

            if self.hardware.rapl_available:
                logger.info("Energy Sensor : RAPL (Exact Hardware Mode Enabled)")
            elif self.hardware.apple_silicon_available:
                logger.info("Energy Sensor : Apple Silicon (powermetrics)")
            else:
                logger.info("Energy Sensor : Prediction Mode (Boavizta Advanced Estimation, ~15-20% error margin)")
                import platform
                if platform.system() == "Linux":
                    logger.warning("RAPL access denied! Run with 'sudo' for 0% deviation exact CPU profiling.")
                elif platform.system() == "Darwin":
                    logger.warning("Apple Silicon powermetrics access denied! Run with 'sudo' to allow exact energy profiling (0% deviation).")

            if self.ram_info and isinstance(self.ram_info, dict):
                ram_gb = self.ram_info.get('total_gb', 0.0)
                ram_type_str = self.ram_info.get('type', 'DDR4')
                logger.info(f"Memory Config : {ram_gb:.1f} GB {ram_type_str}")

            if self.gpu_info and isinstance(self.gpu_info, dict):
                gpu_brand_str = self.gpu_info.get('brand', 'Unknown')
                gpu_tdp_val = self.gpu_info.get('tdp', 0.0)
                logger.info(f"GPU Accelerator: {gpu_brand_str} ({gpu_tdp_val}W TDP)")

            logger.info("-" * 53)
            logger.info("[INFO] Instrumentation sequence finalized.\n")

        # Baseline measurement to subtract background OS noise from subsequent metrics
        self._idle_baseline_pct = self._measure_idle_baseline()

        self._session_start_time = time.perf_counter()
        self._session_summary_enabled = session_summary and not quiet
        if self._session_summary_enabled:
            EcoTrace._instances.add(self)

    # ========================================================================
    # Live Grid API — Intensity Resolution 
    # ========================================================================

    def _resolve_intensity_with_live_fallback(self):
        """Resolves carbon intensity using live API data with static fallback.

        Implements a three-tier resolution strategy:
        1. **Cache Hit**: If a valid cached value exists and is within the
           GRID_CACHE_TTL_S window (1 hour), returns it immediately.
        2. **Live Fetch**: Queries the Electricity Maps API for real-time
           carbon intensity data for the configured region.
        3. **Static Fallback**: If both cache and live fetch fail, falls back
           to the static ``CARBON_INTENSITY_MAP`` from ``constants.json``.

        This method is called during ``__init__`` and can be called again
        at any time to refresh the intensity value.

        Returns:
            float: Carbon intensity in gCO2/kWh from the best available
            source (live API preferred, static data as fallback).
        """
        import time as _time

        # Tier 1: Check memory cache validity (1-hour TTL)
        now = _time.time()
        if (self._grid_cached_intensity is not None
                and (now - self._grid_cache_timestamp) < GRID_CACHE_TTL_S):
            self._intensity_source = "live"
            return self._grid_cached_intensity

        # Tier 2: Attempt live API fetch
        if self.grid_api_key:
            live_intensity = fetch_live_carbon_intensity(
                self.region_code, self.grid_api_key
            )
            if live_intensity is not None:
                # Update cache with fresh value
                self._grid_cached_intensity = live_intensity
                self._grid_cache_timestamp = now
                self._intensity_source = "live"
                logger.info(f"🌍 Live grid data: {live_intensity} gCO2/kWh")
                return live_intensity
            else:
                logger.warning("⚠️ Live grid API unavailable, using static data.")

        # Tier 3: Static fallback from constants.json
        self._intensity_source = "static"
        return resolve_carbon_intensity(self.region_code, self._constants_data)

    def _measure_idle_baseline(self):
        """Captures short baseline measurement for differential carbon tracking.

        Takes a 100ms snapshot of current system utilization to establish the
        idle baseline. This baseline is subtracted from function measurements
        to report only the code's incremental energy cost.

        Returns:
            float: Baseline CPU utilization percentage (0-100), core-normalized.
        """
        baseline_start = time.perf_counter()
        baseline_samples = []

        while (time.perf_counter() - baseline_start) * 1000 < self.BASELINE_MEASUREMENT_MS:
            try:
                cpu_usage = self._current_process.cpu_percent()
                baseline_samples.append(cpu_usage)
                time.sleep(self.MONITOR_INTERVAL_S)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                break

        if not baseline_samples:
            return 0.0

        raw_avg = sum(baseline_samples) / len(baseline_samples)
        core_count = psutil.cpu_count(logical=True) or 1
        return raw_avg / core_count

    def _compute_carbon(self, tdp, utilization_pct, duration_s, energy_delta_j=None, is_gpu=False):
        """Computes carbon emissions from power parameters.

        Args:
            tdp: Thermal Design Power in watts.
            utilization_pct: Average utilization as a percentage (0–100).
            duration_s: Measurement duration in seconds.
            energy_delta_j: Exact hardware energy delta in Joules (optional).
            is_gpu: Flag to differentiate GPU vs CPU workloads.

        Returns:
            float: Estimated carbon emissions in gCO2.
        """
        normalized_utilization = min(max(utilization_pct, 0.0), 100.0)

        # Power calculation (Exact vs Estimated)
        if energy_delta_j is not None:
            main_power_wh = (energy_delta_j / 3600.0) * (normalized_utilization / 100.0)
        else:
            if not is_gpu and tdp == self.cpu_info.get('tdp'):
                power_w = self.hardware.estimate_cpu_power_w(tdp, normalized_utilization)
            else:
                power_w = tdp * (normalized_utilization / 100.0)
            main_power_wh = power_w * duration_s / self.SECONDS_PER_HOUR

        # RAM energy calculation - Process Tree (RSS)
        try:
            total_rss = self._current_process.memory_info().rss
            for child in self._current_process.children(recursive=True):
                try:
                    total_rss += child.memory_info().rss
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue
            ram_usage_gb = total_rss / (1024**3)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            ram_usage_gb = 0.0

        ram_type = 'DDR4'
        if self.ram_info and isinstance(self.ram_info, dict):
            raw_ram_type = self.ram_info.get('type')
            if raw_ram_type and isinstance(raw_ram_type, str):
                ram_type = raw_ram_type.upper()
        default_ram_watt = RAM_WATT_FACTORS.get('DDR4', 0.375)
        ram_watt_factor = RAM_WATT_FACTORS.get(ram_type, default_ram_watt)
        if ram_watt_factor is None:
            ram_watt_factor = default_ram_watt
        ram_power_wh = (ram_watt_factor * ram_usage_gb) * duration_s / self.SECONDS_PER_HOUR

        total_power_wh = main_power_wh + ram_power_wh
        return (total_power_wh / self.WATTS_PER_KILOWATT) * self.carbon_intensity

    def _accumulate_carbon(self, carbon_emitted, func_name, duration, avg_cpu=None, file_path=None, line_number=None):
        """Thread-safe accumulation of carbon emissions with CSV logging.

        Accumulates the emitted carbon into the session total, logs to CSV,
        and enforces carbon budget limits if configured. The library is the
        authority on budget rules — it warns and triggers callbacks here.

        Args:
            carbon_emitted: Carbon value in gCO2 to add to the running total.
            func_name: Name of the measured function for the audit log.
            duration: Execution duration in seconds.
            avg_cpu: Average CPU usage percentage (optional).
            file_path: Absolute path to the source file.
            line_number: Line number where the function is defined.
        """
        with self._carbon_lock:
            if self._paused:
                return
            self.total_carbon += carbon_emitted
            if self.carbon_intensity and self.carbon_intensity > 0:
                self.total_energy_kwh += (carbon_emitted / self.carbon_intensity)
            self._tracked_functions_count += 1
            self._log_to_csv(func_name, duration, carbon_emitted, avg_cpu, file_path, line_number)

            # Dispatch metrics to registered exporters
            if self._exporters:
                exporters = list(self._exporters)

                def _dispatch_exporters(exporters=exporters):
                    has_com = False
                    if sys.platform == "win32":
                        try:
                            import ctypes
                            ctypes.windll.ole32.CoInitialize(None)
                            has_com = True
                        except Exception:
                            pass
                    try:
                        for exporter in exporters:
                            try:
                                exporter.export(
                                    carbon_emitted=carbon_emitted,
                                    func_name=func_name,
                                    duration=duration,
                                    region=self.region_code,
                                    run_id=self._run_id,
                                    run_label=self._run_label
                                )
                            except Exception as e:
                                logger.debug(f"EcoTrace Exporter error: {e}")
                    finally:
                        if has_com:
                            try:
                                import ctypes
                                ctypes.windll.ole32.CoUninitialize()
                            except Exception:
                                pass

                try:
                    self._exporter_pool.submit(_dispatch_exporters)
                except Exception as e:
                    logger.debug(f"EcoTrace Exporter pool unavailable, falling back to synchronous dispatch: {e}")
                    _dispatch_exporters()

            self._enforce_carbon_budget(func_name)

    def add_exporter(self, exporter):
        """Registers a telemetry exporter to receive carbon metrics in real-time.

        Args:
            exporter: An object implementing an `export(carbon_emitted, func_name, duration, region)` method.
        """
        self._exporters.append(exporter)

    def _enforce_carbon_budget(self, func_name):
        """Checks carbon budget thresholds and fires warnings/callbacks.

        Called inside _accumulate_carbon under the carbon lock. Implements a
        two-tier alert system:
            - 80% threshold: WARNING log (fires once)
            - 100% threshold: WARNING log + optional callback (fires once)

        The library is the authority on budget rules. External consumers
        (IDE, CI/CD) read the state via ``remaining_budget``.

        Args:
            func_name: Name of the function that triggered the check.
        """
        if self.carbon_limit is None:
            return

        # 80% early warning threshold
        if not self._budget_warning_fired and self.total_carbon >= self.carbon_limit * 0.8:
            self._budget_warning_fired = True
            remaining = self.carbon_limit - self.total_carbon
            logger.warning(
                f"Carbon budget 80% consumed: {self.total_carbon:.6f} / "
                f"{self.carbon_limit:.6f} gCO2 (remaining: {remaining:.6f} gCO2)"
            )

        # 100% budget exceeded threshold
        if not self._budget_exceeded_fired and self.total_carbon >= self.carbon_limit:
            self._budget_exceeded_fired = True
            logger.warning(
                f"CARBON BUDGET EXCEEDED after '{func_name}': "
                f"{self.total_carbon:.6f} gCO2 (limit: {self.carbon_limit:.6f} gCO2)"
            )
            if self._on_budget_exceeded:
                try:
                    self._on_budget_exceeded(self.total_carbon, self.carbon_limit)
                except Exception as e:
                    logger.debug(f"on_budget_exceeded callback error: {e}")

    @property
    def remaining_budget(self):
        """Returns the remaining carbon budget in gCO2, or None if no limit is set.

        This is the primary data interface for external consumers (IDE sidebar,
        CI/CD gates). The library provides the number; the consumer acts on it.

        Returns:
            float or None: Remaining gCO2 budget, or None if no limit configured.
        """
        if self.carbon_limit is None:
            return None
        return max(0.0, self.carbon_limit - self.total_carbon)

    def pause(self):
        """Temporarily pauses carbon tracking for the session."""
        with self._carbon_lock:
            if not self._paused:
                self._paused = True
                self._paused_at = time.perf_counter()
                logger.info("[EcoTrace] Instrumentation session paused.")

    def resume(self):
        """Resumes carbon tracking for the session."""
        with self._carbon_lock:
            if self._paused:
                self._paused = False
                if self._paused_at is not None:
                    self._total_paused_duration += time.perf_counter() - self._paused_at
                    self._paused_at = None
                logger.info("[EcoTrace] Instrumentation session resumed.")

    def get_summary(self) -> dict:
        """Returns the current session metrics as a structured dictionary.

        Provides programmatic access to all session data — suitable for use
        in notebooks, dashboards, custom reporting pipelines, and test assertions.
        Can be called at any point during or after a session.

        Returns:
            dict: Session summary with keys:
                - ``run_id``: Short unique ID for this session.
                - ``run_label``: Optional human-readable label.
                - ``duration_s``: Elapsed session time in seconds.
                - ``functions_tracked``: Number of tracked function calls.
                - ``total_carbon_gco2``: Cumulative carbon in gCO2.
                - ``region``: ISO region code.
                - ``carbon_intensity``: gCO2/kWh intensity value.
                - ``intensity_source``: ``'live'`` or ``'static'``.
                - ``budget``: Budget status dict (``None`` if no limit set).
                - ``equivalence``: Human-readable carbon comparison string.
                - ``hardware``: CPU/GPU/energy sensor metadata dict.
        """
        session_duration = time.perf_counter() - self._session_start_time
        current_pause = 0.0
        if self._paused and self._paused_at is not None:
            current_pause = time.perf_counter() - self._paused_at
        active_duration = max(0.0, session_duration - self._total_paused_duration - current_pause)

        # Determine energy sensor label
        if self.hardware.rapl_available:
            sensor = "RAPL (Exact Hardware)"
        elif self.hardware.apple_silicon_available:
            sensor = "Apple Silicon (powermetrics)"
        else:
            sensor = "Prediction Mode (Boavizta Estimation, ~15-20% error margin)"

        budget_info = None
        if self.carbon_limit is not None:
            remaining = self.remaining_budget
            used_pct = (self.total_carbon / self.carbon_limit * 100) if self.carbon_limit else 0.0
            budget_info = {
                "limit_gco2": self.carbon_limit,
                "remaining_gco2": remaining,
                "used_pct": round(used_pct, 2),
                "status": "EXCEEDED" if remaining == 0 else "OK",
            }

        return {
            "run_id": self._run_id,
            "run_label": self._run_label,
            "duration_s": round(active_duration, 4),
            "functions_tracked": self._tracked_functions_count,
            "total_carbon_gco2": self.total_carbon,
            "region": self.region_code,
            "carbon_intensity": self.carbon_intensity,
            "intensity_source": self._intensity_source,
            "budget": budget_info,
            "equivalence": self.equivalence(self.total_carbon),
            "hardware": {
                "cpu": self.cpu_info.get("brand", "Unknown"),
                "cores": self.cpu_info.get("cores", 0),
                "tdp_w": self.cpu_info.get("tdp", 0),
                "gpu": self.gpu_info["brand"] if self.gpu_info else None,
                "energy_sensor": sensor,
            },
        }

    def _print_session_summary(self):
        """Prints a summary table when the process exits via atexit.

        Registered in __init__ when session_summary=True and quiet=False.
        Delegates to get_summary() to avoid duplicated logic.
        """
        try:
            # Flush exporter pool before exit
            self._exporter_pool.shutdown(wait=True)

            if self._tracked_functions_count == 0:
                return  # No measurements taken, skip summary

            s = self.get_summary()

            print()
            print("=" * 55)
            print("  EcoTrace — Session Summary")
            print("=" * 55)
            print(f"  Run ID         : {s['run_id']}" + (f" [{s['run_label']}]" if s['run_label'] else ""))
            print(f"  Duration       : {s['duration_s']:.2f}s")
            print(f"  Functions      : {s['functions_tracked']} tracked")
            print(f"  Total Carbon   : {s['total_carbon_gco2']:.8f} gCO2")
            print(f"  Region         : {s['region']} ({s['carbon_intensity']} gCO2/kWh)")

            # --- Carbon Budget Status ----------------------------------------
            if s["budget"] is not None:
                b = s["budget"]
                print(f"  Budget         : {s['total_carbon_gco2']:.6f} / {b['limit_gco2']:.6f} gCO2 ({b['used_pct']:.1f}%) [{b['status']}]")

            if s["equivalence"]:
                print(f"  Equivalent     : {s['equivalence']}")

            print("=" * 55)
        except Exception:
            pass  # Session summary must never crash the application

    def equivalence(self, gco2):
        """Converts a gCO2 value into a human-readable real-world comparison.

        Uses a tiered system: selects the most relatable comparison based on
        the magnitude of the emission value. The library owns this conversion;
        external consumers (IDE, reports) can call it to enrich their display.

        Args:
            gco2: Carbon emissions in grams of CO2.

        Returns:
            str: Human-readable equivalence string, or empty string if the
            value is too small to compare meaningfully.
        """
        if gco2 <= 0:
            return ""

        if gco2 < 0.01:
            searches = gco2 / 0.2
            return f"{searches:.2f} Google searches"
        elif gco2 < 1.0:
            led_minutes = (gco2 / 5.2) * 60
            return f"{led_minutes:.1f} min of LED bulb (10W)"
        elif gco2 < 10.0:
            charges = gco2 / 8.22
            return f"{charges:.2f} smartphone charges"
        elif gco2 < 100.0:
            netflix_min = (gco2 / 36.0) * 60
            return f"{netflix_min:.1f} min of Netflix streaming"
        else:
            km = gco2 / 121.0
            return f"{km:.2f} km of car driving"

    def _cpu_monitor_worker(self):
        """Background thread that continuously samples process-scoped CPU usage.

        Samples at MONITOR_INTERVAL_S intervals using ``psutil.Process``,
        storing ``(timestamp, cpu_percent)`` tuples in a thread-safe deque.
        Exits gracefully if the process is no longer accessible.
        """
        has_com = False
        if sys.platform == "win32":
            try:
                import ctypes
                ctypes.windll.ole32.CoInitialize(None)
                has_com = True
            except Exception:
                pass

        try:
            child_cache = {}
            next_sample_time = time.perf_counter()
            while self._cpu_monitor_active:
                try:
                    total_usage = self._current_process.cpu_percent()

                    try:
                        current_children = self._current_process.children(recursive=True)
                    except (psutil.NoSuchProcess, psutil.AccessDenied):
                        current_children = []

                    active_pids = set()
                    for child in current_children:
                        pid = child.pid
                        active_pids.add(pid)
                        if pid not in child_cache:
                            try:
                                child.cpu_percent()
                                child_cache[pid] = child
                            except (psutil.NoSuchProcess, psutil.AccessDenied):
                                continue
                        else:
                            try:
                                total_usage += child_cache[pid].cpu_percent()
                            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
                                del child_cache[pid]

                    for pid in list(child_cache.keys()):
                        if pid not in active_pids:
                            del child_cache[pid]

                    timestamp = time.perf_counter()
                    with self._cpu_sample_lock:
                        self._cpu_samples.append((timestamp, total_usage))

                    next_sample_time += self._monitor_interval
                    sleep_duration = next_sample_time - time.perf_counter()
                    if sleep_duration > 0:
                        time.sleep(sleep_duration)
                    else:
                        next_sample_time = time.perf_counter()
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    break
        finally:
            if has_com:
                try:
                    import ctypes
                    ctypes.windll.ole32.CoUninitialize()
                except Exception:
                    pass

    def _gpu_monitor_worker(self):
        """Background thread that continuously samples GPU utilization.

        Only active for NVIDIA GPUs with a valid device handle. Samples at
        MONITOR_INTERVAL_S intervals, storing ``(timestamp, gpu_percent)``
        tuples in a thread-safe deque.
        """
        if not self.gpu_info or self.gpu_info.get("handle") is None:
            return

        try:
            import pynvml  # type: ignore
        except ImportError:
            return

        handle = self.gpu_info["handle"]
        while self._gpu_monitor_active:
            try:
                util = pynvml.nvmlDeviceGetUtilizationRates(handle)
                gpu_usage = util.gpu
                power_mw = pynvml.nvmlDeviceGetPowerUsage(handle)
                timestamp = time.perf_counter()
                with self._gpu_sample_lock:
                    self._gpu_samples.append((timestamp, gpu_usage, power_mw / 1000.0))
                time.sleep(self._monitor_interval)
            except Exception:
                break

    def _start_cpu_monitor(self):
        """Spawns the background CPU sampling thread with reference counting."""
        with self._cpu_sample_lock:
            self._cpu_monitor_ref_count += 1
            if self._cpu_monitor_ref_count == 1:
                self._cpu_monitor_active = True
                self._cpu_samples.clear()
                self._cpu_monitor_thread = threading.Thread(target=self._cpu_monitor_worker, daemon=True)
                self._cpu_monitor_thread.start()

    def _stop_cpu_monitor(self):
        """Signals the CPU sampling thread to stop only when ref count hits zero."""
        with self._cpu_sample_lock:
            if self._cpu_monitor_ref_count > 0:
                self._cpu_monitor_ref_count -= 1

            if self._cpu_monitor_ref_count == 0 and self._cpu_monitor_active:
                self._cpu_monitor_active = False
                if self._cpu_monitor_thread:
                    self._cpu_monitor_thread.join(timeout=self.MONITOR_JOIN_TIMEOUT_S)
                    self._cpu_monitor_thread = None

    def _start_gpu_monitor(self):
        """Spawns the background GPU sampling thread with reference counting."""
        with self._gpu_sample_lock:
            self._gpu_monitor_ref_count += 1
            if self._gpu_monitor_ref_count == 1:
                self._gpu_monitor_active = True
                self._gpu_samples.clear()
                self._gpu_monitor_thread = threading.Thread(target=self._gpu_monitor_worker, daemon=True)
                self._gpu_monitor_thread.start()

    def _stop_gpu_monitor(self):
        """Signals the GPU sampling thread to stop only when ref count hits zero."""
        with self._gpu_sample_lock:
            if self._gpu_monitor_ref_count > 0:
                self._gpu_monitor_ref_count -= 1

            if self._gpu_monitor_ref_count == 0 and self._gpu_monitor_active:
                self._gpu_monitor_active = False
                if self._gpu_monitor_thread:
                    self._gpu_monitor_thread.join(timeout=self.MONITOR_JOIN_TIMEOUT_S)
                    self._gpu_monitor_thread = None

    def _get_avg_cpu_in_range(self, start_time, end_time):
        """Computes mean CPU utilization from samples within a time window.

        Applies idle baseline subtraction so the returned value represents
        only the incremental CPU load caused by the measured code, not
        background OS activity.

        Args:
            start_time: Window start as a ``time.perf_counter()`` value.
            end_time: Window end as a ``time.perf_counter()`` value.

        Returns:
            float: Average CPU percentage (baseline-subtracted), or
            FULL_UTILIZATION_PERCENT if no samples were captured.
        """
        with self._cpu_sample_lock:
            relevant_samples = [
                cpu for ts, cpu in self._cpu_samples
                if start_time <= ts <= end_time
            ]
            if not relevant_samples:
                return self.FULL_UTILIZATION_PERCENT

            raw_avg = sum(relevant_samples) / len(relevant_samples)
            core_count = psutil.cpu_count(logical=True) or 1
            normalized = raw_avg / core_count

            return max(0.0, normalized - self._idle_baseline_pct)

    def _get_source_location(self, func):
        """Returns the source file path and line number for a callable.

        Works for wrapped, decorated, and async functions by unwrapping the
        original implementation before querying inspect.
        """
        try:
            target = inspect.unwrap(func)
            file_path = os.path.abspath(inspect.getfile(target))
            line_number = inspect.getsourcelines(target)[1]
            return file_path, line_number
        except Exception:
            return None, None

    def _get_avg_gpu_in_range(self, start_time, end_time):
        """Computes mean GPU utilization and power from samples within a time window.

        Args:
            start_time: Window start as a ``time.perf_counter()`` value.
            end_time: Window end as a ``time.perf_counter()`` value.

        Returns:
            tuple: (Average GPU percentage, Average Power in Watts).
            Returns (FULL_UTILIZATION_PERCENT, None) if no samples were captured.
        """
        with self._gpu_sample_lock:
            relevant_samples = []
            for item in self._gpu_samples:
                if len(item) == 3:
                    ts, gpu, pwr = item
                    if start_time <= ts <= end_time:
                        relevant_samples.append((gpu, pwr))

        if not relevant_samples:
            return self.FULL_UTILIZATION_PERCENT, None

        avg_gpu = sum(s[0] for s in relevant_samples) / len(relevant_samples)
        avg_pwr = sum(s[1] for s in relevant_samples) / len(relevant_samples)
        return avg_gpu, avg_pwr

    @contextmanager
    def cpu_monitor(self):
        """Context manager that brackets a code block with CPU monitoring.

        Yields:
            EcoTrace: The current instance for optional chaining.
        """
        self._start_cpu_monitor()
        try:
            yield self
        finally:
            self._stop_cpu_monitor()

    @contextmanager
    def gpu_monitor(self):
        """Context manager that brackets a code block with GPU monitoring.

        Yields:
            EcoTrace: The current instance for optional chaining.
        """
        self._start_gpu_monitor()
        try:
            yield self
        finally:
            self._stop_gpu_monitor()

    def _log_to_csv(self, func_name, duration, carbon, avg_cpu=None, file_path=None, line_number=None):
        """Appends a single measurement row to the CSV audit log.

        Creates ``ecotrace_log.csv`` with headers if it doesn't exist.

        Args:
            func_name: Name of the tracked function.
            duration: Execution time in seconds.
            carbon: Estimated carbon emissions in gCO2.
            avg_cpu: Average CPU usage percentage (optional).
            file_path: Source file path.
            line_number: Source line number.
        """
        try:
            file_exists = os.path.isfile("ecotrace_log.csv") and os.path.getsize("ecotrace_log.csv") > 0
            with open("ecotrace_log.csv", "a", newline="", encoding="utf-8") as f:
                writer = csv.writer(f)
                if not file_exists:
                    writer.writerow(["Date", "Function", "Duration(s)", "Carbon(gCO2)",
                                     "Region", "AvgCPU(%)", "FilePath", "Line",
                                     "RunID", "RunLabel"])
                avg_cpu_str = f"{avg_cpu:.2f}" if avg_cpu is not None else "N/A"
                writer.writerow([
                    datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                    func_name,
                    f"{duration:.4f}",
                    f"{carbon:.8f}",
                    self.region_code,
                    avg_cpu_str,
                    file_path or "N/A",
                    line_number or "N/A",
                    self._run_id,
                    self._run_label,
                ])
        except Exception as e:
            logger.warning(f"EcoTrace CSV logging failed: {e}")

    def track(self, func):
        """Decorator that measures carbon emissions for any function call.

        Automatically detects whether the target is synchronous or asynchronous
        and selects the appropriate measurement strategy.

        Args:
            func: The function to decorate. Can be sync or async.

        Returns:
            Callable: Wrapped function that measures emissions transparently.
        """
        if inspect.iscoroutinefunction(func):
            @functools.wraps(func)
            async def async_wrapper(*args, **kwargs):
                res = await self.measure_async(func, *args, **kwargs)
                return res["result"] if isinstance(res, dict) else res
            return async_wrapper

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            res = self.measure(func, *args, **kwargs)
            return res["result"] if isinstance(res, dict) else res

        return wrapper

    def track_gpu(self, func):
        """Decorator that measures GPU carbon emissions with real utilization monitoring.

        If no GPU is detected, the wrapped function executes normally without
        measurement. If the GPU becomes unavailable mid-calculation, the function
        result is preserved and a warning is logged.

        Args:
            func: The function to decorate.

        Returns:
            Callable: Wrapped function with GPU carbon measurement.
        """

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if self.gpu_info is None:
                logger.warning(f"No GPU detected, executing '{func.__name__}' without measurement.")
                return func(*args, **kwargs)

            try:
                file_path = os.path.abspath(inspect.getfile(func))
                line_number = inspect.getsourcelines(func)[1]
            except Exception:
                file_path, line_number = None, None

            start_time = time.perf_counter()
            try:
                with self.cpu_monitor():
                    with self.gpu_monitor():
                        result = func(*args, **kwargs)
                return result
            finally:
                end_time = time.perf_counter()
                try:
                    duration = end_time - start_time
                    avg_cpu = self._get_avg_cpu_in_range(start_time, end_time)
                    cpu_carbon = self._compute_carbon(self.cpu_info['tdp'], avg_cpu, duration)

                    avg_gpu_util, avg_gpu_pwr = self._get_avg_gpu_in_range(start_time, end_time)
                    if avg_gpu_pwr is not None:
                        gpu_energy_wh = (avg_gpu_pwr * duration) / self.SECONDS_PER_HOUR
                        gpu_carbon = (gpu_energy_wh / self.WATTS_PER_KILOWATT) * self.carbon_intensity
                    else:
                        gpu_carbon = self._compute_carbon((self.gpu_info or {}).get('tdp', 100.0), avg_gpu_util, duration, is_gpu=True)

                    carbon_emitted = cpu_carbon + gpu_carbon
                    self._accumulate_carbon(carbon_emitted, func.__name__, duration, avg_cpu=avg_cpu, file_path=file_path, line_number=line_number)
                    logger.info(f"GPU/CPU Carbon Emissions: {carbon_emitted:.8f} gCO2")
                    logger.info(f"Duration     : {duration:.4f} sec")
                    logger.info(f"GPU Usage    : {avg_gpu_util:.1f}%")
                    logger.info(f"CO2          : {carbon_emitted:.8f} gCO2")
                except Exception as e:
                    logger.error(f"GPU measurement failed for '{func.__name__}': {e}")
        return wrapper

    def measure(self, func, *args, **kwargs):
        """Executes a synchronous function and measures its CPU carbon emissions.

        Uses continuous background sampling for accurate utilization measurement.
        If the measurement calculation fails, the function result is still returned.

        Args:
            func: Synchronous callable to measure.
            *args: Positional arguments forwarded to ``func``.
            **kwargs: Keyword arguments forwarded to ``func``.

        Returns:
            dict: Keys ``func_name``, ``duration``, ``avg_cpu``, ``carbon``,
            ``cpu_samples``, and ``result``.
        """
        start_time = time.perf_counter()
        result_data = None
        func_success = False
        energy_start = None

        try:
            energy_start = self.hardware.get_cpu_energy_j()
            with self.cpu_monitor():
                if self.gpu_info:
                    with self.gpu_monitor():
                        result_data = func(*args, **kwargs)
                else:
                    result_data = func(*args, **kwargs)
                func_success = True
        finally:
            end_time = time.perf_counter()
            energy_end = self.hardware.get_cpu_energy_j()
            duration = end_time - start_time

            try:
                avg_cpu = self._get_avg_cpu_in_range(start_time, end_time)

                with self._cpu_sample_lock:
                    measurement_samples = list(self._cpu_samples)

                # Capture location info robustly for decorated and async functions
                file_path, line_number = self._get_source_location(func)

                energy_delta_j = None
                if energy_start is not None and energy_end is not None:
                    energy_delta_j = max(0.0, energy_end - energy_start)

                carbon_emitted = self._compute_carbon(self.cpu_info['tdp'], avg_cpu, duration, energy_delta_j=energy_delta_j)
                self._accumulate_carbon(carbon_emitted, func.__name__, duration, avg_cpu, file_path=file_path, line_number=line_number)

                if func_success:
                    return {
                        "func_name": func.__name__,
                        "duration": duration,
                        "avg_cpu": avg_cpu,
                        "carbon": carbon_emitted,
                        "cpu_samples": measurement_samples,
                        "result": result_data
                    }
            except Exception as e:
                logger.error(f"Measurement failed for '{func.__name__}': {e}")
                if func_success:
                    return {
                        "func_name": func.__name__,
                        "duration": duration,
                        "avg_cpu": 0.0,
                        "carbon": 0.0,
                        "cpu_samples": [],
                        "result": result_data
                    }

    async def measure_async(self, func, *args, **kwargs):
        """Executes an async function and measures its CPU carbon emissions.

        Uses continuous background sampling, which is particularly important
        for bursty or I/O-bound async workloads where point-in-time readings
        misrepresent actual utilization.

        Args:
            func: Async callable to measure.
            *args: Positional arguments forwarded to ``func``.
            **kwargs: Keyword arguments forwarded to ``func``.

        Returns:
            dict: Keys ``func_name``, ``duration``, ``avg_cpu``, ``carbon``,
            ``cpu_samples``, and ``result``.

        Raises:
            Exception: Re-raises any exception from the wrapped function after
            completing the measurement teardown.
        """
        start_time = time.perf_counter()
        result_data = None
        func_success = False
        energy_start = None

        try:
            energy_start = self.hardware.get_cpu_energy_j()
            with self.cpu_monitor():
                try:
                    if self.gpu_info:
                        with self.gpu_monitor():
                            result_data = await func(*args, **kwargs)
                    else:
                        result_data = await func(*args, **kwargs)
                    func_success = True
                finally:
                    await asyncio.sleep(self.MONITOR_INTERVAL_S)  # Allow trailing samples to be captured
        finally:
            end_time = time.perf_counter()
            energy_end = self.hardware.get_cpu_energy_j()
            duration = end_time - start_time

            try:
                avg_cpu = self._get_avg_cpu_in_range(start_time, end_time)

                with self._cpu_sample_lock:
                    measurement_samples = list(self._cpu_samples)

                energy_delta_j = None
                if energy_start is not None and energy_end is not None:
                    energy_delta_j = max(0.0, energy_end - energy_start)

                # Capture location info robustly for decorated and async functions
                file_path, line_number = self._get_source_location(func)

                carbon_emitted = self._compute_carbon(self.cpu_info['tdp'], avg_cpu, duration, energy_delta_j=energy_delta_j)
                self._accumulate_carbon(carbon_emitted, func.__name__, duration, avg_cpu, file_path=file_path, line_number=line_number)

                if func_success:
                    return {
                        "func_name": func.__name__,
                        "duration": duration,
                        "avg_cpu": avg_cpu,
                        "carbon": carbon_emitted,
                        "cpu_samples": measurement_samples,
                        "result": result_data
                    }
            except Exception as e:
                logger.error(f"Async measurement failed for '{func.__name__}': {e}")
                if func_success:
                    return {
                        "func_name": func.__name__,
                        "duration": duration,
                        "avg_cpu": 0.0,
                        "carbon": 0.0,
                        "cpu_samples": [],
                        "result": result_data
                    }

    def compare(self, func1, func2):
        """Runs two functions sequentially and compares their carbon footprints.

        Args:
            func1: First callable to measure.
            func2: Second callable to measure.

        Returns:
            dict: Keys ``func1`` and ``func2``, each containing the full
            measurement dict from ``measure()``.
        """
        result1 = self.measure(func1)
        result2 = self.measure(func2)
        if isinstance(result1, dict) and isinstance(result2, dict):
            logger.info(f"Comparison Results:")
            logger.info(f"Function 1: {result1['func_name']} - Duration: {result1['duration']:.4f} sec - CO2: {result1['carbon']:.8f} gCO2")
            logger.info(f"Function 2: {result2['func_name']} - Duration: {result2['duration']:.4f} sec - CO2: {result2['carbon']:.8f} gCO2")
        return {"func1": result1, "func2": result2}

    # ========================================================================
    # Reporting
    # ========================================================================

    def generate_pdf_report(self, filename="ecotrace_full_report.pdf", comparison=None, cpu_samples=None, gpu_samples=None):
        """Generates a comprehensive PDF audit report dynamically.

        If cpu_samples or gpu_samples are not provided, the engine automatically
        snapshots the internal session deques for full-history reporting.
        """
        from .report import generate_pdf_report as generate_pdf

        # CPU Samples snapshot
        final_cpu_samples = None
        if cpu_samples is not None:
            final_cpu_samples = list(cpu_samples)
        else:
            with self._cpu_sample_lock:
                final_cpu_samples = list(self._cpu_samples)

        # GPU Samples snapshot — normalize 3-tuples (ts, util, power) to
        # 2-tuples (ts, util) expected by report chart functions.
        final_gpu_samples = None
        if gpu_samples is not None:
            final_gpu_samples = [(item[0], item[1]) for item in gpu_samples]
        elif self.gpu_info:
            with self._gpu_sample_lock:
                final_gpu_samples = [(item[0], item[1]) for item in self._gpu_samples]

        generate_pdf(
            filename=filename,
            cpu_info=self.cpu_info,
            gpu_info=self.gpu_info,
            region_code=self.region_code,
            comparison=comparison,
            cpu_samples=final_cpu_samples,
            gpu_samples=final_gpu_samples,
            api_key=self.api_key
        )

    def export_json(self, filename="ecotrace_report.json", csv_path="ecotrace_log.csv"):
        """Exports session data to a structured JSON file.

        Combines hardware metadata, measurement history from the CSV audit
        log, and aggregate statistics into a single machine-readable document.
        This output is designed for consumption by the VS Code extension
        sidebar, CI/CD carbon gates, and third-party analytics tools.

        The JSON schema contains three top-level keys:

        - ``meta``: Hardware profile, region, version, and export timestamp.
        - ``measurements``: Array of per-function measurement records from
          the CSV audit log.
        - ``summary``: Aggregate statistics (total carbon, total duration,
          measurement count, top emitters).

        Args:
            filename: Output path for the JSON file. Defaults to
                ``ecotrace_report.json`` in the current working directory.
            csv_path: Path to the CSV audit log to read measurements from.
                Defaults to ``ecotrace_log.csv``.

        Raises:
            IOError: If the output file cannot be written.

        Example::

            eco = EcoTrace(region_code="TR")

            @eco.track
            def my_function():
                pass

            my_function()
            eco.export_json("report.json")
        """
        import json as _json
        from . import __version__

        ram_dict = None
        if self.ram_info and isinstance(self.ram_info, dict):
            ram_total_gb = self.ram_info.get("total_gb")
            ram_type_val = self.ram_info.get("type")
            ram_dict = {
                "total_gb": round(float(ram_total_gb if ram_total_gb is not None else 0.0), 2),
                "type": str(ram_type_val if ram_type_val is not None else "DDR4")
            }

        gpu_dict = None
        if self.gpu_info and isinstance(self.gpu_info, dict):
            gpu_brand = self.gpu_info.get("brand")
            gpu_tdp = self.gpu_info.get("tdp")
            gpu_type = self.gpu_info.get("type")
            gpu_dict = {
                "brand": str(gpu_brand if gpu_brand is not None else "Unknown"),
                "tdp_w": float(gpu_tdp if gpu_tdp is not None else 0.0),
                "type": str(gpu_type if gpu_type is not None else "Unknown")
            }

        cpu_dict = {}
        if self.cpu_info and isinstance(self.cpu_info, dict):
            cpu_brand = self.cpu_info.get("brand")
            cpu_cores = self.cpu_info.get("cores")
            cpu_tdp = self.cpu_info.get("tdp")
            cpu_dict = {
                "brand": str(cpu_brand if cpu_brand is not None else "Unknown"),
                "cores": int(cpu_cores if cpu_cores is not None else 1),
                "tdp_w": float(cpu_tdp if cpu_tdp is not None else 65.0)
            }
        else:
            cpu_dict = {
                "brand": "Unknown",
                "cores": 1,
                "tdp_w": 65.0
            }

        meta = {
            "version": __version__,
            "exported_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "run_id": self._run_id,
            "run_label": self._run_label,
            "region_code": self.region_code,
            "carbon_intensity": self.carbon_intensity,
            "intensity_source": self._intensity_source,
            "cpu": cpu_dict,
            "ram": ram_dict,
            "gpu": gpu_dict
        }

        measurements = []
        total_carbon = 0.0
        total_duration = 0.0
        func_carbon_map = {}

        if os.path.isfile(csv_path):
            try:
                with open(csv_path, "r", encoding="utf-8") as f:
                    reader = csv.DictReader(f)
                    for row in reader:
                        try:
                            carbon_val = float(row.get("Carbon(gCO2)", 0))
                            duration_val = float(row.get("Duration(s)", 0))
                        except (ValueError, TypeError):
                            continue

                        record = {
                            "date": row.get("Date", ""),
                            "function": row.get("Function", "unknown"),
                            "duration_s": duration_val,
                            "carbon_gco2": carbon_val,
                            "region": row.get("Region", ""),
                            "avg_cpu_pct": row.get("AvgCPU(%)", "N/A"),
                            "file_path": row.get("FilePath", "N/A"),
                            "line": row.get("Line", "N/A"),
                            "run_id": row.get("RunID", "N/A"),
                            "run_label": row.get("RunLabel", ""),
                        }
                        measurements.append(record)

                        total_carbon += carbon_val
                        total_duration += duration_val

                        fname = record["function"]
                        if fname not in func_carbon_map:
                            func_carbon_map[fname] = 0.0
                        func_carbon_map[fname] += carbon_val

            except Exception as e:
                logger.warning(f"CSV read error, exporting metadata only: {e}")

        top_emitters = sorted(func_carbon_map.items(), key=lambda x: x[1], reverse=True)[:5]

        summary = {
            "total_carbon_gco2": round(total_carbon, 8),
            "total_duration_s": round(total_duration, 4),
            "measurement_count": len(measurements),
            "session_carbon_gco2": round(self.total_carbon, 8),
            "top_emitters": [
                {"function": name, "carbon_gco2": round(carbon, 8)}
                for name, carbon in top_emitters
            ]
        }

        report = {
            "meta": meta,
            "measurements": measurements,
            "summary": summary
        }

        with open(filename, "w", encoding="utf-8") as f:
            _json.dump(report, f, indent=2, ensure_ascii=False)

        logger.info(f"JSON report written: {filename} ({len(measurements)} records)")

    @contextmanager
    def track_block(self, block_name="custom_block"):
        """Context manager for tracking arbitrary code blocks.

        Usage:
            with eco.track_block("data_processing"):
                result = expensive_operation()

        Args:
            block_name: Name to use for the tracked block in reports.

        Yields:
            None: Control flow continues within the context.
        """
        start_time = time.perf_counter()
        try:
            with self.cpu_monitor():
                if self.gpu_info:
                    with self.gpu_monitor():
                        yield
                else:
                    yield
        finally:
            end_time = time.perf_counter()
            duration = end_time - start_time

            try:
                avg_cpu = self._get_avg_cpu_in_range(start_time, end_time)
                cpu_carbon = self._compute_carbon(self.cpu_info['tdp'], avg_cpu, duration)

                gpu_carbon = 0.0
                if self.gpu_info:
                    avg_gpu, avg_gpu_pwr = self._get_avg_gpu_in_range(start_time, end_time)
                    if avg_gpu_pwr is not None:
                        gpu_energy_wh = (avg_gpu_pwr * duration) / self.SECONDS_PER_HOUR
                        gpu_carbon = (gpu_energy_wh / self.WATTS_PER_KILOWATT) * self.carbon_intensity
                    else:
                        gpu_carbon = self._compute_carbon(self.gpu_info['tdp'], avg_gpu, duration, is_gpu=True)

                carbon_emitted = cpu_carbon + gpu_carbon
                self._accumulate_carbon(carbon_emitted, block_name, duration, avg_cpu)

                logger.info(f"Block '{block_name}': {duration:.3f}s, {avg_cpu:.1f}% CPU, {carbon_emitted:.8f}g CO2")
            except Exception as e:
                logger.error(f"Block measurement failed for '{block_name}': {e}")

    def __del__(self):
        """Ensures all background monitoring threads are stopped and resources released."""
        try:
            self._stop_cpu_monitor()
            self._stop_gpu_monitor()
        except Exception:
            pass

remaining_budget property

Returns the remaining carbon budget in gCO2, or None if no limit is set.

This is the primary data interface for external consumers (IDE sidebar, CI/CD gates). The library provides the number; the consumer acts on it.

Returns:

Type	Description
	float or None: Remaining gCO2 budget, or None if no limit configured.

add_exporter(exporter)

Registers a telemetry exporter to receive carbon metrics in real-time.

Parameters:

Name	Type	Description	Default
exporter		An object implementing an export(carbon_emitted, func_name, duration, region) method.	required

Source code in ecotrace/core.py

def add_exporter(self, exporter):
    """Registers a telemetry exporter to receive carbon metrics in real-time.

    Args:
        exporter: An object implementing an `export(carbon_emitted, func_name, duration, region)` method.
    """
    self._exporters.append(exporter)

pause()

Temporarily pauses carbon tracking for the session.

Source code in ecotrace/core.py

def pause(self):
    """Temporarily pauses carbon tracking for the session."""
    with self._carbon_lock:
        if not self._paused:
            self._paused = True
            self._paused_at = time.perf_counter()
            logger.info("[EcoTrace] Instrumentation session paused.")

resume()

Resumes carbon tracking for the session.

Source code in ecotrace/core.py

def resume(self):
    """Resumes carbon tracking for the session."""
    with self._carbon_lock:
        if self._paused:
            self._paused = False
            if self._paused_at is not None:
                self._total_paused_duration += time.perf_counter() - self._paused_at
                self._paused_at = None
            logger.info("[EcoTrace] Instrumentation session resumed.")

get_summary()

Returns the current session metrics as a structured dictionary.

Provides programmatic access to all session data — suitable for use in notebooks, dashboards, custom reporting pipelines, and test assertions. Can be called at any point during or after a session.

Returns:

Name	Type	Description
dict	dict	Session summary with keys: - run_id: Short unique ID for this session. - run_label: Optional human-readable label. - duration_s: Elapsed session time in seconds. - functions_tracked: Number of tracked function calls. - total_carbon_gco2: Cumulative carbon in gCO2. - region: ISO region code. - carbon_intensity: gCO2/kWh intensity value. - intensity_source: 'live' or 'static'. - budget: Budget status dict (None if no limit set). - equivalence: Human-readable carbon comparison string. - hardware: CPU/GPU/energy sensor metadata dict.

Source code in ecotrace/core.py

def get_summary(self) -> dict:
    """Returns the current session metrics as a structured dictionary.

    Provides programmatic access to all session data — suitable for use
    in notebooks, dashboards, custom reporting pipelines, and test assertions.
    Can be called at any point during or after a session.

    Returns:
        dict: Session summary with keys:
            - ``run_id``: Short unique ID for this session.
            - ``run_label``: Optional human-readable label.
            - ``duration_s``: Elapsed session time in seconds.
            - ``functions_tracked``: Number of tracked function calls.
            - ``total_carbon_gco2``: Cumulative carbon in gCO2.
            - ``region``: ISO region code.
            - ``carbon_intensity``: gCO2/kWh intensity value.
            - ``intensity_source``: ``'live'`` or ``'static'``.
            - ``budget``: Budget status dict (``None`` if no limit set).
            - ``equivalence``: Human-readable carbon comparison string.
            - ``hardware``: CPU/GPU/energy sensor metadata dict.
    """
    session_duration = time.perf_counter() - self._session_start_time
    current_pause = 0.0
    if self._paused and self._paused_at is not None:
        current_pause = time.perf_counter() - self._paused_at
    active_duration = max(0.0, session_duration - self._total_paused_duration - current_pause)

    # Determine energy sensor label
    if self.hardware.rapl_available:
        sensor = "RAPL (Exact Hardware)"
    elif self.hardware.apple_silicon_available:
        sensor = "Apple Silicon (powermetrics)"
    else:
        sensor = "Prediction Mode (Boavizta Estimation, ~15-20% error margin)"

    budget_info = None
    if self.carbon_limit is not None:
        remaining = self.remaining_budget
        used_pct = (self.total_carbon / self.carbon_limit * 100) if self.carbon_limit else 0.0
        budget_info = {
            "limit_gco2": self.carbon_limit,
            "remaining_gco2": remaining,
            "used_pct": round(used_pct, 2),
            "status": "EXCEEDED" if remaining == 0 else "OK",
        }

    return {
        "run_id": self._run_id,
        "run_label": self._run_label,
        "duration_s": round(active_duration, 4),
        "functions_tracked": self._tracked_functions_count,
        "total_carbon_gco2": self.total_carbon,
        "region": self.region_code,
        "carbon_intensity": self.carbon_intensity,
        "intensity_source": self._intensity_source,
        "budget": budget_info,
        "equivalence": self.equivalence(self.total_carbon),
        "hardware": {
            "cpu": self.cpu_info.get("brand", "Unknown"),
            "cores": self.cpu_info.get("cores", 0),
            "tdp_w": self.cpu_info.get("tdp", 0),
            "gpu": self.gpu_info["brand"] if self.gpu_info else None,
            "energy_sensor": sensor,
        },
    }

equivalence(gco2)

Converts a gCO2 value into a human-readable real-world comparison.

Uses a tiered system: selects the most relatable comparison based on the magnitude of the emission value. The library owns this conversion; external consumers (IDE, reports) can call it to enrich their display.

Parameters:

Name	Type	Description	Default
gco2		Carbon emissions in grams of CO2.	required

Returns:

Name	Type	Description
str		Human-readable equivalence string, or empty string if the
		value is too small to compare meaningfully.

Source code in ecotrace/core.py

def equivalence(self, gco2):
    """Converts a gCO2 value into a human-readable real-world comparison.

    Uses a tiered system: selects the most relatable comparison based on
    the magnitude of the emission value. The library owns this conversion;
    external consumers (IDE, reports) can call it to enrich their display.

    Args:
        gco2: Carbon emissions in grams of CO2.

    Returns:
        str: Human-readable equivalence string, or empty string if the
        value is too small to compare meaningfully.
    """
    if gco2 <= 0:
        return ""

    if gco2 < 0.01:
        searches = gco2 / 0.2
        return f"{searches:.2f} Google searches"
    elif gco2 < 1.0:
        led_minutes = (gco2 / 5.2) * 60
        return f"{led_minutes:.1f} min of LED bulb (10W)"
    elif gco2 < 10.0:
        charges = gco2 / 8.22
        return f"{charges:.2f} smartphone charges"
    elif gco2 < 100.0:
        netflix_min = (gco2 / 36.0) * 60
        return f"{netflix_min:.1f} min of Netflix streaming"
    else:
        km = gco2 / 121.0
        return f"{km:.2f} km of car driving"

cpu_monitor()

Context manager that brackets a code block with CPU monitoring.

Yields:

Name	Type	Description
EcoTrace		The current instance for optional chaining.

Source code in ecotrace/core.py

@contextmanager
def cpu_monitor(self):
    """Context manager that brackets a code block with CPU monitoring.

    Yields:
        EcoTrace: The current instance for optional chaining.
    """
    self._start_cpu_monitor()
    try:
        yield self
    finally:
        self._stop_cpu_monitor()

gpu_monitor()

Context manager that brackets a code block with GPU monitoring.

Yields:

Name	Type	Description
EcoTrace		The current instance for optional chaining.

Source code in ecotrace/core.py

@contextmanager
def gpu_monitor(self):
    """Context manager that brackets a code block with GPU monitoring.

    Yields:
        EcoTrace: The current instance for optional chaining.
    """
    self._start_gpu_monitor()
    try:
        yield self
    finally:
        self._stop_gpu_monitor()

track(func)

Decorator that measures carbon emissions for any function call.

Automatically detects whether the target is synchronous or asynchronous and selects the appropriate measurement strategy.

Parameters:

Name	Type	Description	Default
func		The function to decorate. Can be sync or async.	required

Returns:

Name	Type	Description
Callable		Wrapped function that measures emissions transparently.

Source code in ecotrace/core.py

def track(self, func):
    """Decorator that measures carbon emissions for any function call.

    Automatically detects whether the target is synchronous or asynchronous
    and selects the appropriate measurement strategy.

    Args:
        func: The function to decorate. Can be sync or async.

    Returns:
        Callable: Wrapped function that measures emissions transparently.
    """
    if inspect.iscoroutinefunction(func):
        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            res = await self.measure_async(func, *args, **kwargs)
            return res["result"] if isinstance(res, dict) else res
        return async_wrapper

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        res = self.measure(func, *args, **kwargs)
        return res["result"] if isinstance(res, dict) else res

    return wrapper

track_gpu(func)

Decorator that measures GPU carbon emissions with real utilization monitoring.

If no GPU is detected, the wrapped function executes normally without measurement. If the GPU becomes unavailable mid-calculation, the function result is preserved and a warning is logged.

Parameters:

Name	Type	Description	Default
func		The function to decorate.	required

Returns:

Name	Type	Description
Callable		Wrapped function with GPU carbon measurement.

Source code in ecotrace/core.py

def track_gpu(self, func):
    """Decorator that measures GPU carbon emissions with real utilization monitoring.

    If no GPU is detected, the wrapped function executes normally without
    measurement. If the GPU becomes unavailable mid-calculation, the function
    result is preserved and a warning is logged.

    Args:
        func: The function to decorate.

    Returns:
        Callable: Wrapped function with GPU carbon measurement.
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if self.gpu_info is None:
            logger.warning(f"No GPU detected, executing '{func.__name__}' without measurement.")
            return func(*args, **kwargs)

        try:
            file_path = os.path.abspath(inspect.getfile(func))
            line_number = inspect.getsourcelines(func)[1]
        except Exception:
            file_path, line_number = None, None

        start_time = time.perf_counter()
        try:
            with self.cpu_monitor():
                with self.gpu_monitor():
                    result = func(*args, **kwargs)
            return result
        finally:
            end_time = time.perf_counter()
            try:
                duration = end_time - start_time
                avg_cpu = self._get_avg_cpu_in_range(start_time, end_time)
                cpu_carbon = self._compute_carbon(self.cpu_info['tdp'], avg_cpu, duration)

                avg_gpu_util, avg_gpu_pwr = self._get_avg_gpu_in_range(start_time, end_time)
                if avg_gpu_pwr is not None:
                    gpu_energy_wh = (avg_gpu_pwr * duration) / self.SECONDS_PER_HOUR
                    gpu_carbon = (gpu_energy_wh / self.WATTS_PER_KILOWATT) * self.carbon_intensity
                else:
                    gpu_carbon = self._compute_carbon((self.gpu_info or {}).get('tdp', 100.0), avg_gpu_util, duration, is_gpu=True)

                carbon_emitted = cpu_carbon + gpu_carbon
                self._accumulate_carbon(carbon_emitted, func.__name__, duration, avg_cpu=avg_cpu, file_path=file_path, line_number=line_number)
                logger.info(f"GPU/CPU Carbon Emissions: {carbon_emitted:.8f} gCO2")
                logger.info(f"Duration     : {duration:.4f} sec")
                logger.info(f"GPU Usage    : {avg_gpu_util:.1f}%")
                logger.info(f"CO2          : {carbon_emitted:.8f} gCO2")
            except Exception as e:
                logger.error(f"GPU measurement failed for '{func.__name__}': {e}")
    return wrapper

measure(func, *args, **kwargs)

Executes a synchronous function and measures its CPU carbon emissions.

Uses continuous background sampling for accurate utilization measurement. If the measurement calculation fails, the function result is still returned.

Parameters:

Name	Type	Description	Default
func		Synchronous callable to measure.	required
*args		Positional arguments forwarded to func.	()
**kwargs		Keyword arguments forwarded to func.	{}

Returns:

Name	Type	Description
dict		Keys func_name, duration, avg_cpu, carbon,
		cpu_samples, and result.

Source code in ecotrace/core.py

def measure(self, func, *args, **kwargs):
    """Executes a synchronous function and measures its CPU carbon emissions.

    Uses continuous background sampling for accurate utilization measurement.
    If the measurement calculation fails, the function result is still returned.

    Args:
        func: Synchronous callable to measure.
        *args: Positional arguments forwarded to ``func``.
        **kwargs: Keyword arguments forwarded to ``func``.

    Returns:
        dict: Keys ``func_name``, ``duration``, ``avg_cpu``, ``carbon``,
        ``cpu_samples``, and ``result``.
    """
    start_time = time.perf_counter()
    result_data = None
    func_success = False
    energy_start = None

    try:
        energy_start = self.hardware.get_cpu_energy_j()
        with self.cpu_monitor():
            if self.gpu_info:
                with self.gpu_monitor():
                    result_data = func(*args, **kwargs)
            else:
                result_data = func(*args, **kwargs)
            func_success = True
    finally:
        end_time = time.perf_counter()
        energy_end = self.hardware.get_cpu_energy_j()
        duration = end_time - start_time

        try:
            avg_cpu = self._get_avg_cpu_in_range(start_time, end_time)

            with self._cpu_sample_lock:
                measurement_samples = list(self._cpu_samples)

            # Capture location info robustly for decorated and async functions
            file_path, line_number = self._get_source_location(func)

            energy_delta_j = None
            if energy_start is not None and energy_end is not None:
                energy_delta_j = max(0.0, energy_end - energy_start)

            carbon_emitted = self._compute_carbon(self.cpu_info['tdp'], avg_cpu, duration, energy_delta_j=energy_delta_j)
            self._accumulate_carbon(carbon_emitted, func.__name__, duration, avg_cpu, file_path=file_path, line_number=line_number)

            if func_success:
                return {
                    "func_name": func.__name__,
                    "duration": duration,
                    "avg_cpu": avg_cpu,
                    "carbon": carbon_emitted,
                    "cpu_samples": measurement_samples,
                    "result": result_data
                }
        except Exception as e:
            logger.error(f"Measurement failed for '{func.__name__}': {e}")
            if func_success:
                return {
                    "func_name": func.__name__,
                    "duration": duration,
                    "avg_cpu": 0.0,
                    "carbon": 0.0,
                    "cpu_samples": [],
                    "result": result_data
                }

measure_async(func, *args, **kwargs) async

Executes an async function and measures its CPU carbon emissions.

Uses continuous background sampling, which is particularly important for bursty or I/O-bound async workloads where point-in-time readings misrepresent actual utilization.

Parameters:

Name	Type	Description	Default
func		Async callable to measure.	required
*args		Positional arguments forwarded to func.	()
**kwargs		Keyword arguments forwarded to func.	{}

Returns:

Name	Type	Description
dict		Keys func_name, duration, avg_cpu, carbon,
		cpu_samples, and result.

Raises:

Type	Description
Exception	Re-raises any exception from the wrapped function after

Source code in ecotrace/core.py

async def measure_async(self, func, *args, **kwargs):
    """Executes an async function and measures its CPU carbon emissions.

    Uses continuous background sampling, which is particularly important
    for bursty or I/O-bound async workloads where point-in-time readings
    misrepresent actual utilization.

    Args:
        func: Async callable to measure.
        *args: Positional arguments forwarded to ``func``.
        **kwargs: Keyword arguments forwarded to ``func``.

    Returns:
        dict: Keys ``func_name``, ``duration``, ``avg_cpu``, ``carbon``,
        ``cpu_samples``, and ``result``.

    Raises:
        Exception: Re-raises any exception from the wrapped function after
        completing the measurement teardown.
    """
    start_time = time.perf_counter()
    result_data = None
    func_success = False
    energy_start = None

    try:
        energy_start = self.hardware.get_cpu_energy_j()
        with self.cpu_monitor():
            try:
                if self.gpu_info:
                    with self.gpu_monitor():
                        result_data = await func(*args, **kwargs)
                else:
                    result_data = await func(*args, **kwargs)
                func_success = True
            finally:
                await asyncio.sleep(self.MONITOR_INTERVAL_S)  # Allow trailing samples to be captured
    finally:
        end_time = time.perf_counter()
        energy_end = self.hardware.get_cpu_energy_j()
        duration = end_time - start_time

        try:
            avg_cpu = self._get_avg_cpu_in_range(start_time, end_time)

            with self._cpu_sample_lock:
                measurement_samples = list(self._cpu_samples)

            energy_delta_j = None
            if energy_start is not None and energy_end is not None:
                energy_delta_j = max(0.0, energy_end - energy_start)

            # Capture location info robustly for decorated and async functions
            file_path, line_number = self._get_source_location(func)

            carbon_emitted = self._compute_carbon(self.cpu_info['tdp'], avg_cpu, duration, energy_delta_j=energy_delta_j)
            self._accumulate_carbon(carbon_emitted, func.__name__, duration, avg_cpu, file_path=file_path, line_number=line_number)

            if func_success:
                return {
                    "func_name": func.__name__,
                    "duration": duration,
                    "avg_cpu": avg_cpu,
                    "carbon": carbon_emitted,
                    "cpu_samples": measurement_samples,
                    "result": result_data
                }
        except Exception as e:
            logger.error(f"Async measurement failed for '{func.__name__}': {e}")
            if func_success:
                return {
                    "func_name": func.__name__,
                    "duration": duration,
                    "avg_cpu": 0.0,
                    "carbon": 0.0,
                    "cpu_samples": [],
                    "result": result_data
                }

compare(func1, func2)

Runs two functions sequentially and compares their carbon footprints.

Parameters:

Name	Type	Description	Default
func1		First callable to measure.	required
func2		Second callable to measure.	required

Returns:

Name	Type	Description
dict		Keys func1 and func2, each containing the full
		measurement dict from measure().

Source code in ecotrace/core.py

def compare(self, func1, func2):
    """Runs two functions sequentially and compares their carbon footprints.

    Args:
        func1: First callable to measure.
        func2: Second callable to measure.

    Returns:
        dict: Keys ``func1`` and ``func2``, each containing the full
        measurement dict from ``measure()``.
    """
    result1 = self.measure(func1)
    result2 = self.measure(func2)
    if isinstance(result1, dict) and isinstance(result2, dict):
        logger.info(f"Comparison Results:")
        logger.info(f"Function 1: {result1['func_name']} - Duration: {result1['duration']:.4f} sec - CO2: {result1['carbon']:.8f} gCO2")
        logger.info(f"Function 2: {result2['func_name']} - Duration: {result2['duration']:.4f} sec - CO2: {result2['carbon']:.8f} gCO2")
    return {"func1": result1, "func2": result2}

generate_pdf_report(filename='ecotrace_full_report.pdf', comparison=None, cpu_samples=None, gpu_samples=None)

Generates a comprehensive PDF audit report dynamically.

If cpu_samples or gpu_samples are not provided, the engine automatically snapshots the internal session deques for full-history reporting.

Source code in ecotrace/core.py

def generate_pdf_report(self, filename="ecotrace_full_report.pdf", comparison=None, cpu_samples=None, gpu_samples=None):
    """Generates a comprehensive PDF audit report dynamically.

    If cpu_samples or gpu_samples are not provided, the engine automatically
    snapshots the internal session deques for full-history reporting.
    """
    from .report import generate_pdf_report as generate_pdf

    # CPU Samples snapshot
    final_cpu_samples = None
    if cpu_samples is not None:
        final_cpu_samples = list(cpu_samples)
    else:
        with self._cpu_sample_lock:
            final_cpu_samples = list(self._cpu_samples)

    # GPU Samples snapshot — normalize 3-tuples (ts, util, power) to
    # 2-tuples (ts, util) expected by report chart functions.
    final_gpu_samples = None
    if gpu_samples is not None:
        final_gpu_samples = [(item[0], item[1]) for item in gpu_samples]
    elif self.gpu_info:
        with self._gpu_sample_lock:
            final_gpu_samples = [(item[0], item[1]) for item in self._gpu_samples]

    generate_pdf(
        filename=filename,
        cpu_info=self.cpu_info,
        gpu_info=self.gpu_info,
        region_code=self.region_code,
        comparison=comparison,
        cpu_samples=final_cpu_samples,
        gpu_samples=final_gpu_samples,
        api_key=self.api_key
    )

export_json(filename='ecotrace_report.json', csv_path='ecotrace_log.csv')

Exports session data to a structured JSON file.

Combines hardware metadata, measurement history from the CSV audit log, and aggregate statistics into a single machine-readable document. This output is designed for consumption by the VS Code extension sidebar, CI/CD carbon gates, and third-party analytics tools.

The JSON schema contains three top-level keys:

• meta: Hardware profile, region, version, and export timestamp.
• measurements: Array of per-function measurement records from the CSV audit log.
• summary: Aggregate statistics (total carbon, total duration, measurement count, top emitters).

Parameters:

Name	Type	Description	Default
filename		Output path for the JSON file. Defaults to ecotrace_report.json in the current working directory.	'ecotrace_report.json'
csv_path		Path to the CSV audit log to read measurements from. Defaults to ecotrace_log.csv.	'ecotrace_log.csv'

Raises:

Type	Description
IOError	If the output file cannot be written.

Example::

eco = EcoTrace(region_code="TR")

@eco.track
def my_function():
    pass

my_function()
eco.export_json("report.json")

Source code in ecotrace/core.py

def export_json(self, filename="ecotrace_report.json", csv_path="ecotrace_log.csv"):
    """Exports session data to a structured JSON file.

    Combines hardware metadata, measurement history from the CSV audit
    log, and aggregate statistics into a single machine-readable document.
    This output is designed for consumption by the VS Code extension
    sidebar, CI/CD carbon gates, and third-party analytics tools.

    The JSON schema contains three top-level keys:

    - ``meta``: Hardware profile, region, version, and export timestamp.
    - ``measurements``: Array of per-function measurement records from
      the CSV audit log.
    - ``summary``: Aggregate statistics (total carbon, total duration,
      measurement count, top emitters).

    Args:
        filename: Output path for the JSON file. Defaults to
            ``ecotrace_report.json`` in the current working directory.
        csv_path: Path to the CSV audit log to read measurements from.
            Defaults to ``ecotrace_log.csv``.

    Raises:
        IOError: If the output file cannot be written.

    Example::

        eco = EcoTrace(region_code="TR")

        @eco.track
        def my_function():
            pass

        my_function()
        eco.export_json("report.json")
    """
    import json as _json
    from . import __version__

    ram_dict = None
    if self.ram_info and isinstance(self.ram_info, dict):
        ram_total_gb = self.ram_info.get("total_gb")
        ram_type_val = self.ram_info.get("type")
        ram_dict = {
            "total_gb": round(float(ram_total_gb if ram_total_gb is not None else 0.0), 2),
            "type": str(ram_type_val if ram_type_val is not None else "DDR4")
        }

    gpu_dict = None
    if self.gpu_info and isinstance(self.gpu_info, dict):
        gpu_brand = self.gpu_info.get("brand")
        gpu_tdp = self.gpu_info.get("tdp")
        gpu_type = self.gpu_info.get("type")
        gpu_dict = {
            "brand": str(gpu_brand if gpu_brand is not None else "Unknown"),
            "tdp_w": float(gpu_tdp if gpu_tdp is not None else 0.0),
            "type": str(gpu_type if gpu_type is not None else "Unknown")
        }

    cpu_dict = {}
    if self.cpu_info and isinstance(self.cpu_info, dict):
        cpu_brand = self.cpu_info.get("brand")
        cpu_cores = self.cpu_info.get("cores")
        cpu_tdp = self.cpu_info.get("tdp")
        cpu_dict = {
            "brand": str(cpu_brand if cpu_brand is not None else "Unknown"),
            "cores": int(cpu_cores if cpu_cores is not None else 1),
            "tdp_w": float(cpu_tdp if cpu_tdp is not None else 65.0)
        }
    else:
        cpu_dict = {
            "brand": "Unknown",
            "cores": 1,
            "tdp_w": 65.0
        }

    meta = {
        "version": __version__,
        "exported_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "run_id": self._run_id,
        "run_label": self._run_label,
        "region_code": self.region_code,
        "carbon_intensity": self.carbon_intensity,
        "intensity_source": self._intensity_source,
        "cpu": cpu_dict,
        "ram": ram_dict,
        "gpu": gpu_dict
    }

    measurements = []
    total_carbon = 0.0
    total_duration = 0.0
    func_carbon_map = {}

    if os.path.isfile(csv_path):
        try:
            with open(csv_path, "r", encoding="utf-8") as f:
                reader = csv.DictReader(f)
                for row in reader:
                    try:
                        carbon_val = float(row.get("Carbon(gCO2)", 0))
                        duration_val = float(row.get("Duration(s)", 0))
                    except (ValueError, TypeError):
                        continue

                    record = {
                        "date": row.get("Date", ""),
                        "function": row.get("Function", "unknown"),
                        "duration_s": duration_val,
                        "carbon_gco2": carbon_val,
                        "region": row.get("Region", ""),
                        "avg_cpu_pct": row.get("AvgCPU(%)", "N/A"),
                        "file_path": row.get("FilePath", "N/A"),
                        "line": row.get("Line", "N/A"),
                        "run_id": row.get("RunID", "N/A"),
                        "run_label": row.get("RunLabel", ""),
                    }
                    measurements.append(record)

                    total_carbon += carbon_val
                    total_duration += duration_val

                    fname = record["function"]
                    if fname not in func_carbon_map:
                        func_carbon_map[fname] = 0.0
                    func_carbon_map[fname] += carbon_val

        except Exception as e:
            logger.warning(f"CSV read error, exporting metadata only: {e}")

    top_emitters = sorted(func_carbon_map.items(), key=lambda x: x[1], reverse=True)[:5]

    summary = {
        "total_carbon_gco2": round(total_carbon, 8),
        "total_duration_s": round(total_duration, 4),
        "measurement_count": len(measurements),
        "session_carbon_gco2": round(self.total_carbon, 8),
        "top_emitters": [
            {"function": name, "carbon_gco2": round(carbon, 8)}
            for name, carbon in top_emitters
        ]
    }

    report = {
        "meta": meta,
        "measurements": measurements,
        "summary": summary
    }

    with open(filename, "w", encoding="utf-8") as f:
        _json.dump(report, f, indent=2, ensure_ascii=False)

    logger.info(f"JSON report written: {filename} ({len(measurements)} records)")

track_block(block_name='custom_block')

Context manager for tracking arbitrary code blocks.

Usage

with eco.track_block("data_processing"): result = expensive_operation()

Parameters:

Name	Type	Description	Default
block_name		Name to use for the tracked block in reports.	'custom_block'

Yields:

Name	Type	Description
None		Control flow continues within the context.

Source code in ecotrace/core.py

@contextmanager
def track_block(self, block_name="custom_block"):
    """Context manager for tracking arbitrary code blocks.

    Usage:
        with eco.track_block("data_processing"):
            result = expensive_operation()

    Args:
        block_name: Name to use for the tracked block in reports.

    Yields:
        None: Control flow continues within the context.
    """
    start_time = time.perf_counter()
    try:
        with self.cpu_monitor():
            if self.gpu_info:
                with self.gpu_monitor():
                    yield
            else:
                yield
    finally:
        end_time = time.perf_counter()
        duration = end_time - start_time

        try:
            avg_cpu = self._get_avg_cpu_in_range(start_time, end_time)
            cpu_carbon = self._compute_carbon(self.cpu_info['tdp'], avg_cpu, duration)

            gpu_carbon = 0.0
            if self.gpu_info:
                avg_gpu, avg_gpu_pwr = self._get_avg_gpu_in_range(start_time, end_time)
                if avg_gpu_pwr is not None:
                    gpu_energy_wh = (avg_gpu_pwr * duration) / self.SECONDS_PER_HOUR
                    gpu_carbon = (gpu_energy_wh / self.WATTS_PER_KILOWATT) * self.carbon_intensity
                else:
                    gpu_carbon = self._compute_carbon(self.gpu_info['tdp'], avg_gpu, duration, is_gpu=True)

            carbon_emitted = cpu_carbon + gpu_carbon
            self._accumulate_carbon(carbon_emitted, block_name, duration, avg_cpu)

            logger.info(f"Block '{block_name}': {duration:.3f}s, {avg_cpu:.1f}% CPU, {carbon_emitted:.8f}g CO2")
        except Exception as e:
            logger.error(f"Block measurement failed for '{block_name}': {e}")

__del__()

Ensures all background monitoring threads are stopped and resources released.

Source code in ecotrace/core.py

def __del__(self):
    """Ensures all background monitoring threads are stopped and resources released."""
    try:
        self._stop_cpu_monitor()
        self._stop_gpu_monitor()
    except Exception:
        pass

---

CLI Module

Technical details of the Command Line Interface (ecotrace run, ecotrace gate, etc.).

ecotrace.cli

EcoTrace CLI — Terminal interface for carbon-aware script profiling.

Provides four subcommands for headless carbon instrumentation without modifying the target source code:

ecotrace run <script.py>     Run a script under full carbon monitoring
ecotrace analyze             Summarize existing CSV audit logs
ecotrace export --json       Export session data to machine-readable JSON
ecotrace benchmark           Measure EcoTrace's own overhead

Design constraints

• Zero external dependencies (argparse + runpy + json from stdlib)
• Must never crash with ugly tracebacks — all commands are fail-safe
• runpy.run_path keeps us in the same process so psutil isolation works

main()

Main entry point for the ecotrace CLI command.

---

Report Module

Generates PDF and CSV carbon audit reports.

ecotrace.report

create_cpu_usage_chart(samples_data, core_count=1)

Renders a CPU usage line chart and saves it to a temporary PNG file.

Generates a visual representation of CPU utilization normalized against the total available logical cores to prevent inflation on heavily multi-threaded systems.

Parameters:

Name	Type	Description	Default
samples_data	list	List of (timestamp, cpu_percent) float tuples.	required
core_count	int	Number of logical processor cores for scaling reference.	1

Returns:

Type	Description
	str or None: Absolute path to the generated PNG file, or None on failure.

create_gpu_usage_chart(samples_data)

Renders a GPU utilization line chart and saves it as a temporary PNG.

Parameters:

Name	Type	Description	Default
samples_data	list	List of (timestamp, gpu_percent) float tuples.	required

Returns:

Type	Description
	str or None: Path to PNG or None on failure.

get_gemini_insights(api_key, cpu_info, gpu_info, history, region_code)

Fetches dynamic carbon-optimization insights from Google Gemini.

Constructs a detailed prompt containing hardware specs, regional grid intensity, and recent execution history to generate actionable 'Green Coding' advice.

Parameters:

Name	Type	Description	Default
api_key	str	Google Gemini API key.	required
cpu_info	dict	CPU hardware specs.	required
gpu_info	dict	GPU hardware specs (optional).	required
history	list	Recent measurements from CSV.	required
region_code	str	ISO region for grid intensity context.	required

Returns:

Name	Type	Description
str		AI-generated insights or an error message if the call fails.

sanitize_for_pdf(text)

Strips non-ASCII characters for safe PDF rendering.

Parameters:

Name	Type	Description	Default
text	str	Input string potentially containing non-ASCII characters.	required

Returns:

Name	Type	Description
str		ASCII-safe string formatted for FPDF encoding restrictions.

---

Hardware Module

Auto-detects CPU, GPU, and RAM specifications.

ecotrace.hardware

HardwareMonitor

Hardware-level energy monitoring using exact RAPL sensors or advanced estimation.

Implements a hybrid approach to zero-configuration carbon tracking: 1. Hardware Mode (RAPL): Reads raw energy counters on supported hardware (Linux). 2. Advanced Estimation Mode: Falls back to Boavizta's logarithmic load curve to eliminate linear TDP estimation errors on unsupported systems.

estimate_cpu_power_w(tdp, utilization_pct)

Advanced CPU power estimation based on Boavizta's non-linear curve.

Unlike simple linear TDP multiplication (TDP * utilization), this model accounts for baseline idle power and exponential load scaling.

Parameters:

Name	Type	Description	Default
tdp	float	Thermal Design Power in watts.	required
utilization_pct	float	CPU load percentage (0-100).	required

Returns:

Name	Type	Description
float		Estimated power draw in watts.

get_cpu_energy_j()

Reads the current CPU package energy counter in Joules.

Tries RAPL first (Linux), then Apple Silicon powermetrics (macOS arm64).

Returns:

Name	Type	Description
float		Current energy in Joules, or None if hardware is unavailable.

---

CPU Module

CPU TDP lookup and energy estimation logic.

ecotrace.cpu

fetch_raw_cpu_info() cached

Retrieves raw CPU information bounding to py-cpuinfo caching.

Returns:

Name	Type	Description
dict		Standardized CPU properties including architecture and physical identifiers.

get_cpu_info(tdp_db, constants_data)

Detects CPU hardware and resolves Thermal Design Power using a multi-source chain.

Matches available chip strings strictly against Apple Silicon static definitions first, before fuzzy matching against the Boavizta hardware database for x86 chips.

Parameters:

Name	Type	Description	Default
tdp_db	dict	Generated dictionary matching CPU hardware to known TDPs.	required
constants_data	dict	Application-wide constant configurations containing 'TDP_MAP'.	required

Returns:

Name	Type	Description
dict		CPU characteristics comprising: - brand (str): ASCII-cleaned display name for the physical CPU. - cores (int): Count of logical processing threads utilizing the OS scheduler. - tdp (float): Assigned structural TDP boundary in watts.

load_tdp_database(csv_path)

Parses the Boavizta CPU specification dataset into a TDP lookup table.

Parameters:

Name	Type	Description	Default
csv_path	str	File system path targeting 'cpu_data.csv'.	required

Returns:

Name	Type	Description
dict		Hash map linking lowercase exact CPU model strings to their float TDP values.

---

GPU Module

GPU power monitoring (NVIDIA NVML, AMD/Intel WMI).

ecotrace.gpu

get_gpu_info(gpu_index, gpu_tdp_defaults)

Detects GPU hardware and resolves its power limit using a tri-vendor fallback chain.

Initializes NVIDIA NVML for precision TDP tracking. If an NVIDIA GPU is not found, it falls back to WMI (Windows Management Instrumentation) to perform string-based matching and assigns default architectural estimated TDPs.

Parameters:

Name	Type	Description	Default
gpu_index	int	Zero-based index of the target GPU device to monitor.	required
gpu_tdp_defaults	dict	Vendor mapping for missing hardware limits (Intel/AMD).	required

Returns:

Type	Description
	dict or None: GPU metadata dictionary containing: - brand (str): Human-readable device name. - tdp (float): Power management limit in watts. - type (str): Vendor classifier ('nvidia', 'intel', 'amd', 'unknown'). - handle (object): NVML hardware handle if available, otherwise None.
	Returns None if no GPU is detected.

get_gpu_power_w(gpu_info)

Reads exact instantaneous GPU power usage in watts via NVML.

Parameters:

Name	Type	Description	Default
gpu_info	dict	The GPU metadata dictionary from get_gpu_info.	required

Returns:

Name	Type	Description
float		Current power draw in watts, or None if unavailable/unsupported.

---

RAM Module

Process-scoped memory usage and energy calculation.

ecotrace.ram

get_ram_info()

Detects RAM specifications including type, speed, and total capacity.

Performs OS-specific detection using WMIC on Windows and dmidecode on Linux to retrieve the memory speed, which is used to classify the RAM type (DDR4 vs DDR5).

Returns:

Name	Type	Description
dict		Dictionary containing: - total_gb (float): System total memory in gigabytes. - type (str): RAM generation ('DDR4' or 'DDR5'). - speed_mhz (str): Active memory frequency, or 'Unknown'.

---

Config Module

Region validation, carbon intensity resolution, and Live Grid API integration.

ecotrace.config

fetch_live_carbon_intensity(region_code, grid_api_key)

Fetches real-time carbon intensity from the Electricity Maps API.

Queries the Electricity Maps /v3/carbon-intensity/latest endpoint for the specified region's current grid carbon intensity. The API requires a valid authentication token.

This function is designed to be fail-safe: any network error, timeout, authentication failure, or malformed response will cause it to return None, allowing the caller to fall back to static data.

Parameters:

Name	Type	Description	Default
region_code		ISO 3166-1 alpha-2 country code (e.g. 'TR', 'DE'). Converted to an Electricity Maps zone via ZONE_MAPPING.	required
grid_api_key		Electricity Maps API authentication token.	required

Returns:

Type	Description
	float or None: Real-time carbon intensity in gCO2eq/kWh if the
	API query succeeds, or None if any error occurs. The caller
	should fall back to static CARBON_INTENSITY_MAP values
	when None is returned.

Example

intensity = fetch_live_carbon_intensity("TR", "my-api-key") if intensity is not None: ... print(f"Live grid: {intensity} gCO2/kWh") ... else: ... print("Using static fallback data")

get_cli_config_path()

Returns the absolute file system path to the CLI configuration file (~/.ecotrace/config.json).

identify_user_region()

Attempts to auto-detect the user's current region via IP address.

load_cli_config(config_path=None)

Loads stored CLI credentials and settings.

Returns:

Name	Type	Description
dict		Config values (e.g. {'api_key': 'eco_usr_...', 'endpoint': '...'}), or empty dict if not found.

load_constants(json_path)

Loads constants from the JSON configuration file.

Parameters:

Name	Type	Description	Default
json_path	str	Absolute or relative path to the constants.json file.	required

Returns:

Name	Type	Description
dict		Parsed JSON data, or an empty dictionary if loading fails.

load_gpu_tdp_defaults(constants_data)

Retrieves default GPU TDP estimations based on vendor.

Parameters:

Name	Type	Description	Default
constants_data	dict	Data dictionary containing 'GPU_TDP_DEFAULTS'.	required

Returns:

Name	Type	Description
dict		Mapping of vendor names (intel, amd, unknown) to their respective TDPs in watts.

resolve_carbon_intensity(region_code, constants_data)

Resolves the carbon intensity value for the given region.

Parameters:

Name	Type	Description	Default
region_code	str	The validated region code string.	required
constants_data	dict	Data dictionary containing 'CARBON_INTENSITY_MAP'.	required

Returns:

Name	Type	Description
float		Carbon intensity value in gCO2/kWh.

save_cli_config(data, config_path=None)

Saves CLI credentials and settings to ~/.ecotrace/config.json.

validate_region_code(region_code, constants_data)

Validates the provided region code against known carbon mappings.

Parameters:

Name	Type	Description	Default
region_code	str	The ISO 3166-1 alpha-2 country code.	required
constants_data	dict	Data dictionary containing 'CARBON_INTENSITY_MAP'.	required

Returns:

Name	Type	Description
str		Validated uppercase region code, or DEFAULT_REGION if invalid.

---

Exceptions Module

All domain-specific exceptions raised by EcoTrace.

ecotrace.exceptions

EcoTrace - Custom Exceptions

This module defines domain-specific exceptions for EcoTrace to provide meaningful, programmatic error handling instead of generic built-in failures.

AIInsightsError

Bases: EcoTraceError

Raised when Google Gemini fails to generate insights.

CPUMonitoringError

Bases: EcoTraceError

Raised when the CPU/process tree cannot be read due to permission or lifecycle issues.

EcoTraceConfigurationError

Bases: EcoTraceError

Raised when EcoTrace is initialized with invalid parameters (e.g. negative GPU index).

EcoTraceError

Bases: Exception

Base exception for all EcoTrace-related errors.

EcoTraceUpdateError

Bases: EcoTraceError

Raised (or logged) when the auto-updater fails to connect to PyPI.

GPUMonitoringError

Bases: EcoTraceError

Raised when the GPU monitoring thread encounters an unrecoverable failure.

LiveGridAPIError

Bases: EcoTraceError

Raised when the real-time grid intensity API fails or times out.

ReportGenerationError

Bases: EcoTraceError

Raised when the PDF report cannot be compiled or saved to disk.

---

Middleware

Flask (ecotrace.middleware.flask)

ecotrace.middleware.flask

EcoTraceFlask

Flask extension for tracking carbon emissions per request.

Injects 'X-Eco-Carbon-Emitted' and 'X-Eco-Duration' headers into every response.

Parameters:

Name	Type	Description	Default
app		The Flask application.	None
ecotrace_instance	Optional[EcoTrace]	Optional initialized EcoTrace instance. If not provided, it creates one quietly.	None
log_to_csv	bool	Whether to log each request to the ecotrace_log.csv. Default False.	False

init_app(app)

Initializes the extension with the Flask app.

FastAPI / Starlette (ecotrace.middleware.fastapi)

ecotrace.middleware.fastapi

EcoTraceMiddleware

Bases: BaseHTTPMiddleware

FastAPI/Starlette middleware for tracking carbon emissions per request.

Injects 'X-Eco-Carbon-Emitted' and 'X-Eco-Duration' headers into every response.

Parameters:

Name	Type	Description	Default
app		The ASGI application.	required
ecotrace_instance	Optional[EcoTrace]	Optional initialized EcoTrace instance. If not provided, it creates one quietly.	None
log_to_csv	bool	Whether to log each request to the ecotrace_log.csv. Default False.	False

dispatch(request, call_next) async

Instruments a single HTTP request for carbon monitoring.

---

Plugins

pytest Plugin (ecotrace.plugins.pytest_plugin)

ecotrace.plugins.pytest_plugin

pytest_addoption(parser)

Add command-line flag to enable EcoTrace during tests.

pytest_configure(config)

Initialize EcoTrace if the flag is enabled.

pytest_runtest_protocol(item, nextitem)

Wrap test execution for carbon metric resolution.

Snapshots process-scoped CPU utilization across the execution lifecycle of each pytest item.

pytest_terminal_summary(terminalreporter, exitstatus, config)

Print the carbon footprint summary at the end of the test session.