Radxa_MIXER/scripts/input_controls.py

"""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
import time

CONFIG = Path('/etc/radxa-audio/mixer.json')
MEMORY = Path('/etc/radxa-audio/input-memory.json')
PAIRING_UNTIL = Path('/run/radxa-audio/pairing-until')
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 properties(service, path, interface, names):
    """Read selected properties only.

    GetManagedObjects cannot be used for BlueZ: busctl fails to convert some
    advertisement data (e.g. Apple ManufacturerData a{qv}) to JSON, which
    would hide every device.
    """
    out = command('busctl', '--system', '--json=short', 'get-property', service, path, interface, *names)
    return {name: json.loads(line)['data'] for name, line in zip(names, out.splitlines())}


def bluez_paths(pattern):
    return [p for p in command('busctl', '--system', 'tree', '--list', 'org.bluez').split()
            if re.fullmatch(pattern, p)]


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:
        for path, props in pcms().items():
            if not is_input(props): continue
            dev = props['Device']; key = address(dev)
            # A missing name must not hide a working input.
            try: name = properties('org.bluez', dev, 'org.bluez.Device1', ['Alias'])['Alias']
            except (ValueError, KeyError, subprocess.TimeoutExpired): name = key
            volume, muted = decode_volume(int(props['Volume']))
            result.append({'id':key,'name':name,'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 = []
    base = ['Address', 'Alias', 'Paired', 'Bonded', 'Connected']
    for path in bluez_paths(r'/org/bluez/hci\d+/dev_[0-9A-F_]{17}'):
        try:
            try: props = properties('org.bluez', path, 'org.bluez.Device1', base + ['UUIDs'])
            # UUIDs is absent until services are known; devices may also vanish mid-read.
            except ValueError: props = properties('org.bluez', path, 'org.bluez.Device1', base)
        except ValueError: 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 pairing():
    paths = bluez_paths(r'/org/bluez/hci\d+')
    adapter = properties('org.bluez', paths[0], 'org.bluez.Adapter1', ['Discoverable', 'Alias']) if paths else {}
    on = bool(adapter.get('Discoverable'))
    remaining = 0
    try: remaining = max(0, round(float(PAIRING_UNTIL.read_text()) - time.time()))
    except (FileNotFoundError, ValueError): pass
    return {'discoverable': on, 'remaining': remaining if on else 0, 'alias': adapter.get('Alias', '')}


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))
    try: pair = pairing()
    except Exception as e: pair={'discoverable':False,'remaining':0,'alias':''}; issues.append(str(e))
    return {'config':config,'inputs':inputs(),'bluetooth':bluetooth,'dacs':dac,'pairing':pair,'issues':issues}