jsahnandClaude Opus 5.5
655fbc1d29
The board stays hidden by default, so new phones could only pair via an SSH session. Add 'radxa-audio discoverable on|off' backed by a transient radxa-bt-pairing unit that auto-approves and trusts devices pairing in the window, expires on its own, and expose it in the web mixer with a countdown. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
529 lines
22 KiB
Python
Executable File
529 lines
22 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Radxa audio mixer manager. Debian 13 / systemd; Python standard library only."""
|
|
import argparse
|
|
import datetime
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
CONFIG = Path('/etc/radxa-audio/mixer.json')
|
|
BT_SERVICE = 'radxa-bt-output.service'
|
|
PAIRING_UNIT = 'radxa-bt-pairing.service'
|
|
PAIRING_UNTIL = Path('/run/radxa-audio/pairing-until')
|
|
MAC_RE = re.compile(r'(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}\Z')
|
|
PCM_RE = re.compile(r'hw:CARD=[A-Za-z0-9_-]+,DEV=[0-9]+(?:,SUBDEV=[0-9]+)?\Z')
|
|
|
|
|
|
def run(*args, check=True, capture=False, timeout=40):
|
|
return subprocess.run(args, check=check, text=True,
|
|
stdout=subprocess.PIPE if capture else None,
|
|
stderr=subprocess.STDOUT if capture else None, timeout=timeout)
|
|
|
|
|
|
def root():
|
|
if os.geteuid() != 0:
|
|
raise ValueError('sudo로 실행해 주세요.')
|
|
|
|
|
|
def validate(c):
|
|
if not isinstance(c.get('name'), str) or not c['name'].strip() or len(c['name'].encode()) > 80:
|
|
raise ValueError('AirPlay 이름은 1~80바이트여야 합니다.')
|
|
if any(ord(ch) < 32 for ch in c['name']):
|
|
raise ValueError('이름에는 제어 문자를 사용할 수 없습니다.')
|
|
if not PCM_RE.fullmatch(c.get('dac_pcm', '')):
|
|
raise ValueError('DAC 형식 예: hw:CARD=Audio,DEV=0')
|
|
if c.get('output') not in ('dac', 'bluetooth'):
|
|
raise ValueError('output은 dac 또는 bluetooth여야 합니다.')
|
|
addr = c.get('bluetooth_device', '')
|
|
if addr and not MAC_RE.fullmatch(addr):
|
|
raise ValueError('블루투스 주소 형식이 잘못됐습니다.')
|
|
if c['output'] == 'bluetooth' and not addr:
|
|
raise ValueError('먼저 블루투스 출력 장비 주소를 설정해 주세요.')
|
|
return c
|
|
|
|
|
|
def load():
|
|
if not CONFIG.exists():
|
|
raise ValueError('설치되지 않았습니다. install.sh를 먼저 실행해 주세요.')
|
|
return validate(json.loads(CONFIG.read_text()))
|
|
|
|
|
|
def save(path, text, mode=0o644):
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, temp = tempfile.mkstemp(dir=path.parent, prefix='.radxa-')
|
|
try:
|
|
with os.fdopen(fd, 'w') as f:
|
|
f.write(text)
|
|
os.chmod(temp, mode)
|
|
os.replace(temp, path)
|
|
finally:
|
|
if os.path.exists(temp):
|
|
os.unlink(temp)
|
|
|
|
|
|
def mixer(pcm, rate):
|
|
return f'''# Managed by radxa-audio; each input is attenuated by 3 dB.
|
|
pcm.radxa_dmix {{
|
|
type dmix
|
|
ipc_key 553001
|
|
ipc_key_add_uid false
|
|
ipc_perm 0660
|
|
ipc_gid audio
|
|
slave {{
|
|
pcm "{pcm}"
|
|
format S16_LE
|
|
rate {rate}
|
|
channels 2
|
|
period_time 0
|
|
period_size {rate // 100}
|
|
buffer_size {rate // 5}
|
|
}}
|
|
bindings {{ 0 0 1 1 }}
|
|
}}
|
|
pcm.radxa_mix {{
|
|
type plug
|
|
slave.pcm "radxa_dmix"
|
|
ttable {{ 0.0 0.707945784 1.1 0.707945784 }}
|
|
}}
|
|
'''
|
|
|
|
|
|
def templates(c):
|
|
validate(c)
|
|
dac = mixer(c['dac_pcm'], 44100)
|
|
bt = mixer('hw:CARD=Loopback,DEV=0,SUBDEV=0', 48000)
|
|
files = {
|
|
'/etc/radxa-audio/asound-dac.conf': dac,
|
|
'/etc/radxa-audio/asound-bluetooth.conf': bt,
|
|
'/etc/asound.conf': dac if c['output'] == 'dac' else bt,
|
|
str(CONFIG): json.dumps(c, ensure_ascii=False, indent=2) + '\n',
|
|
'/etc/shairport-sync.conf': f'''general = {{
|
|
name = {json.dumps(c['name'], ensure_ascii=False)};
|
|
service_type = "airplay2";
|
|
output_backend = "alsa";
|
|
audio_backend_buffer_desired_length_in_seconds = 0.15;
|
|
}};
|
|
alsa = {{
|
|
output_device = "radxa_mix";
|
|
output_format = "S16_LE";
|
|
}};
|
|
diagnostics = {{ statistics = "yes"; log_verbosity = 1; }};
|
|
''',
|
|
'/etc/systemd/system/bluealsa.service.d/10-radxa-sink.conf': '''[Service]
|
|
ExecStart=
|
|
ExecStart=/usr/bin/bluealsa -S -p a2dp-sink -p a2dp-source --initial-volume=50 --sbc-quality=medium
|
|
''',
|
|
'/etc/systemd/system/bluealsa-aplay.service.d/10-radxa-mixer.conf': '''[Unit]
|
|
Requires=bluealsa.service
|
|
After=bluealsa.service sound.target
|
|
[Service]
|
|
User=shairport-sync
|
|
Group=audio
|
|
PrivateUsers=false
|
|
RemoveIPC=false
|
|
ExecStart=
|
|
ExecStart=/usr/bin/bluealsa-aplay -S --pcm=radxa_mix --volume=software --pcm-period-time=10000 --pcm-buffer-time=200000
|
|
''',
|
|
'/etc/modules-load.d/radxa-audio-loopback.conf': 'snd-aloop\n',
|
|
'/etc/modprobe.d/radxa-audio-loopback.conf': 'options snd-aloop pcm_substreams=1\n',
|
|
}
|
|
if c.get('bluetooth_device'):
|
|
addr = c['bluetooth_device'].upper()
|
|
files['/etc/systemd/system/' + BT_SERVICE] = f'''[Unit]
|
|
Description=Radxa mixed audio to Bluetooth headphones or speaker
|
|
Requires=bluealsa.service
|
|
After=bluealsa.service sound.target systemd-modules-load.service
|
|
PartOf=bluealsa.service
|
|
StartLimitIntervalSec=0
|
|
[Service]
|
|
User=shairport-sync
|
|
Group=audio
|
|
ExecStartPre=-/usr/bin/bluetoothctl --timeout 8 connect {addr} a2dp-sink
|
|
ExecStart=/usr/bin/alsaloop -C hw:Loopback,1,0 -P bluealsa:DEV={addr},PROFILE=a2dp -c 2 -r 48000 -f S16_LE -t 200000 -S 4 -A 2
|
|
Restart=always
|
|
RestartSec=5
|
|
TimeoutStartSec=30
|
|
[Install]
|
|
WantedBy=multi-user.target
|
|
'''
|
|
return files
|
|
|
|
|
|
def snapshot(paths):
|
|
stamp = datetime.datetime.now().strftime('%Y%m%d-%H%M%S-%f')
|
|
directory = Path('/var/backups/radxa-audio') / stamp
|
|
directory.mkdir(parents=True, mode=0o700)
|
|
records = {}
|
|
for name in paths:
|
|
path = Path(name)
|
|
records[name] = os.path.lexists(path)
|
|
if records[name]:
|
|
dest = directory / name.lstrip('/')
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(path, dest, follow_symlinks=False)
|
|
save(directory / 'files.json', json.dumps(records, indent=2) + '\n', 0o600)
|
|
return directory
|
|
|
|
|
|
def restore_files(directory):
|
|
records = json.loads((directory / 'files.json').read_text())
|
|
for name, existed in records.items():
|
|
path = Path(name)
|
|
if existed:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.unlink(missing_ok=True)
|
|
shutil.copy2(directory / name.lstrip('/'), path, follow_symlinks=False)
|
|
else:
|
|
path.unlink(missing_ok=True)
|
|
|
|
|
|
def active(unit):
|
|
return run('systemctl', 'is-active', '--quiet', unit, check=False, capture=True).returncode == 0
|
|
|
|
|
|
def enabled(unit):
|
|
return run('systemctl', 'is-enabled', '--quiet', unit, check=False, capture=True).returncode == 0
|
|
|
|
|
|
def restart_inputs():
|
|
run('systemctl', 'restart', 'shairport-sync', 'bluealsa-aplay')
|
|
|
|
|
|
def check_dac(pcm):
|
|
parts = dict(item.split("=", 1) for item in pcm.removeprefix("hw:").split(","))
|
|
if not (Path("/proc/asound") / parts["CARD"] / ("pcm" + parts["DEV"] + "p")).exists():
|
|
raise ValueError("DAC 출력 장치가 없습니다. aplay -l에서 카드 ID를 확인해 주세요.")
|
|
|
|
|
|
def apply(c, initial=False, reconfigure=False):
|
|
root()
|
|
files = templates(c)
|
|
if reconfigure:
|
|
previous = load()
|
|
if any(c.get(k) != previous.get(k) for k in ('output', 'dac_pcm', 'bluetooth_device')):
|
|
raise ValueError('현재 출력 경로를 바꾸는 작업은 연결 확인이 필요합니다.')
|
|
if c["output"] == "dac" and not reconfigure:
|
|
check_dac(c["dac_pcm"])
|
|
if c['output'] == 'bluetooth' and not reconfigure:
|
|
# Preflight before changing a working DAC route.
|
|
run('modprobe', 'snd-aloop')
|
|
result = run('bluetoothctl', '--timeout', '10', 'connect',
|
|
c['bluetooth_device'], 'a2dp-sink', check=False, capture=True)
|
|
pcm_list = run('bluealsa-aplay', '-l', capture=True).stdout
|
|
playback = pcm_list.split('**** List of CAPTURE')[0]
|
|
if c['bluetooth_device'].upper() not in playback.upper():
|
|
raise ValueError('출력 장비의 A2DP 연결이 없습니다. 전원을 켜고 pair 명령을 먼저 실행해 주세요.\n' + result.stdout)
|
|
units = ['shairport-sync', 'bluealsa-aplay', BT_SERVICE, 'radxa-bose-output.service']
|
|
states = {u: (active(u), enabled(u)) for u in units}
|
|
backup = snapshot(files)
|
|
print('설정 백업:', backup, flush=True)
|
|
try:
|
|
stopping = [u for u in units if states[u][0]]
|
|
if stopping:
|
|
run('systemctl', 'stop', *stopping)
|
|
for path, text in files.items():
|
|
save(path, text)
|
|
run('systemctl', 'daemon-reload')
|
|
run('systemctl', 'disable', 'radxa-bose-output.service', check=False, capture=True)
|
|
if initial:
|
|
run('systemctl', 'enable', 'avahi-daemon', 'bluetooth', 'nqptp',
|
|
'shairport-sync', 'bluealsa', 'bluealsa-aplay')
|
|
run('systemctl', 'restart', 'bluetooth')
|
|
if run('systemctl', 'cat', 'aic-bluetooth.service', check=False, capture=True).returncode == 0:
|
|
run('systemctl', 'restart', 'aic-bluetooth')
|
|
run('systemctl', 'start', 'avahi-daemon', 'nqptp')
|
|
run('systemctl', 'restart', 'bluealsa')
|
|
if c['output'] == 'bluetooth':
|
|
run('systemctl', 'enable', BT_SERVICE)
|
|
run('systemctl', 'start', BT_SERVICE)
|
|
else:
|
|
run('systemctl', 'disable', BT_SERVICE, check=False, capture=True)
|
|
run('systemctl', 'start', 'shairport-sync', 'bluealsa-aplay')
|
|
required = ['shairport-sync', 'bluealsa-aplay']
|
|
if c['output'] == 'bluetooth' and not reconfigure:
|
|
required.append(BT_SERVICE)
|
|
if not all(active(u) for u in required):
|
|
raise ValueError('서비스 시작에 실패했습니다.')
|
|
alias = c['name'] + ' Bluetooth'
|
|
run('bluetoothctl', 'system-alias', alias, check=not reconfigure, capture=reconfigure)
|
|
print('출력:', c['output'], c.get('bluetooth_device', '') if c['output'] == 'bluetooth' else c['dac_pcm'])
|
|
except Exception:
|
|
run('systemctl', 'stop', *units, check=False)
|
|
restore_files(backup)
|
|
run('systemctl', 'daemon-reload', check=False)
|
|
for u, (was_active, was_enabled) in states.items():
|
|
run('systemctl', 'enable' if was_enabled else 'disable', u, check=False, capture=True)
|
|
if was_active:
|
|
run('systemctl', 'start', u, check=False)
|
|
print('설정을 이전 상태로 복구했습니다.', file=sys.stderr)
|
|
raise
|
|
|
|
|
|
def pair(addr, output=False, bond=False):
|
|
root()
|
|
if not MAC_RE.fullmatch(addr):
|
|
raise ValueError('장비 주소 형식이 잘못됐습니다.')
|
|
info = run('bluetoothctl', 'info', addr, check=False, capture=True).stdout
|
|
if 'Paired: yes' not in info or (bond and 'Bonded: yes' not in info):
|
|
controller = run('bluetoothctl', 'show', capture=True).stdout
|
|
was_pairable = 'Pairable: yes' in controller
|
|
if 'Paired: yes' in info:
|
|
# A temporary (unbonded) pairing blocks a new link key; start clean.
|
|
run('bluetoothctl', 'remove', addr, check=False, capture=True)
|
|
try:
|
|
agent_pair(addr)
|
|
finally:
|
|
if not was_pairable:
|
|
run('bluetoothctl', 'pairable', 'off', check=False)
|
|
info = run('bluetoothctl', 'info', addr, capture=True).stdout
|
|
if 'Paired: yes' not in info:
|
|
raise ValueError('페어링되지 않았습니다. 장비를 페어링 모드로 켜 주세요.')
|
|
if bond and 'Bonded: yes' not in info:
|
|
raise ValueError('영구 등록되지 않았습니다. 장비를 페어링 모드로 다시 켠 뒤 재시도해 주세요.')
|
|
run('bluetoothctl', 'trust', addr)
|
|
run('bluetoothctl', '--timeout', '10', 'connect', addr,
|
|
'a2dp-sink' if output else 'a2dp-source', check=False)
|
|
if output:
|
|
c = load()
|
|
c['bluetooth_device'] = addr.upper()
|
|
c['output'] = 'bluetooth'
|
|
apply(c)
|
|
print('페어링/신뢰 등록 완료:', addr)
|
|
|
|
|
|
def agent_pair(addr, scan_seconds=12, pair_seconds=20):
|
|
"""Pair inside one bluetoothctl session so the agent stays registered.
|
|
|
|
A one-shot `bluetoothctl --agent ... pair` can exit before Secure Simple
|
|
Pairing finishes, which leaves the device paired but not bonded.
|
|
"""
|
|
import time
|
|
proc = subprocess.Popen(['bluetoothctl'], stdin=subprocess.PIPE, text=True)
|
|
steps = [('agent NoInputNoOutput', 1), ('default-agent', 1), ('pairable on', 1),
|
|
('scan on', scan_seconds), ('scan off', 1), ('pair ' + addr, pair_seconds)]
|
|
try:
|
|
for command, wait in steps:
|
|
proc.stdin.write(command + '\n')
|
|
proc.stdin.flush()
|
|
time.sleep(wait)
|
|
proc.stdin.write('quit\n')
|
|
proc.stdin.flush()
|
|
proc.wait(timeout=10)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
proc.kill()
|
|
proc.wait()
|
|
|
|
|
|
def discoverable(on, seconds=300):
|
|
"""Show/hide the board to phones; the agent runs as a transient unit."""
|
|
import time
|
|
root()
|
|
run('systemctl', 'stop', PAIRING_UNIT, check=False, capture=True)
|
|
if not on:
|
|
run('bluetoothctl', 'discoverable', 'off', check=False, capture=True)
|
|
run('bluetoothctl', 'pairable', 'off', check=False, capture=True)
|
|
PAIRING_UNTIL.unlink(missing_ok=True)
|
|
print('Bluetooth 검색을 숨겼습니다.')
|
|
return
|
|
if not 30 <= seconds <= 1800: raise ValueError('검색 허용 시간: 30~1800초')
|
|
run('systemd-run', '--unit', PAIRING_UNIT, '--collect',
|
|
'--property', f'RuntimeMaxSec={seconds + 15}',
|
|
# Fallback if the agent is killed before its own cleanup runs.
|
|
'--property', 'ExecStopPost=/usr/bin/bluetoothctl discoverable off',
|
|
'--property', 'ExecStopPost=/usr/bin/bluetoothctl pairable off',
|
|
str(Path(__file__).resolve()), 'pairing-agent', '--seconds', str(seconds), capture=True)
|
|
for _ in range(20):
|
|
time.sleep(0.5)
|
|
if 'Discoverable: yes' in run('bluetoothctl', 'show', check=False, capture=True).stdout:
|
|
print(f'{seconds}초 동안 Bluetooth 검색을 허용합니다.')
|
|
return
|
|
run('systemctl', 'stop', PAIRING_UNIT, check=False, capture=True)
|
|
raise ValueError('Bluetooth 검색 허용을 켜지 못했습니다.')
|
|
|
|
|
|
def pairing_agent(seconds):
|
|
"""Accept phones that pair while the window is open and trust them."""
|
|
import pty
|
|
import select
|
|
import signal
|
|
import time
|
|
root()
|
|
signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
|
|
end = time.monotonic() + seconds
|
|
PAIRING_UNTIL.parent.mkdir(parents=True, exist_ok=True)
|
|
PAIRING_UNTIL.write_text(str(time.time() + seconds))
|
|
pid, fd = pty.fork()
|
|
if pid == 0:
|
|
os.execvp('bluetoothctl', ['bluetoothctl', '--agent', 'NoInputNoOutput'])
|
|
queue = ['default-agent', f'discoverable-timeout {seconds}', 'pairable on', 'discoverable on']
|
|
next_command = time.monotonic() + 1.5
|
|
buf, trusted = '', set()
|
|
try:
|
|
while time.monotonic() < end:
|
|
ready, _, _ = select.select([fd], [], [], 0.25)
|
|
if ready:
|
|
try: data = os.read(fd, 8192)
|
|
except OSError: break
|
|
if not data: break
|
|
buf += re.sub(r'\x1b\[[0-9;]*[A-Za-z]', '', data.decode(errors='replace'))
|
|
# Pairing confirmation and A2DP service authorization prompts.
|
|
while '(yes/no)' in buf:
|
|
head, buf = buf.split('(yes/no)', 1)
|
|
print('승인:', (head.strip().splitlines() or [''])[-1], flush=True)
|
|
os.write(fd, b'yes\n')
|
|
for addr in re.findall(r'Device ((?:[0-9A-F]{2}:){5}[0-9A-F]{2}) (?:Paired|Bonded): yes', buf):
|
|
if addr not in trusted:
|
|
trusted.add(addr)
|
|
print('신뢰 등록:', addr, flush=True)
|
|
os.write(fd, f'trust {addr}\n'.encode())
|
|
buf = buf[-4096:]
|
|
if queue and time.monotonic() >= next_command:
|
|
os.write(fd, (queue.pop(0) + '\n').encode())
|
|
next_command = time.monotonic() + 1.5
|
|
finally:
|
|
for command in (b'discoverable off\n', b'pairable off\n', b'quit\n'):
|
|
try: os.write(fd, command); time.sleep(0.3)
|
|
except OSError: pass
|
|
os.close(fd)
|
|
os.waitpid(pid, 0)
|
|
PAIRING_UNTIL.unlink(missing_ok=True)
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description='Radxa AirPlay + Bluetooth 믹서 관리')
|
|
sub = p.add_subparsers(dest='cmd', required=True)
|
|
sub.add_parser('status')
|
|
sub.add_parser('devices')
|
|
q = sub.add_parser('scan'); q.add_argument('--seconds', type=int, default=30)
|
|
q = sub.add_parser('pair'); q.add_argument('address'); q.add_argument('--output', action='store_true'); q.add_argument('--bond', action='store_true')
|
|
q = sub.add_parser('pair-input'); q.add_argument('--seconds', type=int, default=120)
|
|
q = sub.add_parser('discoverable'); q.add_argument('state', choices=['on', 'off']); q.add_argument('--seconds', type=int, default=300)
|
|
q = sub.add_parser('pairing-agent'); q.add_argument('--seconds', type=int, required=True)
|
|
q = sub.add_parser('output'); q.add_argument('mode', choices=['dac', 'bluetooth', 'bose']); q.add_argument('address', nargs='?'); q.add_argument('--pcm')
|
|
q = sub.add_parser('name'); q.add_argument('name')
|
|
q = sub.add_parser('dac'); q.add_argument('pcm')
|
|
sub.add_parser('apply')
|
|
q = sub.add_parser('init'); q.add_argument('--name', required=True); q.add_argument('--dac', default='hw:CARD=Audio,DEV=0')
|
|
q.add_argument('--adopt', action='store_true'); q.add_argument('--address', default='')
|
|
q.add_argument('--output', choices=['dac', 'bluetooth'], default='dac')
|
|
sub.add_parser('inputs')
|
|
q = sub.add_parser('volume'); q.add_argument('input'); q.add_argument('percent', type=int)
|
|
q = sub.add_parser('mute'); q.add_argument('input'); q.add_argument('state', choices=['on', 'off'])
|
|
a = p.parse_args()
|
|
if a.cmd in ('inputs', 'volume', 'mute'):
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
sys.path.insert(0, '/usr/local/lib/radxa-mixer')
|
|
import input_controls
|
|
if a.cmd == 'inputs': print(json.dumps(input_controls.inputs(), ensure_ascii=False, indent=2))
|
|
elif a.cmd == 'volume': root(); input_controls.set_input(a.input, volume=a.percent)
|
|
else: root(); input_controls.set_input(a.input, mute=a.state == 'on')
|
|
return
|
|
if a.cmd == 'devices':
|
|
run('bluetoothctl', 'devices'); run('aplay', '-l'); run('bluealsa-aplay', '-l')
|
|
elif a.cmd == 'status':
|
|
print(json.dumps(load(), ensure_ascii=False, indent=2), flush=True)
|
|
run('systemctl', 'is-active', 'shairport-sync', 'nqptp', 'bluealsa', 'bluealsa-aplay', BT_SERVICE, check=False)
|
|
run('bluealsa-aplay', '-l')
|
|
elif a.cmd == 'scan':
|
|
root()
|
|
if not 5 <= a.seconds <= 120: raise ValueError('검색 시간: 5~120초')
|
|
# Keep the discovery client alive; stop discovery on timeout/interrupt.
|
|
try:
|
|
run('bluetoothctl', '--timeout', str(a.seconds), 'scan', 'bredr', check=False, timeout=a.seconds + 10)
|
|
finally:
|
|
run('bluetoothctl', 'scan', 'off', check=False)
|
|
run('bluetoothctl', 'devices')
|
|
elif a.cmd == 'pair':
|
|
pair(a.address, a.output, a.bond)
|
|
elif a.cmd == 'discoverable':
|
|
discoverable(a.state == 'on', a.seconds)
|
|
elif a.cmd == 'pairing-agent':
|
|
if not 30 <= a.seconds <= 1800: raise ValueError('검색 허용 시간: 30~1800초')
|
|
pairing_agent(a.seconds)
|
|
elif a.cmd == 'pair-input':
|
|
root()
|
|
if not 15 <= a.seconds <= 900: raise ValueError('페어링 시간: 15~900초')
|
|
run('bluetoothctl', 'discoverable-timeout', str(a.seconds))
|
|
# Interactive agent: approve the intended phone's service with "yes".
|
|
print('휴대폰에서 연결하세요. 서비스 승인 질문에는 yes를 입력하고, 완료 후 quit을 입력하세요.', flush=True)
|
|
try:
|
|
# Relay a PTY so service approval answers still come from the operator.
|
|
interactive_pair(a.seconds)
|
|
finally:
|
|
run('bluetoothctl', 'discoverable', 'off', check=False)
|
|
run('bluetoothctl', 'pairable', 'off', check=False)
|
|
print('서비스를 승인한 휴대폰은 sudo bluetoothctl trust 주소 로 신뢰 등록해 주세요.')
|
|
elif a.cmd == 'init':
|
|
apply(dict(name=a.name, dac_pcm=a.dac, output=a.output, bluetooth_device=a.address), initial=not a.adopt)
|
|
else:
|
|
c = load()
|
|
if a.cmd == 'output':
|
|
c['output'] = 'bluetooth' if a.mode in ('bluetooth', 'bose') else 'dac'
|
|
if a.pcm:
|
|
if c['output'] != 'dac': raise ValueError('--pcm은 DAC 출력에서만 지정합니다.')
|
|
c['dac_pcm'] = a.pcm
|
|
if a.address:
|
|
if c['output'] != 'bluetooth': raise ValueError('장비 주소는 bluetooth 출력에서만 지정합니다.')
|
|
c['bluetooth_device'] = a.address.upper()
|
|
elif a.cmd == 'name': c['name'] = a.name
|
|
elif a.cmd == 'dac': c['dac_pcm'] = a.pcm
|
|
apply(c, reconfigure=a.cmd == 'apply')
|
|
|
|
|
|
def interactive_pair(seconds):
|
|
import pty
|
|
import select
|
|
import time
|
|
pid, fd = pty.fork()
|
|
if pid == 0:
|
|
os.execvp('bluetoothctl', ['bluetoothctl', '--agent', 'NoInputNoOutput'])
|
|
end = time.monotonic() + seconds
|
|
# Do not send asynchronous agent registration commands in a single batch.
|
|
queue = ['default-agent', 'pairable on', 'discoverable on']
|
|
next_command = time.monotonic() + 1.5
|
|
try:
|
|
while time.monotonic() < end:
|
|
ready, _, _ = select.select([fd, sys.stdin], [], [], 0.25)
|
|
if fd in ready:
|
|
try: data = os.read(fd, 8192)
|
|
except OSError: break
|
|
if not data: break
|
|
os.write(sys.stdout.fileno(), data)
|
|
if sys.stdin in ready:
|
|
data = os.read(sys.stdin.fileno(), 8192)
|
|
if not data: break
|
|
os.write(fd, data)
|
|
if queue and time.monotonic() >= next_command:
|
|
os.write(fd, (queue.pop(0) + '\n').encode())
|
|
next_command = time.monotonic() + 1.5
|
|
finally:
|
|
try: os.write(fd, b'quit\n')
|
|
except OSError: pass
|
|
os.close(fd)
|
|
os.waitpid(pid, 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
import fcntl
|
|
if len(sys.argv) > 1 and sys.argv[1] not in ('status', 'devices', 'inputs', 'pairing-agent', '-h', '--help'):
|
|
root()
|
|
Path('/run/lock').mkdir(parents=True, exist_ok=True)
|
|
with open('/run/lock/radxa-audio.lock', 'w') as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
main()
|
|
else:
|
|
main()
|
|
except (ValueError, subprocess.CalledProcessError, subprocess.TimeoutExpired, OSError) as e:
|
|
print('오류:', e, file=sys.stderr)
|
|
sys.exit(1)
|
|
except KeyboardInterrupt:
|
|
print('\n중단했습니다.', file=sys.stderr)
|
|
sys.exit(130)
|