Merge pull request #1 from NeleBiH/codex/locate-and-fix-critical-code-error

Handle text status output from liquidctl

Author: NeleBiH

LiquidctlGUI.py

@@ -701,11 +701,14 @@ def update_status(self):
             return
         try:
             cmd = ["liquidctl", "-m", self.selected_device["description"], "status", "--json"]
-            result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=5)
+            result = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
             try:
+                result.check_returncode()
                 data = json.loads(result.stdout)
                 self.parse_json_status(data)
-            except Exception:
+            except (json.JSONDecodeError, subprocess.CalledProcessError):
+                cmd = ["liquidctl", "-m", self.selected_device["description"], "status"]
+                result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=5)
                 self.parse_text_status(result.stdout)
         except Exception as e:
             self.show_status_message(f"Error updating status: {e}")
@@ -726,8 +729,8 @@ def parse_json_status(self, status_data):
         for item in device_status:
             key = item.get("key", "").lower()
             value = item.get("value", 0)
-            if "fan speed" in key:
-                m = re.search(r"fan speed (\d+)", key)
+            if "fan" in key and "speed" in key:
+                m = re.search(r"fan (\d+) speed", key)
                 if m:
                     fan_num = m.group(1)
                     rpm = int(value) if isinstance(value, (int, float)) else 0
@@ -742,7 +745,28 @@ def parse_json_status(self, status_data):
         self.update_ui_with_status(fan_speeds, pump_speed, water_temp)
 
     def parse_text_status(self, output):
-        pass
+        fan_speeds = {}
+        pump_speed = None
+        water_temp = None
+        for line in output.splitlines():
+            lower = line.strip().lower()
+            m = re.search(r"fan (\d+) speed[:\s]+(\d+)", lower)
+            if m:
+                fan_num = m.group(1)
+                rpm = int(m.group(2))
+                percent = self.rpm_to_percent(rpm)
+                fan_speeds[f"Fan {fan_num}"] = (percent, rpm)
+                continue
+            m = re.search(r"pump speed[:\s]+(\d+)", lower)
+            if m:
+                rpm = int(m.group(1))
+                percent = self.rpm_to_percent(rpm, is_pump=True)
+                pump_speed = (percent, rpm)
+                continue
+            m = re.search(r"(liquid|water) temperature[:\s]+([\d\.]+)", lower)
+            if m:
+                water_temp = float(m.group(2))
+        self.update_ui_with_status(fan_speeds, pump_speed, water_temp)
 
     def update_ui_with_status(self, fan_speeds, pump_speed, water_temp):
         now = time.time()

test_liquidctl_gui.py

@@ -0,0 +1,151 @@
+import subprocess
+import types
+import sys
+from types import SimpleNamespace
+
+import pytest
+from unittest.mock import patch
+
+
+def load_module():
+    """Load LiquidctlGUI with stubbed PyQt6 and without running device checks."""
+    # Stub PyQt6 modules
+    pyqt6 = types.ModuleType("PyQt6")
+    qtwidgets = types.ModuleType("PyQt6.QtWidgets")
+    qtgui = types.ModuleType("PyQt6.QtGui")
+    qtcore = types.ModuleType("PyQt6.QtCore")
+
+    class DummyMeta(type):
+        def __getattr__(cls, name):
+            return Dummy
+
+    class Dummy(metaclass=DummyMeta):
+        def __init__(self, *a, **k):
+            pass
+        def __getattr__(self, name):
+            return Dummy()
+        def __call__(self, *a, **k):
+            return Dummy()
+        def setText(self, *a, **k):
+            pass
+        def show(self):
+            pass
+        def setIcon(self, *a, **k):
+            pass
+        def setContextMenu(self, *a, **k):
+            pass
+        def start(self, *a, **k):
+            pass
+        def timeout(self, *a, **k):
+            return Dummy()
+        def connect(self, *a, **k):
+            pass
+        def blockSignals(self, *a, **k):
+            pass
+        def setValue(self, *a, **k):
+            pass
+
+    widget_names = [
+        "QApplication", "QMainWindow", "QWidget", "QVBoxLayout", "QHBoxLayout",
+        "QPushButton", "QLabel", "QSlider", "QGroupBox", "QComboBox",
+        "QSystemTrayIcon", "QMenu", "QMessageBox", "QDialog",
+        "QFormLayout", "QDialogButtonBox", "QLineEdit", "QInputDialog",
+        "QStyle", "QSpacerItem", "QSizePolicy", "QStatusBar"
+    ]
+    for name in widget_names:
+        setattr(qtwidgets, name, Dummy)
+    for name in ["QIcon", "QAction", "QFont"]:
+        setattr(qtgui, name, Dummy)
+    qtcore.Qt = SimpleNamespace(Orientation=SimpleNamespace(Horizontal=0))
+    qtcore.QTimer = Dummy
+
+    sys.modules.setdefault("PyQt6", pyqt6)
+    sys.modules.setdefault("PyQt6.QtWidgets", qtwidgets)
+    sys.modules.setdefault("PyQt6.QtGui", qtgui)
+    sys.modules.setdefault("PyQt6.QtCore", qtcore)
+
+    # Load module with device access check removed
+    with open("LiquidctlGUI.py", "r") as f:
+        lines = f.readlines()
+    filtered = []
+    skip = {23, 24, 25, 26, 162, 163}
+    for idx, line in enumerate(lines, start=1):
+        if idx in skip:
+            continue
+        filtered.append(line)
+    module = types.ModuleType("LiquidctlGUI")
+    exec("".join(filtered), module.__dict__)
+    return module
+
+
+# Load LiquidctlGUI module
+LiquidctlModule = load_module()
+LiquidCtlGUI = LiquidctlModule.LiquidCtlGUI
+
+
+class DummyLabel:
+    def setText(self, text):
+        self.text = text
+
+
+def make_dummy_gui():
+    gui = SimpleNamespace(
+        selected_device={"description": "test"},
+        parse_json_status=lambda data: None,
+        parse_text_status=lambda text: None,
+        show_status_message=lambda msg: None,
+        cpu_temp_label=DummyLabel(),
+    )
+    return gui
+
+
+def test_update_status_falls_back_on_json_parse_error():
+    gui = make_dummy_gui()
+    sample_text = "Fan 1 Speed: 800\nPump Speed: 2500\nWater Temperature: 31.5"
+    error_cp = subprocess.CompletedProcess(["liquidctl"], 0, stdout="{", stderr="")
+    text_cp = subprocess.CompletedProcess(["liquidctl"], 0, stdout=sample_text, stderr="")
+    gui.parse_text_status = lambda text: setattr(gui, "text_output", text)
+    with patch.object(LiquidctlModule.subprocess, "run", side_effect=[error_cp, text_cp]):
+        with patch.object(LiquidctlModule, "get_cpu_temp", return_value=None):
+            LiquidCtlGUI.update_status(gui)
+    assert getattr(gui, "text_output", None) == sample_text
+
+
+def test_update_status_falls_back_on_json_command_error():
+    gui = make_dummy_gui()
+    sample_text = "Fan 1 Speed: 800\nPump Speed: 2500\nWater Temperature: 31.5"
+    error_cp = subprocess.CompletedProcess(["liquidctl"], 1, stdout="", stderr="boom")
+    text_cp = subprocess.CompletedProcess(["liquidctl"], 0, stdout=sample_text, stderr="")
+    gui.parse_text_status = lambda text: setattr(gui, "text_output", text)
+    with patch.object(LiquidctlModule.subprocess, "run", side_effect=[error_cp, text_cp]):
+        with patch.object(LiquidctlModule, "get_cpu_temp", return_value=None):
+            LiquidCtlGUI.update_status(gui)
+    assert getattr(gui, "text_output", None) == sample_text
+
+
+def test_parse_text_status_extracts_values():
+    class Dummy:
+        min_fan_rpm = 200
+        max_fan_rpm = 2000
+        min_pump_rpm = 1000
+        max_pump_rpm = 2700
+
+        def rpm_to_percent(self, rpm, is_pump=False):
+            return LiquidCtlGUI.rpm_to_percent(self, rpm, is_pump)
+
+        def update_ui_with_status(self, fan_speeds, pump_speed, water_temp):
+            self.fan_speeds = fan_speeds
+            self.pump_speed = pump_speed
+            self.water_temp = water_temp
+
+    dummy = Dummy()
+    sample_text = (
+        "Fan 1 Speed: 800\n"
+        "Fan 2 Speed: 1000\n"
+        "Pump Speed: 2500\n"
+        "Water Temperature: 31.5"
+    )
+    LiquidCtlGUI.parse_text_status(dummy, sample_text)
+    assert dummy.fan_speeds == {"Fan 1": (50, 800), "Fan 2": (60, 1000)}
+    assert dummy.pump_speed == (90, 2500)
+    assert dummy.water_temp == 31.5