mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-18 07:29:45 +00:00
stats: Axelera Metis board temperature poller
The card exposes no utilization counter usable next to a live detector, so the poller reports the board-controller temperature read via axcmd (cached 5s; the read must not race detector init).
This commit is contained in:
parent
1b38f6952c
commit
df95482655
@ -21,6 +21,7 @@ from frigate.detectors.hardware import DEV_ROOT, hardware_prober
|
||||
from frigate.util.services import (
|
||||
get_amd_gpu_stats,
|
||||
get_axcl_npu_stats,
|
||||
get_axelera_board_temp,
|
||||
get_cpu_stats,
|
||||
get_hailo_temps,
|
||||
get_intel_gpu_stats,
|
||||
@ -195,6 +196,18 @@ def _poll_axengine(config: FrigateConfig) -> HardwarePollResult:
|
||||
return HardwarePollResult(npu={"axengine": npu_usage})
|
||||
|
||||
|
||||
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
|
||||
board_temp = get_axelera_board_temp()
|
||||
|
||||
if board_temp is None:
|
||||
return HardwarePollResult(ok=False)
|
||||
|
||||
return HardwarePollResult(npu={"axelera": {"mem": "-%", "temp": board_temp}})
|
||||
|
||||
|
||||
POLLERS: dict[str, Callable[[FrigateConfig], HardwarePollResult]] = {
|
||||
"nvidia": _poll_nvidia,
|
||||
"jetson": _poll_jetson,
|
||||
@ -204,6 +217,7 @@ POLLERS: dict[str, Callable[[FrigateConfig], HardwarePollResult]] = {
|
||||
"rpi": _poll_rpi,
|
||||
"intel_npu": _poll_intel_npu,
|
||||
"axengine": _poll_axengine,
|
||||
"axelera": _poll_axelera,
|
||||
}
|
||||
|
||||
|
||||
@ -379,6 +393,8 @@ class HardwareStats:
|
||||
names.add("rockchip")
|
||||
elif spec.detector == "axengine":
|
||||
names.add("axengine")
|
||||
elif spec.detector == "axelera":
|
||||
names.add("axelera")
|
||||
elif spec.detector == "tensorrt":
|
||||
names.add("jetson")
|
||||
elif spec.detector == "openvino":
|
||||
@ -468,5 +484,11 @@ def get_hardware_temperatures(detector_type: str) -> list[float | None]:
|
||||
elif detector_type == "hailo8l":
|
||||
hailo_temps = get_hailo_temps()
|
||||
return [hailo_temps[name] for name in sorted(hailo_temps.keys())]
|
||||
elif detector_type == "axelera":
|
||||
# Metis reports one board temperature through the SDK's axcmd tool
|
||||
board_temp = get_axelera_board_temp()
|
||||
|
||||
if board_temp is not None:
|
||||
return [board_temp]
|
||||
|
||||
return []
|
||||
|
||||
@ -831,6 +831,70 @@ def get_axcl_npu_stats() -> dict[str, str | float] | None:
|
||||
return None
|
||||
|
||||
|
||||
_axelera_board_temp_cache: tuple[float, float] | None = None
|
||||
_AXELERA_BOARD_TEMP_TTL = 5.0
|
||||
|
||||
|
||||
def get_axelera_board_temp() -> float | None:
|
||||
"""Get the Axelera Metis board temperature through the SDK's axcmd tool.
|
||||
|
||||
The read goes through the board controller and does not open an inference
|
||||
stream, so it is safe to call while a detector is running. ``axcmd`` ships
|
||||
with the runtime wheel; ``find_tool`` prefers the runtime-installed copy
|
||||
under the user base. Results are cached for a few seconds so one stats tick
|
||||
spawns axcmd at most once.
|
||||
|
||||
NOTE: this function is blocking (up to ~5s on a hung board).
|
||||
"""
|
||||
global _axelera_board_temp_cache
|
||||
|
||||
now = time.monotonic()
|
||||
if (
|
||||
_axelera_board_temp_cache is not None
|
||||
and now - _axelera_board_temp_cache[0] < _AXELERA_BOARD_TEMP_TTL
|
||||
):
|
||||
return _axelera_board_temp_cache[1]
|
||||
|
||||
try:
|
||||
from frigate.util.runtime_deps import find_tool
|
||||
|
||||
axcmd_path = find_tool("axcmd")
|
||||
except Exception: # noqa: BLE001
|
||||
axcmd_path = shutil.which("axcmd")
|
||||
|
||||
if axcmd_path is None or not os.path.exists(axcmd_path):
|
||||
return None
|
||||
|
||||
try:
|
||||
result = sp.run(
|
||||
[axcmd_path, "--board-temp"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, sp.TimeoutExpired):
|
||||
return None
|
||||
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
|
||||
# anchor on the board-controller line ("N: temp_sensorX@YY = <value>") so
|
||||
# unrelated "= <number>" output (firmware/version preambles) is not picked up
|
||||
match = re.search(r"temp_sensor[^=\n]*=\s*([\d.]+)", result.stdout)
|
||||
|
||||
if match is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
temp = round(float(match.group(1)), 1)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
_axelera_board_temp_cache = (now, temp)
|
||||
return temp
|
||||
|
||||
|
||||
def try_get_info(f, h, default="N/A", sensor=None):
|
||||
try:
|
||||
if h:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user