mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-03-28 19:18:22 +03:00
feat: add axera npu load (#22662)
This commit is contained in:
parent
03d0139497
commit
06ad72860c
@ -19,6 +19,7 @@ from frigate.types import StatsTrackingTypes
|
||||
from frigate.util.services import (
|
||||
calculate_shm_requirements,
|
||||
get_amd_gpu_stats,
|
||||
get_axcl_npu_stats,
|
||||
get_bandwidth_stats,
|
||||
get_cpu_stats,
|
||||
get_fs_type,
|
||||
@ -324,6 +325,10 @@ async def set_npu_usages(config: FrigateConfig, all_stats: dict[str, Any]) -> No
|
||||
# OpenVINO NPU usage
|
||||
ov_usage = get_openvino_npu_stats()
|
||||
stats["openvino"] = ov_usage
|
||||
elif detector.type == "axengine":
|
||||
# AXERA NPU usage
|
||||
axcl_usage = get_axcl_npu_stats()
|
||||
stats["axengine"] = axcl_usage
|
||||
|
||||
if stats:
|
||||
all_stats["npu_usages"] = stats
|
||||
|
||||
@ -488,6 +488,43 @@ def get_rockchip_npu_stats() -> Optional[dict[str, float | str]]:
|
||||
return stats
|
||||
|
||||
|
||||
def get_axcl_npu_stats() -> Optional[dict[str, str | float]]:
|
||||
"""Get NPU stats using axcl."""
|
||||
# Check if axcl-smi exists
|
||||
axcl_smi_path = "/usr/bin/axcl/axcl-smi"
|
||||
if not os.path.exists(axcl_smi_path):
|
||||
return None
|
||||
|
||||
try:
|
||||
# Run axcl-smi command to get NPU stats
|
||||
axcl_command = [axcl_smi_path, "sh", "cat", "/proc/ax_proc/npu/top"]
|
||||
p = sp.run(
|
||||
axcl_command,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
if p.returncode != 0:
|
||||
pass
|
||||
else:
|
||||
utilization = None
|
||||
|
||||
for line in p.stdout.strip().splitlines():
|
||||
line = line.strip()
|
||||
if line.startswith("utilization:"):
|
||||
match = re.search(r"utilization:(\d+)%", line)
|
||||
if match:
|
||||
utilization = float(match.group(1))
|
||||
|
||||
if utilization is not None:
|
||||
stats: dict[str, str | float] = {"npu": utilization, "mem": "-%"}
|
||||
return stats
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def try_get_info(f, h, default="N/A", sensor=None):
|
||||
try:
|
||||
if h:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user