diff --git a/bumble/hfp.py b/bumble/hfp.py index 27bb0976..5adfcb90 100644 --- a/bumble/hfp.py +++ b/bumble/hfp.py @@ -22,7 +22,8 @@ import dataclasses import enum import traceback import pyee -from typing import Dict, List, Union, Set, Any, Optional, TYPE_CHECKING +from typing import Dict, List, Union, Set, Any, Optional, Type, TYPE_CHECKING +from typing_extensions import Self from bumble import at from bumble import rfcomm @@ -417,17 +418,21 @@ class AtResponseType(enum.Enum): MULTIPLE = 2 +@dataclasses.dataclass class AtResponse: code: str parameters: list - def __init__(self, response: bytearray): - code_and_parameters = response.split(b':') + @classmethod + def parse_from(cls: Type[Self], buffer: bytearray) -> Self: + code_and_parameters = buffer.split(b':') parameters = ( code_and_parameters[1] if len(code_and_parameters) > 1 else bytearray() ) - self.code = code_and_parameters[0].decode() - self.parameters = at.parse_parameters(parameters) + return cls( + code=code_and_parameters[0].decode(), + parameters=at.parse_parameters(parameters), + ) @dataclasses.dataclass @@ -530,7 +535,7 @@ class HfProtocol(pyee.EventEmitter): # Isolate the AT response code and parameters. raw_response = self.read_buffer[header + 2 : trailer] - response = AtResponse(raw_response) + response = AtResponse.parse_from(raw_response) logger.debug(f"<<< {raw_response.decode()}") # Consume the response bytes. @@ -1006,7 +1011,9 @@ class EscoParameters: transmit_coding_format: CodingFormat receive_coding_format: CodingFormat packet_type: HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType - retransmission_effort: HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort + retransmission_effort: ( + HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort + ) max_latency: int # Common @@ -1014,12 +1021,12 @@ class EscoParameters: output_coding_format: CodingFormat = CodingFormat(CodecID.LINEAR_PCM) input_coded_data_size: int = 16 output_coded_data_size: int = 16 - input_pcm_data_format: HCI_Enhanced_Setup_Synchronous_Connection_Command.PcmDataFormat = ( - HCI_Enhanced_Setup_Synchronous_Connection_Command.PcmDataFormat.TWOS_COMPLEMENT - ) - output_pcm_data_format: HCI_Enhanced_Setup_Synchronous_Connection_Command.PcmDataFormat = ( - HCI_Enhanced_Setup_Synchronous_Connection_Command.PcmDataFormat.TWOS_COMPLEMENT - ) + input_pcm_data_format: ( + HCI_Enhanced_Setup_Synchronous_Connection_Command.PcmDataFormat + ) = HCI_Enhanced_Setup_Synchronous_Connection_Command.PcmDataFormat.TWOS_COMPLEMENT + output_pcm_data_format: ( + HCI_Enhanced_Setup_Synchronous_Connection_Command.PcmDataFormat + ) = HCI_Enhanced_Setup_Synchronous_Connection_Command.PcmDataFormat.TWOS_COMPLEMENT input_pcm_sample_payload_msb_position: int = 0 output_pcm_sample_payload_msb_position: int = 0 input_data_path: HCI_Enhanced_Setup_Synchronous_Connection_Command.DataPath = ( diff --git a/examples/hfp_handsfree.html b/examples/hfp_handsfree.html index a86fc4ab..30be04e4 100644 --- a/examples/hfp_handsfree.html +++ b/examples/hfp_handsfree.html @@ -1,79 +1,132 @@ - - - - - - Server Port
- AT Command
- Dial Phone Number
- - - -
-
- + + - function answer() { - send({ type:'at_command', command: 'ATA' }) - } - - function hangup() { - send({ type:'at_command', command: 'AT+CHUP' }) - } - - function dial() { - send({ type:'at_command', command: `ATD${dialNumberInput.value}` }) - } - - function start_voice_assistant() { - send(({ type:'at_command', command: 'AT+BVRA=1' })) - } - - function stop_voice_assistant() { - send(({ type:'at_command', command: 'AT+BVRA=0' })) - } - - - + \ No newline at end of file diff --git a/examples/run_hfp_handsfree.py b/examples/run_hfp_handsfree.py index f4e445ee..40d36bde 100644 --- a/examples/run_hfp_handsfree.py +++ b/examples/run_hfp_handsfree.py @@ -16,6 +16,7 @@ # Imports # ----------------------------------------------------------------------------- import asyncio +import contextlib import sys import os import logging @@ -31,39 +32,16 @@ from bumble.transport import open_transport_or_link from bumble import hfp from bumble.hfp import HfProtocol - -# ----------------------------------------------------------------------------- -class UiServer: - protocol: Optional[HfProtocol] = None - - async def start(self): - """Start a Websocket server to receive events from a web page.""" - - async def serve(websocket, _path): - while True: - try: - message = await websocket.recv() - print('Received: ', str(message)) - - parsed = json.loads(message) - message_type = parsed['type'] - if message_type == 'at_command': - if self.protocol is not None: - await self.protocol.execute_command(parsed['command']) - - except websockets.exceptions.ConnectionClosedOK: - pass - - # pylint: disable=no-member - await websockets.serve(serve, 'localhost', 8989) +ws: Optional[websockets.WebSocketServerProtocol] = None +hf_protocol: Optional[HfProtocol] = None # ----------------------------------------------------------------------------- def on_dlc(dlc: rfcomm.DLC, configuration: hfp.Configuration): print('*** DLC connected', dlc) - protocol = HfProtocol(dlc, configuration) - UiServer.protocol = protocol - asyncio.create_task(protocol.run()) + global hf_protocol + hf_protocol = HfProtocol(dlc, configuration) + asyncio.create_task(hf_protocol.run()) def on_sco_request(connection: Connection, link_type: int, protocol: HfProtocol): if connection == protocol.dlc.multiplexer.l2cap_channel.connection: @@ -88,7 +66,7 @@ def on_dlc(dlc: rfcomm.DLC, configuration: hfp.Configuration): ), ) - handler = functools.partial(on_sco_request, protocol=protocol) + handler = functools.partial(on_sco_request, protocol=hf_protocol) dlc.multiplexer.l2cap_channel.connection.device.on('sco_request', handler) dlc.multiplexer.l2cap_channel.once( 'close', @@ -97,6 +75,13 @@ def on_dlc(dlc: rfcomm.DLC, configuration: hfp.Configuration): ), ) + def on_ag_indicator(indicator): + global ws + if ws: + asyncio.create_task(ws.send(str(indicator))) + + hf_protocol.on('ag_indicator', on_ag_indicator) + # ----------------------------------------------------------------------------- async def main(): @@ -154,8 +139,30 @@ async def main(): await device.set_connectable(True) # Start the UI websocket server to offer a few buttons and input boxes - ui_server = UiServer() - await ui_server.start() + async def serve(websocket: websockets.WebSocketServerProtocol, _path): + global ws + ws = websocket + async for message in websocket: + with contextlib.suppress(websockets.exceptions.ConnectionClosedOK): + print('Received: ', str(message)) + + parsed = json.loads(message) + message_type = parsed['type'] + if message_type == 'at_command': + if hf_protocol is not None: + response = str( + await hf_protocol.execute_command( + parsed['command'], + response_type=hfp.AtResponseType.MULTIPLE, + ) + ) + await websocket.send(response) + elif message_type == 'query_call': + if hf_protocol: + response = str(await hf_protocol.query_current_calls()) + await websocket.send(response) + + await websockets.serve(serve, 'localhost', 8989) await hci_source.wait_for_termination()