commit 002721c63c5a72f61bdc9878f88bc1d7c33ca4f5 Author: Sean Doran Date: Fri Feb 27 10:08:09 2026 -0500 Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..44c0ca0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,45 @@ +# Python +__pycache__/ +*.py[cod] +*.pyo +*.pyd +.Python + +# Distribution / packaging +dist/ +build/ +*.egg-info/ +*.egg +.eggs/ + +# Virtual environments +.venv/ +venv/ +env/ +ENV/ + +# pytest / coverage +.pytest_cache/ +.coverage +htmlcov/ +.tox/ + +# mypy / pyright +.mypy_cache/ +.pyright/ + +# Build artefacts (makepkg — these are created next to PKGBUILD) +pkg/ +*.tar.zst +*.tar.gz +*.tar.xz + +# User config (not tracked — see contrib/config.example.json) +config.json + +# IDE / editor +.vscode/ +.idea/ +*.swp +*.swo +*~ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..27191c5 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Sean Doran + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/PKGBUILD b/PKGBUILD new file mode 100644 index 0000000..3d1e511 --- /dev/null +++ b/PKGBUILD @@ -0,0 +1,48 @@ +# Maintainer: Sean Doran +pkgname=turnup +pkgver=0.1.0 +pkgrel=1 +pkgdesc="USB serial knob/button mixer daemon for PipeWire/PulseAudio on Linux" +arch=('any') +url="https://github.com/sean351/turn-up-arch" +license=('MIT') +depends=( + 'python>=3.10' + 'python-pyserial' + 'python-pulsectl' + 'pipewire-pulse' # or pulseaudio — provides the PulseAudio socket +) +makedepends=( + 'python-build' + 'python-installer' + 'python-hatchling' +) +optdepends=( + 'playerctl: media key support via button commands' +) +backup=() +source=("$pkgname-$pkgver.tar.gz::https://github.com/sean351/turn-up-arch/archive/refs/tags/v$pkgver.tar.gz") +sha256sums=('SKIP') # Replace with actual checksum before submitting to AUR + +build() { + cd "$pkgname-$pkgver" + python -m build --wheel --no-isolation +} + +package() { + cd "$pkgname-$pkgver" + + python -m installer --destdir="$pkgdir" dist/*.whl + + # Systemd user service + install -Dm644 contrib/turnupd.service \ + "$pkgdir/usr/lib/systemd/user/turnupd.service" + + # License + install -Dm644 LICENSE \ + "$pkgdir/usr/share/licenses/$pkgname/LICENSE" + + # Documentation + install -Dm644 README.md \ + "$pkgdir/usr/share/doc/$pkgname/README.md" +} diff --git a/README.md b/README.md new file mode 100644 index 0000000..bac0a30 --- /dev/null +++ b/README.md @@ -0,0 +1,131 @@ +# turnup + +A lightweight daemon that bridges a USB serial device (physical knobs and +buttons) to PipeWire/PulseAudio on Linux. Map each knob to per-sink, +per-source, or per-application volume; assign buttons to mute toggles or +arbitrary shell commands. + +## Requirements + +- Python 3.10+ +- [pyserial](https://pypi.org/project/pyserial/) +- [pulsectl](https://pypi.org/project/pulsectl/) +- PipeWire (with `pipewire-pulse`) or PulseAudio +- `playerctl` *(optional — for media key bindings)* + +## Installation + +### Arch Linux (AUR) + +```sh +yay -S turnup +``` + +Or manually: + +```sh +git clone https://aur.archlinux.org/turnup.git +cd turnup +makepkg -si +``` + +### From source + +```sh +git clone https://github.com/sean351/turn-up-arch.git +cd turn-up-arch +pip install . +``` + +### From source (pip) + +```sh +pip install . +``` + +## Configuration + +On first run `turnupd` writes a default config to +`~/.config/turnup/config.json`. Edit it to match your device layout. +See [`contrib/config.example.json`](contrib/config.example.json) for a +fully commented example. + +```jsonc +{ + "port": "/dev/ttyACM0", + "baud": 115200, + + "knobs": { + "0": { "action": "sink_volume", "target": "default" }, + "1": { "action": "group_volume", "targets": ["vlc", "spotify"] }, + "2": { "action": "app_volume", "target": "Brave" }, + "3": { "action": "source_volume","target": "default" } + }, + + "buttons": { + "0": { "action": "mute_sink", "target": "default" }, + "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" } + } +} +``` + +### Knob actions + +| Action | Description | +|---|---| +| `sink_volume` | Output device volume (0–150 %) | +| `source_volume` | Mic / input volume (0–100 %) | +| `app_volume` | Single application volume, matched by name or binary | +| `group_volume` | Multiple applications at once — use `"targets": [...]` | + +### Button actions + +| Action | Description | +|---|---| +| `mute_sink` | Toggle output mute | +| `mute_source` | Toggle mic mute | +| `command` | Run an arbitrary shell command | + +## Running as a service + +A systemd user service unit is included: + +```sh +# After install via AUR or pip: +systemctl --user enable --now turnupd.service +``` + +To view logs: + +```sh +journalctl --user -u turnupd -f +``` + +## Running manually + +```sh +turnupd +``` + +Pass a custom config path with the `TURNUP_CONFIG` environment variable +*(planned — currently edit `~/.config/turnup/config.json` directly)*. + +## Hardware + +The daemon expects a USB-serial device speaking a simple binary protocol: + +| Frame | Bytes | Description | +|---|---|---| +| Heartbeat | `FE 02 FF` | Keepalive | +| Button | `FE 06/07 FF` | `06` = press, `07` = release | +| Knob | `FE 03 FF` | 10-bit ADC value, big-endian | + +Tested with an RP2040-based board. Any microcontroller that enumerates as a +USB-CDC serial port and speaks the protocol above will work. + +## License + +MIT — see [LICENSE](LICENSE). diff --git a/config.py b/config.py new file mode 100644 index 0000000..fbb50f8 --- /dev/null +++ b/config.py @@ -0,0 +1,89 @@ +#!/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 new file mode 100644 index 0000000..8674356 --- /dev/null +++ b/contrib/config.example.json @@ -0,0 +1,22 @@ +{ + "port": "/dev/ttyACM0", + "baud": 115200, + + "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" } + }, + + "buttons": { + "0": { "action": "mute_sink", "target": "default" }, + "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" } + } +} + + diff --git a/contrib/turnupd.service b/contrib/turnupd.service new file mode 100644 index 0000000..a5fb1b5 --- /dev/null +++ b/contrib/turnupd.service @@ -0,0 +1,21 @@ +[Unit] +Description=Turn Up mixer daemon +Documentation=https://github.com/sean351/turn-up-arch +After=pipewire-pulse.service pulseaudio.service +Wants=pipewire-pulse.service + +[Service] +Type=simple +ExecStart=%h/.local/bin/turnupd +Restart=on-failure +RestartSec=5 +# Give the audio server a moment to start before connecting +ExecStartPre=/bin/sleep 2 + +# Logging +StandardOutput=journal +StandardError=journal +SyslogIdentifier=turnupd + +[Install] +WantedBy=default.target diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..30fcfae --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,35 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "turnup" +version = "0.1.0" +description = "USB serial knob/button mixer daemon for PipeWire/PulseAudio on Linux" +readme = "README.md" +license = { text = "MIT" } +requires-python = ">=3.10" +dependencies = [ + "pyserial>=3.5", + "pulsectl>=22.0", +] + +[project.scripts] +turnupd = "turnup.turnupd:main" + +[project.urls] +Homepage = "https://github.com/sean351/turn-up-arch" +Repository = "https://github.com/sean351/turn-up-arch" +Issues = "https://github.com/sean351/turn-up-arch/issues" + +[tool.hatch.build.targets.wheel] +packages = ["src/turnup"] + +[tool.hatch.build.targets.sdist] +include = [ + "src/", + "contrib/", + "README.md", + "LICENSE", + "pyproject.toml", +] diff --git a/src/turnup/__init__.py b/src/turnup/__init__.py new file mode 100644 index 0000000..8070061 --- /dev/null +++ b/src/turnup/__init__.py @@ -0,0 +1,3 @@ +"""Turn Up — USB serial mixer daemon for PipeWire/PulseAudio.""" + +__version__ = "0.1.0" diff --git a/src/turnup/config.py b/src/turnup/config.py new file mode 100644 index 0000000..fbb50f8 --- /dev/null +++ b/src/turnup/config.py @@ -0,0 +1,89 @@ +#!/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/src/turnup/turnupd.py b/src/turnup/turnupd.py new file mode 100755 index 0000000..d72fbba --- /dev/null +++ b/src/turnup/turnupd.py @@ -0,0 +1,310 @@ +#!/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 turnup.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() diff --git a/turnupd.py b/turnupd.py new file mode 100755 index 0000000..ca1b069 --- /dev/null +++ b/turnupd.py @@ -0,0 +1,310 @@ +#!/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()