Add web mixer with native input volume and output selection

scripts/input_controls.py

@@ -0,0 +1,167 @@
+"""Native input volume controls. Never writes a DAC/headphone output volume."""
+import json
+import math
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+import os
+
+CONFIG = Path('/etc/radxa-audio/mixer.json')
+MEMORY = Path('/etc/radxa-audio/input-memory.json')
+SPS = 'org.gnome.ShairportSync'
+SPS_PATH = '/org/gnome/ShairportSync'
+MAC = re.compile(r'(?:[0-9A-F]{2}:){5}[0-9A-F]{2}\Z')
+
+
+def command(*args):
+    p = subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=12)
+    if p.returncode:
+        raise ValueError(p.stderr.strip() or p.stdout.strip() or '장치 제어에 실패했습니다.')
+    return p.stdout
+
+
+def bus(*args):
+    result = json.loads(command('busctl', '--system', '--json=short', *args))
+    # Method replies contain a tuple; properties contain their direct value.
+    return result['data'][0] if args[0] == 'call' else result['data']
+
+
+def objects(service, path):
+    raw = bus('call', service, path, 'org.freedesktop.DBus.ObjectManager', 'GetManagedObjects')
+    return {p: {i: {k: v['data'] for k, v in props.items()} for i, props in interfaces.items()}
+            for p, interfaces in raw.items()}
+
+
+def pcms():
+    return {p: interfaces['org.bluealsa.PCM1'] for p, interfaces in objects('org.bluealsa', '/org/bluealsa').items()
+            if 'org.bluealsa.PCM1' in interfaces}
+
+
+def is_input(props):
+    return props.get('Mode') == 'source' and props.get('Transport') == 'A2DP-sink'
+
+
+def memory():
+    try: return json.loads(MEMORY.read_text())
+    except FileNotFoundError: return {}
+
+
+def remember(key, percent):
+    data = memory(); data[key] = percent
+    MEMORY.parent.mkdir(parents=True, exist_ok=True)
+    fd, name = tempfile.mkstemp(dir=MEMORY.parent)
+    try:
+        with os.fdopen(fd, 'w') as f: json.dump(data, f)
+        os.chmod(name, 0o644); os.replace(name, MEMORY)
+    finally:
+        if os.path.exists(name): os.unlink(name)
+
+
+def address(device_path):
+    match = re.search(r'/dev_([0-9A-F_]{17})$', device_path)
+    if not match: raise ValueError('장비 주소를 읽을 수 없습니다.')
+    return match[1].replace('_', ':')
+
+
+def airplay():
+    props = bus('call', SPS, SPS_PATH, 'org.freedesktop.DBus.Properties', 'GetAll', 's', SPS)
+    props = {k: v['data'] for k,v in props.items()}
+    value = props['Volume']
+    if props.get('Active'):
+        try:
+            value = bus('get-property', SPS, SPS_PATH, SPS+'.RemoteControl', 'AirplayVolume')
+        except (ValueError, subprocess.TimeoutExpired):
+            pass
+    muted = value <= -100
+    percent = round((value + 30) / 30 * 100) if not muted else memory().get('airplay', 75)
+    return {'id':'airplay', 'name':'AirPlay', 'kind':'airplay', 'volume':max(0,min(100,percent)),
+            'muted':muted, 'playing':bool(props.get('Active')), 'available':True}
+
+
+def decode_volume(value):
+    levels = [(value >> 8) & 127, value & 127]
+    return round(sum(levels) / 254 * 100), bool(value & 0x8080)
+
+
+def encode_volume(percent, muted):
+    byte = round(percent * 127 / 100) | (128 if muted else 0)
+    return (byte << 8) | byte
+
+
+def inputs():
+    result = []
+    try: result.append(airplay())
+    except (ValueError, KeyError, subprocess.TimeoutExpired):
+        result.append({'id':'airplay','name':'AirPlay','kind':'airplay','volume':0,
+                       'muted':False,'playing':False,'available':False})
+    try:
+        devices = objects('org.bluez', '/')
+        for path, props in pcms().items():
+            if not is_input(props): continue
+            dev = props['Device']; key = address(dev)
+            d = devices.get(dev, {}).get('org.bluez.Device1', {})
+            volume, muted = decode_volume(int(props['Volume']))
+            result.append({'id':key,'name':d.get('Alias', key),'kind':'bluetooth','volume':volume,
+                           'muted':muted,'playing':bool(props.get('Running')),'available':True})
+    except (ValueError, KeyError, subprocess.TimeoutExpired): pass
+    return result
+
+
+def set_input(key, volume=None, mute=None):
+    if volume is not None and (type(volume) is not int or not 0 <= volume <= 100):
+        raise ValueError('입력 볼륨은 0~100의 정수여야 합니다.')
+    if mute is not None and type(mute) is not bool: raise ValueError('음소거 값이 잘못됐습니다.')
+    if key == 'airplay':
+        current = airplay()
+        percent = current['volume'] if volume is None else volume
+        muted = current['muted'] if mute is None else mute
+        if volume is not None: muted = percent == 0
+        if mute is True: remember(key, current['volume'])
+        if not muted and percent: remember(key, percent)
+        value = -144.0 if muted or percent == 0 else -30.0 + percent * 0.3
+        command('busctl','--system','--','set-property',SPS,SPS_PATH,SPS,'Volume','d',str(value))
+        return
+    key = str(key).upper()
+    if not MAC.fullmatch(key): raise ValueError('입력 장비 주소가 잘못됐습니다.')
+    matches = [(p, v) for p,v in pcms().items() if is_input(v) and address(v['Device']) == key]
+    if not matches: raise ValueError('이 장비는 연결된 Bluetooth 입력이 아닙니다.')
+    for path, props in matches:
+        old, old_mute = decode_volume(int(props['Volume']))
+        percent = old if volume is None else volume
+        muted = old_mute if mute is None else mute
+        if volume is not None: muted = percent == 0
+        # Only the validated source PCM is written; output PCMs cannot pass this guard.
+        command('busctl','--system','--','set-property','org.bluealsa',path,'org.bluealsa.PCM1','SoftVolume','b','true')
+        command('busctl','--system','--','set-property','org.bluealsa',path,'org.bluealsa.PCM1','Volume','q',
+                str(encode_volume(percent, muted)))
+
+
+def devices():
+    result = []
+    for path, interfaces in objects('org.bluez', '/').items():
+        props = interfaces.get('org.bluez.Device1')
+        if not props: continue
+        result.append({'address': props['Address'],'name': props.get('Alias',props['Address']),
+                       'bonded': bool(props.get('Bonded')), 'paired': bool(props.get('Paired')),'connected':bool(props.get('Connected')),
+                       'output': '0000110b-0000-1000-8000-00805f9b34fb' in props.get('UUIDs',[])})
+    return result
+
+
+def dacs():
+    result = []
+    for line in command('aplay','-l').splitlines():
+        match = re.match(r'card \d+: ([\w-]+) \[([^]]+)\], device (\d+): ([^[]+)',line)
+        if match and match[1] != 'Loopback':
+            result.append({'pcm':f'hw:CARD={match[1]},DEV={match[3]}','name':match[2]+' · '+match[4].strip()})
+    return result
+
+
+def state():
+    config = json.loads(CONFIG.read_text())
+    issues = []
+    try: bluetooth = devices()
+    except Exception as e: bluetooth=[]; issues.append(str(e))
+    try: dac = dacs()
+    except Exception as e: dac=[]; issues.append(str(e))
+    return {'config':config,'inputs':inputs(),'bluetooth':bluetooth,'dacs':dac,'issues':issues}