nik/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-10-18 07:29:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

axelera: publish the run duty cycle through a shared Value

Browse Source 
Replace the tmp-file duty-cycle handoff (and the FRIGATE_TMP env var
nothing else reads) with a third multiprocessing.Value alongside
avg_inference_speed and detection_start: ObjectDetectProcess creates it,
AsyncDetectorRunner passes it to the detector, and the plugin publishes
busy/wall of the blocking ModelInstance.run over a 2s window, mirroring
how memryx receives its shared state. No filesystem path, no staleness
window, no atomic rename, and stats/hardware.py no longer imports the
plugin module to find the file.

The producer also ticks the window while the queue is empty, so the duty
cycle falls to 0 when the card idles instead of sticking at its last
value.

Name it what it is: host-side occupancy of the run call, including DMA
and driver time, not an AIPU core counter (the runtime exposes no
device-side one). The NPU entry now reads temperature from the board
controller plus the detector-reported run duty cycle folded in where
the detector processes are in reach.
This commit is contained in:
patrykorwat 2026-09-05 05:44:29 +00:00
parent fec2a1c73a
commit f50a2dc1ed
4 changed files with 76 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
frigate/detectors/plugins/axelera.py
View File

@ -36,7 +36,6 @@ import logging
import os
import queue
import re
import tempfile
import threading
import time
import zipfile
@ -56,12 +55,9 @@ 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"
)
# window length over which the producer thread averages ModelInstance.run
# time before publishing the run duty cycle to the shared stats Value
USAGE_WINDOW_SECONDS = 2.0
# 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
@ -417,7 +413,10 @@ class _AxeleraRuntimeInference:
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
self._usage_window_start = time.monotonic()
# multiprocessing.Value("d") handed in via set_duty_value; None means
# nobody reads the duty cycle (tests, standalone use)
self._duty_value = None
if runtime_factory is None:
try:
@ -653,6 +652,8 @@ class _AxeleraRuntimeInference:
try:
connection_id, frame = self._frame_q.get(timeout=0.25)
except queue.Empty:
# keep the duty cycle window turning while the card idles
self._publish_duty(time.monotonic())
continue
try:
in_buf, out_bufs = self._pool.get(timeout=1.0)
@ -672,28 +673,25 @@ class _AxeleraRuntimeInference:
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.
"""Accumulate run 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.
ModelInstance.run blocks until the AIPU completes, so busy/wall over
a window is the detector thread's run duty cycle: host-side occupancy
of the blocking call (including DMA and driver time), which the
runtime exposes no device-side counter to read instead. The value
rides a multiprocessing.Value to the main process's stats poller.
"""
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:
self._publish_duty(time.monotonic())
def _publish_duty(self, now: float) -> None:
if now - self._usage_window_start < USAGE_WINDOW_SECONDS:
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
if window > 0 and self._duty_value is not None:
self._duty_value.value = min(
100.0, round(100.0 * self._usage_busy / window, 1)
)
self._usage_window_start = now
self._usage_busy = 0.0
@ -951,6 +949,10 @@ class AxeleraDetector(DetectionApi):
"""Return (connection_id, rows) for the next completed inference."""
return self._inference.collect()
def set_duty_value(self, duty_value) -> None:
"""Hand the shared stats Value to the producer thread's duty meter."""
self._inference._duty_value = duty_value
def shutdown(self) -> None:
"""Gracefully shut down the runtime and release the Metis device."""
self._inference.stop()

15
frigate/object_detection/base.py
View File

@ -45,6 +45,7 @@ class BaseLocalDetector(ObjectDetector):
detector_config: BaseDetectorConfig | None = None,
labels: str | None = None,
stop_event: MpEvent | None = None,
duty_value: Any = None,
) -> None:
self.fps = EventsPerSecond()
if labels is None:
@ -65,6 +66,10 @@ class BaseLocalDetector(ObjectDetector):
if hasattr(self.detect_api, "set_stop_event") and stop_event:
self.detect_api.set_stop_event(stop_event)
# If the detector publishes a run duty cycle, hand it the shared Value
if hasattr(self.detect_api, "set_duty_value") and duty_value is not None:
self.detect_api.set_duty_value(duty_value)
def _transform_input(self, tensor_input: np.ndarray) -> np.ndarray:
if self.input_transform:
tensor_input = np.transpose(tensor_input, self.input_transform)
@ -196,6 +201,7 @@ class AsyncDetectorRunner(FrigateProcess):
config: FrigateConfig,
detector_config: BaseDetectorConfig,
stop_event: MpEvent,
duty_value: Any = None,
) -> None:
super().__init__(stop_event, PROCESS_PRIORITY_HIGH, name=name, daemon=True)
self.detection_queue = detection_queue
@ -204,6 +210,7 @@ class AsyncDetectorRunner(FrigateProcess):
self.start_time = start_time
self.config = config
self.detector_config = detector_config
self.duty_value = duty_value
self.outputs: dict[str, Any] = {}
self._frame_manager: SharedMemoryFrameManager | None = None
self._publisher: ObjectDetectorPublisher | None = None
@ -282,7 +289,9 @@ class AsyncDetectorRunner(FrigateProcess):
self._frame_manager = SharedMemoryFrameManager()
self._publisher = ObjectDetectorPublisher()
self._detector = AsyncLocalObjectDetector(
detector_config=self.detector_config, stop_event=self.stop_event
detector_config=self.detector_config,
stop_event=self.stop_event,
duty_value=self.duty_value,
)
for name in self.cameras:
@ -330,6 +339,9 @@ class ObjectDetectProcess:
self.detection_queue = detection_queue
self.avg_inference_speed = Value("d", 0.01)
self.detection_start = Value("d", 0.0)
# async detectors that time their accelerator runs publish a windowed
# duty cycle here (see AsyncDetectorRunner); read by the stats poller
self.run_duty_cycle = Value("d", -1.0)
self.detect_process: FrigateProcess | None = None
self.config = config
self.detector_config = detector_config
@ -382,6 +394,7 @@ class ObjectDetectProcess:
self.config,
self.detector_config,
self.stop_event,
duty_value=self.run_duty_cycle,
)
else:
self.detect_process = DetectorRunner(

29
frigate/stats/hardware.py
View File

@ -196,39 +196,16 @@ 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 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
# the runtime exposes no device-side usage counter; the board controller
# temperature is the only honest device readout. The run duty cycle is a
# detector-process stat, added to npu_usages in stats_snapshot
board_temp = get_axelera_board_temp()
if board_temp is None:
return HardwarePollResult(ok=False)
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})

32
frigate/stats/util.py
View File

@ -64,6 +64,30 @@ def stats_init(
return stats_tracking
def _axelera_run_duty_cycle(
detectors: dict[str, ObjectDetectProcess],
) -> float | None:
"""Highest run duty cycle published by the configured axelera detectors.
The detector process times each blocking ModelInstance.run and publishes
busy/wall for the last window through a shared Value (a negative value
means the detector has not produced a window yet and nothing is shown).
This is host-side occupancy of the run call, including DMA and driver
time, not an AIPU core counter: the runtime exposes no device-side one.
"""
cycles = [
detector.run_duty_cycle.value
for detector in detectors.values()
if detector.detector_config.type == "axelera"
and detector.run_duty_cycle.value >= 0.0
]
if not cycles:
return None
return round(max(cycles), 1)
def get_detector_stats(
stats_tracking: StatsTrackingTypes,
) -> dict[str, dict[str, Any]]:
@ -247,6 +271,14 @@ def stats_snapshot(
hardware_stats.update_stats(stats)
# the axelera run duty cycle is a detector-process stat (a shared Value,
# not a device counter); fold it into the polled NPU entry if one exists
axelera_npu = stats.get("npu_usages", {}).get("axelera")
if axelera_npu is not None:
duty = _axelera_run_duty_cycle(stats_tracking["detectors"])
if duty is not None:
axelera_npu["npu"] = duty
if config.telemetry.stats.network_bandwidth:
bandwidth_stats = get_bandwidth_stats(config)