diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..3cb8471 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,43 @@ +# Changelog + +All notable changes to this project are documented in this file. + +## [Unreleased] + +### Added +- **`active_window_volume`** knob action — controls the volume of whatever + application currently has window focus. No `target` is needed; the focused + window is resolved dynamically on every turn. +- **`mute_active_window`** button action — toggles mute for whatever + application currently has window focus. +- **`mute_group`** button action — toggles mute for multiple applications at + once (`targets = ["app1", "app2"]`), syncing every matched stream to the + same mute state so repeated presses can't leave the group half-muted. +- **`binary:` needle syntax** for `app_volume`/`group_volume`/ + `app`-matching button targets. Some applications (generic Electron apps + that don't set their own PipeWire identity, e.g. Feishin) report the same + `application.name` as an unrelated app and are only distinguishable by + `application.process.binary`. A `binary:`-prefixed needle matches that + field exactly instead of substring-matching name-or-binary, avoiding + cross-talk between knobs/groups that would otherwise both match the + generic name. +- New dependency: `kdotool` (AUR), required for focused-window detection + under KWin/Wayland. + +### Changed +- `PulseController.set_app_volume()` / `get_app_volume_norm()` now share a + single `stream_matches()` helper (in `audio.py`) instead of duplicating + the name/binary substring check inline, and skip the MPRIS/`playerctl` + lookup for `binary:`-prefixed needles (no meaningful player name in that + case). +- `reapply_app_volumes()`'s periodic PA-correction loop uses the same + `stream_matches()` helper, fixing a bug where two knobs/groups whose + needles both matched a generic app name (e.g. "chromium" matching both the + real browser and a bare-Electron app reporting as "Chromium") would fight + over that stream's volume every ~1s. +- Web UI's TOML serializer (`ui/server.py`) now writes `targets = [...]` for + `mute_group` buttons, mirroring how knobs already serialize + `group_volume`. + +## [2.2.2] - previous release +See git history prior to this file's introduction. diff --git a/PKGBUILD b/PKGBUILD index 5b28663..6d4083c 100644 --- a/PKGBUILD +++ b/PKGBUILD @@ -15,6 +15,7 @@ depends=( 'uvicorn' 'pipewire-pulse' 'playerctl' + 'kdotool' ) makedepends=( 'python-build' diff --git a/README.md b/README.md index 081e970..ca5f381 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,13 @@ The daemon expects a USB-serial device speaking a simple binary protocol: Tested with an RP2040-based board. Any microcontroller that enumerates as a USB-CDC serial port and speaks the protocol above will work. +## Credits + +Originally created by [Sean Doran](https://github.com/sean351) — +[sean351/turn-up-arch](https://github.com/sean351/turn-up-arch). This copy +tracks a personal fork with local additions (focused-window volume/mute +control, app-group muting, etc.). + ## License MIT — see [LICENSE](LICENSE). diff --git a/src/turnup/audio.py b/src/turnup/audio.py index b7b5390..44b90e9 100644 --- a/src/turnup/audio.py +++ b/src/turnup/audio.py @@ -30,6 +30,29 @@ log = logging.getLogger("turnupd") # Imported by callers that need the ceiling constant. VOLUME_MAX: float = 1.0 +# Prefix for needles that must match application.process.binary *exactly* +# rather than as a case-insensitive substring of name-or-binary. Needed for +# generic Electron apps (e.g. Feishin) that don't set their own PipeWire +# application.name and so report the same "Chromium" name as the real +# browser — only the binary field ("electron" vs "chromium") tells them +# apart, and a substring match against name would still collide. +BINARY_EXACT_PREFIX: str = "binary:" + + +def stream_matches(proplist: dict, needle: str) -> bool: + """Return ``True`` if a PA/PW stream's *proplist* matches *needle*. + + A needle of the form ``"binary:"`` matches only when + ``application.process.binary`` equals ```` exactly (case-insensitive). + Any other needle is a case-insensitive substring match against + ``application.name`` OR ``application.process.binary``, as before. + """ + binary = proplist.get("application.process.binary", "").lower() + if needle.startswith(BINARY_EXACT_PREFIX): + return binary == needle[len(BINARY_EXACT_PREFIX):] + name = proplist.get("application.name", "").lower() + return needle in name or needle in binary + # ── MPRIS2 controller ────────────────────────────────────────────────────────── @@ -110,6 +133,49 @@ class MPRISController: return ok +# ── Active-window controller ──────────────────────────────────────────────────── + +class ActiveWindowController: + """Uses *kdotool* to read the currently focused window's app id (KWin/Wayland). + + Caches the result for ``_CACHE_TTL`` seconds so a fast knob turn (which can + generate 20-50 ADC samples in quick succession) doesn't spawn a subprocess + per sample. + """ + + _CACHE_TTL: float = 0.3 # seconds + + def __init__(self) -> None: + self._app: str | None = None + self._ts: float = 0.0 + self._lock = threading.Lock() + + def get_active_app(self) -> str | None: + """Return the lower-cased resource class of the focused window, or ``None``.""" + now = time.monotonic() + with self._lock: + if (now - self._ts) < self._CACHE_TTL: + return self._app + + app: str | None = None + try: + result = subprocess.run( + ["kdotool", "getactivewindow", "getwindowclassname"], + capture_output=True, + text=True, + timeout=1.0, + ) + if result.returncode == 0: + app = result.stdout.strip().lower() or None + except (FileNotFoundError, subprocess.TimeoutExpired) as exc: + log.debug("kdotool call failed: %s", exc) + + with self._lock: + self._app = app + self._ts = now + return app + + # ── PulseAudio / PipeWire controller ────────────────────────────────────────── class PulseController: @@ -231,6 +297,49 @@ class PulseController: except Exception as exc: log.warning("toggle_mute_source(%r) failed: %s", source_name, exc) + # ── App / group mute ───────────────────────────────────────────────────── + + def toggle_mute_app(self, app_name: str) -> None: + """Toggle mute for every sink input matching *app_name* (see :func:`stream_matches`).""" + needle = app_name.lower() + try: + matched = False + for inp in self._pulse.sink_input_list(): + if stream_matches(inp.proplist, needle): + self._pulse.mute(inp, not inp.mute) + matched = True + if matched: + log.info("App %r mute toggled", app_name) + else: + log.debug("App %r not found in sink inputs", app_name) + except Exception as exc: + log.warning("toggle_mute_app(%r) failed: %s", app_name, exc) + + def toggle_mute_group(self, app_names: list[str]) -> None: + """Toggle mute for every sink input matching any of *app_names*. + + All matched streams are set to the same mute state (the inverse of + whichever the first live match currently has) so repeated presses + can't leave the group in a mixed muted/unmuted state. + """ + needles = [a.lower() for a in app_names if a] + if not needles: + return + try: + matches = [ + inp for inp in self._pulse.sink_input_list() + if any(stream_matches(inp.proplist, n) for n in needles) + ] + if not matches: + log.debug("Group %s not found in sink inputs", app_names) + return + new_mute = not matches[0].mute + for inp in matches: + self._pulse.mute(inp, new_mute) + log.info("Group %s mute set to %s", app_names, new_mute) + except Exception as exc: + log.warning("toggle_mute_group(%r) failed: %s", app_names, exc) + # ── App volume (MPRIS-first, PA fallback) ───────────────────────────────── def set_app_volume(self, app_name: str, volume: float) -> None: @@ -242,7 +351,10 @@ class PulseController: # may coincidentally have an MPRIS player whose name matches the needle, but # their actual output volume lives on the PA stream. Always apply the PA # correction so both MPRIS-capable and PA-only apps are handled correctly. - if self._mpris: + # "binary:" needles disambiguate generic-Electron apps that share a + # PipeWire application.name with something else — they have no + # meaningful MPRIS player name, so skip that lookup entirely. + if self._mpris and not app_name.startswith(BINARY_EXACT_PREFIX): if self._mpris.set_volume(app_name, volume): log.debug("MPRIS set_volume: %r = %.4f", app_name, volume) @@ -251,9 +363,7 @@ class PulseController: found = False 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(): + if stream_matches(inp.proplist, needle): self._pulse.volume_set_all_chans(inp, volume) found = True if not found: @@ -287,8 +397,9 @@ class PulseController: def get_app_volume_norm(self, app_name: str) -> float | None: """Return the current app volume normalised to 0.0–1.0, or None if not found.""" - # Prefer MPRIS — more accurate for apps like Spotify. - if self._mpris: + # Prefer MPRIS — more accurate for apps like Spotify. Skip for + # "binary:" needles — see set_app_volume for why. + if self._mpris and not app_name.startswith(BINARY_EXACT_PREFIX): vol = self._mpris.get_volume(app_name) if vol is not None: return vol @@ -297,9 +408,7 @@ class PulseController: 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(): + if stream_matches(inp.proplist, needle): return min(1.0, inp.volume.value_flat / VOLUME_MAX) except Exception: pass diff --git a/src/turnup/config.py b/src/turnup/config.py index 768da10..1af43a2 100755 --- a/src/turnup/config.py +++ b/src/turnup/config.py @@ -62,10 +62,11 @@ high_color = [0, 255, 0] # green at 100 % # ── Knobs ───────────────────────────────────────────────────────────────────── # Available actions: -# sink_volume — output device volume (0–150 %) -# source_volume — mic / input volume (0–100 %) -# app_volume — single application; set target = "AppName" -# group_volume — multiple applications at once; set targets = ["app1", "app2"] +# sink_volume — output device volume (0–150 %) +# source_volume — mic / input volume (0–100 %) +# app_volume — single application; set target = "AppName" +# group_volume — multiple applications at once; set targets = ["app1", "app2"] +# active_window_volume — whatever app currently has window focus (no target needed) # # Each knob can optionally include a [knobs.N.led] block to override the # global LED colours for that specific knob. @@ -97,9 +98,11 @@ target = "default" # ── Buttons ─────────────────────────────────────────────────────────────────── # Available actions: -# mute_sink — toggle output mute -# mute_source — toggle mic mute -# command — run an arbitrary shell command; set target = "command args" +# mute_sink — toggle output mute +# mute_source — toggle mic mute +# mute_active_window — toggle mute for whatever app currently has window focus +# mute_group — toggle mute for multiple apps at once; set targets = ["app1", "app2"] +# command — run an arbitrary shell command; set target = "command args" [buttons.0] # Button 0 — mute output @@ -128,10 +131,10 @@ target = "default" """ VALID_KNOB_ACTIONS: frozenset = frozenset( - {"sink_volume", "source_volume", "app_volume", "group_volume"} + {"sink_volume", "source_volume", "app_volume", "group_volume", "active_window_volume"} ) VALID_BUTTON_ACTIONS: frozenset = frozenset( - {"mute_sink", "mute_source", "command"} + {"mute_sink", "mute_source", "command", "mute_active_window", "mute_group"} ) VALID_LED_MODES: frozenset = frozenset({"volume", "static", "off"}) diff --git a/src/turnup/turnupd.py b/src/turnup/turnupd.py index 3fbe48a..5aabb4f 100755 --- a/src/turnup/turnupd.py +++ b/src/turnup/turnupd.py @@ -21,7 +21,13 @@ import time import serial -from turnup.audio import VOLUME_MAX, MPRISController, PulseController +from turnup.audio import ( + VOLUME_MAX, + ActiveWindowController, + MPRISController, + PulseController, + stream_matches, +) from turnup.config import DEFAULT_CONFIG_PATH, get_knob_led_cfg, get_led_color, load_config logging.basicConfig( @@ -231,6 +237,8 @@ def reapply_app_volumes(config: dict, pulse: PulseController, knob_norms: list[f mpris = pulse._mpris if mpris: for app_name, vol in app_volumes.items(): + if app_name.startswith("binary:"): + continue # no meaningful MPRIS player name — see audio.py if mpris.set_volume(app_name, vol): log.debug("reapply MPRIS: %r → %.4f", app_name, vol) @@ -238,10 +246,8 @@ def reapply_app_volumes(config: dict, pulse: PulseController, knob_norms: list[f # PA-only apps (Brave, Discord, Electron) are not silently skipped. try: for inp in pulse._pulse.sink_input_list(): - name = inp.proplist.get("application.name", "").lower() - binary = inp.proplist.get("application.process.binary", "").lower() for needle, vol in app_volumes.items(): - if needle in name or needle in binary: + if stream_matches(inp.proplist, needle): current = inp.volume.value_flat if abs(current - vol) > 0.01: pulse._pulse.volume_set_all_chans(inp, vol) @@ -267,6 +273,7 @@ def handle_knob( knob_norms: list[float], last_led_colors: list[tuple[int, int, int]], last_knob_event: list[float], + active_window: ActiveWindowController, ) -> None: """Dispatch a knob event, update PulseAudio, then refresh the LEDs. @@ -307,6 +314,13 @@ def handle_knob( pulse.set_app_volume(t, vol) log.info("Knob %d → group %s = %.2f", knob_id, knob_cfg.get("targets"), vol) + elif action == "active_window_volume": + vol = knob_to_volume(value) + app = active_window.get_active_app() + if app: + pulse.set_app_volume(app, vol) + log.info("Knob %d → active window %r = %.2f", knob_id, app, vol) + knob_norms[knob_id] = norm last_knob_event[0] = time.monotonic() @@ -321,7 +335,11 @@ def handle_knob( def handle_button( - button_id: int, action: str, config: dict, pulse: PulseController + button_id: int, + action: str, + config: dict, + pulse: PulseController, + active_window: ActiveWindowController, ) -> None: """Dispatch a button press event.""" if action != "press": @@ -338,6 +356,12 @@ def handle_button( pulse.toggle_mute_sink(target) elif btn_action == "mute_source": pulse.toggle_mute_source(target) + elif btn_action == "mute_active_window": + app = active_window.get_active_app() + if app: + pulse.toggle_mute_app(app) + elif btn_action == "mute_group": + pulse.toggle_mute_group(btn_cfg.get("targets", [])) elif btn_action == "command": try: subprocess.Popen(target, shell=True) # noqa: S602 @@ -355,8 +379,9 @@ def main() -> None: log.info("Turn Up daemon starting — %s @ %d baud", port, baud) - mpris = MPRISController() - pulse = PulseController(mpris) + mpris = MPRISController() + pulse = PulseController(mpris) + active_window = ActiveWindowController() pulse.start_watching() knob_norms = init_knob_norms(config, pulse) buf = bytearray() @@ -407,10 +432,11 @@ def main() -> None: msg["id"], msg["value"], config, pulse, ser, knob_norms, last_led_colors, last_knob_event, + active_window, ) elif msg["type"] == "button": handle_button( - msg["id"], msg["action"], config, pulse + msg["id"], msg["action"], config, pulse, active_window ) elif msg["type"] == "heartbeat": new_colors = all_led_colors(config, knob_norms) diff --git a/src/turnup/ui/server.py b/src/turnup/ui/server.py index 6a6cfd0..a4437af 100644 --- a/src/turnup/ui/server.py +++ b/src/turnup/ui/server.py @@ -86,9 +86,14 @@ def config_to_toml(cfg: dict) -> str: btn = buttons.get(str(i)) if not btn: continue + btn_action = btn.get("action", "mute_sink") lines.append(f"[buttons.{i}]") - lines.append(f'action = {_s(btn.get("action", "mute_sink"))}') - lines.append(f'target = {_s(btn.get("target", "default"))}') + lines.append(f'action = {_s(btn_action)}') + if btn_action == "mute_group": + tgts = btn.get("targets") or [] + lines.append(f'targets = [{", ".join(_s(t) for t in tgts)}]') + else: + lines.append(f'target = {_s(btn.get("target", "default"))}') lines.append("") return "\n".join(lines)