fix(WebServer): fix threshold alarm function not working properly

Author: FASTSHIFT

Tools/WebServer/monitor.py

@@ -252,15 +252,22 @@ def cleanup_audio_meter(device):
             logger.debug(f"Audio recorder cleanup: {e}")
 
 
-def check_threshold_alarm(device, value, current_mode):
-    """Check if value exceeds threshold and trigger alarm."""
+def check_threshold_alarm(device):
+    """Check if threshold is exceeded and trigger alarm.
+
+    The threshold check is independent of the current monitor mode.
+    It fetches the value based on device.threshold_mode separately.
+    """
     if not device.threshold_enable:
         return
 
-    if current_mode != device.threshold_mode:
+    threshold_mode = device.threshold_mode
+    if not threshold_mode or threshold_mode == "none":
         return
 
-    if value is None:
+    # Get value based on threshold_mode (independent of monitor mode)
+    value, error = get_monitor_value(threshold_mode)
+    if error or value is None:
         return
 
     now = time.time()
@@ -271,7 +278,7 @@ def check_threshold_alarm(device, value, current_mode):
             serial_write_direct(device, cmd)
         logger = logging.getLogger(__name__)
         logger.info(
-            f"Threshold alarm triggered: {value:.1f}% > {device.threshold_value}%"
+            f"Threshold alarm triggered: {threshold_mode}={value:.1f}% > {device.threshold_value}%"
         )
 
 
@@ -404,9 +411,8 @@ def monitor_tick():
         # 更新 legacy last_percent (用于兼容)
         device.last_percent = percent_0 if percent_0 is not None else (percent_1 or 0)
 
-        # 阈值报警检查 (使用 CH0 的值)
-        if percent_0 is not None:
-            check_threshold_alarm(device, percent_0, mode_0)
+        # 阈值报警检查 (独立于监控模式)
+        check_threshold_alarm(device)
 
     return monitor_tick
 

Tools/WebServer/tests/test_monitor.py

@@ -8,7 +8,7 @@
 """
 
 import time
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 
 class TestGetMonitorValue:
@@ -280,49 +280,56 @@ def test_threshold_disabled(self):
 
         device = DeviceState("test", "Test")
         device.threshold_enable = False
+        device.threshold_mode = "cpu-usage"
         # Should not raise any exception
-        check_threshold_alarm(device, 100, "cpu-usage")
+        check_threshold_alarm(device)
 
-    def test_threshold_mode_mismatch(self):
-        """Test when mode doesn't match threshold mode."""
+    def test_threshold_mode_none(self):
+        """Test when threshold mode is none."""
         from monitor import check_threshold_alarm
         from state import DeviceState
 
         device = DeviceState("test", "Test")
         device.threshold_enable = True
-        device.threshold_mode = "cpu-usage"
-        # Should not trigger alarm for different mode
-        check_threshold_alarm(device, 100, "mem-usage")
+        device.threshold_mode = "none"
+        # Should not trigger alarm
+        check_threshold_alarm(device)
 
-    def test_threshold_value_none(self):
-        """Test when value is None."""
+    @patch("monitor.get_monitor_value")
+    def test_threshold_value_none(self, mock_get_value):
+        """Test when get_monitor_value returns None."""
         from monitor import check_threshold_alarm
         from state import DeviceState
 
+        mock_get_value.return_value = (None, "Error")
         device = DeviceState("test", "Test")
         device.threshold_enable = True
         device.threshold_mode = "cpu-usage"
         # Should not raise any exception
-        check_threshold_alarm(device, None, "cpu-usage")
+        check_threshold_alarm(device)
 
-    def test_threshold_below_value(self):
+    @patch("monitor.get_monitor_value")
+    def test_threshold_below_value(self, mock_get_value):
         """Test when value is below threshold."""
         from monitor import check_threshold_alarm
         from state import DeviceState
 
+        mock_get_value.return_value = (50, None)
         device = DeviceState("test", "Test")
         device.threshold_enable = True
         device.threshold_mode = "cpu-usage"
         device.threshold_value = 80
         device.last_alarm_time = 0
         # Should not trigger alarm
-        check_threshold_alarm(device, 50, "cpu-usage")
+        check_threshold_alarm(device)
 
-    def test_threshold_triggered(self):
+    @patch("monitor.get_monitor_value")
+    def test_threshold_triggered(self, mock_get_value):
         """Test when threshold is exceeded."""
         from monitor import check_threshold_alarm
         from state import DeviceState
 
+        mock_get_value.return_value = (90, None)
         device = DeviceState("test", "Test")
         device.threshold_enable = True
         device.threshold_mode = "cpu-usage"
@@ -332,15 +339,17 @@ def test_threshold_triggered(self):
         device.last_alarm_time = 0
         device.ser = None  # No serial, but should still update last_alarm_time
 
-        check_threshold_alarm(device, 90, "cpu-usage")
+        check_threshold_alarm(device)
         # last_alarm_time should be updated
         assert device.last_alarm_time > 0
 
-    def test_threshold_cooldown(self):
+    @patch("monitor.get_monitor_value")
+    def test_threshold_cooldown(self, mock_get_value):
         """Test threshold cooldown period."""
         from monitor import check_threshold_alarm
         from state import DeviceState
 
+        mock_get_value.return_value = (90, None)
         device = DeviceState("test", "Test")
         device.threshold_enable = True
         device.threshold_mode = "cpu-usage"
@@ -351,15 +360,17 @@ def test_threshold_cooldown(self):
         device.ser = None
 
         old_time = device.last_alarm_time
-        check_threshold_alarm(device, 90, "cpu-usage")
+        check_threshold_alarm(device)
         # Should not update due to cooldown
         assert device.last_alarm_time == old_time
 
-    def test_threshold_with_serial(self):
+    @patch("monitor.get_monitor_value")
+    def test_threshold_with_serial(self, mock_get_value):
         """Test threshold alarm with serial port."""
         from monitor import check_threshold_alarm
         from state import DeviceState
 
+        mock_get_value.return_value = (90, None)
         device = DeviceState("test", "Test")
         device.threshold_enable = True
         device.threshold_mode = "cpu-usage"
@@ -370,7 +381,7 @@ def test_threshold_with_serial(self):
         device.ser = MagicMock()
         device.ser.isOpen.return_value = True
 
-        check_threshold_alarm(device, 90, "cpu-usage")
+        check_threshold_alarm(device)
         assert device.last_alarm_time > 0