"""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}