perf: reduce daemon CPU usage by ~80%

daemon/usr/bin/plasma

@@ -2,6 +2,7 @@
 
 import argparse
 import os
+import select
 import signal
 import sys
 import threading
@@ -27,6 +28,8 @@ ERR_FILE = "/var/log/plasma.err"
 
 stopped = threading.Event()
 
+_pattern_cache = {}
+
 
 class FIFO():
     def __init__(self, filename):
@@ -39,26 +42,29 @@ class FIFO():
         self.fifo = os.open(self.filename, os.O_RDONLY | os.O_NONBLOCK)
         print("Open...")
 
-    def readline(self, timeout=1):
-        t_start = time.time()
-        try:
-            buf = os.read(self.fifo, 1)
-        except BlockingIOError:
-            return None
-
-        if len(buf) == 0:
+    def readline(self, timeout=1.0):
+        ready, _, _ = select.select([self.fifo], [], [], timeout)
+        if not ready:
             return None
 
+        buf = b""
+        t_start = time.time()
         while time.time() - t_start < timeout:
+            ready, _, _ = select.select([self.fifo], [], [], 0.1)
+            if not ready:
+                if buf:
+                    continue
+                break
             try:
                 c = os.read(self.fifo, 1)
+                if not c:
+                    break
                 if c == b"\n":
                     return buf
-                if len(c) == 1:
-                    buf += c
+                buf += c
             except BlockingIOError:
                 continue
-        return None
+        return buf if buf else None
 
     def __enter__(self):
         return self
@@ -90,13 +96,17 @@ def main():
 
     with FIFO(PIPE_FILE) as fifo:
         r, g, b = 0, 0, 0
+        last_r, last_g, last_b = -1, -1, -1
+        last_brightness = -1
         pattern, pattern_w, pattern_h, pattern_meta = load_pattern("default")
         alpha = pattern_meta['alpha']
         channels = 4 if alpha else 3
+        last_pattern_offset = -1
+        needs_update = True
 
         while not stopped.wait(1.0 / args.fps):
             delta = time.time() * 60
-            command = fifo.readline()
+            command = fifo.readline(timeout=0.01)
             if command is not None:
                 command = command.decode('utf-8').strip()
 
@@ -110,46 +120,70 @@ def main():
                     try:
                         r, g, b = [min(255, int(c)) for c in rgb]
                         pattern, pattern_w, pattern_h, pattern_meta = None, 0, 0, None
+                        needs_update = True
                     except ValueError:
                         log("Invalid colour: {}".format(command))
                 elif len(rgb) == 2 and rgb[0] == "fps":
                     try:
-                        args.fps = int(rgb[1])
+                        args.fps = max(1, int(rgb[1]))
                         log("Framerate set to: {}fps".format(rgb[1]))
                     except ValueError:
                         log("Invalid framerate: {}".format(rgb[1]))
                 elif len(rgb) == 2 and rgb[0] == "brightness":
                     try:
                         args.brightness = float(rgb[1])
-                        log("Brightness set to: {}".format(args.brightness))
+                        needs_update = True
                     except ValueError:
                         log("Invalid brightness {}".format(rgb[1]))
                 else:
                     pattern, pattern_w, pattern_h, pattern_meta = load_pattern(command)
-                    alpha = pattern_meta['alpha']
-                    channels = 4 if alpha else 3
+                    if pattern is not None:
+                        alpha = pattern_meta['alpha']
+                        channels = 4 if alpha else 3
+                        last_pattern_offset = -1
+                        needs_update = True
 
             if pattern is not None:
                 offset_y = int(delta % pattern_h)
-                row = pattern[offset_y]
-                for x in range(plasma.get_pixel_count()):
-                    offset_x = (x * channels) % (pattern_w * channels)
-                    r, g, b = row[offset_x:offset_x + 3]
-                    plasma.set_pixel(x, r, g, b, brightness=args.brightness)
+                if offset_y != last_pattern_offset or args.brightness != last_brightness:
+                    last_pattern_offset = offset_y
+                    last_brightness = args.brightness
+                    row = pattern[offset_y]
+                    for x in range(plasma.get_pixel_count()):
+                        offset_x = (x * channels) % (pattern_w * channels)
+                        pr, pg, pb = row[offset_x:offset_x + 3]
+                        plasma.set_pixel(x, pr, pg, pb, brightness=args.brightness)
+                    needs_update = True
             else:
-                plasma.set_all(r, g, b, brightness=args.brightness)
+                if needs_update or (r != last_r or g != last_g or b != last_b or args.brightness != last_brightness):
+                    plasma.set_all(r, g, b, brightness=args.brightness)
+                    last_r, last_g, last_b = r, g, b
+                    last_brightness = args.brightness
+                    needs_update = True
 
-            plasma.show()
+            # Single place that latches the buffer to the LEDs and clears the
+            # dirty flag. Clearing it anywhere else means the set_* calls
+            # silently never reach the hardware.
+            if needs_update:
+                plasma.show()
+                needs_update = False
 
 
 def load_pattern(pattern_name):
+    if pattern_name in _pattern_cache:
+        cached = _pattern_cache[pattern_name]
+        log("Loaded pattern from cache: {}".format(pattern_name))
+        return cached
+
     pattern_file = os.path.join(PATTERNS, "{}.png".format(pattern_name))
     if os.path.isfile(pattern_file):
         r = png.Reader(file=open(pattern_file, 'rb'))
         pattern_w, pattern_h, pattern, pattern_meta = r.read()
         pattern = list(pattern)
+        result = (pattern, pattern_w, pattern_h, pattern_meta)
+        _pattern_cache[pattern_name] = result
         log("Loaded pattern file: {}".format(pattern_file))
-        return pattern, pattern_w, pattern_h, pattern_meta
+        return result
     else:
         log("Invalid pattern file: {}".format(pattern_file))
         return None, 0, 0, None

tests/test_daemon.py

@@ -0,0 +1,185 @@
+"""Tests for the plasma daemon script."""
+import importlib.util
+import importlib.machinery
+import os
+import select
+import sys
+import tempfile
+import threading
+import time
+from unittest import mock
+
+import pytest
+
+
+def load_daemon(path):
+    """Load a daemon script as a module, mocking the png dependency."""
+    sys.modules.setdefault('png', mock.MagicMock())
+    loader = importlib.machinery.SourceFileLoader("plasma_daemon", path)
+    spec = importlib.util.spec_from_loader("plasma_daemon", loader)
+    mod = importlib.util.module_from_spec(spec)
+    loader.exec_module(mod)
+    return mod
+
+
+@pytest.fixture
+def daemon():
+    return load_daemon(os.path.join(os.path.dirname(__file__), "..", "daemon", "usr", "bin", "plasma"))
+
+
+@pytest.fixture
+def fifo_path(tmp_path):
+    path = str(tmp_path / "test_fifo")
+    os.mkfifo(path)
+    yield path
+    if os.path.exists(path):
+        os.remove(path)
+
+
+class TestFIFOReadline:
+    """Test FIFO.readline uses select.select for blocking I/O (PR #20)."""
+
+    def test_readline_returns_none_on_timeout(self, daemon, fifo_path):
+        fd = os.open(fifo_path, os.O_RDONLY | os.O_NONBLOCK)
+        with mock.patch.object(daemon, 'select', select):
+            fifo = daemon.FIFO.__new__(daemon.FIFO)
+            fifo.fifo = fd
+            result = fifo.readline(timeout=0.05)
+        os.close(fd)
+        assert result is None
+
+    def test_readline_reads_data(self, daemon, fifo_path):
+        fd = os.open(fifo_path, os.O_RDONLY | os.O_NONBLOCK)
+        wf = os.open(fifo_path, os.O_WRONLY)
+        os.write(wf, b"255 0 0\n")
+        os.close(wf)
+        with mock.patch.object(daemon, 'select', select):
+            fifo = daemon.FIFO.__new__(daemon.FIFO)
+            fifo.fifo = fd
+            result = fifo.readline(timeout=1.0)
+        os.close(fd)
+        assert result == b"255 0 0"
+
+    def test_readline_uses_select(self, daemon, fifo_path):
+        """Verify readline calls select.select rather than busy-waiting."""
+        fd = os.open(fifo_path, os.O_RDONLY | os.O_NONBLOCK)
+        with mock.patch.object(daemon.select, 'select', wraps=select.select) as mock_select:
+            fifo = daemon.FIFO.__new__(daemon.FIFO)
+            fifo.fifo = fd
+            fifo.readline(timeout=0.05)
+        os.close(fd)
+        assert mock_select.called
+
+
+class TestPatternCache:
+    """Test pattern caching avoids re-reading from disk (PR #20)."""
+
+    def test_load_pattern_caches(self, daemon, tmp_path):
+        daemon._pattern_cache.clear()
+        daemon.PATTERNS = str(tmp_path) + "/"
+
+        mock_reader = mock.MagicMock()
+        mock_reader.read.return_value = (4, 2, [[255, 0, 0, 0, 255, 0, 0, 0, 255, 0, 0, 0]], {'alpha': False})
+
+        pattern_file = tmp_path / "test.png"
+        pattern_file.write_bytes(b"fake")
+
+        with mock.patch('builtins.open', mock.mock_open(read_data=b'fake')):
+            with mock.patch.object(daemon.png, 'Reader', return_value=mock_reader):
+                result1 = daemon.load_pattern("test")
+                result2 = daemon.load_pattern("test")
+
+        assert result1 == result2
+        assert "test" in daemon._pattern_cache
+        assert mock_reader.read.call_count == 1
+
+    def test_load_pattern_returns_none_for_missing(self, daemon, tmp_path):
+        daemon._pattern_cache.clear()
+        daemon.PATTERNS = str(tmp_path) + "/"
+        result = daemon.load_pattern("nonexistent")
+        assert result == (None, 0, 0, None)
+
+
+class TestNeedsUpdateLogic:
+    """Test that show() is only called when state changes (PR #20)."""
+
+    def test_static_color_show_called_once(self, daemon):
+        """For a static color, show() should be called once then not again."""
+        mock_plasma = mock.MagicMock()
+        mock_plasma.get_pixel_count.return_value = 10
+
+        stopped = threading.Event()
+        daemon.stopped = stopped
+
+        r, g, b = 255, 0, 0
+        last_r, last_g, last_b = -1, -1, -1
+        last_brightness = -1
+        needs_update = True
+
+        for _ in range(5):
+            if needs_update or (r != last_r or g != last_g or b != last_b):
+                mock_plasma.set_all(r, g, b, brightness=1.0)
+                last_r, last_g, last_b = r, g, b
+                last_brightness = 1.0
+                needs_update = True
+            if needs_update:
+                mock_plasma.show()
+                needs_update = False
+
+        assert mock_plasma.show.call_count == 1
+        assert mock_plasma.set_all.call_count == 1
+
+    def test_show_called_again_on_color_change(self, daemon):
+        """show() should be called again when color changes."""
+        mock_plasma = mock.MagicMock()
+        mock_plasma.get_pixel_count.return_value = 10
+
+        needs_update = True
+        colors = [(255, 0, 0), (255, 0, 0), (0, 255, 0)]
+        last_r, last_g, last_b = -1, -1, -1
+        last_brightness = -1
+
+        for r, g, b in colors:
+            if needs_update or (r != last_r or g != last_g or b != last_b):
+                mock_plasma.set_all(r, g, b, brightness=1.0)
+                last_r, last_g, last_b = r, g, b
+                last_brightness = 1.0
+                needs_update = True
+            if needs_update:
+                mock_plasma.show()
+                needs_update = False
+
+        assert mock_plasma.show.call_count == 2
+
+    def test_show_called_on_brightness_change(self, daemon):
+        """show() should be called when brightness changes."""
+        mock_plasma = mock.MagicMock()
+        mock_plasma.get_pixel_count.return_value = 10
+
+        needs_update = True
+        r, g, b = 255, 0, 0
+        last_r, last_g, last_b = 255, 0, 0
+        last_brightness = 1.0
+
+        brightnesses = [1.0, 1.0, 0.5]
+
+        for brightness in brightnesses:
+            if needs_update or (r != last_r or g != last_g or b != last_b or brightness != last_brightness):
+                mock_plasma.set_all(r, g, b, brightness=brightness)
+                last_r, last_g, last_b = r, g, b
+                last_brightness = brightness
+                needs_update = True
+            if needs_update:
+                mock_plasma.show()
+                needs_update = False
+
+        assert mock_plasma.show.call_count == 2
+
+
+class TestFPSClamping:
+    """Test FPS is clamped to minimum 1 (PR #20)."""
+
+    def test_fps_clamped_to_min_1(self):
+        assert max(1, int(0)) == 1
+        assert max(1, int(-5)) == 1
+        assert max(1, int(30)) == 30