Add npu usages as a statistic

frigate/stats/util.py

@@ -24,6 +24,7 @@ from frigate.util.services import (
     get_intel_gpu_stats,
     get_jetson_stats,
     get_nvidia_gpu_stats,
+    get_rockchip_npu_stats,
     is_vaapi_amd_driver,
 )
 from frigate.version import VERSION
@@ -109,6 +110,7 @@ def get_processing_stats(
         stats_tasks = [
             asyncio.create_task(set_gpu_stats(config, stats, hwaccel_errors)),
             asyncio.create_task(set_cpu_stats(stats)),
+            asyncio.create_task(set_npu_usages(config, stats)),
         ]
 
         if config.telemetry.stats.network_bandwidth:
@@ -238,6 +240,21 @@ async def set_gpu_stats(
         all_stats["gpu_usages"] = stats
 
 
+async def set_npu_usages(
+    config: FrigateConfig, all_stats: dict[str, Any]
+) -> dict[str, Any]:
+    stats: dict[str, dict] = {}
+
+    for detector in config.detectors.values():
+        if detector.type == "rknn":
+            # Rockchip NPU usage
+            rk_usage = get_rockchip_npu_stats()
+            stats["rockchip"] = rk_usage
+
+    if stats:
+        all_stats["npu_usages"] = stats
+
+
 def stats_snapshot(
     config: FrigateConfig, stats_tracking: StatsTrackingTypes, hwaccel_errors: list[str]
 ) -> dict[str, Any]:

frigate/util/services.py

@@ -3,6 +3,7 @@
 import asyncio
 import json
 import logging
+import math
 import os
 import re
 import signal
@@ -382,6 +383,23 @@ def get_intel_gpu_stats(sriov: bool) -> dict[str, str]:
         return results
 
 
+def get_rockchip_npu_stats() -> dict[str, str]:
+    """Get stats using rk."""
+    try:
+        with open("/sys/kernel/debug/rknpu/load", "r") as f:
+            npu_output = f.read()
+            core_loads = re.findall(r"Core\d+:\s*(\d+)%", npu_output)
+    except FileNotFoundError:
+        core_loads = None
+
+    if not core_loads:
+        return None
+
+    percentages = [int(load) for load in core_loads]
+    mean = round(sum(percentages) / len(percentages), 2)
+    return {"npu": mean, "mem": "-"}
+
+
 def try_get_info(f, h, default="N/A"):
     try:
         if h: