From e7c00947d2a73cfff269f9b258103816273114c2 Mon Sep 17 00:00:00 2001
From: patrykorwat <patrykorwat@gmail.com>
Date: Thu, 3 Sep 2026 13:30:53 +0000
Subject: [PATCH] axelera: report AIPU utilization as the detector run duty
 cycle

The runtime exposes no device-side usage counter (mvm_utilisation is a
configuration limit, not a readout), so the NPU card rendered empty.
ModelInstance.run blocks until the AIPU completes: the detector process
now times each run and publishes busy/wall over a 2s window through a
small file under the tmp cache dir; the stats poller reads it and sets
the npu key. A stale file (detector idle/restarting) reports nothing
rather than a stale number.
---
 frigate/detectors/plugins/axelera.py | 39 ++++++++++++++++++++++++++++
 frigate/stats/hardware.py            | 33 ++++++++++++++++++++---
 2 files changed, 68 insertions(+), 4 deletions(-)

diff --git a/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py
index 92aaf00a8b..3af13d383d 100644
--- a/frigate/detectors/plugins/axelera.py
+++ b/frigate/detectors/plugins/axelera.py
@@ -36,7 +36,9 @@ import logging
 import os
 import queue
 import re
+import tempfile
 import threading
+import time
 import zipfile
 from typing import ClassVar, Literal
 
@@ -54,6 +56,13 @@ logger = logging.getLogger(__name__)
 
 DETECTOR_KEY = "axelera"
 
+# windowed duty cycle of the AIPU, written by the detector process and read
+# by the main process's hardware poller (same tmpfs cache dir as the rest of
+# the runtime's scratch state); "<pct> <monotonic-epoch>" text
+AXELERA_USAGE_PATH = os.path.join(
+    os.environ.get("FRIGATE_TMP", tempfile.gettempdir()), "frigate-axelera-usage"
+)
+
 # The Axelera runtime SDK is installed at first start rather than shipped in
 # the image. The wheels come from Axelera's public Artifactory PyPI index
 # (readable without credentials) and are pinned by sha256. The runtime wheel
@@ -407,6 +416,8 @@ class _AxeleraRuntimeInference:
         self._stop = threading.Event()
         self._producer_thread: threading.Thread | None = None
         self._consumer_thread: threading.Thread | None = None
+        self._usage_busy = 0.0
+        self._usage_window_start: float | None = None
 
         if runtime_factory is None:
             try:
@@ -629,13 +640,41 @@ class _AxeleraRuntimeInference:
                 continue
             try:
                 self._preprocess(frame, in_buf)
+                run_t0 = time.monotonic()
                 self._instance.run([in_buf], out_bufs)
+                self._note_run(run_t0, time.monotonic())
                 self._decode_q.put_nowait((connection_id, in_buf, out_bufs))
             except Exception:
                 logger.exception("axelera: inference failed for %s", connection_id)
                 self._pool.put((in_buf, out_bufs))
                 self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32))
 
+    def _note_run(self, t0: float, t1: float) -> None:
+        """Accumulate card busy time and publish a windowed duty cycle.
+
+        ModelInstance.run blocks until the AIPU completes, so busy/wall over a
+        window is the card's true utilization; the runtime exposes no usage
+        counter to read instead. The main stats process reads the file.
+        """
+        now = time.monotonic()
+        self._usage_busy += t1 - t0
+        if self._usage_window_start is None:
+            self._usage_window_start = t0
+        if now - self._usage_window_start < 2.0:
+            return
+        window = now - self._usage_window_start
+        if window > 0:
+            pct = min(100.0, round(100.0 * self._usage_busy / window, 1))
+            try:
+                tmp = AXELERA_USAGE_PATH + ".tmp"
+                with open(tmp, "w") as f:
+                    f.write(f"{pct} {time.time():.3f}")
+                os.replace(tmp, AXELERA_USAGE_PATH)
+            except OSError:
+                pass
+        self._usage_window_start = now
+        self._usage_busy = 0.0
+
     def _consumer(self) -> None:
         """Decode the raw int8 heads (dequant + onnx + NMS) to (20,6) rows."""
         while not self._stop.is_set():
diff --git a/frigate/stats/hardware.py b/frigate/stats/hardware.py
index ac5fdef043..343c3f1789 100644
--- a/frigate/stats/hardware.py
+++ b/frigate/stats/hardware.py
@@ -196,16 +196,41 @@ def _poll_axengine(config: FrigateConfig) -> HardwarePollResult:
     return HardwarePollResult(npu={"axengine": npu_usage})
 
 
+def _axelera_board_utilization() -> float | None:
+    """Duty-cycle of the AIPU published by the detector process.
+
+    The detector times each blocking ModelInstance.run and writes
+    busy/wall for the last window to a small file; a stale file means the
+    detector has not run inference recently and no number is reported.
+    """
+    try:
+        from frigate.detectors.plugins.axelera import AXELERA_USAGE_PATH
+
+        with open(AXELERA_USAGE_PATH) as f:
+            pct, stamp = f.read().split()
+        if time.time() - float(stamp) > 10:
+            return None
+        return max(0.0, float(pct))
+    except (OSError, ValueError, ImportError):
+        return None
+
+
 def _poll_axelera(config: FrigateConfig) -> HardwarePollResult:
-    # the Metis card exposes no utilization counter without a broker that
-    # cannot run alongside a live detector; the board controller temperature
-    # is the one honest readout
+    # the runtime exposes no device-side usage counter; the board
+    # controller temperature and the detector process's run-time duty cycle
+    # are the two honest readouts
     board_temp = get_axelera_board_temp()
 
     if board_temp is None:
         return HardwarePollResult(ok=False)
 
-    return HardwarePollResult(npu={"axelera": {"mem": "-%", "temp": board_temp}})
+    entry: dict[str, Any] = {"mem": "-%", "temp": board_temp}
+    util = _axelera_board_utilization()
+
+    if util is not None:
+        entry["npu"] = util
+
+    return HardwarePollResult(npu={"axelera": entry})
 
 
 POLLERS: dict[str, Callable[[FrigateConfig], HardwarePollResult]] = {