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.
This commit is contained in:
patrykorwat 2026-09-03 13:30:53 +00:00
parent 5cd1350067
commit e7c00947d2
2 changed files with 68 additions and 4 deletions

View File

@ -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():

View File

@ -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]] = {