diff --git a/bumble/crypto.py b/bumble/crypto.py index 757594fc..852c675a 100644 --- a/bumble/crypto.py +++ b/bumble/crypto.py @@ -23,22 +23,18 @@ # ----------------------------------------------------------------------------- import logging import operator -import platform -if platform.system() != 'Emscripten': - import secrets - from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes - from cryptography.hazmat.primitives.asymmetric.ec import ( - generate_private_key, - ECDH, - EllipticCurvePublicNumbers, - EllipticCurvePrivateNumbers, - SECP256R1, - ) - from cryptography.hazmat.primitives import cmac -else: - # TODO: implement stubs - pass +import secrets +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives.asymmetric.ec import ( + generate_private_key, + ECDH, + EllipticCurvePublicNumbers, + EllipticCurvePrivateNumbers, + SECP256R1, +) +from cryptography.hazmat.primitives import cmac + # ----------------------------------------------------------------------------- # Logging diff --git a/bumble/device.py b/bumble/device.py index 6ab0a12a..591e385f 100644 --- a/bumble/device.py +++ b/bumble/device.py @@ -2262,17 +2262,21 @@ class Device(CompositeEventEmitter): return keys.ltk_peripheral.value async def get_link_key(self, address: Address) -> Optional[bytes]: - # Look for the key in the keystore - if self.keystore is not None: - keys = await self.keystore.get(str(address)) - if keys is not None: - logger.debug('found keys in the key store') - if keys.link_key is None: - logger.warning('no link key') - return None + if self.keystore is None: + return None - return keys.link_key.value - return None + # Look for the key in the keystore + keys = await self.keystore.get(str(address)) + if keys is None: + logger.debug(f'no keys found for {address}') + return None + + logger.debug('found keys in the key store') + if keys.link_key is None: + logger.warning('no link key') + return None + + return keys.link_key.value # [Classic only] async def authenticate(self, connection): @@ -2391,6 +2395,18 @@ class Device(CompositeEventEmitter): 'connection_encryption_failure', on_encryption_failure ) + async def update_keys(self, address: str, keys: PairingKeys) -> None: + if self.keystore is None: + return + + try: + await self.keystore.update(address, keys) + await self.device.refresh_resolving_list() + except Exception as error: + logger.warning(f'!!! error while storing keys: {error}') + else: + self.emit('key_store_update') + # [Classic only] async def switch_role(self, connection: Connection, role: int): pending_role_change = asyncio.get_running_loop().create_future() @@ -2485,13 +2501,7 @@ class Device(CompositeEventEmitter): value=link_key, authenticated=authenticated ) - async def store_keys(): - try: - await self.keystore.update(str(bd_addr), pairing_keys) - except Exception as error: - logger.warning(f'!!! error while storing keys: {error}') - - self.abort_on('flush', store_keys()) + self.abort_on('flush', self.update_keys(str(bd_addr), pairing_keys)) if connection := self.find_connection_by_bd_addr( bd_addr, transport=BT_BR_EDR_TRANSPORT diff --git a/bumble/smp.py b/bumble/smp.py index 9588a5ac..55b83590 100644 --- a/bumble/smp.py +++ b/bumble/smp.py @@ -1832,8 +1832,9 @@ class Manager(EventEmitter): ) -> None: # Store the keys in the key store if self.device.keystore and identity_address is not None: - await self.device.keystore.update(str(identity_address), keys) - await self.device.refresh_resolving_list() + self.device.abort_on( + 'flush', self.device.update_keys(str(identity_address), keys) + ) # Notify the device self.device.on_pairing(session.connection, identity_address, keys, session.sc) diff --git a/bumble/transport/__init__.py b/bumble/transport/__init__.py index 840b3e59..c7222833 100644 --- a/bumble/transport/__init__.py +++ b/bumble/transport/__init__.py @@ -69,6 +69,7 @@ async def open_transport(name: str) -> Transport: * usb * pyusb * android-emulator + * android-netsim """ return _wrap_transport(await _open_transport(name)) diff --git a/bumble/transport/ws_server.py b/bumble/transport/ws_server.py index c7b7c6e2..ddebef23 100644 --- a/bumble/transport/ws_server.py +++ b/bumble/transport/ws_server.py @@ -43,7 +43,7 @@ async def open_ws_server_transport(spec): def __init__(self): source = ParserSource() sink = PumpedPacketSink(self.send_packet) - self.connection = asyncio.get_running_loop().create_future() + self.connection = None self.server = None super().__init__(source, sink) @@ -63,7 +63,7 @@ async def open_ws_server_transport(spec): f'new connection on {connection.local_address} ' f'from {connection.remote_address}' ) - self.connection.set_result(connection) + self.connection = connection # pylint: disable=no-member try: async for packet in connection: @@ -74,12 +74,14 @@ async def open_ws_server_transport(spec): except websockets.WebSocketException as error: logger.debug(f'exception while receiving packet: {error}') - # Wait for a new connection - self.connection = asyncio.get_running_loop().create_future() + # We're now disconnected + self.connection = None async def send_packet(self, packet): - connection = await self.connection - return await connection.send(packet) + if self.connection is None: + logger.debug('no connection, dropping packet') + return + return await self.connection.send(packet) local_host, local_port = spec.split(':') transport = WsServerTransport() diff --git a/setup.cfg b/setup.cfg index a7a09d63..1072accc 100644 --- a/setup.cfg +++ b/setup.cfg @@ -32,17 +32,17 @@ package_dir = include_package_data = True install_requires = aiohttp ~= 3.8; platform_system!='Emscripten' - appdirs >= 1.4 - bt-test-interfaces >= 0.0.2 + appdirs >= 1.4; platform_system!='Emscripten' + bt-test-interfaces >= 0.0.2; platform_system!='Emscripten' click == 8.1.3; platform_system!='Emscripten' - cryptography == 35; platform_system!='Emscripten' + cryptography == 39; platform_system!='Emscripten' grpcio == 1.51.1; platform_system!='Emscripten' - humanize >= 4.6.0 + humanize >= 4.6.0; platform_system!='Emscripten' libusb1 >= 2.0.1; platform_system!='Emscripten' libusb-package == 1.0.26.1; platform_system!='Emscripten' prompt_toolkit >= 3.0.16; platform_system!='Emscripten' - prettytable >= 3.6.0 - protobuf >= 3.12.4 + prettytable >= 3.6.0; platform_system!='Emscripten' + protobuf >= 3.12.4; platform_system!='Emscripten' pyee >= 8.2.2 pyserial-asyncio >= 0.5; platform_system!='Emscripten' pyserial >= 3.5; platform_system!='Emscripten' diff --git a/speaker.html b/speaker.html deleted file mode 100644 index 05cc31f8..00000000 --- a/speaker.html +++ /dev/null @@ -1,28 +0,0 @@ - - - - Audio WAV Player - - -

Audio WAV Player

- - - - - diff --git a/tasks.py b/tasks.py index 3a3a01a9..4df0f736 100644 --- a/tasks.py +++ b/tasks.py @@ -177,3 +177,31 @@ project_tasks.add_task(lint) project_tasks.add_task(format_code, name="format") project_tasks.add_task(check_types, name="check-types") project_tasks.add_task(pre_commit) + + +# ----------------------------------------------------------------------------- +# Web +# ----------------------------------------------------------------------------- +web_tasks = Collection() +ns.add_collection(web_tasks, name="web") + + +# ----------------------------------------------------------------------------- +@task +def serve(ctx, port=8000): + """ + Run a simple HTTP server for the examples under the `web` directory. + """ + import http.server + + address = ("", port) + class Handler(http.server.SimpleHTTPRequestHandler): + def __init__(self, *args, **kwargs): + super().__init__(*args, directory="web", **kwargs) + server = http.server.HTTPServer(address, Handler) + print(f"Now serving on port {port} 🕸️") + server.serve_forever() + + +# ----------------------------------------------------------------------------- +web_tasks.add_task(serve) diff --git a/web/bumble.js b/web/bumble.js new file mode 100644 index 00000000..c5fd6a3b --- /dev/null +++ b/web/bumble.js @@ -0,0 +1,92 @@ +function bufferToHex(buffer) { + return [...new Uint8Array(buffer)].map(x => x.toString(16).padStart(2, '0')).join(''); +} + +class PacketSource { + constructor(pyodide) { + this.parser = pyodide.runPython(` + from bumble.transport.common import PacketParser + class ProxiedPacketParser(PacketParser): + def feed_data(self, js_data): + super().feed_data(bytes(js_data.to_py())) + ProxiedPacketParser() + `); + } + + set_packet_sink(sink) { + this.parser.set_packet_sink(sink); + } + + data_received(data) { + console.log(`HCI[controller->host]: ${bufferToHex(data)}`); + this.parser.feed_data(data); + } +} + +class PacketSink { + constructor(writer) { + this.writer = writer; + } + + on_packet(packet) { + const buffer = packet.toJs({create_proxies : false}); + packet.destroy(); + console.log(`HCI[host->controller]: ${bufferToHex(buffer)}`); + // TODO: create an async queue here instead of blindly calling write without awaiting + this.writer(buffer); + } +} + +export async function connectWebSocketTransport(pyodide, hciWsUrl) { + return new Promise((resolve, reject) => { + let resolved = false; + + let ws = new WebSocket(hciWsUrl); + ws.binaryType = "arraybuffer"; + + ws.onopen = () => { + console.log("WebSocket open"); + resolve({ + packet_source, + packet_sink + }); + resolved = true; + } + + ws.onclose = () => { + console.log("WebSocket close"); + if (!resolved) { + reject(`Failed to connect to ${hciWsUrl}`) + } + } + + ws.onmessage = (event) => { + packet_source.data_received(event.data); + } + + const packet_source = new PacketSource(pyodide); + const packet_sink = new PacketSink((packet) => ws.send(packet)); + }) +} + +export async function loadBumble(pyodide, bumblePackage) { + // Load the Bumble module + await pyodide.loadPackage("micropip"); + await pyodide.runPythonAsync(` + import micropip + await micropip.install("cryptography") + await micropip.install("${bumblePackage}") + package_list = micropip.list() + print(package_list) + `) + + // Mount a filesystem so that we can persist data like the Key Store + let mountDir = "/bumble"; + pyodide.FS.mkdir(mountDir); + pyodide.FS.mount(pyodide.FS.filesystems.IDBFS, { root: "." }, mountDir); + + // Sync previously persisted filesystem data into memory + pyodide.FS.syncfs(true, () => { + console.log("FS synced in") + }); +} \ No newline at end of file diff --git a/web/index.html b/web/index.html deleted file mode 100644 index 4374db02..00000000 --- a/web/index.html +++ /dev/null @@ -1,131 +0,0 @@ - - - - - - - - -
-
-
Output:
- - - - - diff --git a/web/scanner/scanner.html b/web/scanner/scanner.html new file mode 100644 index 00000000..12c65dd6 --- /dev/null +++ b/web/scanner/scanner.html @@ -0,0 +1,129 @@ + + + + + + + + + +
+
+
Log Output

+ +
+ + + + + \ No newline at end of file diff --git a/web/scanner.py b/web/scanner/scanner.py similarity index 54% rename from web/scanner.py rename to web/scanner/scanner.py index 59eda67e..dd53050b 100644 --- a/web/scanner.py +++ b/web/scanner/scanner.py @@ -15,50 +15,38 @@ # ----------------------------------------------------------------------------- # Imports # ----------------------------------------------------------------------------- +import time + from bumble.device import Device -from bumble.transport.common import PacketParser + + +# ----------------------------------------------------------------------------- +class ScanEntry: + def __init__(self, advertisement): + self.address = str(advertisement.address).replace("/P", "") + self.address_type = ('Public', 'Random', 'Public Identity', 'Random Identity')[ + advertisement.address.address_type + ] + self.rssi = advertisement.rssi + self.data = advertisement.data.to_string("\n") # ----------------------------------------------------------------------------- class ScannerListener(Device.Listener): + def __init__(self, callback): + self.callback = callback + self.entries = {} + def on_advertisement(self, advertisement): - address_type_string = ('P', 'R', 'PI', 'RI')[advertisement.address.address_type] - print( - f'>>> {advertisement.address} [{address_type_string}]: RSSI={advertisement.rssi}, {advertisement.ad_data}' - ) - - -class HciSource: - def __init__(self, host_source): - self.parser = PacketParser() - host_source.delegate = self - - def set_packet_sink(self, sink): - self.parser.set_packet_sink(sink) - - # host source delegation - def data_received(self, data): - print('*** DATA from JS:', data) - buffer = bytes(data.to_py()) - self.parser.feed_data(buffer) - - -# class HciSink: -# def __init__(self, host_sink): -# self.host_sink = host_sink - -# def on_packet(self, packet): -# print(f'>>> PACKET from Python: {packet}') -# self.host_sink.on_packet(packet) + self.entries[advertisement.address] = ScanEntry(advertisement) + self.callback(list(self.entries.values())) # ----------------------------------------------------------------------------- -async def main(host_source, host_sink): +async def main(hci_source, hci_sink, callback): print('### Starting Scanner') - hci_source = HciSource(host_source) - hci_sink = host_sink device = Device.with_hci('Bumble', 'F0:F1:F2:F3:F4:F5', hci_source, hci_sink) - device.listener = ScannerListener() + device.listener = ScannerListener(callback) await device.power_on() await device.start_scanning() diff --git a/web/speaker/logo.svg b/web/speaker/logo.svg new file mode 100644 index 00000000..70ef7a90 --- /dev/null +++ b/web/speaker/logo.svg @@ -0,0 +1,42 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/web/speaker/speaker.css b/web/speaker/speaker.css new file mode 100644 index 00000000..988392a0 --- /dev/null +++ b/web/speaker/speaker.css @@ -0,0 +1,76 @@ +body, h1, h2, h3, h4, h5, h6 { + font-family: sans-serif; +} + +#controlsDiv { + margin: 6px; +} + +#errorText { + background-color: rgb(239, 89, 75); + border: none; + border-radius: 4px; + padding: 8px; + display: inline-block; + margin: 4px; +} + +#startButton { + padding: 4px; + margin: 6px; +} + +#fftCanvas { + border-radius: 16px; + margin: 6px; +} + +#bandwidthCanvas { + border: grey; + border-style: solid; + border-radius: 8px; + margin: 6px; +} + +#streamStateText { + background-color: rgb(93, 165, 93); + border: none; + border-radius: 8px; + padding: 10px 20px; + display: inline-block; + margin: 6px; +} + +#connectionStateText { + background-color: rgb(112, 146, 206); + border: none; + border-radius: 8px; + padding: 10px 20px; + display: inline-block; + margin: 6px; +} + +#propertiesTable { + border: grey; + border-style: solid; + border-radius: 4px; + padding: 4px; + margin: 6px; + margin-left: 0px; +} + +th, td { + padding-left: 6px; + padding-right: 6px; +} + +.properties td:nth-child(even) { + background-color: #D6EEEE; + font-family: monospace; +} + +.properties td:nth-child(odd) { + font-weight: bold; +} + +.properties tr td:nth-child(2) { width: 150px; } \ No newline at end of file diff --git a/web/speaker/speaker.html b/web/speaker/speaker.html new file mode 100644 index 00000000..a20f084d --- /dev/null +++ b/web/speaker/speaker.html @@ -0,0 +1,34 @@ + + + + Bumble Speaker + + + + + +

Bumble Virtual Speaker

+
+
+ + + +
+ + + + +
Codec
Packets
Bytes
+ 		+ Bandwidth Graph +
+ IDLE + NOT CONNECTED +
+ +
+ Audio Frequencies Animation + +
+ + \ No newline at end of file diff --git a/web/speaker/speaker.js b/web/speaker/speaker.js new file mode 100644 index 00000000..b94180fa --- /dev/null +++ b/web/speaker/speaker.js @@ -0,0 +1,289 @@ +import { loadBumble, connectWebSocketTransport } from "../bumble.js"; + +(function () { + 'use strict'; + + let codecText; + let packetsReceivedText; + let bytesReceivedText; + let streamStateText; + let connectionStateText; + let errorText; + let audioOnButton; + let mediaSource; + let sourceBuffer; + let audioElement; + let audioContext; + let audioAnalyzer; + let audioFrequencyBinCount; + let audioFrequencyData; + let packetsReceived = 0; + let bytesReceived = 0; + let audioState = "stopped"; + let streamState = "IDLE"; + let fftCanvas; + let fftCanvasContext; + let bandwidthCanvas; + let bandwidthCanvasContext; + let bandwidthBinCount; + let bandwidthBins = []; + let pyodide; + + const FFT_WIDTH = 800; + const FFT_HEIGHT = 256; + const BANDWIDTH_WIDTH = 500; + const BANDWIDTH_HEIGHT = 100; + + + function init() { + initUI(); + initMediaSource(); + initAudioElement(); + initAnalyzer(); + initBumble(); + } + + function initUI() { + audioOnButton = document.getElementById("audioOnButton"); + codecText = document.getElementById("codecText"); + packetsReceivedText = document.getElementById("packetsReceivedText"); + bytesReceivedText = document.getElementById("bytesReceivedText"); + streamStateText = document.getElementById("streamStateText"); + errorText = document.getElementById("errorText"); + connectionStateText = document.getElementById("connectionStateText"); + + audioOnButton.onclick = () => startAudio(); + + codecText.innerText = "AAC"; + setErrorText(""); + + requestAnimationFrame(onAnimationFrame); + } + + function initMediaSource() { + mediaSource = new MediaSource(); + mediaSource.onsourceopen = onMediaSourceOpen; + mediaSource.onsourceclose = onMediaSourceClose; + mediaSource.onsourceended = onMediaSourceEnd; + } + + function initAudioElement() { + audioElement = document.getElementById("audio"); + audioElement.src = URL.createObjectURL(mediaSource); + // audioElement.controls = true; + } + + function initAnalyzer() { + fftCanvas = document.getElementById("fftCanvas"); + fftCanvas.width = FFT_WIDTH + fftCanvas.height = FFT_HEIGHT + fftCanvasContext = fftCanvas.getContext('2d'); + fftCanvasContext.fillStyle = "rgb(0, 0, 0)"; + fftCanvasContext.fillRect(0, 0, FFT_WIDTH, FFT_HEIGHT); + + bandwidthCanvas = document.getElementById("bandwidthCanvas"); + bandwidthCanvas.width = BANDWIDTH_WIDTH + bandwidthCanvas.height = BANDWIDTH_HEIGHT + bandwidthCanvasContext = bandwidthCanvas.getContext('2d'); + bandwidthCanvasContext.fillStyle = "rgb(255, 255, 255)"; + bandwidthCanvasContext.fillRect(0, 0, BANDWIDTH_WIDTH, BANDWIDTH_HEIGHT); + } + + async function initBumble() { + // Load pyodide + console.log("Loading Pyodide"); + pyodide = await loadPyodide(); + + // Load Bumble + console.log("Loading Bumble"); + const params = (new URL(document.location)).searchParams; + const bumblePackage = params.get("package") || "bumble"; + await loadBumble(pyodide, bumblePackage); + + console.log("Ready!") + + const hciWsUrl = params.get("hci") || "ws://localhost:9922/hci"; + try { + // Create a WebSocket HCI transport + let transport + try { + transport = await connectWebSocketTransport(pyodide, hciWsUrl); + } catch (error) { + console.error(error); + setErrorText(error); + return; + } + + // Run the scanner example + const script = await (await fetch("speaker.py")).text(); + await pyodide.runPythonAsync(script); + const pythonMain = pyodide.globals.get("main"); + console.log("Starting speaker..."); + await pythonMain(transport.packet_source, transport.packet_sink, onEvent); + console.log("Speaker running"); + } catch (err) { + console.log(err); + } + } + + function startAnalyzer() { + // FFT + if (audioElement.captureStream !== undefined) { + audioContext = new AudioContext(); + audioAnalyzer = audioContext.createAnalyser(); + audioAnalyzer.fftSize = 128; + audioFrequencyBinCount = audioAnalyzer.frequencyBinCount; + audioFrequencyData = new Uint8Array(audioFrequencyBinCount); + const stream = audioElement.captureStream(); + const source = audioContext.createMediaStreamSource(stream); + source.connect(audioAnalyzer); + } + + // Bandwidth + bandwidthBinCount = BANDWIDTH_WIDTH / 2; + bandwidthBins = []; + } + + function setErrorText(message) { + errorText.innerText = message; + if (message.length == 0) { + errorText.style.display = "none"; + } else { + errorText.style.display = "inline-block"; + } + } + + function setStreamState(state) { + streamState = state; + streamStateText.innerText = streamState; + } + + function onAnimationFrame() { + // FFT + if (audioAnalyzer !== undefined) { + audioAnalyzer.getByteFrequencyData(audioFrequencyData); + fftCanvasContext.fillStyle = "rgb(0, 0, 0)"; + fftCanvasContext.fillRect(0, 0, FFT_WIDTH, FFT_HEIGHT); + const barCount = audioFrequencyBinCount; + const barWidth = (FFT_WIDTH / audioFrequencyBinCount) - 1; + for (let bar = 0; bar < barCount; bar++) { + const barHeight = audioFrequencyData[bar]; + fftCanvasContext.fillStyle = `rgb(${barHeight / 256 * 200 + 50}, 50, ${50 + 2 * bar})`; + fftCanvasContext.fillRect(bar * (barWidth + 1), FFT_HEIGHT - barHeight, barWidth, barHeight); + } + } + + // Bandwidth + bandwidthCanvasContext.fillStyle = "rgb(255, 255, 255)"; + bandwidthCanvasContext.fillRect(0, 0, BANDWIDTH_WIDTH, BANDWIDTH_HEIGHT); + bandwidthCanvasContext.fillStyle = `rgb(100, 100, 100)`; + for (let t = 0; t < bandwidthBins.length; t++) { + const lineHeight = (bandwidthBins[t] / 1000) * BANDWIDTH_HEIGHT; + bandwidthCanvasContext.fillRect(t * 2, BANDWIDTH_HEIGHT - lineHeight, 2, lineHeight); + } + + // Display again at the next frame + requestAnimationFrame(onAnimationFrame); + } + + function onMediaSourceOpen() { + console.log(this.readyState); + sourceBuffer = mediaSource.addSourceBuffer("audio/aac"); + } + + function onMediaSourceClose() { + console.log(this.readyState); + } + + function onMediaSourceEnd() { + console.log(this.readyState); + } + + async function startAudio() { + try { + console.log("starting audio..."); + audioOnButton.disabled = true; + audioState = "starting"; + await audioElement.play(); + console.log("audio started"); + audioState = "playing"; + startAnalyzer(); + } catch (error) { + console.error(`play failed: ${error}`); + audioState = "stopped"; + audioOnButton.disabled = false; + } + } + + async function onEvent(name, params) { + // Dispatch the message. + const handlerName = `on${name.charAt(0).toUpperCase()}${name.slice(1)}` + const handler = eventHandlers[handlerName]; + if (handler !== undefined) { + handler(params); + } else { + console.warn(`unhandled event: ${name}`) + } + } + + function onStart() { + setStreamState("STARTED"); + } + + function onStop() { + setStreamState("STOPPED"); + } + + function onSuspend() { + setStreamState("SUSPENDED"); + } + + function onConnection(params) { + connectionStateText.innerText = `CONNECTED: ${params.get('peer_name')} (${params.get('peer_address')})`; + } + + function onDisconnection(params) { + connectionStateText.innerText = "DISCONNECTED"; + } + + function onAudio(python_packet) { + const packet = python_packet.toJs({create_proxies : false}); + python_packet.destroy(); + if (audioState != "stopped") { + // Queue the audio packet. + sourceBuffer.appendBuffer(packet); + } + + packetsReceived += 1; + packetsReceivedText.innerText = packetsReceived; + bytesReceived += packet.byteLength; + bytesReceivedText.innerText = bytesReceived; + + bandwidthBins[bandwidthBins.length] = packet.byteLength; + if (bandwidthBins.length > bandwidthBinCount) { + bandwidthBins.shift(); + } + } + + function onKeystoreupdate() { + // Sync the FS + pyodide.FS.syncfs(() => { + console.log("FS synced out") + }); + } + + const eventHandlers = { + onStart, + onStop, + onSuspend, + onConnection, + onDisconnection, + onAudio, + onKeystoreupdate + } + + window.onload = (event) => { + init(); + } + +}()); \ No newline at end of file diff --git a/web/speaker/speaker.py b/web/speaker/speaker.py new file mode 100644 index 00000000..ddc20864 --- /dev/null +++ b/web/speaker/speaker.py @@ -0,0 +1,321 @@ +# Copyright 2021-2023 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# ----------------------------------------------------------------------------- +# Imports +# ----------------------------------------------------------------------------- +from __future__ import annotations +import enum +import logging +from typing import Dict, List + +from bumble.core import BT_BR_EDR_TRANSPORT, CommandTimeoutError +from bumble.device import Device, DeviceConfiguration +from bumble.pairing import PairingConfig +from bumble.sdp import ServiceAttribute +from bumble.avdtp import ( + AVDTP_AUDIO_MEDIA_TYPE, + Listener, + MediaCodecCapabilities, + MediaPacket, + Protocol, +) +from bumble.a2dp import ( + make_audio_sink_service_sdp_records, + MPEG_2_AAC_LC_OBJECT_TYPE, + A2DP_SBC_CODEC_TYPE, + A2DP_MPEG_2_4_AAC_CODEC_TYPE, + SBC_MONO_CHANNEL_MODE, + SBC_DUAL_CHANNEL_MODE, + SBC_SNR_ALLOCATION_METHOD, + SBC_LOUDNESS_ALLOCATION_METHOD, + SBC_STEREO_CHANNEL_MODE, + SBC_JOINT_STEREO_CHANNEL_MODE, + SbcMediaCodecInformation, + AacMediaCodecInformation, +) +from bumble.utils import AsyncRunner +from bumble.codecs import AacAudioRtpPacket + + +# ----------------------------------------------------------------------------- +# Logging +# ----------------------------------------------------------------------------- +logger = logging.getLogger(__name__) + + +# ----------------------------------------------------------------------------- +class AudioExtractor: + @staticmethod + def create(codec: str): + if codec == 'aac': + return AacAudioExtractor() + if codec == 'sbc': + return SbcAudioExtractor() + + def extract_audio(self, packet: MediaPacket) -> bytes: + raise NotImplementedError() + + +# ----------------------------------------------------------------------------- +class AacAudioExtractor: + def extract_audio(self, packet: MediaPacket) -> bytes: + return AacAudioRtpPacket(packet.payload).to_adts() + + +# ----------------------------------------------------------------------------- +class SbcAudioExtractor: + def extract_audio(self, packet: MediaPacket) -> bytes: + # header = packet.payload[0] + # fragmented = header >> 7 + # start = (header >> 6) & 0x01 + # last = (header >> 5) & 0x01 + # number_of_frames = header & 0x0F + + # TODO: support fragmented payloads + return packet.payload[1:] + + +# ----------------------------------------------------------------------------- +class Speaker: + class StreamState(enum.Enum): + IDLE = 0 + STOPPED = 1 + STARTED = 2 + SUSPENDED = 3 + + def __init__(self, hci_source, hci_sink, emit_event, codec, discover): + self.hci_source = hci_source + self.hci_sink = hci_sink + self.emit_event = emit_event + self.codec = codec + self.discover = discover + self.device = None + self.connection = None + self.listener = None + self.packets_received = 0 + self.bytes_received = 0 + self.stream_state = Speaker.StreamState.IDLE + self.audio_extractor = AudioExtractor.create(codec) + + def sdp_records(self) -> Dict[int, List[ServiceAttribute]]: + service_record_handle = 0x00010001 + return { + service_record_handle: make_audio_sink_service_sdp_records( + service_record_handle + ) + } + + def codec_capabilities(self) -> MediaCodecCapabilities: + if self.codec == 'aac': + return self.aac_codec_capabilities() + + if self.codec == 'sbc': + return self.sbc_codec_capabilities() + + raise RuntimeError('unsupported codec') + + def aac_codec_capabilities(self) -> MediaCodecCapabilities: + return MediaCodecCapabilities( + media_type=AVDTP_AUDIO_MEDIA_TYPE, + media_codec_type=A2DP_MPEG_2_4_AAC_CODEC_TYPE, + media_codec_information=AacMediaCodecInformation.from_lists( + object_types=[MPEG_2_AAC_LC_OBJECT_TYPE], + sampling_frequencies=[48000, 44100], + channels=[1, 2], + vbr=1, + bitrate=256000, + ), + ) + + def sbc_codec_capabilities(self) -> MediaCodecCapabilities: + return MediaCodecCapabilities( + media_type=AVDTP_AUDIO_MEDIA_TYPE, + media_codec_type=A2DP_SBC_CODEC_TYPE, + media_codec_information=SbcMediaCodecInformation.from_lists( + sampling_frequencies=[48000, 44100, 32000, 16000], + channel_modes=[ + SBC_MONO_CHANNEL_MODE, + SBC_DUAL_CHANNEL_MODE, + SBC_STEREO_CHANNEL_MODE, + SBC_JOINT_STEREO_CHANNEL_MODE, + ], + block_lengths=[4, 8, 12, 16], + subbands=[4, 8], + allocation_methods=[ + SBC_LOUDNESS_ALLOCATION_METHOD, + SBC_SNR_ALLOCATION_METHOD, + ], + minimum_bitpool_value=2, + maximum_bitpool_value=53, + ), + ) + + def on_key_store_update(self): + print("Key Store updated") + self.emit_event('keystoreupdate', None) + + def on_bluetooth_connection(self, connection): + print(f'Connection: {connection}') + self.connection = connection + connection.on('disconnection', self.on_bluetooth_disconnection) + peer_name = '' if connection.peer_name is None else connection.peer_name + peer_address = str(connection.peer_address).replace('/P', '') + self.emit_event( + 'connection', {'peer_name': peer_name, 'peer_address': peer_address} + ) + + def on_bluetooth_disconnection(self, reason): + print(f'Disconnection ({reason})') + self.connection = None + AsyncRunner.spawn(self.advertise()) + self.emit_event('disconnection', None) + + def on_avdtp_connection(self, protocol): + print('Audio Stream Open') + + # Add a sink endpoint to the server + sink = protocol.add_sink(self.codec_capabilities()) + sink.on('start', self.on_sink_start) + sink.on('stop', self.on_sink_stop) + sink.on('suspend', self.on_sink_suspend) + sink.on('configuration', lambda: self.on_sink_configuration(sink.configuration)) + sink.on('rtp_packet', self.on_rtp_packet) + sink.on('rtp_channel_open', self.on_rtp_channel_open) + sink.on('rtp_channel_close', self.on_rtp_channel_close) + + # Listen for close events + protocol.on('close', self.on_avdtp_close) + + # Discover all endpoints on the remote device is requested + if self.discover: + AsyncRunner.spawn(self.discover_remote_endpoints(protocol)) + + def on_avdtp_close(self): + print("Audio Stream Closed") + + def on_sink_start(self): + print("Sink Started") + self.stream_state = self.StreamState.STARTED + self.emit_event('start', None) + + def on_sink_stop(self): + print("Sink Stopped") + self.stream_state = self.StreamState.STOPPED + self.emit_event('stop', None) + + def on_sink_suspend(self): + print("Sink Suspended") + self.stream_state = self.StreamState.SUSPENDED + self.emit_event('suspend', None) + + def on_sink_configuration(self, config): + print("Sink Configuration:") + print('\n'.join([" " + str(capability) for capability in config])) + + def on_rtp_channel_open(self): + print("RTP Channel Open") + + def on_rtp_channel_close(self): + print("RTP Channel Closed") + self.stream_state = self.StreamState.IDLE + + def on_rtp_packet(self, packet): + self.packets_received += 1 + self.bytes_received += len(packet.payload) + self.emit_event("audio", self.audio_extractor.extract_audio(packet)) + + async def advertise(self): + await self.device.set_discoverable(True) + await self.device.set_connectable(True) + + async def connect(self, address): + # Connect to the source + print(f'=== Connecting to {address}...') + connection = await self.device.connect(address, transport=BT_BR_EDR_TRANSPORT) + print(f'=== Connected to {connection.peer_address}') + + # Request authentication + print('*** Authenticating...') + await connection.authenticate() + print('*** Authenticated') + + # Enable encryption + print('*** Enabling encryption...') + await connection.encrypt() + print('*** Encryption on') + + protocol = await Protocol.connect(connection) + self.listener.set_server(connection, protocol) + self.on_avdtp_connection(protocol) + + async def discover_remote_endpoints(self, protocol): + endpoints = await protocol.discover_remote_endpoints() + print(f'@@@ Found {len(endpoints)} endpoints') + for endpoint in endpoints: + print('@@@', endpoint) + + async def run(self, connect_address): + # Create a device + device_config = DeviceConfiguration() + device_config.name = "Bumble Speaker" + device_config.class_of_device = 0x240414 + device_config.keystore = "JsonKeyStore:/bumble/keystore.json" + device_config.classic_enabled = True + device_config.le_enabled = False + self.device = Device.from_config_with_hci( + device_config, self.hci_source, self.hci_sink + ) + + # Setup the SDP to expose the sink service + self.device.sdp_service_records = self.sdp_records() + + # Don't require MITM when pairing. + self.device.pairing_config_factory = lambda connection: PairingConfig( + mitm=False + ) + + # Start the controller + await self.device.power_on() + + # Listen for Bluetooth connections + self.device.on('connection', self.on_bluetooth_connection) + + # Listen for changes to the key store + self.device.on('key_store_update', self.on_key_store_update) + + # Create a listener to wait for AVDTP connections + self.listener = Listener(Listener.create_registrar(self.device)) + self.listener.on('connection', self.on_avdtp_connection) + + print(f'Speaker ready to play, codec={self.codec}') + + if connect_address: + # Connect to the source + try: + await self.connect(connect_address) + except CommandTimeoutError: + print("Connection timed out") + return + else: + # Start being discoverable and connectable + print("Waiting for connection...") + await self.advertise() + + +# ----------------------------------------------------------------------------- +async def main(hci_source, hci_sink, emit_event): + # logging.basicConfig(level='DEBUG') + speaker = Speaker(hci_source, hci_sink, emit_event, "aac", False) + await speaker.run(None)