diff --git a/config.py b/config.py deleted file mode 100644 index fbb50f8..0000000 --- a/config.py +++ /dev/null @@ -1,89 +0,0 @@ -#!/usr/bin/env python3 -""" -config.py — Load and validate turnup configuration from config.json -""" - -import json -import logging -import os -import sys - -log = logging.getLogger("turnupd") - -DEFAULT_CONFIG: dict = { - "port": "/dev/ttyACM0", - "baud": 115200, - "knobs": { - "0": {"action": "sink_volume", "target": "default"}, - "1": {"action": "sink_volume", "target": "default"}, - "2": {"action": "source_volume", "target": "default"}, - "3": {"action": "sink_volume", "target": "default"}, - "4": {"action": "sink_volume", "target": "default"}, - }, - "buttons": { - "0": {"action": "mute_sink", "target": "default"}, - "1": {"action": "mute_source", "target": "default"}, - "2": {"action": "command", "target": ""}, - "3": {"action": "command", "target": ""}, - "4": {"action": "command", "target": ""}, - }, -} - -VALID_KNOB_ACTIONS: frozenset = frozenset( - {"sink_volume", "source_volume", "app_volume", "group_volume"} -) -VALID_BUTTON_ACTIONS: frozenset = frozenset( - {"mute_sink", "mute_source", "command"} -) - -# Default config search path: ~/.config/turnup/config.json -_XDG_CONFIG_DIR = os.path.join( - os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")), - "turnup", -) -DEFAULT_CONFIG_PATH = os.path.join(_XDG_CONFIG_DIR, "config.json") - - -def load_config(path: str | None = None) -> dict: - """Load configuration from *path*. - - If *path* is ``None`` the XDG-compliant location - ``~/.config/turnup/config.json`` is used, falling back to the directory - that contains this module for backwards compatibility. - - Returns the parsed configuration dictionary. Exits with status 1 on - malformed JSON. - """ - if path is None: - # Prefer XDG location; fall back to the directory next to this module. - if os.path.exists(DEFAULT_CONFIG_PATH): - path = DEFAULT_CONFIG_PATH - else: - script_dir = os.path.dirname(os.path.abspath(__file__)) - path = os.path.join(script_dir, "config.json") - - if not os.path.exists(path): - log.warning("No config.json found at %s — writing defaults", path) - _write_default(path) - return dict(DEFAULT_CONFIG) - - try: - with open(path) as f: - cfg: dict = json.load(f) - log.info("Loaded config from %s", path) - return cfg - except json.JSONDecodeError as exc: - log.error("Invalid JSON in %s: %s", path, exc) - sys.exit(1) - - -def _write_default(path: str) -> None: - """Write the default configuration to *path*, creating directories as needed.""" - try: - os.makedirs(os.path.dirname(path), exist_ok=True) - with open(path, "w") as f: - json.dump(DEFAULT_CONFIG, f, indent=2) - f.write("\n") - log.info("Created default config at %s", path) - except OSError as exc: - log.warning("Could not write default config: %s", exc) diff --git a/contrib/config.example.json b/contrib/config.example.json old mode 100644 new mode 100755 index 8674356..079ff1a --- a/contrib/config.example.json +++ b/contrib/config.example.json @@ -2,21 +2,55 @@ "port": "/dev/ttyACM0", "baud": 115200, + "leds": { + "mode": "volume", + "low_color": [255, 0, 0], + "high_color": [0, 255, 0] + }, + "knobs": { - "0": { "action": "sink_volume", "target": "default" }, - "1": { "action": "group_volume", "targets": ["vlc", "spotify", "Cider"] }, - "2": { "action": "app_volume", "target": "Brave" }, - "3": { "action": "group_volume", "targets": ["electron", "discord"] }, - "4": { "action": "sink_volume", "target": "default" } + "0": { + "action": "sink_volume", + "target": "default" + }, + "1": { + "action": "group_volume", + "targets": ["vlc", "spotify", "Cider"], + "led": { + "mode": "volume", + "low_color": [255, 0, 0], + "high_color": [0, 0, 255] + } + }, + "2": { + "action": "app_volume", + "target": "Brave", + "led": { + "mode": "static", + "high_color": [255, 165, 0] + } + }, + "3": { + "action": "group_volume", + "targets": ["electron", "discord"], + "led": { "mode": "off" } + }, + "4": { + "action": "source_volume", + "target": "default", + "led": { + "mode": "volume", + "low_color": [255, 0, 0], + "high_color": [0, 255, 255] + } + } }, "buttons": { "0": { "action": "mute_sink", "target": "default" }, - "1": { "action": "command", "target": "playerctl previous" }, + "1": { "action": "command", "target": "playerctl previous" }, "2": { "action": "command", "target": "playerctl play-pause" }, "3": { "action": "command", "target": "playerctl next" }, - "4": { "action": "mute_source", "target": "default" } + "4": { "action": "mute_source", "target": "default" } } } - - diff --git a/src/turnup/config.py b/src/turnup/config.py old mode 100644 new mode 100755 index fbb50f8..4c07161 --- a/src/turnup/config.py +++ b/src/turnup/config.py @@ -13,6 +13,18 @@ log = logging.getLogger("turnupd") DEFAULT_CONFIG: dict = { "port": "/dev/ttyACM0", "baud": 115200, + "leds": { + # Global default LED behaviour — overridable per-knob via a "led": {} + # block inside each knob entry. + # + # mode: + # "volume" — interpolate low_color→high_color based on knob position + # "static" — always show high_color regardless of volume + # "off" — LEDs disabled + "mode": "volume", + "low_color": [255, 0, 0], # red at volume 0.0 + "high_color": [0, 255, 0], # green at volume 1.0 + }, "knobs": { "0": {"action": "sink_volume", "target": "default"}, "1": {"action": "sink_volume", "target": "default"}, @@ -35,8 +47,8 @@ VALID_KNOB_ACTIONS: frozenset = frozenset( VALID_BUTTON_ACTIONS: frozenset = frozenset( {"mute_sink", "mute_source", "command"} ) +VALID_LED_MODES: frozenset = frozenset({"volume", "static", "off"}) -# Default config search path: ~/.config/turnup/config.json _XDG_CONFIG_DIR = os.path.join( os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")), "turnup", @@ -44,6 +56,106 @@ _XDG_CONFIG_DIR = os.path.join( DEFAULT_CONFIG_PATH = os.path.join(_XDG_CONFIG_DIR, "config.json") +# ── LED helpers ──────────────────────────────────────────────────────────────── + +def _validate_color(color: object, name: str, fallback: list) -> list: + """Return *color* if it is a valid [R, G, B] list, else *fallback*.""" + if ( + isinstance(color, (list, tuple)) + and len(color) == 3 + and all(isinstance(c, int) and 0 <= c <= 255 for c in color) + ): + return list(color) + log.warning("Invalid LED %s %r — using default", name, color) + return fallback + + +def _validate_leds(leds: object, context: str = "leds") -> dict: + """Validate and normalise a LED config block. + + *context* is used only in warning messages (e.g. ``"knobs.2.led"``). + Falls back field-by-field to the global defaults so partial overrides work. + """ + global_defaults = DEFAULT_CONFIG["leds"] + if not isinstance(leds, dict): + if leds is not None: + log.warning("%s is not a dict — using defaults", context) + return dict(global_defaults) + + mode = leds.get("mode", global_defaults["mode"]) + if mode not in VALID_LED_MODES: + log.warning("%s: unknown mode %r — falling back to 'volume'", context, mode) + mode = "volume" + + return { + "mode": mode, + "low_color": _validate_color( + leds.get("low_color"), "low_color", global_defaults["low_color"] + ), + "high_color": _validate_color( + leds.get("high_color"), "high_color", global_defaults["high_color"] + ), + } + + +def get_knob_led_cfg(config: dict, knob_id: int) -> dict: + """Return the effective LED config for *knob_id*. + + If the knob entry contains a ``"led"`` key it is used (merged with global + defaults for any missing fields); otherwise the top-level ``"leds"`` block + is returned. + """ + global_leds = config.get("leds", DEFAULT_CONFIG["leds"]) + knob_cfg = config.get("knobs", {}).get(str(knob_id), {}) + knob_led = knob_cfg.get("led") + + if knob_led is None: + return global_leds + + # Knob has its own led block — validate it, falling back to global values + # for any field it doesn't specify. + merged = { + "mode": global_leds.get("mode", DEFAULT_CONFIG["leds"]["mode"]), + "low_color": global_leds.get("low_color", DEFAULT_CONFIG["leds"]["low_color"]), + "high_color": global_leds.get("high_color", DEFAULT_CONFIG["leds"]["high_color"]), + } + if isinstance(knob_led, dict): + if "mode" in knob_led: + merged["mode"] = knob_led["mode"] + if "low_color" in knob_led: + merged["low_color"] = knob_led["low_color"] + if "high_color" in knob_led: + merged["high_color"] = knob_led["high_color"] + + return _validate_leds(merged, context=f"knobs.{knob_id}.led") + + +def get_led_color(led_cfg: dict, norm: float) -> tuple[int, int, int]: + """Return an ``(r, g, b)`` tuple for a knob at normalised position *norm* (0.0-1.0). + + Behaviour depends on ``led_cfg["mode"]``: + + * ``"off"`` -> ``(0, 0, 0)`` + * ``"static"`` -> ``high_color`` always + * ``"volume"`` -> linear interpolation between ``low_color`` and ``high_color`` + """ + mode = led_cfg.get("mode", "volume") + + if mode == "off": + return (0, 0, 0) + + high = led_cfg.get("high_color", DEFAULT_CONFIG["leds"]["high_color"]) + + if mode == "static": + return tuple(high) + + low = led_cfg.get("low_color", DEFAULT_CONFIG["leds"]["low_color"]) + t = max(0.0, min(1.0, norm)) + return tuple(int(low[i] + (high[i] - low[i]) * t) for i in range(3)) + + +# ── Config I/O ───────────────────────────────────────────────────────────────── + def load_config(path: str | None = None) -> dict: """Load configuration from *path*. @@ -51,11 +163,10 @@ def load_config(path: str | None = None) -> dict: ``~/.config/turnup/config.json`` is used, falling back to the directory that contains this module for backwards compatibility. - Returns the parsed configuration dictionary. Exits with status 1 on - malformed JSON. + Returns the parsed and validated configuration dictionary. + Exits with status 1 on malformed JSON. """ if path is None: - # Prefer XDG location; fall back to the directory next to this module. if os.path.exists(DEFAULT_CONFIG_PATH): path = DEFAULT_CONFIG_PATH else: @@ -70,6 +181,7 @@ def load_config(path: str | None = None) -> dict: try: with open(path) as f: cfg: dict = json.load(f) + cfg["leds"] = _validate_leds(cfg.get("leds", {})) log.info("Loaded config from %s", path) return cfg except json.JSONDecodeError as exc: diff --git a/src/turnup/turnupd.py b/src/turnup/turnupd.py index d72fbba..9302900 100755 --- a/src/turnup/turnupd.py +++ b/src/turnup/turnupd.py @@ -5,6 +5,10 @@ turnupd — Turn Up mixer daemon for Linux Bridges a USB serial device (knobs + buttons) to PipeWire/PulseAudio via pulsectl, mapping hardware inputs to per-sink, per-source, and per-app volume control as well as mute toggles and arbitrary shell commands. + +LED feedback: after every knob move the device's RGB LEDs are updated to +reflect the current volume using the per-knob (or global) colour scheme +from config. """ import logging @@ -16,7 +20,7 @@ import time import pulsectl import serial -from turnup.config import load_config +from turnup.config import get_knob_led_cfg, get_led_color, load_config logging.basicConfig( level=logging.INFO, @@ -29,23 +33,16 @@ log = logging.getLogger("turnupd") KNOB_MAX: int = 1012 # Maximum output volume multiplier (1.5 = 150 %). VOLUME_MAX: float = 1.5 +# Number of physical knobs (and therefore LED groups). +NUM_KNOBS: int = 5 +# Number of LEDs per knob. +LEDS_PER_KNOB: int = 3 # ── Protocol parser ──────────────────────────────────────────────────────────── def parse_messages(buf: bytearray) -> tuple[list[dict], bytearray]: - """Parse framed messages out of *buf* and return ``(messages, remainder)``. - - Supported frame formats (all delimited by ``0xFE`` … ``0xFF``): - - +-----------+----------------------------+ - | Heartbeat | ``FE 02 FF`` | - +-----------+----------------------------+ - | Button | ``FE 06/07 FF`` | - +-----------+----------------------------+ - | Knob | ``FE 03 FF``| - +-----------+----------------------------+ - """ + """Parse framed messages out of *buf* and return ``(messages, remainder)``.""" messages: list[dict] = [] i = 0 while i < len(buf): @@ -55,12 +52,10 @@ def parse_messages(buf: bytearray) -> tuple[list[dict], bytearray]: remaining = len(buf) - i - # Heartbeat: FE 02 FF (3 bytes) if remaining >= 3 and buf[i + 1] == 0x02 and buf[i + 2] == 0xFF: messages.append({"type": "heartbeat"}) i += 3 - # Button press/release: FE 06/07 FF (4 bytes) elif ( remaining >= 4 and buf[i + 1] in (0x06, 0x07) @@ -73,7 +68,6 @@ def parse_messages(buf: bytearray) -> tuple[list[dict], bytearray]: }) i += 4 - # Knob value: FE 03 FF (6 bytes) elif ( remaining >= 6 and buf[i + 1] == 0x03 @@ -102,6 +96,39 @@ def knob_to_norm(value: int) -> float: return round(value / KNOB_MAX, 4) +# ── LED control ──────────────────────────────────────────────────────────────── + +def build_led_packet(colors: list[tuple[int, int, int]]) -> bytes: + """Build the 47-byte LED packet for all 5 knobs. + + Frame format: ``FE 05 [R G B * LEDS_PER_KNOB] * NUM_KNOBS FF`` + """ + assert len(colors) == NUM_KNOBS + payload = bytearray([0xFE, 0x05]) + for r, g, b in colors: + payload += bytes([r, g, b]) * LEDS_PER_KNOB + payload.append(0xFF) + return bytes(payload) + + +def send_leds(ser: serial.Serial, colors: list[tuple[int, int, int]]) -> None: + """Write an LED packet to the open serial port, swallowing any I/O errors.""" + try: + ser.write(build_led_packet(colors)) + except serial.SerialException as exc: + log.warning("LED write failed: %s", exc) + + +def all_led_colors( + config: dict, knob_norms: list[float] +) -> list[tuple[int, int, int]]: + """Return one ``(r, g, b)`` per knob based on each knob's LED config.""" + return [ + get_led_color(get_knob_led_cfg(config, i), knob_norms[i]) + for i in range(NUM_KNOBS) + ] + + # ── PulseAudio / PipeWire controller ────────────────────────────────────────── class PulseController: @@ -113,10 +140,7 @@ class PulseController: def close(self) -> None: self._pulse.close() - # -- Sink (output) --------------------------------------------------------- - def set_sink_volume(self, sink_name: str, volume: float) -> None: - """Set output volume for *sink_name*. *volume* is clamped to 0.0–1.5.""" volume = max(0.0, min(VOLUME_MAX, volume)) try: if sink_name == "default": @@ -129,23 +153,18 @@ class PulseController: log.warning("set_sink_volume(%r) failed: %s", sink_name, exc) def toggle_mute_sink(self, sink_name: str) -> None: - """Toggle the mute state of *sink_name*.""" try: if sink_name == "default": info = self._pulse.server_info() sink = self._pulse.get_sink_by_name(info.default_sink_name) else: sink = self._pulse.get_sink_by_name(sink_name) - new_mute = not sink.mute - self._pulse.mute(sink, new_mute) - log.info("Sink %r mute → %s", sink_name, new_mute) + self._pulse.mute(sink, not sink.mute) + log.info("Sink %r mute toggled", sink_name) except Exception as exc: log.warning("toggle_mute_sink(%r) failed: %s", sink_name, exc) - # -- Source (input / mic) -------------------------------------------------- - def set_source_volume(self, source_name: str, volume: float) -> None: - """Set mic/input volume for *source_name*. *volume* is clamped to 0.0–1.0.""" volume = max(0.0, min(1.0, volume)) try: if source_name == "default": @@ -158,27 +177,18 @@ class PulseController: log.warning("set_source_volume(%r) failed: %s", source_name, exc) def toggle_mute_source(self, source_name: str) -> None: - """Toggle the mute state of *source_name*.""" try: if source_name == "default": info = self._pulse.server_info() source = self._pulse.get_source_by_name(info.default_source_name) else: source = self._pulse.get_source_by_name(source_name) - new_mute = not source.mute - self._pulse.mute(source, new_mute) - log.info("Source %r mute → %s", source_name, new_mute) + self._pulse.mute(source, not source.mute) + log.info("Source %r mute toggled", source_name) except Exception as exc: log.warning("toggle_mute_source(%r) failed: %s", source_name, exc) - # -- Sink inputs (per-app) ------------------------------------------------- - def set_app_volume(self, app_name: str, volume: float) -> None: - """Set volume for the sink input whose ``application.name`` or - ``application.process.binary`` contains *app_name* (case-insensitive). - - *volume* is clamped to 0.0–1.5. - """ volume = max(0.0, min(VOLUME_MAX, volume)) needle = app_name.lower() try: @@ -196,15 +206,21 @@ class PulseController: # ── Event handlers ───────────────────────────────────────────────────────────── def handle_knob( - knob_id: int, value: int, config: dict, pulse: PulseController + knob_id: int, + value: int, + config: dict, + pulse: PulseController, + ser: serial.Serial, + knob_norms: list[float], ) -> None: - """Dispatch a knob event according to the loaded configuration.""" + """Dispatch a knob event, update PulseAudio, then refresh the LEDs.""" knob_cfg = config.get("knobs", {}).get(str(knob_id)) if not knob_cfg: return action = knob_cfg.get("action", "sink_volume") target = knob_cfg.get("target", "default") + norm = knob_to_norm(value) if action == "sink_volume": vol = knob_to_volume(value) @@ -212,9 +228,8 @@ def handle_knob( log.info("Knob %d → sink %r = %.2f", knob_id, target, vol) elif action == "source_volume": - vol = knob_to_norm(value) - pulse.set_source_volume(target, vol) - log.info("Knob %d → source %r = %.2f", knob_id, target, vol) + pulse.set_source_volume(target, norm) + log.info("Knob %d → source %r = %.2f", knob_id, target, norm) elif action == "app_volume": vol = knob_to_volume(value) @@ -223,20 +238,18 @@ def handle_knob( elif action == "group_volume": vol = knob_to_volume(value) - targets: list[str] = knob_cfg.get("targets", []) - for t in targets: + for t in knob_cfg.get("targets", []): pulse.set_app_volume(t, vol) - log.info("Knob %d → group %s = %.2f", knob_id, targets, vol) + log.info("Knob %d → group %s = %.2f", knob_id, knob_cfg.get("targets"), vol) + + knob_norms[knob_id] = norm + send_leds(ser, all_led_colors(config, knob_norms)) def handle_button( button_id: int, action: str, config: dict, pulse: PulseController ) -> None: - """Dispatch a button event according to the loaded configuration. - - Only ``press`` events are acted upon; ``release`` events are silently - ignored. - """ + """Dispatch a button press event.""" if action != "press": return @@ -249,10 +262,8 @@ def handle_button( if btn_action == "mute_sink": pulse.toggle_mute_sink(target) - elif btn_action == "mute_source": pulse.toggle_mute_source(target) - elif btn_action == "command": try: subprocess.Popen(target, shell=True) # noqa: S602 @@ -265,13 +276,14 @@ def handle_button( def main() -> None: config = load_config() - port: str = config.get("port", "/dev/ttyACM0") - baud: int = config.get("baud", 115200) + port: str = config.get("port", "/dev/ttyACM0") + baud: int = config.get("baud", 115200) log.info("Turn Up daemon starting — %s @ %d baud", port, baud) - pulse = PulseController() - buf = bytearray() + pulse = PulseController() + knob_norms = [0.0] * NUM_KNOBS + buf = bytearray() def _shutdown(sig: int, _frame: object) -> None: log.info("Received signal %d — shutting down", sig) @@ -286,6 +298,8 @@ def main() -> None: with serial.Serial(port, baud, timeout=0.1) as ser: log.info("Connected to %s", port) buf.clear() + send_leds(ser, all_led_colors(config, knob_norms)) + while True: data = ser.read(64) if not data: @@ -294,9 +308,14 @@ def main() -> None: messages, buf = parse_messages(buf) for msg in messages: if msg["type"] == "knob": - handle_knob(msg["id"], msg["value"], config, pulse) + handle_knob( + msg["id"], msg["value"], + config, pulse, ser, knob_norms, + ) elif msg["type"] == "button": - handle_button(msg["id"], msg["action"], config, pulse) + handle_button( + msg["id"], msg["action"], config, pulse + ) except serial.SerialException as exc: log.warning("Serial error: %s — retrying in 3 s", exc) diff --git a/turnupd.py b/turnupd.py deleted file mode 100755 index ca1b069..0000000 --- a/turnupd.py +++ /dev/null @@ -1,310 +0,0 @@ -#!/usr/bin/env python3 -""" -turnupd — Turn Up mixer daemon for Linux - -Bridges a USB serial device (knobs + buttons) to PipeWire/PulseAudio via -pulsectl, mapping hardware inputs to per-sink, per-source, and per-app -volume control as well as mute toggles and arbitrary shell commands. -""" - -import logging -import signal -import subprocess -import sys -import time - -import pulsectl -import serial - -from config import load_config - -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(message)s", - datefmt="%H:%M:%S", -) -log = logging.getLogger("turnupd") - -# Maximum raw ADC value reported by the hardware. -KNOB_MAX: int = 1012 -# Maximum output volume multiplier (1.5 = 150 %). -VOLUME_MAX: float = 1.5 - - -# ── Protocol parser ──────────────────────────────────────────────────────────── - -def parse_messages(buf: bytearray) -> tuple[list[dict], bytearray]: - """Parse framed messages out of *buf* and return ``(messages, remainder)``. - - Supported frame formats (all delimited by ``0xFE`` … ``0xFF``): - - +-----------+----------------------------+ - | Heartbeat | ``FE 02 FF`` | - +-----------+----------------------------+ - | Button | ``FE 06/07 FF`` | - +-----------+----------------------------+ - | Knob | ``FE 03 FF``| - +-----------+----------------------------+ - """ - messages: list[dict] = [] - i = 0 - while i < len(buf): - if buf[i] != 0xFE: - i += 1 - continue - - remaining = len(buf) - i - - # Heartbeat: FE 02 FF (3 bytes) - if remaining >= 3 and buf[i + 1] == 0x02 and buf[i + 2] == 0xFF: - messages.append({"type": "heartbeat"}) - i += 3 - - # Button press/release: FE 06/07 FF (4 bytes) - elif ( - remaining >= 4 - and buf[i + 1] in (0x06, 0x07) - and buf[i + 3] == 0xFF - ): - messages.append({ - "type": "button", - "action": "press" if buf[i + 1] == 0x06 else "release", - "id": buf[i + 2], - }) - i += 4 - - # Knob value: FE 03 FF (6 bytes) - elif ( - remaining >= 6 - and buf[i + 1] == 0x03 - and buf[i + 5] == 0xFF - ): - messages.append({ - "type": "knob", - "id": buf[i + 2], - "value": (buf[i + 3] << 8) | buf[i + 4], - }) - i += 6 - - else: - i += 1 - - return messages, bytearray(buf[i:]) - - -def knob_to_volume(value: int) -> float: - """Convert a raw knob value (0–``KNOB_MAX``) to a sink volume (0.0–1.5).""" - return round((value / KNOB_MAX) * VOLUME_MAX, 4) - - -def knob_to_norm(value: int) -> float: - """Convert a raw knob value (0–``KNOB_MAX``) to a normalised float (0.0–1.0).""" - return round(value / KNOB_MAX, 4) - - -# ── PulseAudio / PipeWire controller ────────────────────────────────────────── - -class PulseController: - """Thin wrapper around :class:`pulsectl.Pulse` for volume and mute control.""" - - def __init__(self) -> None: - self._pulse = pulsectl.Pulse("turnupd") - - def close(self) -> None: - self._pulse.close() - - # -- Sink (output) --------------------------------------------------------- - - def set_sink_volume(self, sink_name: str, volume: float) -> None: - """Set output volume for *sink_name*. *volume* is clamped to 0.0–1.5.""" - volume = max(0.0, min(VOLUME_MAX, volume)) - try: - if sink_name == "default": - info = self._pulse.server_info() - sink = self._pulse.get_sink_by_name(info.default_sink_name) - else: - sink = self._pulse.get_sink_by_name(sink_name) - self._pulse.volume_set_all_chans(sink, volume) - except Exception as exc: - log.warning("set_sink_volume(%r) failed: %s", sink_name, exc) - - def toggle_mute_sink(self, sink_name: str) -> None: - """Toggle the mute state of *sink_name*.""" - try: - if sink_name == "default": - info = self._pulse.server_info() - sink = self._pulse.get_sink_by_name(info.default_sink_name) - else: - sink = self._pulse.get_sink_by_name(sink_name) - new_mute = not sink.mute - self._pulse.mute(sink, new_mute) - log.info("Sink %r mute → %s", sink_name, new_mute) - except Exception as exc: - log.warning("toggle_mute_sink(%r) failed: %s", sink_name, exc) - - # -- Source (input / mic) -------------------------------------------------- - - def set_source_volume(self, source_name: str, volume: float) -> None: - """Set mic/input volume for *source_name*. *volume* is clamped to 0.0–1.0.""" - volume = max(0.0, min(1.0, volume)) - try: - if source_name == "default": - info = self._pulse.server_info() - source = self._pulse.get_source_by_name(info.default_source_name) - else: - source = self._pulse.get_source_by_name(source_name) - self._pulse.volume_set_all_chans(source, volume) - except Exception as exc: - log.warning("set_source_volume(%r) failed: %s", source_name, exc) - - def toggle_mute_source(self, source_name: str) -> None: - """Toggle the mute state of *source_name*.""" - try: - if source_name == "default": - info = self._pulse.server_info() - source = self._pulse.get_source_by_name(info.default_source_name) - else: - source = self._pulse.get_source_by_name(source_name) - new_mute = not source.mute - self._pulse.mute(source, new_mute) - log.info("Source %r mute → %s", source_name, new_mute) - except Exception as exc: - log.warning("toggle_mute_source(%r) failed: %s", source_name, exc) - - # -- Sink inputs (per-app) ------------------------------------------------- - - def set_app_volume(self, app_name: str, volume: float) -> None: - """Set volume for the sink input whose ``application.name`` or - ``application.process.binary`` contains *app_name* (case-insensitive). - - *volume* is clamped to 0.0–1.5. - """ - volume = max(0.0, min(VOLUME_MAX, volume)) - needle = app_name.lower() - try: - for inp in self._pulse.sink_input_list(): - name = inp.proplist.get("application.name", "") - binary = inp.proplist.get("application.process.binary", "") - if needle in name.lower() or needle in binary.lower(): - self._pulse.volume_set_all_chans(inp, volume) - return - log.debug("App %r not found in sink inputs", app_name) - except Exception as exc: - log.warning("set_app_volume(%r) failed: %s", app_name, exc) - - -# ── Event handlers ───────────────────────────────────────────────────────────── - -def handle_knob( - knob_id: int, value: int, config: dict, pulse: PulseController -) -> None: - """Dispatch a knob event according to the loaded configuration.""" - knob_cfg = config.get("knobs", {}).get(str(knob_id)) - if not knob_cfg: - return - - action = knob_cfg.get("action", "sink_volume") - target = knob_cfg.get("target", "default") - - if action == "sink_volume": - vol = knob_to_volume(value) - pulse.set_sink_volume(target, vol) - log.info("Knob %d → sink %r = %.2f", knob_id, target, vol) - - elif action == "source_volume": - vol = knob_to_norm(value) - pulse.set_source_volume(target, vol) - log.info("Knob %d → source %r = %.2f", knob_id, target, vol) - - elif action == "app_volume": - vol = knob_to_volume(value) - pulse.set_app_volume(target, vol) - log.info("Knob %d → app %r = %.2f", knob_id, target, vol) - - elif action == "group_volume": - vol = knob_to_volume(value) - targets: list[str] = knob_cfg.get("targets", []) - for t in targets: - pulse.set_app_volume(t, vol) - log.info("Knob %d → group %s = %.2f", knob_id, targets, vol) - - -def handle_button( - button_id: int, action: str, config: dict, pulse: PulseController -) -> None: - """Dispatch a button event according to the loaded configuration. - - Only ``press`` events are acted upon; ``release`` events are silently - ignored. - """ - if action != "press": - return - - btn_cfg = config.get("buttons", {}).get(str(button_id)) - if not btn_cfg: - return - - btn_action = btn_cfg.get("action", "") - target = btn_cfg.get("target", "default") - - if btn_action == "mute_sink": - pulse.toggle_mute_sink(target) - - elif btn_action == "mute_source": - pulse.toggle_mute_source(target) - - elif btn_action == "command": - try: - subprocess.Popen(target, shell=True) # noqa: S602 - log.info("Button %d → command %r", button_id, target) - except Exception as exc: - log.warning("Button %d command failed: %s", button_id, exc) - - -# ── Main loop ────────────────────────────────────────────────────────────────── - -def main() -> None: - config = load_config() - port: str = config.get("port", "/dev/ttyACM0") - baud: int = config.get("baud", 115200) - - log.info("Turn Up daemon starting — %s @ %d baud", port, baud) - - pulse = PulseController() - buf = bytearray() - - def _shutdown(sig: int, _frame: object) -> None: - log.info("Received signal %d — shutting down", sig) - pulse.close() - sys.exit(0) - - signal.signal(signal.SIGINT, _shutdown) - signal.signal(signal.SIGTERM, _shutdown) - - while True: - try: - with serial.Serial(port, baud, timeout=0.1) as ser: - log.info("Connected to %s", port) - buf.clear() - while True: - data = ser.read(64) - if not data: - continue - buf.extend(data) - messages, buf = parse_messages(buf) - for msg in messages: - if msg["type"] == "knob": - handle_knob(msg["id"], msg["value"], config, pulse) - elif msg["type"] == "button": - handle_button(msg["id"], msg["action"], config, pulse) - - except serial.SerialException as exc: - log.warning("Serial error: %s — retrying in 3 s", exc) - time.sleep(3) - except Exception as exc: # noqa: BLE001 - log.error("Unexpected error: %s — retrying in 3 s", exc) - time.sleep(3) - - -if __name__ == "__main__": - main()