feat: initial release
Fixed volume steps for Bluetooth audio devices on PipeWire. Corrects the coarse internal volume grid of devices such as AirPods Pro (16 AVRCP steps, 6.25 % per swipe) to the desktop's own step size. The device grid is measured per device; the desktop step size is read from Plasma. English and German messages.
This commit is contained in:
Felitendo committed
commit
6b36b4f742
7 files changed
+841
No files matched your search
@@ -0,0 +1,2 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
@@ -0,0 +1,28 @@
|
||||
BSD 3-Clause License
|
||||
|
||||
Copyright (c) 2026, Felitendo
|
||||
|
||||
Redistribution and use in source and binary forms, with or without
|
||||
modification, are permitted provided that the following conditions are met:
|
||||
|
||||
1. Redistributions of source code must retain the above copyright notice, this
|
||||
list of conditions and the following disclaimer.
|
||||
|
||||
2. Redistributions in binary form must reproduce the above copyright notice,
|
||||
this list of conditions and the following disclaimer in the documentation
|
||||
and/or other materials provided with the distribution.
|
||||
|
||||
3. Neither the name of the copyright holder nor the names of its
|
||||
contributors may be used to endorse or promote products derived from
|
||||
this software without specific prior written permission.
|
||||
|
||||
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
|
||||
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
||||
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
|
||||
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
|
||||
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
|
||||
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
|
||||
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
|
||||
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
|
||||
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
@@ -0,0 +1,21 @@
|
||||
PREFIX ?= /usr
|
||||
DESTDIR ?=
|
||||
|
||||
BINDIR = $(DESTDIR)$(PREFIX)/bin
|
||||
UNITDIR = $(DESTDIR)$(PREFIX)/lib/systemd/user
|
||||
|
||||
.PHONY: install uninstall check
|
||||
|
||||
install:
|
||||
install -Dm755 bt-volume-step $(BINDIR)/bt-volume-step
|
||||
install -Dm644 systemd/bt-volume-step.service \
|
||||
$(UNITDIR)/bt-volume-step.service
|
||||
sed -i 's|^ExecStart=.*|ExecStart=$(PREFIX)/bin/bt-volume-step|' \
|
||||
$(UNITDIR)/bt-volume-step.service
|
||||
|
||||
uninstall:
|
||||
rm -f $(BINDIR)/bt-volume-step
|
||||
rm -f $(UNITDIR)/bt-volume-step.service
|
||||
|
||||
check:
|
||||
python3 tests/test_bt_volume_step.py
|
||||
@@ -0,0 +1,159 @@
|
||||
# bt-volume-step
|
||||
|
||||
Fixed volume steps for Bluetooth audio devices on PipeWire.
|
||||
|
||||
## The problem
|
||||
|
||||
Many Bluetooth devices carry a coarse internal volume grid. AirPods Pro expose
|
||||
only 16 AVRCP steps, so every swipe on the stem moves the system volume by
|
||||
6.25 %, and the on-screen display walks through
|
||||
|
||||
```
|
||||
6 · 13 · 19 · 25 · 31 · 38 · 44 · 50 · 56 · 63 · 69 · 75 · 81 · 88 · 94 · 100
|
||||
```
|
||||
|
||||
instead of clean multiples of five. A speaker's volume buttons do the same
|
||||
thing with whatever grid that speaker happens to use.
|
||||
|
||||
This is not a desktop misconfiguration — KDE's own volume step is already
|
||||
exactly 5 %. The grid lives in the device firmware, and it cannot be changed:
|
||||
the device sends *absolute* volume values over AVRCP, not increments.
|
||||
|
||||
## What this does
|
||||
|
||||
`bt-volume-step` watches for device-initiated volume changes, takes only their
|
||||
*direction* into account, and applies a clean step of its own instead. One
|
||||
press or swipe on the device becomes one step on your desktop's grid.
|
||||
|
||||
It works because such devices adopt a volume written back over AVRCP silently,
|
||||
without reporting it, so there is no feedback loop. As a side effect the
|
||||
device's internal position follows the corrected value, which keeps it from
|
||||
running into the end of its own scale.
|
||||
|
||||
## Requirements
|
||||
|
||||
- PipeWire with `pactl` (`libpulse`)
|
||||
- Python 3.9 or newer
|
||||
- Optional: `kreadconfig6` (KDE Plasma) to follow the desktop's own step size
|
||||
- Optional: `bluez-utils` for device names in `--show`
|
||||
|
||||
## Installation
|
||||
|
||||
### Arch Linux
|
||||
|
||||
```bash
|
||||
yay -S bt-volume-step
|
||||
```
|
||||
|
||||
### From source
|
||||
|
||||
```bash
|
||||
sudo make install
|
||||
```
|
||||
|
||||
`make install` honours `PREFIX` and `DESTDIR`; for a per-user install use
|
||||
`make install PREFIX=$HOME/.local`.
|
||||
|
||||
## Usage
|
||||
|
||||
Enable it for your user session:
|
||||
|
||||
```bash
|
||||
systemctl --user enable --now bt-volume-step
|
||||
```
|
||||
|
||||
That is the whole setup. Connect a Bluetooth device and use its volume
|
||||
control.
|
||||
|
||||
## Calibration
|
||||
|
||||
The daemon measures each device's grid on its own. **While a device's grid is
|
||||
unknown it does not intervene at all** — it only watches. Once the same jump
|
||||
has repeated three times, that jump is accepted as the device step and
|
||||
remembered, and corrections start from then on. In practice: press volume-up
|
||||
three or four times on a new device and it is set up.
|
||||
|
||||
Show what has been measured:
|
||||
|
||||
```bash
|
||||
bt-volume-step --show
|
||||
```
|
||||
|
||||
```
|
||||
file: /home/you/.local/state/bt-volume-step/devsteps.json
|
||||
AirPods Pro (30_0E_43_04_52_19) = 6.25 % (16 steps)
|
||||
```
|
||||
|
||||
Discard a measurement (all devices, or one MAC):
|
||||
|
||||
```bash
|
||||
bt-volume-step --reset
|
||||
bt-volume-step --reset 30_0E_43_04_52_19
|
||||
```
|
||||
|
||||
To keep the daemon from calibrating itself onto keyboard input, jumps that
|
||||
match the desktop's own step size are excluded from measurement, as are jumps
|
||||
below 1.5 % or above 20 %. A device whose grid happens to equal the desktop
|
||||
step therefore never calibrates, and is left alone.
|
||||
|
||||
## Configuration
|
||||
|
||||
On KDE Plasma the step size comes from *System Settings → Audio → Volume step*
|
||||
(`plasmaparc [General] VolumeStep`, read through `kreadconfig6` so KDE's
|
||||
configuration cascade applies). Changes take effect within two seconds, with
|
||||
no restart. On other desktops, or without `kreadconfig6`, it falls back to
|
||||
5 %.
|
||||
|
||||
Everything can be overridden through the environment — put these in a drop-in
|
||||
(`systemctl --user edit bt-volume-step`):
|
||||
|
||||
| Variable | Effect |
|
||||
|---|---|
|
||||
| `BT_VOL_STEP` | fixed step size in percent; ignores the desktop setting |
|
||||
| `BT_VOL_DEVSTEP` | fixed device grid in percent; skips measurement |
|
||||
| `BT_VOL_MAC` | only watch this device (MAC with `_` instead of `:`) |
|
||||
| `BT_VOL_MAX` | upper limit in percent (default 100) |
|
||||
|
||||
The daemon handles several connected Bluetooth devices at once, keeping state
|
||||
and calibration per device.
|
||||
|
||||
## Localisation
|
||||
|
||||
Messages follow `LC_ALL` / `LC_MESSAGES` / `LANG`. English and German are
|
||||
included; other languages fall back to English. To add one, extend
|
||||
`TRANSLATIONS` in the script — English strings are the keys.
|
||||
|
||||
## Limitations
|
||||
|
||||
**Foreign volume changes can be misread.** PipeWire events carry no
|
||||
information about where a change came from, so a jump that happens to match
|
||||
one or two device steps is indistinguishable from a button press. The daemon
|
||||
accepts at most two steps per event to keep that window narrow, but it cannot
|
||||
close it entirely.
|
||||
|
||||
**Devices that report their volume back are not supported.** The approach
|
||||
assumes a device accepts a written volume silently. One that echoes a snapped
|
||||
value back would oscillate. To check, set a volume and watch it for a few
|
||||
seconds:
|
||||
|
||||
```bash
|
||||
pactl set-sink-volume bluez_output.XX_XX_XX_XX_XX_XX.1 40%
|
||||
sleep 5 && pactl get-sink-volume bluez_output.XX_XX_XX_XX_XX_XX.1
|
||||
```
|
||||
|
||||
If the value drifts on its own, this tool is the wrong approach; disabling
|
||||
hardware volume (`bluez5.enable-hw-volume = false` in WirePlumber) is the
|
||||
alternative, at the cost of the device's own control.
|
||||
|
||||
## Tests
|
||||
|
||||
```bash
|
||||
make check
|
||||
```
|
||||
|
||||
Covers the decision logic and the measurement, including simulated devices
|
||||
with 8, 16 and 32 steps.
|
||||
|
||||
## License
|
||||
|
||||
BSD 3-Clause. See [LICENSE](LICENSE).
|
||||
Executable
+475
@@ -0,0 +1,475 @@
|
||||
#!/usr/bin/env python3
|
||||
# SPDX-License-Identifier: BSD-3-Clause
|
||||
# Copyright (c) 2026, Felitendo
|
||||
"""bt-volume-step - fixed volume steps for Bluetooth audio devices.
|
||||
|
||||
Many Bluetooth devices carry a coarse internal volume grid. AirPods Pro, for
|
||||
example, expose only 16 AVRCP steps, so every swipe on the stem moves the
|
||||
volume by 6.25 %: 6, 13, 19, 25, 31 ... Pressing volume-up on a speaker has
|
||||
the same effect with whatever grid that speaker uses.
|
||||
|
||||
This daemon watches for device-initiated volume changes, takes only their
|
||||
*direction* into account and applies a clean step of its own instead.
|
||||
|
||||
It works because such devices adopt a volume written over AVRCP silently,
|
||||
without reporting it back, so there is no feedback loop. A device that does
|
||||
report back would make the volume oscillate; check for that before use by
|
||||
setting a volume and watching it for a few seconds.
|
||||
|
||||
On KDE Plasma the step size is taken from the desktop's own setting (System
|
||||
Settings > Audio > "Volume step", stored as plasmaparc [General] VolumeStep).
|
||||
Changes there take effect immediately, without restarting the daemon.
|
||||
|
||||
The *device* grid is measured per device: while it is unknown the daemon
|
||||
keeps its hands off. Only once the same jump has repeated several times is it
|
||||
accepted as the device step and remembered.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
__version__ = "1.0.0"
|
||||
|
||||
# ---------------------------------------------------------------- i18n
|
||||
|
||||
def _detect_language():
|
||||
for var in ("LC_ALL", "LC_MESSAGES", "LANG"):
|
||||
value = os.environ.get(var)
|
||||
if value:
|
||||
return value.split(".")[0].split("_")[0].lower()
|
||||
return "en"
|
||||
|
||||
|
||||
LANG = _detect_language()
|
||||
|
||||
# English strings are the keys. To add a language, add another table and
|
||||
# extend TRANSLATIONS; untranslated strings fall back to English.
|
||||
TRANSLATIONS = {
|
||||
"de": {
|
||||
"Fixed volume steps for Bluetooth audio devices.":
|
||||
"Feste Lautstärkeschritte für Bluetooth-Audiogeräte.",
|
||||
"show measured device steps and exit":
|
||||
"gemessene Gerätestufen anzeigen und beenden",
|
||||
"discard calibration (without MAC: all devices)":
|
||||
"Kalibrierung verwerfen (ohne MAC: alle Geräte)",
|
||||
"step size from Plasma": "Schrittweite aus Plasma",
|
||||
"step size fixed at {step:g} %": "Schrittweite fest auf {step:g} %",
|
||||
", device step fixed at {devstep:g} %":
|
||||
", Gerätestufe fest auf {devstep:g} %",
|
||||
"step size: {step:g} %": "Schrittweite: {step:g} %",
|
||||
"cannot read the Plasma step size ({error}), using {fallback:g} %":
|
||||
"Plasma-Schrittweite nicht lesbar ({error}), nutze {fallback:g} %",
|
||||
"pactl list sinks failed: {error}":
|
||||
"pactl list sinks fehlgeschlagen: {error}",
|
||||
"cannot save the calibration: {error}":
|
||||
"Kalibrierung nicht speicherbar: {error}",
|
||||
"known device steps:": "bekannte Gerätestufen:",
|
||||
"watching {device} at {volume:.1f} %":
|
||||
"beobachte {device} bei {volume:.1f} %",
|
||||
"{device}: measured device step = {step:.2f} % ({count} steps) - active from now on":
|
||||
"{device}: Gerätestufe gemessen = {step:.2f} % ({count} Stufen) – ab jetzt aktiv",
|
||||
"{device}: {last:.1f} % -> device reported {cur:.1f} % -> set {target:.1f} %":
|
||||
"{device}: {last:.1f} % -> Gerät meldete {cur:.1f} % -> gesetzt {target:.1f} %",
|
||||
"file: {path}": "Datei: {path}",
|
||||
"no device step measured yet": "noch keine Gerätestufe gemessen",
|
||||
"{step:g} % ({count} steps)": "{step:g} % ({count} Stufen)",
|
||||
"all calibrations discarded": "alle Kalibrierungen verworfen",
|
||||
"calibration for {device} discarded":
|
||||
"Kalibrierung für {device} verworfen",
|
||||
"no calibration for {mac}": "keine Kalibrierung für {mac}",
|
||||
"restart the service: systemctl --user restart bt-volume-step":
|
||||
"Dienst neu starten: systemctl --user restart bt-volume-step",
|
||||
},
|
||||
}
|
||||
|
||||
_TABLE = TRANSLATIONS.get(LANG, {})
|
||||
|
||||
|
||||
def _(text):
|
||||
"""Translate a string into the current language, English as fallback."""
|
||||
return _TABLE.get(text, text)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- settings
|
||||
|
||||
MAC_FILTER = os.environ.get("BT_VOL_MAC", "")
|
||||
STEP_FIXED = os.environ.get("BT_VOL_STEP")
|
||||
DEVSTEP_FORCED = os.environ.get("BT_VOL_DEVSTEP")
|
||||
MAXV = float(os.environ.get("BT_VOL_MAX", "100"))
|
||||
|
||||
NORM = 65536.0 # PA_VOLUME_NORM
|
||||
NOISE = 1.0 # % below this: rounding noise, or our own write
|
||||
# Higher values make foreign changes harder to tell apart: with a fine device
|
||||
# grid (32 steps = 3.125 %) an ordinary 10 % jump already reads as three
|
||||
# button presses.
|
||||
MAXMULT = 2 # device steps a single event may combine
|
||||
TOL_REL = 0.35 # share of the device step tolerated when matching
|
||||
STEP_TTL = 2.0 # s the Plasma step size is cached for
|
||||
STEP_FALLBACK = 5.0 # % Plasma's own default
|
||||
|
||||
# --- measuring the device step ---
|
||||
DEV_MIN = 1.5 # % below this the device is fine-grained; leave it alone
|
||||
DEV_MAX = 20.0 # % above this it is not a volume button press
|
||||
# 0.12 covers the spread seen in practice: on AirPods Pro the jumps ranged
|
||||
# from 5.9 to 6.7 % around a step of 6.25 %.
|
||||
CLUSTER_REL = 0.12 # relative tolerance for two samples to count as equal
|
||||
NEED = 3 # matching samples required before locking the step
|
||||
SAMPLE_CAP = 12 # older samples expire
|
||||
|
||||
STORE = (
|
||||
Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local/state"))
|
||||
/ "bt-volume-step" / "devsteps.json"
|
||||
)
|
||||
|
||||
# C.UTF-8 rather than C: keeps pactl's messages English (we parse them)
|
||||
# without Qt tools such as kreadconfig6 complaining about a non-UTF-8 locale.
|
||||
ENV = {**os.environ, "LC_ALL": "C.UTF-8"}
|
||||
|
||||
_step_cache = (0.0, None) # (expires_at, value); None = never read
|
||||
|
||||
|
||||
def log(msg):
|
||||
print(msg, file=sys.stderr, flush=True)
|
||||
|
||||
|
||||
def half_up(x):
|
||||
"""Round half away from zero instead of Python's round-half-to-even.
|
||||
|
||||
Otherwise snap(65, 10) would be 60 while snap(75, 10) is 80 - that is,
|
||||
unpredictable for values sitting exactly between two grid points.
|
||||
"""
|
||||
return math.floor(x + 0.5)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Plasma
|
||||
|
||||
def current_step():
|
||||
"""The step size from Plasma, briefly cached.
|
||||
|
||||
kreadconfig6 rather than parsing plasmaparc directly, so that KDE's
|
||||
configuration cascade (/etc/xdg, locked-down settings) applies. On a
|
||||
desktop without kreadconfig6 the fallback is Plasma's own default.
|
||||
"""
|
||||
global _step_cache
|
||||
|
||||
if STEP_FIXED:
|
||||
return float(STEP_FIXED)
|
||||
|
||||
expires, value = _step_cache
|
||||
now = time.monotonic()
|
||||
if now < expires:
|
||||
return value
|
||||
|
||||
try:
|
||||
out = subprocess.run(
|
||||
["kreadconfig6", "--file", "plasmaparc", "--group", "General",
|
||||
"--key", "VolumeStep", "--default", str(STEP_FALLBACK)],
|
||||
capture_output=True, text=True, env=ENV, timeout=5,
|
||||
).stdout.strip()
|
||||
step = float(out)
|
||||
if not 0 < step <= 100:
|
||||
raise ValueError(f"implausible: {step}")
|
||||
except (subprocess.SubprocessError, ValueError, OSError, FileNotFoundError) as exc:
|
||||
log(_("cannot read the Plasma step size ({error}), using {fallback:g} %")
|
||||
.format(error=exc, fallback=STEP_FALLBACK))
|
||||
step = STEP_FALLBACK
|
||||
|
||||
if step != value:
|
||||
log(_("step size: {step:g} %").format(step=step))
|
||||
_step_cache = (now + STEP_TTL, step)
|
||||
return step
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- PipeWire
|
||||
|
||||
def bt_sinks():
|
||||
"""All Bluetooth outputs as (sink_name, mac, percent, description)."""
|
||||
try:
|
||||
out = subprocess.run(
|
||||
["pactl", "-f", "json", "list", "sinks"],
|
||||
capture_output=True, text=True, env=ENV, timeout=5,
|
||||
).stdout
|
||||
sinks = json.loads(out)
|
||||
except (subprocess.SubprocessError, json.JSONDecodeError, OSError) as exc:
|
||||
log(_("pactl list sinks failed: {error}").format(error=exc))
|
||||
return []
|
||||
|
||||
found = []
|
||||
for sink in sinks:
|
||||
name = sink.get("name", "")
|
||||
if not name.startswith("bluez_output."):
|
||||
continue
|
||||
parts = name.split(".")
|
||||
if len(parts) < 2:
|
||||
continue
|
||||
mac = parts[1]
|
||||
if MAC_FILTER and mac != MAC_FILTER:
|
||||
continue
|
||||
values = [ch["value"] for ch in sink.get("volume", {}).values()]
|
||||
if not values:
|
||||
continue
|
||||
description = (sink.get("description") or "").strip()
|
||||
found.append((name, mac, max(values) / NORM * 100.0, description))
|
||||
return found
|
||||
|
||||
|
||||
def set_volume(name, pct):
|
||||
raw = int(round(pct / 100.0 * NORM))
|
||||
subprocess.run(["pactl", "set-sink-volume", name, str(raw)], env=ENV, timeout=5)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- device names
|
||||
|
||||
def paired_names():
|
||||
"""MAC (underscore form) -> name for every paired BlueZ device.
|
||||
|
||||
Used by --show, which also lists devices that are not connected right now
|
||||
and therefore have no sink to read a description from.
|
||||
"""
|
||||
try:
|
||||
out = subprocess.run(
|
||||
["bluetoothctl", "devices"],
|
||||
capture_output=True, text=True, env=ENV, timeout=5,
|
||||
).stdout
|
||||
except (subprocess.SubprocessError, OSError):
|
||||
return {}
|
||||
|
||||
names = {}
|
||||
for line in out.splitlines():
|
||||
parts = line.split(maxsplit=2)
|
||||
if len(parts) == 3 and parts[0] == "Device":
|
||||
names[parts[1].replace(":", "_")] = parts[2].strip()
|
||||
return names
|
||||
|
||||
|
||||
def label(mac, name=None):
|
||||
"""Human-readable device label: 'Name (MAC)', or just the MAC."""
|
||||
return f"{name} ({mac})" if name else mac
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- calibration
|
||||
|
||||
def load_store():
|
||||
"""Read the calibration store, tolerating the flat v1 format."""
|
||||
try:
|
||||
raw = json.loads(STORE.read_text())
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return {}
|
||||
|
||||
store = {}
|
||||
for mac, value in raw.items():
|
||||
if isinstance(value, (int, float)):
|
||||
store[mac] = {"step": float(value)} # v1: bare number
|
||||
elif isinstance(value, dict) and "step" in value:
|
||||
store[mac] = value
|
||||
return store
|
||||
|
||||
|
||||
def save_store(store):
|
||||
try:
|
||||
STORE.parent.mkdir(parents=True, exist_ok=True)
|
||||
STORE.write_text(json.dumps(store, indent=2, sort_keys=True) + "\n")
|
||||
except OSError as exc:
|
||||
log(_("cannot save the calibration: {error}").format(error=exc))
|
||||
|
||||
|
||||
def calibrate(delta, samples, plasma_step):
|
||||
"""Record a sample; return the device step once it is certain, else None.
|
||||
|
||||
A volume button press on the device always produces the same jump. Deltas
|
||||
that match the desktop's own step size most likely come from the keyboard
|
||||
and are skipped - otherwise the daemon would calibrate itself onto them.
|
||||
"""
|
||||
mag = abs(delta)
|
||||
if not DEV_MIN <= mag <= DEV_MAX:
|
||||
return None
|
||||
if abs(mag - plasma_step) <= max(0.15, plasma_step * 0.02):
|
||||
return None
|
||||
|
||||
samples.append(mag)
|
||||
del samples[:-SAMPLE_CAP]
|
||||
|
||||
cluster = [s for s in samples if abs(s - mag) <= mag * CLUSTER_REL]
|
||||
if len(cluster) < NEED:
|
||||
return None
|
||||
return sum(cluster) / len(cluster)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- decision
|
||||
|
||||
def snap(pct, step):
|
||||
"""Round to the nearest multiple of the step size, clamped to 0..MAXV."""
|
||||
return min(MAXV, max(0.0, half_up(pct / step) * step))
|
||||
|
||||
|
||||
def decide(last, cur, step, devstep):
|
||||
"""Target volume for an observed change last -> cur, or None.
|
||||
|
||||
A button press on the device always moves the volume by a multiple of the
|
||||
device step. Only then do we intervene and replace the movement with our
|
||||
own grid; anything else is a foreign, absolute change and is accepted.
|
||||
"""
|
||||
delta = cur - last
|
||||
if abs(delta) < NOISE:
|
||||
return None
|
||||
|
||||
mult = half_up(abs(delta) / devstep)
|
||||
if 1 <= mult <= MAXMULT and abs(abs(delta) - mult * devstep) <= devstep * TOL_REL:
|
||||
direction = 1 if delta > 0 else -1
|
||||
target = snap(last, step) + direction * mult * step
|
||||
else:
|
||||
target = snap(cur, step)
|
||||
return min(MAXV, max(0.0, target))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- daemon
|
||||
|
||||
def run():
|
||||
store = load_store()
|
||||
|
||||
banner = (_("step size fixed at {step:g} %").format(step=float(STEP_FIXED))
|
||||
if STEP_FIXED else _("step size from Plasma"))
|
||||
if DEVSTEP_FORCED:
|
||||
banner += _(", device step fixed at {devstep:g} %").format(
|
||||
devstep=float(DEVSTEP_FORCED))
|
||||
log(banner)
|
||||
current_step()
|
||||
|
||||
if store:
|
||||
log(_("known device steps:") + " " + ", ".join(
|
||||
f"{label(mac, entry.get('name'))} = {entry['step']:g} %"
|
||||
for mac, entry in sorted(store.items())))
|
||||
|
||||
state = {} # mac -> {"last": float, "samples": [float]}
|
||||
|
||||
proc = subprocess.Popen(
|
||||
["pactl", "subscribe"],
|
||||
stdout=subprocess.PIPE, text=True, env=ENV, bufsize=1,
|
||||
)
|
||||
|
||||
for line in proc.stdout:
|
||||
if "on sink #" not in line:
|
||||
continue
|
||||
|
||||
plasma_step = current_step()
|
||||
seen = set()
|
||||
|
||||
for sink_name, mac, cur, description in bt_sinks():
|
||||
seen.add(mac)
|
||||
entry = store.setdefault(mac, {})
|
||||
if description and entry.get("name") != description:
|
||||
entry["name"] = description
|
||||
if "step" in entry:
|
||||
save_store(store)
|
||||
device = label(mac, entry.get("name"))
|
||||
|
||||
st = state.setdefault(mac, {"last": None, "samples": []})
|
||||
if st["last"] is None:
|
||||
st["last"] = cur
|
||||
log(_("watching {device} at {volume:.1f} %")
|
||||
.format(device=device, volume=cur))
|
||||
continue
|
||||
|
||||
devstep = float(DEVSTEP_FORCED) if DEVSTEP_FORCED else entry.get("step")
|
||||
|
||||
if devstep is None:
|
||||
# Not calibrated yet: let the change through, only measure.
|
||||
delta = cur - st["last"]
|
||||
if abs(delta) >= NOISE:
|
||||
found = calibrate(delta, st["samples"], plasma_step)
|
||||
if found:
|
||||
entry["step"] = round(found, 3)
|
||||
save_store(store)
|
||||
log(_("{device}: measured device step = {step:.2f} % "
|
||||
"({count} steps) - active from now on")
|
||||
.format(device=device, step=found,
|
||||
count=round(100 / found)))
|
||||
st["last"] = cur
|
||||
continue
|
||||
|
||||
target = decide(st["last"], cur, plasma_step, devstep)
|
||||
if target is None:
|
||||
st["last"] = cur
|
||||
continue
|
||||
|
||||
if abs(target - cur) >= 0.05:
|
||||
set_volume(sink_name, target)
|
||||
log(_("{device}: {last:.1f} % -> device reported {cur:.1f} % "
|
||||
"-> set {target:.1f} %")
|
||||
.format(device=device, last=st["last"], cur=cur, target=target))
|
||||
st["last"] = target
|
||||
|
||||
# Forget disconnected devices so they are re-read on reconnect.
|
||||
for mac in set(state) - seen:
|
||||
del state[mac]
|
||||
|
||||
return proc.wait()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- CLI
|
||||
|
||||
def cmd_show():
|
||||
store = load_store()
|
||||
names = paired_names()
|
||||
print(_("file: {path}").format(path=STORE))
|
||||
if not store:
|
||||
print(_("no device step measured yet"))
|
||||
return 0
|
||||
for mac, entry in sorted(store.items()):
|
||||
step = entry["step"]
|
||||
name = entry.get("name") or names.get(mac)
|
||||
print(f" {label(mac, name)} = "
|
||||
+ _("{step:g} % ({count} steps)").format(
|
||||
step=step, count=round(100 / step)))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_reset(target):
|
||||
store = load_store()
|
||||
if target == "*":
|
||||
store = {}
|
||||
print(_("all calibrations discarded"))
|
||||
elif target in store:
|
||||
name = store[target].get("name")
|
||||
del store[target]
|
||||
print(_("calibration for {device} discarded")
|
||||
.format(device=label(target, name)))
|
||||
else:
|
||||
print(_("no calibration for {mac}").format(mac=target))
|
||||
return 1
|
||||
save_store(store)
|
||||
print(_("restart the service: systemctl --user restart bt-volume-step"))
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
ap = argparse.ArgumentParser(
|
||||
prog="bt-volume-step",
|
||||
description=_("Fixed volume steps for Bluetooth audio devices."),
|
||||
)
|
||||
ap.add_argument("--show", action="store_true",
|
||||
help=_("show measured device steps and exit"))
|
||||
ap.add_argument("--reset", metavar="MAC", nargs="?", const="*",
|
||||
help=_("discard calibration (without MAC: all devices)"))
|
||||
ap.add_argument("--version", action="version",
|
||||
version=f"%(prog)s {__version__}")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
if args.show:
|
||||
return cmd_show()
|
||||
if args.reset:
|
||||
return cmd_reset(args.reset)
|
||||
return run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
@@ -0,0 +1,14 @@
|
||||
[Unit]
|
||||
Description=Fixed volume steps for Bluetooth audio devices
|
||||
Documentation=https://github.com/Felitendo/bt-volume-step
|
||||
After=pipewire-pulse.service
|
||||
Wants=pipewire-pulse.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/bt-volume-step
|
||||
Restart=always
|
||||
RestartSec=3
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
@@ -0,0 +1,142 @@
|
||||
#!/usr/bin/env python3
|
||||
# SPDX-License-Identifier: BSD-3-Clause
|
||||
"""Tests for the decision and measurement logic, without touching audio."""
|
||||
|
||||
from importlib.machinery import SourceFileLoader
|
||||
from pathlib import Path
|
||||
|
||||
SCRIPT = Path(__file__).resolve().parent.parent / "bt-volume-step"
|
||||
m = SourceFileLoader("btvs", str(SCRIPT)).load_module()
|
||||
|
||||
AIRPODS = 6.25 # 16 steps
|
||||
BOX32 = 3.125 # 32 steps
|
||||
BOX8 = 12.5 # 8 steps
|
||||
|
||||
fails = 0
|
||||
|
||||
|
||||
def check(ok, text):
|
||||
global fails
|
||||
if not ok:
|
||||
fails += 1
|
||||
print(f"{'ok ' if ok else 'FAIL'} {text}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- decide()
|
||||
# (description, step size, device step, last, cur, expected target)
|
||||
CASES = [
|
||||
# --- AirPods, 5 % steps: deltas taken from a real capture
|
||||
("swipe up from 60", 5, AIRPODS, 60.0, 66.1, 65.0),
|
||||
("swipe up from 45", 5, AIRPODS, 45.0, 51.2, 50.0),
|
||||
("swipe down from 65", 5, AIRPODS, 65.0, 59.1, 60.0),
|
||||
("swipe down from 35", 5, AIRPODS, 35.0, 28.3, 30.0),
|
||||
("swipe from an off-grid value", 5, AIRPODS, 65.35, 59.1, 60.0),
|
||||
# Keyboard: Plasma sets exactly +/-5 %, so target == cur, no double step
|
||||
("key up 55->60", 5, AIRPODS, 55.0, 60.0, 60.0),
|
||||
("key down 60->55", 5, AIRPODS, 60.0, 55.0, 55.0),
|
||||
# Foreign absolute changes must pass through
|
||||
("foreign set 60->50", 5, AIRPODS, 60.0, 50.0, 50.0),
|
||||
("foreign set 60->20", 5, AIRPODS, 60.0, 20.0, 20.0),
|
||||
# Ambiguous: -13 % is indistinguishable from two swipes (-12.5 %)
|
||||
("foreign jump 60->47 (ambiguous)", 5, AIRPODS, 60.0, 47.0, 50.0),
|
||||
# Noise / our own echo
|
||||
("noise 65->65.35", 5, AIRPODS, 65.0, 65.35, None),
|
||||
("no change", 5, AIRPODS, 60.0, 60.0, None),
|
||||
# Two fast swipes coalesced into one event
|
||||
("double swipe up", 5, AIRPODS, 60.0, 72.5, 70.0),
|
||||
("double swipe down", 5, AIRPODS, 60.0, 47.5, 50.0),
|
||||
# Limits
|
||||
("swipe up at maximum", 5, AIRPODS, 100.0, 100.0, None),
|
||||
("swipe up near maximum", 5, AIRPODS, 96.9, 100.0, 100.0),
|
||||
("swipe down at minimum", 5, AIRPODS, 0.0, 0.0, None),
|
||||
("swipe down near zero", 5, AIRPODS, 3.1, 0.0, 0.0),
|
||||
|
||||
# --- other desktop step sizes
|
||||
("10%: swipe up from 60", 10, AIRPODS, 60.0, 66.1, 70.0),
|
||||
("10%: swipe down from 60", 10, AIRPODS, 60.0, 53.5, 50.0),
|
||||
("10%: swipe from off-grid", 10, AIRPODS, 65.0, 71.1, 80.0),
|
||||
("10%: key up 60->70", 10, AIRPODS, 60.0, 70.0, 70.0),
|
||||
# 61 sits exactly between; half_up locks onto 62, one step down -> 60
|
||||
("2%: swipe down from 61", 2, AIRPODS, 61.0, 54.8, 60.0),
|
||||
("2%: swipe up from 60", 2, AIRPODS, 60.0, 66.1, 62.0),
|
||||
|
||||
# --- speaker with 32 steps (3.125 %)
|
||||
("32-step: key up from 60", 5, BOX32, 60.0, 63.1, 65.0),
|
||||
("32-step: key down from 60", 5, BOX32, 60.0, 56.9, 55.0),
|
||||
("32-step: double press up", 5, BOX32, 60.0, 66.3, 70.0),
|
||||
# A 10 % foreign jump is no multiple of 3.125 within MAXMULT -> pass through
|
||||
("32-step: foreign jump 60->50", 5, BOX32, 60.0, 50.0, 50.0),
|
||||
|
||||
# --- speaker with 8 steps (12.5 %)
|
||||
("8-step: key up from 50", 5, BOX8, 50.0, 62.5, 55.0),
|
||||
("8-step: key down from 50", 5, BOX8, 50.0, 37.5, 45.0),
|
||||
]
|
||||
|
||||
for desc, step, devstep, last, cur, want in CASES:
|
||||
got = m.decide(last, cur, float(step), devstep)
|
||||
ok = (got is None and want is None) or (
|
||||
got is not None and want is not None and abs(got - want) < 0.01
|
||||
)
|
||||
check(ok, f"{desc:34s} {last:6.2f} -> {cur:6.2f} want {want} got {got}")
|
||||
|
||||
print()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- calibrate()
|
||||
|
||||
def feed(deltas, plasma_step=5.0):
|
||||
"""Feed samples one by one; return the first locked-in device step."""
|
||||
samples = []
|
||||
for d in deltas:
|
||||
found = m.calibrate(d, samples, plasma_step)
|
||||
if found:
|
||||
return found
|
||||
return None
|
||||
|
||||
|
||||
got = feed([6.1, -6.4, 6.2])
|
||||
check(got is not None and abs(got - 6.25) < 0.3,
|
||||
f"AirPods spread calibrates -> {got}")
|
||||
|
||||
got = feed([5.9, 6.7, 6.1, 6.4])
|
||||
check(got is not None and abs(got - 6.25) < 0.4,
|
||||
f"wider spread calibrates -> {got}")
|
||||
|
||||
got = feed([3.1, 3.2, 3.1])
|
||||
check(got is not None and abs(got - 3.125) < 0.2,
|
||||
f"32-step speaker calibrates -> {got}")
|
||||
|
||||
got = feed([5.0, 5.0, 5.0, 5.0, 5.0])
|
||||
check(got is None, f"keyboard deltas (= desktop step) do NOT calibrate -> {got}")
|
||||
|
||||
got = feed([10.0, 10.0, 10.0], plasma_step=10.0)
|
||||
check(got is None, f"keyboard at a 10 % desktop step does NOT calibrate -> {got}")
|
||||
|
||||
got = feed([0.8, 0.79, 0.8, 0.8])
|
||||
check(got is None, f"fine-grained device (<1.5 %) is ignored -> {got}")
|
||||
|
||||
got = feed([25.0, 25.0, 25.0])
|
||||
check(got is None, f"jumps that are too large (>20 %) are ignored -> {got}")
|
||||
|
||||
got = feed([6.2, 13.0, 2.0, 6.1, 19.0, 6.3])
|
||||
check(got is not None and abs(got - 6.25) < 0.4,
|
||||
f"measurement finds the cluster despite noise -> {got}")
|
||||
|
||||
got = feed([6.2, 6.2])
|
||||
check(got is None, f"two samples are not enough yet -> {got}")
|
||||
|
||||
print()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- misc
|
||||
|
||||
check(m.label("AA_BB", "Speaker") == "Speaker (AA_BB)", "label with name")
|
||||
check(m.label("AA_BB", None) == "AA_BB", "label without name")
|
||||
check(m.half_up(6.5) == 7 and m.half_up(7.5) == 8, "half_up rounds consistently")
|
||||
check(abs(m.snap(65, 10) - 70) < 0.01 and abs(m.snap(75, 10) - 80) < 0.01,
|
||||
"snap is consistent for exact midpoints")
|
||||
|
||||
print()
|
||||
total = len(CASES) + 9 + 4
|
||||
print(f"{total - fails}/{total} passed")
|
||||
raise SystemExit(1 if fails else 0)
|
||||
Reference in new issue
Block a user