# Copyright 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 # ----------------------------------------------------------------------------- import collections.abc import logging import asyncio import dataclasses import enum import traceback import pyee 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 from bumble.colors import color from bumble.core import ( ProtocolError, BT_GENERIC_AUDIO_SERVICE, BT_HANDSFREE_SERVICE, BT_L2CAP_PROTOCOL_ID, BT_RFCOMM_PROTOCOL_ID, ) from bumble.hci import ( HCI_Enhanced_Setup_Synchronous_Connection_Command, CodingFormat, CodecID, ) from bumble.sdp import ( DataElement, ServiceAttribute, SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID, ) # ----------------------------------------------------------------------------- # Logging # ----------------------------------------------------------------------------- logger = logging.getLogger(__name__) # ----------------------------------------------------------------------------- # Error # ----------------------------------------------------------------------------- class HfpProtocolError(ProtocolError): def __init__(self, error_name: str = '', details: str = ''): super().__init__(None, 'hfp', error_name, details) # ----------------------------------------------------------------------------- # Protocol Support # ----------------------------------------------------------------------------- # ----------------------------------------------------------------------------- class HfpProtocol: dlc: rfcomm.DLC buffer: str lines: collections.deque lines_available: asyncio.Event def __init__(self, dlc: rfcomm.DLC) -> None: self.dlc = dlc self.buffer = '' self.lines = collections.deque() self.lines_available = asyncio.Event() dlc.sink = self.feed def feed(self, data: Union[bytes, str]) -> None: # Convert the data to a string if needed if isinstance(data, bytes): data = data.decode('utf-8') logger.debug(f'<<< Data received: {data}') # Add to the buffer and look for lines self.buffer += data while (separator := self.buffer.find('\r')) >= 0: line = self.buffer[:separator].strip() self.buffer = self.buffer[separator + 1 :] if len(line) > 0: self.on_line(line) def on_line(self, line: str) -> None: self.lines.append(line) self.lines_available.set() def send_command_line(self, line: str) -> None: logger.debug(color(f'>>> {line}', 'yellow')) self.dlc.write(line + '\r') def send_response_line(self, line: str) -> None: logger.debug(color(f'>>> {line}', 'yellow')) self.dlc.write('\r\n' + line + '\r\n') async def next_line(self) -> str: await self.lines_available.wait() line = self.lines.popleft() if not self.lines: self.lines_available.clear() logger.debug(color(f'<<< {line}', 'green')) return line # ----------------------------------------------------------------------------- # Normative protocol definitions # ----------------------------------------------------------------------------- class HfFeature(enum.IntFlag): """ HF supported features (AT+BRSF=) (normative). Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007. """ EC_NR = 0x001 # Echo Cancel & Noise reduction THREE_WAY_CALLING = 0x002 CLI_PRESENTATION_CAPABILITY = 0x004 VOICE_RECOGNITION_ACTIVATION = 0x008 REMOTE_VOLUME_CONTROL = 0x010 ENHANCED_CALL_STATUS = 0x020 ENHANCED_CALL_CONTROL = 0x040 CODEC_NEGOTIATION = 0x080 HF_INDICATORS = 0x100 ESCO_S4_SETTINGS_SUPPORTED = 0x200 ENHANCED_VOICE_RECOGNITION_STATUS = 0x400 VOICE_RECOGNITION_TEST = 0x800 class AgFeature(enum.IntFlag): """ AG supported features (+BRSF:) (normative). Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007. """ THREE_WAY_CALLING = 0x001 EC_NR = 0x002 # Echo Cancel & Noise reduction VOICE_RECOGNITION_FUNCTION = 0x004 IN_BAND_RING_TONE_CAPABILITY = 0x008 VOICE_TAG = 0x010 # Attach a number to voice tag REJECT_CALL = 0x020 # Ability to reject a call ENHANCED_CALL_STATUS = 0x040 ENHANCED_CALL_CONTROL = 0x080 EXTENDED_ERROR_RESULT_CODES = 0x100 CODEC_NEGOTIATION = 0x200 HF_INDICATORS = 0x400 ESCO_S4_SETTINGS_SUPPORTED = 0x800 ENHANCED_VOICE_RECOGNITION_STATUS = 0x1000 VOICE_RECOGNITION_TEST = 0x2000 class AudioCodec(enum.IntEnum): """ Audio Codec IDs (normative). Hands-Free Profile v1.9, 11 Appendix B """ CVSD = 0x01 # Support for CVSD audio codec MSBC = 0x02 # Support for mSBC audio codec LC3_SWB = 0x03 # Support for LC3-SWB audio codec class HfIndicator(enum.IntEnum): """ HF Indicators (normative). Bluetooth Assigned Numbers, 6.10.1 HF Indicators. """ ENHANCED_SAFETY = 0x01 # Enhanced safety feature BATTERY_LEVEL = 0x02 # Battery level feature class CallHoldOperation(enum.IntEnum): """ Call Hold supported operations (normative). AT Commands Reference Guide, 3.5.2.3.12 +CHLD - Call Holding Services. """ RELEASE_ALL_HELD_CALLS = 0 # Release all held calls RELEASE_ALL_ACTIVE_CALLS = 1 # Release all active calls, accept other HOLD_ALL_ACTIVE_CALLS = 2 # Place all active calls on hold, accept other ADD_HELD_CALL = 3 # Adds a held call to conversation class ResponseHoldStatus(enum.IntEnum): """ Response Hold status (normative). Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007. """ INC_CALL_HELD = 0 # Put incoming call on hold HELD_CALL_ACC = 1 # Accept a held incoming call HELD_CALL_REJ = 2 # Reject a held incoming call class AgIndicator(enum.Enum): """ Values for the AG indicator (normative). Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007. """ SERVICE = 'service' CALL = 'call' CALL_SETUP = 'callsetup' CALL_HELD = 'callheld' SIGNAL = 'signal' ROAM = 'roam' BATTERY_CHARGE = 'battchg' class CallSetupAgIndicator(enum.IntEnum): """ Values for the Call Setup AG indicator (normative). Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007. """ NOT_IN_CALL_SETUP = 0 INCOMING_CALL_PROCESS = 1 OUTGOING_CALL_SETUP = 2 REMOTE_ALERTED = 3 # Remote party alerted in an outgoing call class CallHeldAgIndicator(enum.IntEnum): """ Values for the Call Held AG indicator (normative). Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007. """ NO_CALLS_HELD = 0 # Call is placed on hold or active/held calls swapped # (The AG has both an active AND a held call) CALL_ON_HOLD_AND_ACTIVE_CALL = 1 CALL_ON_HOLD_NO_ACTIVE_CALL = 2 # Call on hold, no active call class CallInfoDirection(enum.IntEnum): """ Call Info direction (normative). AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls. """ MOBILE_ORIGINATED_CALL = 0 MOBILE_TERMINATED_CALL = 1 class CallInfoStatus(enum.IntEnum): """ Call Info status (normative). AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls. """ ACTIVE = 0 HELD = 1 DIALING = 2 ALERTING = 3 INCOMING = 4 WAITING = 5 class CallInfoMode(enum.IntEnum): """ Call Info mode (normative). AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls. """ VOICE = 0 DATA = 1 FAX = 2 UNKNOWN = 9 class CallInfoMultiParty(enum.IntEnum): """ Call Info Multi-Party state (normative). AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls. """ NOT_IN_CONFERENCE = 0 IN_CONFERENCE = 1 @dataclasses.dataclass class CallInfo: """ Enhanced call status. AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls. """ index: int direction: CallInfoDirection status: CallInfoStatus mode: CallInfoMode multi_party: CallInfoMultiParty number: Optional[int] = None type: Optional[int] = None # ----------------------------------------------------------------------------- # Hands-Free Control Interoperability Requirements # ----------------------------------------------------------------------------- # Response codes. RESPONSE_CODES = [ "+APLSIRI", "+BAC", "+BCC", "+BCS", "+BIA", "+BIEV", "+BIND", "+BINP", "+BLDN", "+BRSF", "+BTRH", "+BVRA", "+CCWA", "+CHLD", "+CHUP", "+CIND", "+CLCC", "+CLIP", "+CMEE", "+CMER", "+CNUM", "+COPS", "+IPHONEACCEV", "+NREC", "+VGM", "+VGS", "+VTS", "+XAPL", "A", "D", ] # Unsolicited responses and statuses. UNSOLICITED_CODES = [ "+APLSIRI", "+BCS", "+BIND", "+BSIR", "+BTRH", "+BVRA", "+CCWA", "+CIEV", "+CLIP", "+VGM", "+VGS", "BLACKLISTED", "BUSY", "DELAYED", "NO ANSWER", "NO CARRIER", "RING", ] # Status codes STATUS_CODES = [ "+CME ERROR", "BLACKLISTED", "BUSY", "DELAYED", "ERROR", "NO ANSWER", "NO CARRIER", "OK", ] @dataclasses.dataclass class Configuration: supported_hf_features: List[HfFeature] supported_hf_indicators: List[HfIndicator] supported_audio_codecs: List[AudioCodec] class AtResponseType(enum.Enum): """ Indicates if a response is expected from an AT command, and if multiple responses are accepted. """ NONE = 0 SINGLE = 1 MULTIPLE = 2 @dataclasses.dataclass class AtResponse: code: str parameters: list @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() ) return cls( code=code_and_parameters[0].decode(), parameters=at.parse_parameters(parameters), ) @dataclasses.dataclass class AgIndicatorState: description: str index: int supported_values: Set[int] current_status: int @dataclasses.dataclass class HfIndicatorState: supported: bool = False enabled: bool = False class HfProtocol(pyee.EventEmitter): """ Implementation for the Hands-Free side of the Hands-Free profile. Reference specification Hands-Free Profile v1.8. Emitted events: codec_negotiation: When codec is renegotiated, notify the new codec. Args: active_codec: AudioCodec ag_indicator: When AG update their indicators, notify the new state. Args: ag_indicator: AgIndicator """ supported_hf_features: int supported_audio_codecs: List[AudioCodec] supported_ag_features: int supported_ag_call_hold_operations: List[CallHoldOperation] ag_indicators: List[AgIndicatorState] hf_indicators: Dict[HfIndicator, HfIndicatorState] dlc: rfcomm.DLC command_lock: asyncio.Lock if TYPE_CHECKING: response_queue: asyncio.Queue[AtResponse] unsolicited_queue: asyncio.Queue[AtResponse] else: response_queue: asyncio.Queue unsolicited_queue: asyncio.Queue read_buffer: bytearray active_codec: AudioCodec def __init__(self, dlc: rfcomm.DLC, configuration: Configuration) -> None: super().__init__() # Configure internal state. self.dlc = dlc self.command_lock = asyncio.Lock() self.response_queue = asyncio.Queue() self.unsolicited_queue = asyncio.Queue() self.read_buffer = bytearray() self.active_codec = AudioCodec.CVSD # Build local features. self.supported_hf_features = sum(configuration.supported_hf_features) self.supported_audio_codecs = configuration.supported_audio_codecs self.hf_indicators = { indicator: HfIndicatorState() for indicator in configuration.supported_hf_indicators } # Clear remote features. self.supported_ag_features = 0 self.supported_ag_call_hold_operations = [] self.ag_indicators = [] # Bind the AT reader to the RFCOMM channel. self.dlc.sink = self._read_at def supports_hf_feature(self, feature: HfFeature) -> bool: return (self.supported_hf_features & feature) != 0 def supports_ag_feature(self, feature: AgFeature) -> bool: return (self.supported_ag_features & feature) != 0 def _read_at(self, data: bytes): """ Reads AT messages from the RFCOMM channel. Enqueues AT commands, responses, unsolicited responses to their respective queues, and set the corresponding event. """ # Append to the read buffer. self.read_buffer.extend(data) # Locate header and trailer. header = self.read_buffer.find(b'\r\n') trailer = self.read_buffer.find(b'\r\n', header + 2) if header == -1 or trailer == -1: return # Isolate the AT response code and parameters. raw_response = self.read_buffer[header + 2 : trailer] response = AtResponse.parse_from(raw_response) logger.debug(f"<<< {raw_response.decode()}") # Consume the response bytes. self.read_buffer = self.read_buffer[trailer + 2 :] # Forward the received code to the correct queue. if self.command_lock.locked() and ( response.code in STATUS_CODES or response.code in RESPONSE_CODES ): self.response_queue.put_nowait(response) elif response.code in UNSOLICITED_CODES: self.unsolicited_queue.put_nowait(response) else: logger.warning(f"dropping unexpected response with code '{response.code}'") async def execute_command( self, cmd: str, timeout: float = 1.0, response_type: AtResponseType = AtResponseType.NONE, ) -> Union[None, AtResponse, List[AtResponse]]: """ Sends an AT command and wait for the peer response. Wait for the AT responses sent by the peer, to the status code. Args: cmd: the AT command in string to execute. timeout: timeout in float seconds. response_type: type of response. Raises: asyncio.TimeoutError: the status is not received after a timeout (default 1 second). ProtocolError: the status is not OK. """ async with self.command_lock: logger.debug(f">>> {cmd}") self.dlc.write(cmd + '\r') responses: List[AtResponse] = [] while True: result = await asyncio.wait_for( self.response_queue.get(), timeout=timeout ) if result.code == 'OK': if response_type == AtResponseType.SINGLE and len(responses) != 1: raise HfpProtocolError("NO ANSWER") if response_type == AtResponseType.MULTIPLE: return responses if response_type == AtResponseType.SINGLE: return responses[0] return None if result.code in STATUS_CODES: raise HfpProtocolError(result.code) responses.append(result) async def initiate_slc(self): """4.2.1 Service Level Connection Initialization.""" # 4.2.1.1 Supported features exchange # First, in the initialization procedure, the HF shall send the # AT+BRSF= command to the AG to both notify # the AG of the supported features in the HF, as well as to retrieve the # supported features in the AG using the +BRSF result code. response = await self.execute_command( f"AT+BRSF={self.supported_hf_features}", response_type=AtResponseType.SINGLE ) self.supported_ag_features = int(response.parameters[0]) logger.info(f"supported AG features: {self.supported_ag_features}") for feature in AgFeature: if self.supports_ag_feature(feature): logger.info(f" - {feature.name}") # 4.2.1.2 Codec Negotiation # Secondly, in the initialization procedure, if the HF supports the # Codec Negotiation feature, it shall check if the AT+BRSF command # response from the AG has indicated that it supports the Codec # Negotiation feature. if self.supports_hf_feature( HfFeature.CODEC_NEGOTIATION ) and self.supports_ag_feature(AgFeature.CODEC_NEGOTIATION): # If both the HF and AG do support the Codec Negotiation feature # then the HF shall send the AT+BAC= command to # the AG to notify the AG of the available codecs in the HF. codecs = [str(c) for c in self.supported_audio_codecs] await self.execute_command(f"AT+BAC={','.join(codecs)}") # 4.2.1.3 AG Indicators # After having retrieved the supported features in the AG, the HF shall # determine which indicators are supported by the AG, as well as the # ordering of the supported indicators. This is because, according to # the 3GPP 27.007 specification [2], the AG may support additional # indicators not provided for by the Hands-Free Profile, and because the # ordering of the indicators is implementation specific. The HF uses # the AT+CIND=? Test command to retrieve information about the supported # indicators and their ordering. response = await self.execute_command( "AT+CIND=?", response_type=AtResponseType.SINGLE ) self.ag_indicators = [] for index, indicator in enumerate(response.parameters): description = indicator[0].decode() supported_values = [] for value in indicator[1]: value = value.split(b'-') value = [int(v) for v in value] value_min = value[0] value_max = value[1] if len(value) > 1 else value[0] supported_values.extend([v for v in range(value_min, value_max + 1)]) self.ag_indicators.append( AgIndicatorState(description, index, set(supported_values), 0) ) # Once the HF has the necessary supported indicator and ordering # information, it shall retrieve the current status of the indicators # in the AG using the AT+CIND? Read command. response = await self.execute_command( "AT+CIND?", response_type=AtResponseType.SINGLE ) for index, indicator in enumerate(response.parameters): self.ag_indicators[index].current_status = int(indicator) # After having retrieved the status of the indicators in the AG, the HF # shall then enable the "Indicators status update" function in the AG by # issuing the AT+CMER command, to which the AG shall respond with OK. await self.execute_command("AT+CMER=3,,,1") if self.supports_hf_feature( HfFeature.THREE_WAY_CALLING ) and self.supports_ag_feature(HfFeature.THREE_WAY_CALLING): # After the HF has enabled the “Indicators status update” function in # the AG, and if the “Call waiting and 3-way calling” bit was set in the # supported features bitmap by both the HF and the AG, the HF shall # issue the AT+CHLD=? test command to retrieve the information about how # the call hold and multiparty services are supported in the AG. The HF # shall not issue the AT+CHLD=? test command in case either the HF or # the AG does not support the "Three-way calling" feature. response = await self.execute_command( "AT+CHLD=?", response_type=AtResponseType.SINGLE ) self.supported_ag_call_hold_operations = [ CallHoldOperation(int(operation)) for operation in response.parameters[0] if not b'x' in operation ] # 4.2.1.4 HF Indicators # If the HF supports the HF indicator feature, it shall check the +BRSF # response to see if the AG also supports the HF Indicator feature. if self.supports_hf_feature( HfFeature.HF_INDICATORS ) and self.supports_ag_feature(AgFeature.HF_INDICATORS): # If both the HF and AG support the HF Indicator feature, then the HF # shall send the AT+BIND= command to the AG # to notify the AG of the supported indicators’ assigned numbers in the # HF. The AG shall respond with OK indicators = [str(i) for i in self.hf_indicators.keys()] await self.execute_command(f"AT+BIND={','.join(indicators)}") # After having provided the AG with the HF indicators it supports, # the HF shall send the AT+BIND=? to request HF indicators supported # by the AG. The AG shall reply with the +BIND response listing all # HF indicators that it supports followed by an OK. response = await self.execute_command( "AT+BIND=?", response_type=AtResponseType.SINGLE ) logger.info("supported HF indicators:") for indicator in response.parameters[0]: indicator = HfIndicator(int(indicator)) logger.info(f" - {indicator.name}") if indicator in self.hf_indicators: self.hf_indicators[indicator].supported = True # Once the HF receives the supported HF indicators list from the AG, # the HF shall send the AT+BIND? command to determine which HF # indicators are enabled. The AG shall respond with one or more # +BIND responses. The AG shall terminate the list with OK. # (See Section 4.36.1.3). responses = await self.execute_command( "AT+BIND?", response_type=AtResponseType.MULTIPLE ) logger.info("enabled HF indicators:") for response in responses: indicator = HfIndicator(int(response.parameters[0])) enabled = int(response.parameters[1]) != 0 logger.info(f" - {indicator.name}: {enabled}") if indicator in self.hf_indicators: self.hf_indicators[indicator].enabled = True logger.info("SLC setup completed") async def setup_audio_connection(self): """4.11.2 Audio Connection Setup by HF.""" # When the HF triggers the establishment of the Codec Connection it # shall send the AT command AT+BCC to the AG. The AG shall respond with # OK if it will start the Codec Connection procedure, and with ERROR # if it cannot start the Codec Connection procedure. await self.execute_command("AT+BCC") async def setup_codec_connection(self, codec_id: int): """4.11.3 Codec Connection Setup.""" # The AG shall send a +BCS= unsolicited response to the HF. # The HF shall then respond to the incoming unsolicited response with # the AT command AT+BCS=. The ID shall be the same as in the # unsolicited response code as long as the ID is supported. # If the received ID is not available, the HF shall respond with # AT+BAC with its available codecs. if codec_id not in self.supported_audio_codecs: codecs = [str(c) for c in self.supported_audio_codecs] await self.execute_command(f"AT+BAC={','.join(codecs)}") return await self.execute_command(f"AT+BCS={codec_id}") # After sending the OK response, the AG shall open the # Synchronous Connection with the settings that are determined by the # ID. The HF shall be ready to accept the synchronous connection # establishment as soon as it has sent the AT commands AT+BCS=. self.active_codec = AudioCodec(codec_id) self.emit('codec_negotiation', self.active_codec) logger.info("codec connection setup completed") async def answer_incoming_call(self): """4.13.1 Answer Incoming Call from the HF - In-Band Ringing.""" # The user accepts the incoming voice call by using the proper means # provided by the HF. The HF shall then send the ATA command # (see Section 4.34) to the AG. The AG shall then begin the procedure for # accepting the incoming call. await self.execute_command("ATA") async def reject_incoming_call(self): """4.14.1 Reject an Incoming Call from the HF.""" # The user rejects the incoming call by using the User Interface on the # Hands-Free unit. The HF shall then send the AT+CHUP command # (see Section 4.34) to the AG. This may happen at any time during the # procedures described in Sections 4.13.1 and 4.13.2. await self.execute_command("AT+CHUP") async def terminate_call(self): """4.15.1 Terminate a Call Process from the HF.""" # The user may abort the ongoing call process using whatever means # provided by the Hands-Free unit. The HF shall send AT+CHUP command # (see Section 4.34) to the AG, and the AG shall then start the # procedure to terminate or interrupt the current call procedure. # The AG shall then send the OK indication followed by the +CIEV result # code, with the value indicating (call=0). await self.execute_command("AT+CHUP") async def query_current_calls(self) -> List[CallInfo]: """4.32.1 Query List of Current Calls in AG. Return: List of current calls in AG. """ responses = await self.execute_command( "AT+CLCC", response_type=AtResponseType.MULTIPLE ) assert isinstance(responses, list) calls = [] for response in responses: call_info = CallInfo( index=int(response.parameters[0]), direction=CallInfoDirection(int(response.parameters[1])), status=CallInfoStatus(int(response.parameters[2])), mode=CallInfoMode(int(response.parameters[3])), multi_party=CallInfoMultiParty(int(response.parameters[4])), ) if len(response.parameters) >= 7: call_info.number = int(response.parameters[5]) call_info.type = int(response.parameters[6]) calls.append(call_info) return calls async def update_ag_indicator(self, index: int, value: int): # CIEV is in 1-index, while ag_indicators is in 0-index. ag_indicator = self.ag_indicators[index - 1] ag_indicator.current_status = value self.emit('ag_indicator', ag_indicator) logger.info(f"AG indicator updated: {ag_indicator.description}, {value}") async def handle_unsolicited(self): """Handle unsolicited result codes sent by the audio gateway.""" result = await self.unsolicited_queue.get() if result.code == "+BCS": await self.setup_codec_connection(int(result.parameters[0])) elif result.code == "+CIEV": await self.update_ag_indicator( int(result.parameters[0]), int(result.parameters[1]) ) else: logging.info(f"unhandled unsolicited response {result.code}") async def run(self): """ Main routine for the Hands-Free side of the HFP protocol. Initiates the service level connection then loops handling unsolicited AG responses. """ try: await self.initiate_slc() while True: await self.handle_unsolicited() except Exception: logger.error("HFP-HF protocol failed with the following error:") logger.error(traceback.format_exc()) # ----------------------------------------------------------------------------- # Normative SDP definitions # ----------------------------------------------------------------------------- class ProfileVersion(enum.IntEnum): """ Profile version (normative). Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements. """ V1_5 = 0x0105 V1_6 = 0x0106 V1_7 = 0x0107 V1_8 = 0x0108 V1_9 = 0x0109 class HfSdpFeature(enum.IntFlag): """ HF supported features (normative). Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements. """ EC_NR = 0x01 # Echo Cancel & Noise reduction THREE_WAY_CALLING = 0x02 CLI_PRESENTATION_CAPABILITY = 0x04 VOICE_RECOGNITION_ACTIVATION = 0x08 REMOTE_VOLUME_CONTROL = 0x10 WIDE_BAND = 0x20 # Wide band speech ENHANCED_VOICE_RECOGNITION_STATUS = 0x40 VOICE_RECOGNITION_TEST = 0x80 class AgSdpFeature(enum.IntFlag): """ AG supported features (normative). Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements. """ THREE_WAY_CALLING = 0x01 EC_NR = 0x02 # Echo Cancel & Noise reduction VOICE_RECOGNITION_FUNCTION = 0x04 IN_BAND_RING_TONE_CAPABILITY = 0x08 VOICE_TAG = 0x10 # Attach a number to voice tag WIDE_BAND = 0x20 # Wide band speech ENHANCED_VOICE_RECOGNITION_STATUS = 0x40 VOICE_RECOGNITION_TEST = 0x80 def sdp_records( service_record_handle: int, rfcomm_channel: int, configuration: Configuration ) -> List[ServiceAttribute]: """ Generates the SDP record for HFP Hands-Free support. The record exposes the features supported in the input configuration, and the allocated RFCOMM channel. """ hf_supported_features = 0 if HfFeature.EC_NR in configuration.supported_hf_features: hf_supported_features |= HfSdpFeature.EC_NR if HfFeature.THREE_WAY_CALLING in configuration.supported_hf_features: hf_supported_features |= HfSdpFeature.THREE_WAY_CALLING if HfFeature.CLI_PRESENTATION_CAPABILITY in configuration.supported_hf_features: hf_supported_features |= HfSdpFeature.CLI_PRESENTATION_CAPABILITY if HfFeature.VOICE_RECOGNITION_ACTIVATION in configuration.supported_hf_features: hf_supported_features |= HfSdpFeature.VOICE_RECOGNITION_ACTIVATION if HfFeature.REMOTE_VOLUME_CONTROL in configuration.supported_hf_features: hf_supported_features |= HfSdpFeature.REMOTE_VOLUME_CONTROL if ( HfFeature.ENHANCED_VOICE_RECOGNITION_STATUS in configuration.supported_hf_features ): hf_supported_features |= HfSdpFeature.ENHANCED_VOICE_RECOGNITION_STATUS if HfFeature.VOICE_RECOGNITION_TEST in configuration.supported_hf_features: hf_supported_features |= HfSdpFeature.VOICE_RECOGNITION_TEST if AudioCodec.MSBC in configuration.supported_audio_codecs: hf_supported_features |= HfSdpFeature.WIDE_BAND return [ ServiceAttribute( SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, DataElement.unsigned_integer_32(service_record_handle), ), ServiceAttribute( SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, DataElement.sequence( [ DataElement.uuid(BT_HANDSFREE_SERVICE), DataElement.uuid(BT_GENERIC_AUDIO_SERVICE), ] ), ), ServiceAttribute( SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, DataElement.sequence( [ DataElement.sequence([DataElement.uuid(BT_L2CAP_PROTOCOL_ID)]), DataElement.sequence( [ DataElement.uuid(BT_RFCOMM_PROTOCOL_ID), DataElement.unsigned_integer_8(rfcomm_channel), ] ), ] ), ), ServiceAttribute( SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID, DataElement.sequence( [ DataElement.sequence( [ DataElement.uuid(BT_HANDSFREE_SERVICE), DataElement.unsigned_integer_16(ProfileVersion.V1_8), ] ) ] ), ), ServiceAttribute( SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID, DataElement.unsigned_integer_16(hf_supported_features), ), ] # ----------------------------------------------------------------------------- # ESCO Codec Default Parameters # ----------------------------------------------------------------------------- # Hands-Free Profile v1.8, 5.7 Codec Interoperability Requirements class DefaultCodecParameters(enum.IntEnum): SCO_CVSD_D0 = enum.auto() SCO_CVSD_D1 = enum.auto() ESCO_CVSD_S1 = enum.auto() ESCO_CVSD_S2 = enum.auto() ESCO_CVSD_S3 = enum.auto() ESCO_CVSD_S4 = enum.auto() ESCO_MSBC_T1 = enum.auto() ESCO_MSBC_T2 = enum.auto() @dataclasses.dataclass class EscoParameters: # Codec specific 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 ) max_latency: int # Common input_coding_format: CodingFormat = CodingFormat(CodecID.LINEAR_PCM) 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_sample_payload_msb_position: int = 0 output_pcm_sample_payload_msb_position: int = 0 input_data_path: HCI_Enhanced_Setup_Synchronous_Connection_Command.DataPath = ( HCI_Enhanced_Setup_Synchronous_Connection_Command.DataPath.HCI ) output_data_path: HCI_Enhanced_Setup_Synchronous_Connection_Command.DataPath = ( HCI_Enhanced_Setup_Synchronous_Connection_Command.DataPath.HCI ) input_transport_unit_size: int = 0 output_transport_unit_size: int = 0 input_bandwidth: int = 16000 output_bandwidth: int = 16000 transmit_bandwidth: int = 8000 receive_bandwidth: int = 8000 transmit_codec_frame_size: int = 60 receive_codec_frame_size: int = 60 def asdict(self) -> Dict[str, Any]: # dataclasses.asdict() will recursively deep-copy the entire object, # which is expensive and breaks CodingFormat object, so let it simply copy here. return self.__dict__ _ESCO_PARAMETERS_CVSD_D0 = EscoParameters( transmit_coding_format=CodingFormat(CodecID.CVSD), receive_coding_format=CodingFormat(CodecID.CVSD), max_latency=0xFFFF, packet_type=HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.HV1, retransmission_effort=HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort.NO_RETRANSMISSION, ) _ESCO_PARAMETERS_CVSD_D1 = EscoParameters( transmit_coding_format=CodingFormat(CodecID.CVSD), receive_coding_format=CodingFormat(CodecID.CVSD), max_latency=0xFFFF, packet_type=HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.HV3, retransmission_effort=HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort.NO_RETRANSMISSION, ) _ESCO_PARAMETERS_CVSD_S1 = EscoParameters( transmit_coding_format=CodingFormat(CodecID.CVSD), receive_coding_format=CodingFormat(CodecID.CVSD), max_latency=0x0007, packet_type=( HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV5 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV5 ), retransmission_effort=HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort.OPTIMIZE_FOR_POWER, ) _ESCO_PARAMETERS_CVSD_S2 = EscoParameters( transmit_coding_format=CodingFormat(CodecID.CVSD), receive_coding_format=CodingFormat(CodecID.CVSD), max_latency=0x0007, packet_type=( HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV5 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV5 ), retransmission_effort=HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort.OPTIMIZE_FOR_POWER, ) _ESCO_PARAMETERS_CVSD_S3 = EscoParameters( transmit_coding_format=CodingFormat(CodecID.CVSD), receive_coding_format=CodingFormat(CodecID.CVSD), max_latency=0x000A, packet_type=( HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV5 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV5 ), retransmission_effort=HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort.OPTIMIZE_FOR_POWER, ) _ESCO_PARAMETERS_CVSD_S4 = EscoParameters( transmit_coding_format=CodingFormat(CodecID.CVSD), receive_coding_format=CodingFormat(CodecID.CVSD), max_latency=0x000C, packet_type=( HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV5 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV5 ), retransmission_effort=HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort.OPTIMIZE_FOR_QUALITY, ) _ESCO_PARAMETERS_MSBC_T1 = EscoParameters( transmit_coding_format=CodingFormat(CodecID.MSBC), receive_coding_format=CodingFormat(CodecID.MSBC), max_latency=0x0008, packet_type=( HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV5 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV5 ), input_bandwidth=32000, output_bandwidth=32000, retransmission_effort=HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort.OPTIMIZE_FOR_QUALITY, ) _ESCO_PARAMETERS_MSBC_T2 = EscoParameters( transmit_coding_format=CodingFormat(CodecID.MSBC), receive_coding_format=CodingFormat(CodecID.MSBC), max_latency=0x000D, packet_type=( HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV3 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_2_EV5 | HCI_Enhanced_Setup_Synchronous_Connection_Command.PacketType.NO_3_EV5 ), input_bandwidth=32000, output_bandwidth=32000, retransmission_effort=HCI_Enhanced_Setup_Synchronous_Connection_Command.RetransmissionEffort.OPTIMIZE_FOR_QUALITY, ) ESCO_PARAMETERS = { DefaultCodecParameters.SCO_CVSD_D0: _ESCO_PARAMETERS_CVSD_D0, DefaultCodecParameters.SCO_CVSD_D1: _ESCO_PARAMETERS_CVSD_D1, DefaultCodecParameters.ESCO_CVSD_S1: _ESCO_PARAMETERS_CVSD_S1, DefaultCodecParameters.ESCO_CVSD_S2: _ESCO_PARAMETERS_CVSD_S2, DefaultCodecParameters.ESCO_CVSD_S3: _ESCO_PARAMETERS_CVSD_S3, DefaultCodecParameters.ESCO_CVSD_S4: _ESCO_PARAMETERS_CVSD_S4, DefaultCodecParameters.ESCO_MSBC_T1: _ESCO_PARAMETERS_MSBC_T1, DefaultCodecParameters.ESCO_MSBC_T2: _ESCO_PARAMETERS_MSBC_T2, }