auracaster/bumble-auracast
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Adds check if the volume is set correctly.

Browse Source
This commit is contained in:
Pbopbo
2026-04-08 11:50:50 +02:00
parent c3a36dc252
commit c63151dd50
1 changed files with 34 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/auracast/server/multicast_server.py
View File

@@ -3,6 +3,7 @@ TODO: in the future the multicaster objects should run in their own threads or e
"""
import os
import re
import logging as log
import json
from datetime import datetime
@@ -262,15 +263,46 @@ async def _set_adc_level(gain_db_left: float = 0.0, gain_db_right: float = 0.0)
)
stdout, stderr = await proc.communicate()
if proc.returncode != 0:
log.warning(
log.error(
"amixer ADC level command failed (rc=%s): %s",
proc.returncode,
(stderr or b"" ).decode(errors="ignore").strip(),
)
else:
log.info("amixer ADC level set successfully: %s", (stdout or b"" ).decode(errors="ignore").strip())
read_proc = await asyncio.create_subprocess_exec(
"amixer", "-c", "2", "sget", "ADC",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
read_stdout, read_stderr = await read_proc.communicate()
if read_proc.returncode != 0:
log.error(
"amixer ADC sget failed (rc=%s): %s",
read_proc.returncode,
(read_stderr or b"").decode(errors="ignore").strip(),
)
else:
sget_output = (read_stdout or b"").decode(errors="ignore")
actual = {}
for line in sget_output.splitlines():
for ch_key, ch_name in (("left", "Front Left"), ("right", "Front Right")):
if ch_name in line:
m = re.search(r'\[(-?\d+(?:\.\d+)?)dB\]', line)
if m:
actual[ch_key] = round(float(m.group(1)))
expected_left = int(gain_db_left)
expected_right = int(gain_db_right)
if actual.get("left") != expected_left or actual.get("right") != expected_right:
mismatch = (
f"ADC level mismatch after set: expected L={expected_left}dB R={expected_right}dB, "
f"got L={actual.get('left')}dB R={actual.get('right')}dB"
)
log.error(mismatch)
else:
log.info("ADC level set successfully: L=%sdB R=%sdB", expected_left, expected_right)
except Exception as e:
log.warning("Exception running amixer ADC level command: %s", e, exc_info=True)
log.error("Exception running amixer ADC level command: %s", e, exc_info=True)
async def _stop_all() -> bool: