cdnutter
/
turn-up-arch
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add focused-window volume/mute control and app-group muting 

- New active_window_volume/mute_active_window actions using kdotool to
  resolve the focused window under KWin/Wayland.
- New mute_group button action for muting multiple apps at once.
- New "binary:<name>" needle syntax to disambiguate generic Electron apps
  that share a PipeWire application.name with an unrelated app; fixes a bug
  where two knobs/groups matching the same generic name would fight over
  a stream's volume via the periodic reapply loop.
- Add kdotool to PKGBUILD depends, credit original author in README.
main
Chris Nutter 2026-08-13 18:06:53 -07:00
parent 3c0c9adc83
commit 743c504918
7 changed files with 222 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
CHANGELOG.md Normal file
View File

@ -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:<name>` 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.

1
PKGBUILD
View File

@ -15,6 +15,7 @@ depends=(
'uvicorn'
'pipewire-pulse'
'playerctl'
'kdotool'
)
makedepends=(
'python-build'

7
README.md
View File

@ -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).

127
src/turnup/audio.py
View File

@ -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:<name>"`` matches only when
``application.process.binary`` equals ``<name>`` 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

7
src/turnup/config.py
View File

@ -66,6 +66,7 @@ high_color = [0, 255, 0] # green at 100 %
# 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.
@ -99,6 +100,8 @@ target = "default"
# Available actions:
# 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]
@ -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"})

38
src/turnup/turnupd.py
View File

@ -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
@ -357,6 +381,7 @@ def main() -> None:
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)

7
src/turnup/ui/server.py
View File

@ -86,8 +86,13 @@ 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'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("")