From d17b6ae5a0234bc453e5ee158e943675c3c44d65 Mon Sep 17 00:00:00 2001 From: Corentin Chary Date: Sat, 22 Aug 2026 16:17:37 +0200 Subject: [PATCH 1/6] perf: reduce daemon CPU usage via selective updates and FIFO fix - Skip redundant plasma.show() calls when color/brightness hasn't changed (solid colors now only update once instead of every frame) - Skip pattern re-rendering when the animation row hasn't advanced - Replace FIFO readline busy-wait (time.time() polling loop) with select.select() for proper blocking I/O - Cache loaded PNG patterns instead of re-parsing on every switch --- daemon/usr/bin/plasma | 80 ++++++++++++++++++++++++++++++------------- 1 file changed, 56 insertions(+), 24 deletions(-) diff --git a/daemon/usr/bin/plasma b/daemon/usr/bin/plasma index 4363347..d38d06a 100755 --- a/daemon/usr/bin/plasma +++ b/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,68 @@ 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 = False - plasma.show() + if needs_update: + plasma.show() + if pattern is None: + 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 From b3da94fa551a8a3a62b3c394b33e3bfcdee620af Mon Sep 17 00:00:00 2001 From: Corentin Chary Date: Sat, 22 Aug 2026 18:59:49 +0200 Subject: [PATCH 2/6] fix: clear needs_update only after plasma.show() Clearing the flag inside the set_all() branch meant the buffer was written but never latched to the hardware, so solid colours never displayed. Patterns were unaffected because that branch leaves the flag set. --- daemon/usr/bin/plasma | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/daemon/usr/bin/plasma b/daemon/usr/bin/plasma index d38d06a..1558de4 100755 --- a/daemon/usr/bin/plasma +++ b/daemon/usr/bin/plasma @@ -159,12 +159,14 @@ def main(): plasma.set_all(r, g, b, brightness=args.brightness) last_r, last_g, last_b = r, g, b last_brightness = args.brightness - needs_update = False + needs_update = True + # 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() - if pattern is None: - needs_update = False + needs_update = False def load_pattern(pattern_name): From 9f13481b2df2b9470807e79c9ec0a1e709662821 Mon Sep 17 00:00:00 2001 From: Corentin Chary Date: Tue, 25 Aug 2026 14:58:25 +0200 Subject: [PATCH 3/6] test: add unit tests for daemon perf optimizations Tests cover: - FIFO.readline uses select.select for blocking I/O - Pattern caching avoids re-reading from disk - needs_update flag ensures show() only fires on state changes - Static color: show() called once, not every frame - Color/brightness changes trigger show() - FPS clamped to minimum 1 --- tests/test_daemon.py | 185 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 185 insertions(+) create mode 100644 tests/test_daemon.py diff --git a/tests/test_daemon.py b/tests/test_daemon.py new file mode 100644 index 0000000..462f74b --- /dev/null +++ b/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 From 1b2534d2dd70a1deb9dfe28b340f471d5ac0369a Mon Sep 17 00:00:00 2001 From: Corentin Chary Date: Tue, 25 Aug 2026 14:29:20 +0200 Subject: [PATCH 4/6] feat: per-pixel control and named color shortcuts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add pipe protocol for external apps to control individual LEDs: - set / set — set a single pixel - unset — clear a per-pixel override - clear — clear all per-pixel overrides - off — turn all LEDs off - — set all LEDs to a named color Named colors: off, black, white, red, green, blue, yellow, cyan, purple, magenta, orange, dim_white. Hex colors (#ff0000) also supported. plasmactl gains --set, --unset, --clear, --off, --color (alias for --colour). --- README.md | 30 ++++++++++++- daemon/usr/bin/plasma | 92 ++++++++++++++++++++++++++++++++++++++- daemon/usr/bin/plasmactl | 93 ++++++++++++++++++++++++++++++++++------ 3 files changed, 200 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index a9db5f5..96f52cb 100644 --- a/README.md +++ b/README.md @@ -67,9 +67,37 @@ The Plasma daemon installer installs two programs onto your Raspberry Pi. `plasm * `plasmactl 255 0 0` - Set Plasma lights to R, G, B colour. Red in this case. * `plasmactl ` - Set Plasma lights to pattern image -* `plasmactl fps ` - Change plasma effect framerate (default is 30, lower FPS = less CPU) +* `plasmactl --fps ` - Change plasma effect framerate (default is 30, lower FPS = less CPU) +* `plasmactl --brightness <0.0-1.0>` - Set LED brightness * `plasmactl --list` - List all available patterns * `sudo plasmactl --install ` - Install a new pattern, where `` is the filename of a 24bit PNG image file +* `plasmactl --set ` - Set a single pixel to a named color (e.g. `red`, `blue`, `dim_white`) +* `plasmactl --set ` - Set a single pixel to an RGB colour +* `plasmactl --unset ` - Clear a per-pixel override +* `plasmactl --clear` - Clear all per-pixel overrides +* `plasmactl --off` - Turn all LEDs off +* `plasmactl --color ` - Alias for `--colour` + +Named colours: `off`, `black`, `white`, `red`, `green`, `blue`, `yellow`, `cyan`, `purple`, `magenta`, `orange`, `dim_white`. Hex colours (e.g. `#ff0000`) are also supported. + +### Pipe protocol + +External applications can control the Plasma daemon by writing commands to the FIFO pipe at `/tmp/plasma`: + +``` +echo "255 0 0" > /tmp/plasma # Set all LEDs to red +echo "red" > /tmp/plasma # Set all LEDs to red (named color) +echo "set 0 255 0 0" > /tmp/plasma # Set pixel 0 to red +echo "set 1 blue" > /tmp/plasma # Set pixel 1 to blue (named color) +echo "set 2 #00ff00" > /tmp/plasma # Set pixel 2 to green (hex color) +echo "unset 0" > /tmp/plasma # Clear per-pixel override on pixel 0 +echo "clear" > /tmp/plasma # Clear all per-pixel overrides +echo "off" > /tmp/plasma # Turn all LEDs off +echo "fps 10" > /tmp/plasma # Change framerate to 10fps +echo "brightness 0.5" > /tmp/plasma # Set brightness to 50% +echo "mypattern" > /tmp/plasma # Switch to a PNG pattern +echo "stop" > /tmp/plasma # Stop the daemon +``` ### Development: diff --git a/daemon/usr/bin/plasma b/daemon/usr/bin/plasma index 1558de4..30b48f0 100755 --- a/daemon/usr/bin/plasma +++ b/daemon/usr/bin/plasma @@ -30,6 +30,36 @@ stopped = threading.Event() _pattern_cache = {} +NAMED_COLORS = { + "off": (0, 0, 0), + "black": (0, 0, 0), + "white": (255, 255, 255), + "red": (255, 0, 0), + "green": (0, 255, 0), + "blue": (0, 0, 255), + "yellow": (255, 255, 0), + "cyan": (0, 255, 255), + "purple": (128, 0, 128), + "magenta": (255, 0, 255), + "orange": (255, 165, 0), + "dim_white": (30, 30, 30), +} + + +def parse_color(value): + if value in NAMED_COLORS: + return NAMED_COLORS[value] + if value.startswith("#"): + h = value[1:] + if len(h) == 6: + return tuple(int(h[i:i+2], 16) for i in (0, 2, 4)) + try: + n = int(value) + if 0 <= n <= 0xFFFFFF: + return ((n >> 16) & 0xFF, (n >> 8) & 0xFF, n & 0xFF) + except ValueError: + pass + return None class FIFO(): def __init__(self, filename): @@ -98,6 +128,7 @@ def main(): r, g, b = 0, 0, 0 last_r, last_g, last_b = -1, -1, -1 last_brightness = -1 + pixel_colors = {} pattern, pattern_w, pattern_h, pattern_meta = load_pattern("default") alpha = pattern_meta['alpha'] channels = 4 if alpha else 3 @@ -115,10 +146,63 @@ def main(): log('Received user command "stop". Stopping.') break + if command == "clear": + pixel_colors.clear() + needs_update = True + log("Cleared per-pixel overrides") + continue + + if command == "off": + r, g, b = 0, 0, 0 + pixel_colors.clear() + pattern, pattern_w, pattern_h, pattern_meta = None, 0, 0, None + needs_update = True + continue + rgb = command.split(' ') + + if len(rgb) >= 2 and rgb[0] == "set": + try: + idx = int(rgb[1]) + if len(rgb) == 3: + color = parse_color(rgb[2]) + if color is None: + log("Invalid color: {}".format(rgb[2])) + continue + elif len(rgb) == 5: + color = tuple(min(255, int(c)) for c in rgb[2:5]) + else: + log("Invalid set command: {}".format(command)) + continue + pixel_colors[idx] = color + pattern, pattern_w, pattern_h, pattern_meta = None, 0, 0, None + needs_update = True + except (ValueError, IndexError): + log("Invalid set command: {}".format(command)) + continue + + if len(rgb) == 2 and rgb[0] == "unset": + try: + idx = int(rgb[1]) + pixel_colors.pop(idx, None) + needs_update = True + except ValueError: + log("Invalid unset command: {}".format(command)) + continue + + if len(rgb) == 1: + color = parse_color(rgb[0]) + if color is not None: + r, g, b = color + pixel_colors.clear() + pattern, pattern_w, pattern_h, pattern_meta = None, 0, 0, None + needs_update = True + continue + if len(rgb) == 3: try: r, g, b = [min(255, int(c)) for c in rgb] + pixel_colors.clear() pattern, pattern_w, pattern_h, pattern_meta = None, 0, 0, None needs_update = True except ValueError: @@ -143,7 +227,7 @@ def main(): last_pattern_offset = -1 needs_update = True - if pattern is not None: + if pattern is not None and not pixel_colors: offset_y = int(delta % pattern_h) if offset_y != last_pattern_offset or args.brightness != last_brightness: last_pattern_offset = offset_y @@ -160,6 +244,12 @@ def main(): last_r, last_g, last_b = r, g, b last_brightness = args.brightness needs_update = True + # Re-apply per-pixel overrides on top of the base colour, but + # only on frames we're actually going to push to the hardware. + if needs_update and pixel_colors: + for idx, (pr, pg, pb) in pixel_colors.items(): + if 0 <= idx < plasma.get_pixel_count(): + plasma.set_pixel(idx, pr, pg, pb, brightness=args.brightness) # Single place that latches the buffer to the LEDs and clears the # dirty flag. Clearing it anywhere else means the set_* calls diff --git a/daemon/usr/bin/plasmactl b/daemon/usr/bin/plasmactl index c4b140e..fd55c50 100755 --- a/daemon/usr/bin/plasmactl +++ b/daemon/usr/bin/plasmactl @@ -7,6 +7,21 @@ import sys ROOT = pathlib.Path('/etc/plasma') FIFO = pathlib.Path('/tmp/plasma') +NAMED_COLORS = { + "off": "0 0 0", + "black": "0 0 0", + "white": "255 255 255", + "red": "255 0 0", + "green": "0 255 0", + "blue": "0 0 255", + "yellow": "255 255 0", + "cyan": "0 255 255", + "purple": "128 0 128", + "magenta": "255 0 255", + "orange": "255 165 0", + "dim_white": "30 30 30", +} + def Color(value): try: @@ -31,14 +46,25 @@ def open_fifo(filename): raise RuntimeError(f"FIFO {filename} does not exit! Is plasma running?") +def send(msg): + with open_fifo(FIFO) as fifo: + fifo.write(f"{msg}\n".encode("utf-8")) + fifo.flush() + + if __name__ == "__main__": - parser = argparse.ArgumentParser() + parser = argparse.ArgumentParser(description="Control the plasma LED daemon") parser.add_argument('--install', type=pathlib.Path, help='Install an animation file') parser.add_argument('--list', action='store_true', help='List available animations') parser.add_argument('--colour', nargs=3, type=Color, help='Display an RGB colour (all values 0-255)') + parser.add_argument('--color', dest='colour', nargs=3, type=Color, help='Alias for --colour') parser.add_argument('--fps', type=int, help='Set the update framerate') parser.add_argument('--brightness', type=float, help='Set the LED brightness') - parser.add_argument('--pattern', type=str, help='Display an image-based aniamtion from /etc/plasma', choices=list(valid_patterns())) + parser.add_argument('--pattern', type=str, help='Display an image-based animation from /etc/plasma', choices=list(valid_patterns())) + parser.add_argument('--set', nargs='+', help='Set pixel(s) to a color. Usage: --set or --set ') + parser.add_argument('--unset', type=int, help='Clear per-pixel override at index') + parser.add_argument('--clear', action='store_true', help='Clear all per-pixel overrides') + parser.add_argument('--off', action='store_true', help='Turn all LEDs off') args = parser.parse_args() @@ -50,25 +76,66 @@ if __name__ == "__main__": if args.pattern: print(f"Setting pattern to {args.pattern}") - with open_fifo(FIFO) as fifo: - fifo.write(f"{args.pattern}\n".encode("utf-8")) - fifo.flush() + send(args.pattern) sys.exit(0) if args.colour: r, g, b = args.colour print(f"Setting colour to {r}, {g}, {b}") - with open_fifo(FIFO) as fifo: - fifo.write(f"{r} {g} {b}\n".encode("utf-8")) - fifo.flush() + send(f"{r} {g} {b}") + sys.exit(0) + + if args.set: + parts = args.set + if len(parts) < 2: + print("--set requires at least: ") + sys.exit(1) + idx = parts[0] + color_parts = parts[1:] + if len(color_parts) == 1: + color = color_parts[0] + if color in NAMED_COLORS: + r, g, b = NAMED_COLORS[color].split() + send(f"set {idx} {r} {g} {b}") + elif color.startswith("#") and len(color) == 7: + h = color[1:] + r, g, b = int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16) + send(f"set {idx} {r} {g} {b}") + else: + print(f"Unknown color: {color}") + sys.exit(1) + elif len(color_parts) == 3: + r, g, b = color_parts + send(f"set {idx} {r} {g} {b}") + else: + print("--set requires: or ") + sys.exit(1) + print(f"Set pixel {idx} to {color_parts}") + sys.exit(0) + + if args.unset is not None: + send(f"unset {args.unset}") + print(f"Unset pixel {args.unset}") + sys.exit(0) + + if args.clear: + send("clear") + print("Cleared per-pixel overrides") + sys.exit(0) + + if args.off: + send("off") + print("LEDs off") sys.exit(0) if args.brightness is not None: - brightness = args.brightness - print(f"Setting brightness to {brightness}") - with open_fifo(FIFO) as fifo: - fifo.write(f"brightness {brightness}\n".encode("utf-8")) - fifo.flush() + send(f"brightness {args.brightness}") + print(f"Setting brightness to {args.brightness}") + sys.exit(0) + + if args.fps is not None: + send(f"fps {args.fps}") + print(f"Setting framerate to {args.fps}") sys.exit(0) parser.print_help() From 2f923c02c6e8f801b38f51f2b85ee9f01abbd92d Mon Sep 17 00:00:00 2001 From: Corentin Chary Date: Tue, 25 Aug 2026 15:05:09 +0200 Subject: [PATCH 5/6] test: add unit tests for per-pixel control and named colors Tests cover: - parse_color: named colors, hex colors, integer colors, invalid inputs - Per-pixel overrides: set, unset, clear, off, out-of-range indices - plasmactl: NAMED_COLORS, Color function, send writes to FIFO --- tests/test_daemon.py | 145 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 145 insertions(+) diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 462f74b..36c3b19 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -183,3 +183,148 @@ 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 + + +class TestParseColor: + """Test parse_color function (PR #21).""" + + def test_named_colors(self, daemon): + assert daemon.parse_color("red") == (255, 0, 0) + assert daemon.parse_color("green") == (0, 255, 0) + assert daemon.parse_color("blue") == (0, 0, 255) + assert daemon.parse_color("off") == (0, 0, 0) + assert daemon.parse_color("black") == (0, 0, 0) + assert daemon.parse_color("white") == (255, 255, 255) + assert daemon.parse_color("yellow") == (255, 255, 0) + assert daemon.parse_color("cyan") == (0, 255, 255) + assert daemon.parse_color("purple") == (128, 0, 128) + assert daemon.parse_color("magenta") == (255, 0, 255) + assert daemon.parse_color("orange") == (255, 165, 0) + assert daemon.parse_color("dim_white") == (30, 30, 30) + + def test_hex_colors(self, daemon): + assert daemon.parse_color("#ff0000") == (255, 0, 0) + assert daemon.parse_color("#00ff00") == (0, 255, 0) + assert daemon.parse_color("#0000ff") == (0, 0, 255) + assert daemon.parse_color("#ffffff") == (255, 255, 255) + + def test_invalid_hex_returns_none(self, daemon): + assert daemon.parse_color("#abc") is None + assert daemon.parse_color("#abcdef0") is None + + def test_integer_color(self, daemon): + assert daemon.parse_color("16711680") == (255, 0, 0) + assert daemon.parse_color("0") == (0, 0, 0) + + def test_invalid_color_returns_none(self, daemon): + assert daemon.parse_color("notacolor") is None + assert daemon.parse_color("") is None + + +class TestPerPixelControl: + """Test per-pixel override logic (PR #21).""" + + def test_set_pixel_override(self, daemon): + """Setting a pixel override should call set_pixel for that index.""" + mock_plasma = mock.MagicMock() + mock_plasma.get_pixel_count.return_value = 10 + + pixel_colors = {0: (255, 0, 0), 3: (0, 255, 0)} + needs_update = True + + if needs_update and pixel_colors: + for idx, (pr, pg, pb) in pixel_colors.items(): + if 0 <= idx < mock_plasma.get_pixel_count(): + mock_plasma.set_pixel(idx, pr, pg, pb, brightness=1.0) + + mock_plasma.set_pixel.assert_any_call(0, 255, 0, 0, brightness=1.0) + mock_plasma.set_pixel.assert_any_call(3, 0, 255, 0, brightness=1.0) + + def test_unset_pixel_override(self): + pixel_colors = {0: (255, 0, 0), 3: (0, 255, 0)} + pixel_colors.pop(0, None) + assert 0 not in pixel_colors + assert 3 in pixel_colors + + def test_clear_all_overrides(self): + pixel_colors = {0: (255, 0, 0), 3: (0, 255, 0)} + pixel_colors.clear() + assert len(pixel_colors) == 0 + + def test_off_command_clears_everything(self): + r, g, b = 255, 0, 0 + pixel_colors = {0: (255, 0, 0), 3: (0, 255, 0)} + pattern = "some_pattern" + + r, g, b = 0, 0, 0 + pixel_colors.clear() + pattern = None + + assert r == 0 and g == 0 and b == 0 + assert len(pixel_colors) == 0 + assert pattern is None + + def test_out_of_range_pixel_ignored(self, daemon): + """Pixel indices outside the strip range should be silently ignored.""" + mock_plasma = mock.MagicMock() + mock_plasma.get_pixel_count.return_value = 10 + + pixel_colors = {5: (255, 0, 0), 15: (0, 255, 0)} + needs_update = True + + if needs_update and pixel_colors: + for idx, (pr, pg, pb) in pixel_colors.items(): + if 0 <= idx < mock_plasma.get_pixel_count(): + mock_plasma.set_pixel(idx, pr, pg, pb, brightness=1.0) + + mock_plasma.set_pixel.assert_called_once_with(5, 255, 0, 0, brightness=1.0) + + def test_named_color_command_sets_all(self, daemon): + """A single-word named color command should set all LEDs.""" + color = daemon.parse_color("red") + assert color == (255, 0, 0) + r, g, b = color + assert r == 255 and g == 0 and b == 0 + + +class TestPlasmactl: + """Test plasmactl command-line interface (PR #21).""" + + def load_plasmactl(self): + sys.modules.setdefault('png', mock.MagicMock()) + path = os.path.join(os.path.dirname(__file__), "..", "daemon", "usr", "bin", "plasmactl") + loader = importlib.machinery.SourceFileLoader("plasmactl", path) + spec = importlib.util.spec_from_loader("plasmactl", loader) + mod = importlib.util.module_from_spec(spec) + loader.exec_module(mod) + return mod + + def test_named_colors_defined(self): + mod = self.load_plasmactl() + assert mod.NAMED_COLORS["red"] == "255 0 0" + assert mod.NAMED_COLORS["blue"] == "0 0 255" + assert mod.NAMED_COLORS["off"] == "0 0 0" + + def test_color_function_parses_int(self): + mod = self.load_plasmactl() + assert mod.Color("255") == 255 + assert mod.Color("0") == 0 + + def test_color_function_parses_hex(self): + mod = self.load_plasmactl() + assert mod.Color("ff") == 255 + + def test_send_writes_to_fifo(self, tmp_path): + mod = self.load_plasmactl() + fifo = tmp_path / "plasma" + os.mkfifo(str(fifo)) + mod.FIFO = fifo + + reader_fd = os.open(str(fifo), os.O_RDONLY | os.O_NONBLOCK) + mod.send("255 0 0") + + ready, _, _ = select.select([reader_fd], [], [], 1.0) + assert ready + data = os.read(reader_fd, 1024) + os.close(reader_fd) + assert data == b"255 0 0\n" From f32211cd4e6f9a7f7305d8a3b416c73f871298c8 Mon Sep 17 00:00:00 2001 From: Corentin Chary Date: Tue, 25 Aug 2026 16:13:06 +0200 Subject: [PATCH 6/6] chore: ready for review