From f15f09d54297e27d2f54983f56463166f048a1ff Mon Sep 17 00:00:00 2001
From: Nicolas Mowen <nickmowen213@gmail.com>
Date: Sat, 19 Apr 2025 08:02:04 -0600
Subject: [PATCH] Add npu usages as a statistic

---
 frigate/stats/util.py    | 17 +++++++++++++++++
 frigate/util/services.py | 18 ++++++++++++++++++
 2 files changed, 35 insertions(+)

diff --git a/frigate/stats/util.py b/frigate/stats/util.py
index 7bdf92bf26..9fae1178a9 100644
--- a/frigate/stats/util.py
+++ b/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]:
diff --git a/frigate/util/services.py b/frigate/util/services.py
index ce7041c268..4249522589 100644
--- a/frigate/util/services.py
+++ b/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: