examples/run_asha_sink.py

# Copyright 2021-2022 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
# -----------------------------------------------------------------------------
import asyncio
import struct
import sys
import os
import logging

from bumble.core import AdvertisingData
from bumble.device import Device
from bumble.transport import open_transport_or_link
from bumble.hci import UUID
from bumble.gatt import (
    Service,
    Characteristic,
    CharacteristicValue
)


# -----------------------------------------------------------------------------
# Constants
# -----------------------------------------------------------------------------
ASHA_SERVICE                             = UUID.from_16_bits(0xFDF0, 'Audio Streaming for Hearing Aid')
ASHA_READ_ONLY_PROPERTIES_CHARACTERISTIC = UUID('6333651e-c481-4a3e-9169-7c902aad37bb', 'ReadOnlyProperties')
ASHA_AUDIO_CONTROL_POINT_CHARACTERISTIC  = UUID('f0d4de7e-4a88-476c-9d9f-1937b0996cc0', 'AudioControlPoint')
ASHA_AUDIO_STATUS_CHARACTERISTIC         = UUID('38663f1a-e711-4cac-b641-326b56404837', 'AudioStatus')
ASHA_VOLUME_CHARACTERISTIC               = UUID('00e4ca9e-ab14-41e4-8823-f9e70c7e91df', 'Volume')
ASHA_LE_PSM_OUT_CHARACTERISTIC           = UUID('2d410339-82b6-42aa-b34e-e2e01df8cc1a', 'LE_PSM_OUT')


# -----------------------------------------------------------------------------
async def main():
    if len(sys.argv) != 4:
        print('Usage: python run_asha_sink.py <device-config> <transport-spec> <audio-file>')
        print('example: python run_asha_sink.py device1.json usb:0 audio_out.g722')
        return

    audio_out = open(sys.argv[3], 'wb')

    async with await open_transport_or_link(sys.argv[2]) as (hci_source, hci_sink):
        device = Device.from_config_file_with_hci(sys.argv[1], hci_source, hci_sink)

        # Handler for audio control commands
        def on_audio_control_point_write(connection, value):
            print('--- AUDIO CONTROL POINT Write:', value.hex())
            opcode = value[0]
            if opcode == 1:
                # Start
                audio_type = ('Unknown', 'Ringtone', 'Phone Call', 'Media')[value[2]]
                print(f'### START: codec={value[1]}, audio_type={audio_type}, volume={value[3]}, otherstate={value[4]}')
            elif opcode == 2:
                print('### STOP')
            elif opcode == 3:
                print(f'### STATUS: connected={value[1]}')

            # Respond with a status
            asyncio.create_task(device.notify_subscribers(audio_status_characteristic, force=True))

        # Handler for volume control
        def on_volume_write(connection, value):
            print('--- VOLUME Write:', value[0])

        # Register an L2CAP CoC server
        def on_coc(channel):
            def on_data(data):
                print('<<< Voice data received:', data.hex())
                audio_out.write(data)

            channel.sink = on_data

        psm = device.register_l2cap_channel_server(0, on_coc, 8)
        print(f'### LE_PSM_OUT = {psm}')

        # Add the ASHA service to the GATT server
        read_only_properties_characteristic = Characteristic(
            ASHA_READ_ONLY_PROPERTIES_CHARACTERISTIC,
            Characteristic.READ,
            Characteristic.READABLE,
            bytes([
                0x01,        # Version
                0x00,        # Device Capabilities [Left, Monaural]
                0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08,  # HiSyncId
                0x01,        # Feature Map [LE CoC audio output streaming supported]
                0x00, 0x00,  # Render Delay
                0x00, 0x00,  # RFU
                0x02, 0x00   # Codec IDs [G.722 at 16 kHz]
            ])
        )
        audio_control_point_characteristic = Characteristic(
            ASHA_AUDIO_CONTROL_POINT_CHARACTERISTIC,
            Characteristic.WRITE | Characteristic.WRITE_WITHOUT_RESPONSE,
            Characteristic.WRITEABLE,
            CharacteristicValue(write=on_audio_control_point_write)
        )
        audio_status_characteristic = Characteristic(
            ASHA_AUDIO_STATUS_CHARACTERISTIC,
            Characteristic.READ | Characteristic.NOTIFY,
            Characteristic.READABLE,
            bytes([0])
        )
        volume_characteristic = Characteristic(
            ASHA_VOLUME_CHARACTERISTIC,
            Characteristic.WRITE_WITHOUT_RESPONSE,
            Characteristic.WRITEABLE,
            CharacteristicValue(write=on_volume_write)
        )
        le_psm_out_characteristic = Characteristic(
            ASHA_LE_PSM_OUT_CHARACTERISTIC,
            Characteristic.READ,
            Characteristic.READABLE,
            struct.pack('<H', psm)
        )
        device.add_service(Service(
            ASHA_SERVICE,
            [
                read_only_properties_characteristic,
                audio_control_point_characteristic,
                audio_status_characteristic,
                volume_characteristic,
                le_psm_out_characteristic
            ]
        ))

        # Set the advertising data
        device.advertising_data = bytes(
            AdvertisingData([
                (AdvertisingData.COMPLETE_LOCAL_NAME, bytes(device.name, 'utf-8')),
                (AdvertisingData.FLAGS, bytes([0x06])),
                (AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, bytes(ASHA_SERVICE)),
                (AdvertisingData.SERVICE_DATA_16_BIT_UUID, bytes(ASHA_SERVICE) + bytes([
                    0x01,  # Protocol Version
                    0x00,  # Capability
                    0x01, 0x02, 0x03, 0x04  # Truncated HiSyncID
                ]))
            ])
        )

        # Go!
        await device.power_on()
        await device.start_advertising(auto_restart=True)

        await hci_source.wait_for_termination()

# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())