asha: import ASHA Pandora service from AOSP

bumble/pandora/__init__.py

@@ -24,8 +24,10 @@ import grpc.aio
 
 from .config import Config
 from .device import PandoraDevice
+from .asha import AshaService
 from .host import HostService
 from .security import SecurityService, SecurityStorageService
+from pandora.asha_grpc_aio import add_ASHAServicer_to_server
 from pandora.host_grpc_aio import add_HostServicer_to_server
 from pandora.security_grpc_aio import (
     add_SecurityServicer_to_server,
@@ -68,6 +70,7 @@ async def serve(
             config.load_from_dict(bumble.config.get('server', {}))
 
             # add Pandora services to the gRPC server.
+            add_ASHAServicer_to_server(AshaService(bumble.device), server)
             add_HostServicer_to_server(
                 HostService(server, bumble.device, config), server
             )

bumble/pandora/asha.py

@@ -0,0 +1,96 @@
+# Copyright 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.
+
+import asyncio
+import grpc
+import logging
+
+from bumble.decoder import G722Decoder
+from bumble.device import Connection, Device
+from bumble.pandora import utils
+from bumble.profiles import asha_service
+from google.protobuf.empty_pb2 import Empty  # pytype: disable=pyi-error
+from pandora.asha_grpc_aio import ASHAServicer
+from pandora.asha_pb2 import CaptureAudioRequest, CaptureAudioResponse, RegisterRequest
+from typing import AsyncGenerator, Optional
+
+
+class AshaService(ASHAServicer):
+    DECODE_FRAME_LENGTH = 80
+
+    device: Device
+    asha_service: Optional[asha_service.AshaService]
+
+    def __init__(self, device: Device) -> None:
+        self.log = utils.BumbleServerLoggerAdapter(
+            logging.getLogger(), {"service_name": "Asha", "device": device}
+        )
+        self.device = device
+        self.asha_service = None
+
+    @utils.rpc
+    async def Register(
+        self, request: RegisterRequest, context: grpc.ServicerContext
+    ) -> Empty:
+        logging.info("Register")
+        if self.asha_service:
+            self.asha_service.capability = request.capability
+            self.asha_service.hisyncid = request.hisyncid
+        else:
+            self.asha_service = asha_service.AshaService(
+                request.capability, request.hisyncid, self.device
+            )
+            self.device.add_service(self.asha_service)  # type: ignore[no-untyped-call]
+        return Empty()
+
+    @utils.rpc
+    async def CaptureAudio(
+        self, request: CaptureAudioRequest, context: grpc.ServicerContext
+    ) -> AsyncGenerator[CaptureAudioResponse, None]:
+        connection_handle = int.from_bytes(request.connection.cookie.value, "big")
+        logging.info(f"CaptureAudioData connection_handle:{connection_handle}")
+
+        if not (connection := self.device.lookup_connection(connection_handle)):
+            raise RuntimeError(
+                f"Unknown connection for connection_handle:{connection_handle}"
+            )
+
+        decoder = G722Decoder()  # type: ignore
+        queue: asyncio.Queue[bytes] = asyncio.Queue()
+
+        def on_data(asha_connection: Connection, data: bytes) -> None:
+            if asha_connection == connection:
+                queue.put_nowait(data)
+
+        self.asha_service.on("data", on_data)  # type: ignore
+
+        try:
+            while data := await queue.get():
+                output_bytes = bytearray()
+                # First byte is sequence number, last 160 bytes are audio payload.
+                audio_payload = data[1:]
+                data_length = int(len(audio_payload) / AshaService.DECODE_FRAME_LENGTH)
+                for i in range(0, data_length):
+                    input_data = audio_payload[
+                        i
+                        * AshaService.DECODE_FRAME_LENGTH : i
+                        * AshaService.DECODE_FRAME_LENGTH
+                        + AshaService.DECODE_FRAME_LENGTH
+                    ]
+                    decoded_data = decoder.decode_frame(input_data)
+                    output_bytes.extend(decoded_data)
+
+                yield CaptureAudioResponse(data=bytes(output_bytes))
+        finally:
+            self.asha_service.remove_listener("data", on_data)  # type: ignore

bumble/profiles/asha_service.py

@@ -32,6 +32,7 @@ from ..gatt import (
     Characteristic,
     CharacteristicValue,
 )
+from ..l2cap import Channel
 from ..utils import AsyncRunner
 
 # -----------------------------------------------------------------------------
@@ -52,46 +53,48 @@ class AshaService(TemplateService):
     SUPPORTED_CODEC_ID = [0x02, 0x01]  # Codec IDs [G.722 at 16 kHz]
     RENDER_DELAY = [00, 00]
 
-    def __init__(self, capability: int, hisyncid: List[int], device: Device, psm=0):
+    def __init__(
+        self, capability: int, hisyncid: List[int], device: Device, psm: int = 0
+    ) -> None:
         self.hisyncid = hisyncid
         self.capability = capability  # Device Capabilities [Left, Monaural]
         self.device = device
-        self.audio_out_data = b''
-        self.psm = psm  # a non-zero psm is mainly for testing purpose
+        self.audio_out_data = b""
+        self.psm: int = psm  # a non-zero psm is mainly for testing purpose
 
         # Handler for volume control
-        def on_volume_write(connection, value):
-            logger.info(f'--- VOLUME Write:{value[0]}')
-            self.emit('volume', connection, value[0])
+        def on_volume_write(connection: Connection, value: bytes) -> None:
+            logger.info(f"--- VOLUME Write:{value[0]}")
+            self.emit("volume", connection, value[0])
 
         # Handler for audio control commands
-        def on_audio_control_point_write(connection: Connection, value):
-            logger.info(f'--- AUDIO CONTROL POINT Write:{value.hex()}')
+        def on_audio_control_point_write(connection: Connection, value: bytes) -> None:
+            logger.info(f"--- AUDIO CONTROL POINT Write:{value.hex()}")
             opcode = value[0]
             if opcode == AshaService.OPCODE_START:
                 # Start
-                audio_type = ('Unknown', 'Ringtone', 'Phone Call', 'Media')[value[2]]
+                audio_type = ("Unknown", "Ringtone", "Phone Call", "Media")[value[2]]
                 logger.info(
-                    f'### START: codec={value[1]}, '
-                    f'audio_type={audio_type}, '
-                    f'volume={value[3]}, '
-                    f'otherstate={value[4]}'
+                    f"### START: codec={value[1]}, "
+                    f"audio_type={audio_type}, "
+                    f"volume={value[3]}, "
+                    f"otherstate={value[4]}"
                 )
                 self.emit(
-                    'start',
+                    "start",
                     connection,
                     {
-                        'codec': value[1],
-                        'audiotype': value[2],
-                        'volume': value[3],
-                        'otherstate': value[4],
+                        "codec": value[1],
+                        "audiotype": value[2],
+                        "volume": value[3],
+                        "otherstate": value[4],
                     },
                 )
             elif opcode == AshaService.OPCODE_STOP:
-                logger.info('### STOP')
-                self.emit('stop', connection)
+                logger.info("### STOP")
+                self.emit("stop", connection)
             elif opcode == AshaService.OPCODE_STATUS:
-                logger.info(f'### STATUS: connected={value[1]}')
+                logger.info(f"### STATUS: connected={value[1]}")
 
             # OPCODE_STATUS does not need audio status point update
             if opcode != AshaService.OPCODE_STATUS:
@@ -101,49 +104,59 @@ class AshaService(TemplateService):
                     )
                 )
 
+        def on_read_only_properties_read(connection: Connection) -> bytes:
+            value = (
+                bytes(
+                    [
+                        AshaService.PROTOCOL_VERSION,  # Version
+                        self.capability,
+                    ]
+                )
+                + bytes(self.hisyncid)
+                + bytes(AshaService.FEATURE_MAP)
+                + bytes(AshaService.RENDER_DELAY)
+                + bytes(AshaService.RESERVED_FOR_FUTURE_USE)
+                + bytes(AshaService.SUPPORTED_CODEC_ID)
+            )
+            self.emit("read_only_properties", connection, value)
+            return value
+
+        def on_le_psm_out_read(connection: Connection) -> bytes:
+            self.emit("le_psm_out", connection, self.psm)
+            return struct.pack("<H", self.psm)
+
         self.read_only_properties_characteristic = Characteristic(
             GATT_ASHA_READ_ONLY_PROPERTIES_CHARACTERISTIC,
-            Characteristic.Properties.READ,
+            Characteristic.READ,
             Characteristic.READABLE,
-            bytes(
-                [
-                    AshaService.PROTOCOL_VERSION,  # Version
-                    self.capability,
-                ]
-            )
-            + bytes(self.hisyncid)
-            + bytes(AshaService.FEATURE_MAP)
-            + bytes(AshaService.RENDER_DELAY)
-            + bytes(AshaService.RESERVED_FOR_FUTURE_USE)
-            + bytes(AshaService.SUPPORTED_CODEC_ID),
+            CharacteristicValue(read=on_read_only_properties_read),
         )
 
         self.audio_control_point_characteristic = Characteristic(
             GATT_ASHA_AUDIO_CONTROL_POINT_CHARACTERISTIC,
-            Characteristic.Properties.WRITE
-            | Characteristic.Properties.WRITE_WITHOUT_RESPONSE,
+            Characteristic.WRITE | Characteristic.WRITE_WITHOUT_RESPONSE,
             Characteristic.WRITEABLE,
             CharacteristicValue(write=on_audio_control_point_write),
         )
         self.audio_status_characteristic = Characteristic(
             GATT_ASHA_AUDIO_STATUS_CHARACTERISTIC,
-            Characteristic.Properties.READ | Characteristic.Properties.NOTIFY,
+            Characteristic.READ | Characteristic.NOTIFY,
             Characteristic.READABLE,
             bytes([0]),
         )
         self.volume_characteristic = Characteristic(
             GATT_ASHA_VOLUME_CHARACTERISTIC,
-            Characteristic.Properties.WRITE_WITHOUT_RESPONSE,
+            Characteristic.WRITE_WITHOUT_RESPONSE,
             Characteristic.WRITEABLE,
             CharacteristicValue(write=on_volume_write),
         )
 
         # Register an L2CAP CoC server
-        def on_coc(channel):
-            def on_data(data):
-                logging.debug(f'<<< data received:{data}')
+        def on_coc(channel: Channel) -> None:
+            def on_data(data: bytes) -> None:
+                logging.debug(f"data received:{data.hex()}")
 
-                self.emit('data', channel.connection, data)
+                self.emit("data", channel.connection, data)
                 self.audio_out_data += data
 
             channel.sink = on_data
@@ -152,9 +165,9 @@ class AshaService(TemplateService):
         self.psm = self.device.register_l2cap_channel_server(self.psm, on_coc, 8)
         self.le_psm_out_characteristic = Characteristic(
             GATT_ASHA_LE_PSM_OUT_CHARACTERISTIC,
-            Characteristic.Properties.READ,
+            Characteristic.READ,
             Characteristic.READABLE,
-            struct.pack('<H', self.psm),
+            CharacteristicValue(read=on_le_psm_out_read),
         )
 
         characteristics = [
@@ -167,7 +180,7 @@ class AshaService(TemplateService):
 
         super().__init__(characteristics)
 
-    def get_advertising_data(self):
+    def get_advertising_data(self) -> bytes:
         # Advertisement only uses 4 least significant bytes of the HiSyncId.
         return bytes(
             AdvertisingData(