Add web toggle for Bluetooth discoverable window
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>
This commit is contained in:
+85
-1
@@ -13,6 +13,8 @@ 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')
|
||||
|
||||
@@ -318,6 +320,81 @@ def agent_pair(addr, scan_seconds=12, pair_seconds=20):
|
||||
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)
|
||||
@@ -326,6 +403,8 @@ def main():
|
||||
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')
|
||||
@@ -362,6 +441,11 @@ def main():
|
||||
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초')
|
||||
@@ -428,7 +512,7 @@ def interactive_pair(seconds):
|
||||
if __name__ == '__main__':
|
||||
try:
|
||||
import fcntl
|
||||
if len(sys.argv) > 1 and sys.argv[1] not in ('status', 'devices', 'inputs', '-h', '--help'):
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user