feat(fastapi): measurement capability tiers for per-request energy

codecarbon/emissions_tracker.py

@@ -968,7 +968,12 @@ def finish_http_request(
 
             task_emission_data = dataclasses.replace(emissions_at_stop)
             request_duration = time.perf_counter() - baseline.started_at
-            task_emission_data.duration = Time.from_seconds(request_duration).seconds
+            # compute_delta_emission subtracts previous.duration, so pass the
+            # absolute elapsed here; the delta is then the request duration and
+            # emissions_rate is computed against it (not against a negative).
+            task_emission_data.duration = (
+                baseline.duration_at_start + Time.from_seconds(request_duration).seconds
+            )
             task_emission_data.compute_delta_emission(previous)
 
             task.emissions_data = task_emission_data
@@ -996,6 +1001,22 @@ def persist_completed_task(self, task_name: str) -> None:
                 handler.task_out(task_payload, self._experiment_name)
                 task.uploaded_to_api = True
 
+    def discard_task(self, task_name: str) -> None:
+        """Drop a finished task record so ``_tasks`` stays bounded.
+
+        Long-lived servers create one task per HTTP request; without eviction
+        ``_tasks`` grows for the process lifetime. Call after
+        :meth:`persist_completed_task` (data already left the tracker by then).
+
+        Args:
+            task_name: Name of the task to forget. Unknown names are ignored.
+        """
+        with self._http_task_lock:
+            task = self._tasks.get(task_name)
+            if task is None or task.is_active:
+                return
+            del self._tasks[task_name]
+
     @suppress(Exception)
     def flush(self) -> Optional[float]:
         """

codecarbon/integrations/fastapi/__init__.py

@@ -11,6 +11,13 @@
         log_request_complete,
         shutdown_codecarbon_middleware,
     )
+    from codecarbon.integrations.fastapi.tiers import (
+        EndpointTotals,
+        MeasurementTier,
+        RequestMeasurement,
+        TierDetection,
+        detect_measurement_tier,
+    )
 except ImportError as exc:
     raise ImportError(
         "CodeCarbon FastAPI integration requires Starlette (installed with FastAPI). "
@@ -19,6 +26,11 @@
 
 __all__ = [
     "CodeCarbonMiddleware",
+    "EndpointTotals",
+    "MeasurementTier",
+    "RequestMeasurement",
+    "TierDetection",
+    "detect_measurement_tier",
     "add_codecarbon_middleware",
     "compose_lifespans",
     "create_codecarbon_lifespan",

codecarbon/integrations/fastapi/middleware.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import collections
+import dataclasses
 import threading
 from collections.abc import Awaitable, Callable, Iterable, Sequence
 from concurrent import futures
@@ -21,6 +22,13 @@
     build_endpoint_key,
     should_track_request,
 )
+from codecarbon.integrations.fastapi.tiers import (
+    EndpointTotals,
+    MeasurementTier,
+    RequestMeasurement,
+    TierDetection,
+    detect_measurement_tier,
+)
 from codecarbon.output_methods.emissions_data import EmissionsData
 
 DEFAULT_TRACKER_KWARGS: dict[str, Any] = {
@@ -65,21 +73,75 @@ def _resolve_header_fields(
 
 def _inject_emission_headers(
     message: Message,
-    emissions_data: EmissionsData | None,
+    measurement: RequestMeasurement | None,
     fields: Sequence[str],
 ) -> Message:
-    if not fields or emissions_data is None:
+    """Add ``X-CodeCarbon-*`` headers.
+
+    Headers are opt-in (``response_headers``); when enabled they always carry
+    ``X-CodeCarbon-Tier`` next to the numbers, and report ``unavailable``
+    rather than ``0`` when the backend cannot resolve the request.
+    """
+    if not fields or measurement is None:
         return message
     headers = list(message.get("headers", []))
+    headers.append((b"X-CodeCarbon-Tier", measurement.tier.value.encode("latin-1")))
+    emissions_data = measurement.emissions_data
     for field in fields:
-        if not hasattr(emissions_data, field):
-            continue
         name = _codecarbon_header_name(field)
-        value = str(getattr(emissions_data, field))
+        if emissions_data is None:
+            value = "unavailable"
+        elif hasattr(emissions_data, field):
+            value = str(getattr(emissions_data, field))
+        else:
+            continue
         headers.append((name.encode("latin-1"), value.encode("latin-1")))
     return {**message, "headers": headers}
 
 
+def _carbon_intensity(tracker: EmissionsTracker | None) -> float:
+    """Effective kg CO2 per kWh implied by the tracker's running totals."""
+    total_energy = getattr(getattr(tracker, "_total_energy", None), "kWh", 0.0)
+    total_emissions = getattr(tracker, "_total_emissions", 0.0)
+    try:
+        if total_energy and total_energy > 0:
+            return float(total_emissions) / float(total_energy)
+    except (TypeError, ValueError):
+        pass
+    return 0.0
+
+
+def _estimate_from_duration(
+    emissions_data: EmissionsData, intensity: float
+) -> EmissionsData | None:
+    """ESTIMATED tier: energy is analytic (``P x elapsed``) at any resolution.
+
+    The sampled delta can still be 0 when the request fell between scheduler
+    ticks; with a constant power model the honest figure is power x duration
+    rather than 0. Returns ``None`` when carbon intensity is not yet known,
+    so the caller reports unavailable instead of a zero.
+    """
+    duration = emissions_data.duration or 0.0
+    if duration <= 0 or intensity <= 0:
+        return None
+    hours = duration / 3600.0
+    cpu = (emissions_data.cpu_power or 0.0) / 1000.0 * hours
+    gpu = (emissions_data.gpu_power or 0.0) / 1000.0 * hours
+    ram = (emissions_data.ram_power or 0.0) / 1000.0 * hours
+    energy = cpu + gpu + ram
+    if energy <= 0:
+        return None
+    return dataclasses.replace(
+        emissions_data,
+        cpu_energy=cpu,
+        gpu_energy=gpu,
+        ram_energy=ram,
+        energy_consumed=energy,
+        emissions=energy * intensity,
+        emissions_rate=energy * intensity / duration,
+    )
+
+
 class _TrackerRunner:
     """Single tracker thread: request-path jobs first, then pending finalization."""
 
@@ -242,6 +304,90 @@ def __init__(
         self._app_tracker: EmissionsTracker | None = None
         self._tracker_init_lock = threading.Lock()
         self._tracker_runner = _TrackerRunner()
+        self._tier_detection: TierDetection | None = None
+        self._totals: dict[str, EndpointTotals] = {}
+        self._totals_lock = threading.Lock()
+
+    @property
+    def measurement_tier(self) -> MeasurementTier | None:
+        """Tier resolved from the tracker's hardware, or ``None`` before first use."""
+        return None if self._tier_detection is None else self._tier_detection.tier
+
+    @property
+    def tier_detection(self) -> TierDetection | None:
+        """Full detection result (overall tier plus per-component tiers)."""
+        return self._tier_detection
+
+    def endpoint_totals(self) -> dict[str, EndpointTotals]:
+        """Per-endpoint aggregates. The only energy output in AGGREGATE_ONLY.
+
+        Idle/baseline power is charged to requests, not subtracted, so these
+        totals sum to the tracker total.
+        """
+        with self._totals_lock:
+            return {
+                key: dataclasses.replace(value) for key, value in self._totals.items()
+            }
+
+    def _resolve_tier(self, tracker: EmissionsTracker) -> TierDetection:
+        """Resolve the tier once, from the hardware the tracker actually detected."""
+        detection = self._tier_detection
+        if detection is None:
+            detection = detect_measurement_tier(getattr(tracker, "_hardware", None))
+            self._tier_detection = detection
+            logger.debug("CodeCarbon measurement tier: %s", detection.describe())
+        return detection
+
+    def _build_measurement(
+        self,
+        detection: TierDetection,
+        endpoint: str,
+        task_name: str,
+        emissions_data: EmissionsData | None,
+        tracker: EmissionsTracker | None = None,
+    ) -> RequestMeasurement:
+        tier = detection.tier
+        duration = float(getattr(emissions_data, "duration", 0.0) or 0.0)
+        if emissions_data is None:
+            return RequestMeasurement(
+                tier, task_name, endpoint, duration, None, "no measurement returned"
+            )
+        if tier is MeasurementTier.AGGREGATE_ONLY:
+            return RequestMeasurement(
+                tier,
+                task_name,
+                endpoint,
+                duration,
+                None,
+                "backend resolution is coarser than a request; use endpoint_totals()",
+            )
+        if tier is MeasurementTier.ESTIMATED and not emissions_data.energy_consumed:
+            emissions_data = _estimate_from_duration(
+                emissions_data, _carbon_intensity(tracker)
+            )
+        if emissions_data is None or not emissions_data.energy_consumed:
+            return RequestMeasurement(
+                tier,
+                task_name,
+                endpoint,
+                duration,
+                None,
+                "request spanned no completed sampling window",
+            )
+        return RequestMeasurement(tier, task_name, endpoint, duration, emissions_data)
+
+    def _record_totals(
+        self,
+        detection: TierDetection,
+        endpoint: str,
+        emissions_data: EmissionsData | None,
+    ) -> None:
+        with self._totals_lock:
+            totals = self._totals.get(endpoint)
+            if totals is None:
+                totals = EndpointTotals(endpoint=endpoint, tier=detection.tier)
+                self._totals[endpoint] = totals
+            totals.add(emissions_data)
 
     def shutdown_tracker_executor(self, *, wait: bool = True) -> None:
         """Shut down the tracker background thread (idempotent).
@@ -328,7 +474,8 @@ def _finalize_on_worker(
         response: Response,
         run_callback: bool,
         baseline: HttpRequestBaseline | None,
-    ) -> EmissionsData | None:
+    ) -> RequestMeasurement:
+        detection = self._resolve_tier(tracker)
         if baseline is not None:
             emissions_data = tracker.finish_http_request(baseline)
             resolved_task = baseline.task_name
@@ -337,9 +484,24 @@ def _finalize_on_worker(
             resolved_task = active_task if isinstance(active_task, str) else task_name
             emissions_data = tracker.stop_task(resolved_task)
         tracker.persist_completed_task(resolved_task)
+        # Bound tracker._tasks: the record is persisted, keeping it would leak
+        # one Task per request for the lifetime of the server.
+        discard = getattr(tracker, "discard_task", None)
+        if callable(discard):
+            discard(resolved_task)
+        self._record_totals(detection, task_name, emissions_data)
+        measurement = self._build_measurement(
+            detection, task_name, resolved_task, emissions_data, tracker
+        )
+        try:
+            request.state.codecarbon = measurement
+        except Exception:  # pragma: no cover - exotic scopes without state
+            logger.debug("CodeCarbon: could not attach measurement to request.state")
         if run_callback:
-            self._run_request_complete(request, response, emissions_data, resolved_task)
-        return emissions_data
+            self._run_request_complete(
+                request, response, measurement.emissions_data, resolved_task
+            )
+        return measurement
 
     def _run_request_complete(
         self,
@@ -370,7 +532,7 @@ async def _finalize_after_response(
         baseline: HttpRequestBaseline | None,
         *,
         run_callback: bool,
-    ) -> EmissionsData | None:
+    ) -> RequestMeasurement:
         return await self._run_finalize_tracker(
             self._finalize_on_worker,
             tracker,
@@ -513,7 +675,7 @@ async def send_wrapper(message: Message) -> None:
                 return
             status_code = message["status"]
             response = Response(status_code=status_code)
-            emissions_data = await self._finalize_after_response(
+            measurement = await self._finalize_after_response(
                 tracker,
                 task_name,
                 request,
@@ -523,7 +685,7 @@ async def send_wrapper(message: Message) -> None:
             )
             finalized = True
             await send(
-                _inject_emission_headers(message, emissions_data, self.header_fields)
+                _inject_emission_headers(message, measurement, self.header_fields)
             )
 
         error: BaseException | None = None