| # 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 warnings |
| from typing import Dict, List, Union, Set, TYPE_CHECKING |
| |
| from . import at |
| from . 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.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: |
| warnings.warn("See HfProtocol", DeprecationWarning) |
| 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 |
| # ----------------------------------------------------------------------------- |
| |
| |
| # 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 |
| class HfFeature(enum.IntFlag): |
| 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 |
| |
| |
| # 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 |
| class AgFeature(enum.IntFlag): |
| 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 |
| |
| |
| # Audio Codec IDs (normative). |
| # Hands-Free Profile v1.8, 10 Appendix B |
| class AudioCodec(enum.IntEnum): |
| CVSD = 0x01 # Support for CVSD audio codec |
| MSBC = 0x02 # Support for mSBC audio codec |
| |
| |
| # HF Indicators (normative). |
| # Bluetooth Assigned Numbers, 6.10.1 HF Indicators |
| class HfIndicator(enum.IntEnum): |
| ENHANCED_SAFETY = 0x01 # Enhanced safety feature |
| BATTERY_LEVEL = 0x02 # Battery level feature |
| |
| |
| # Call Hold supported operations (normative). |
| # AT Commands Reference Guide, 3.5.2.3.12 +CHLD - Call Holding Services |
| class CallHoldOperation(enum.IntEnum): |
| 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 |
| |
| |
| # Response Hold status (normative). |
| # Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 |
| # and 3GPP 27.007 |
| class ResponseHoldStatus(enum.IntEnum): |
| 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 |
| |
| |
| # 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 |
| class CallSetupAgIndicator(enum.IntEnum): |
| NOT_IN_CALL_SETUP = 0 |
| INCOMING_CALL_PROCESS = 1 |
| OUTGOING_CALL_SETUP = 2 |
| REMOTE_ALERTED = 3 # Remote party alerted in an outgoing call |
| |
| |
| # 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 |
| class CallHeldAgIndicator(enum.IntEnum): |
| 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 |
| |
| |
| # Call Info direction (normative). |
| # AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls |
| class CallInfoDirection(enum.IntEnum): |
| MOBILE_ORIGINATED_CALL = 0 |
| MOBILE_TERMINATED_CALL = 1 |
| |
| |
| # Call Info status (normative). |
| # AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls |
| class CallInfoStatus(enum.IntEnum): |
| ACTIVE = 0 |
| HELD = 1 |
| DIALING = 2 |
| ALERTING = 3 |
| INCOMING = 4 |
| WAITING = 5 |
| |
| |
| # Call Info mode (normative). |
| # AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls |
| class CallInfoMode(enum.IntEnum): |
| VOICE = 0 |
| DATA = 1 |
| FAX = 2 |
| UNKNOWN = 9 |
| |
| |
| # ----------------------------------------------------------------------------- |
| # 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): |
| """Indicate if a response is expected from an AT command, and if multiple |
| responses are accepted.""" |
| |
| NONE = 0 |
| SINGLE = 1 |
| MULTIPLE = 2 |
| |
| |
| class AtResponse: |
| code: str |
| parameters: list |
| |
| def __init__(self, response: bytearray): |
| code_and_parameters = response.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) |
| |
| |
| @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: |
| """Implementation for the Hands-Free side of the Hands-Free profile. |
| Reference specification Hands-Free Profile v1.8""" |
| |
| 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 |
| |
| def __init__(self, dlc: rfcomm.DLC, configuration: Configuration): |
| # 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() |
| |
| # 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 |
| |
| # Read AT messages from the RFCOMM channel. |
| # Enqueue AT commands, responses, unsolicited responses to their |
| # respective queues, and set the corresponding event. |
| def _read_at(self, data: bytes): |
| # 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(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}'") |
| |
| # Send an AT command and wait for the peer response. |
| # Wait for the AT responses sent by the peer, to the status code. |
| # Raises asyncio.TimeoutError if the status is not received |
| # after a timeout (default 1 second). |
| # Raises ProtocolError if the status is not OK. |
| async def execute_command( |
| self, |
| cmd: str, |
| timeout: float = 1.0, |
| response_type: AtResponseType = AtResponseType.NONE, |
| ) -> Union[None, AtResponse, List[AtResponse]]: |
| 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) |
| |
| # 4.2.1 Service Level Connection Initialization. |
| async def initiate_slc(self): |
| # 4.2.1.1 Supported features exchange |
| # First, in the initialization procedure, the HF shall send the |
| # AT+BRSF=<HF supported features> 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=<HF available codecs> 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=<HF supported HF indicators> 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") |
| |
| # 4.11.2 Audio Connection Setup by HF |
| async def setup_audio_connection(self): |
| # 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") |
| |
| # 4.11.3 Codec Connection Setup |
| async def setup_codec_connection(self, codec_id: int): |
| # The AG shall send a +BCS=<Codec ID> unsolicited response to the HF. |
| # The HF shall then respond to the incoming unsolicited response with |
| # the AT command AT+BCS=<Codec ID>. 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=<Codec ID>. |
| |
| logger.info("codec connection setup completed") |
| |
| # 4.13.1 Answer Incoming Call from the HF – In-Band Ringing |
| async def answer_incoming_call(self): |
| # 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") |
| |
| # 4.14.1 Reject an Incoming Call from the HF |
| async def reject_incoming_call(self): |
| # 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") |
| |
| # 4.15.1 Terminate a Call Process from the HF |
| async def terminate_call(self): |
| # 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 update_ag_indicator(self, index: int, value: int): |
| self.ag_indicators[index].current_status = value |
| logger.info( |
| f"AG indicator updated: {self.ag_indicators[index].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 rountine 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 |
| # ----------------------------------------------------------------------------- |
| |
| |
| # Profile version (normative). |
| # Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements |
| class ProfileVersion(enum.IntEnum): |
| V1_5 = 0x0105 |
| V1_6 = 0x0106 |
| V1_7 = 0x0107 |
| V1_8 = 0x0108 |
| V1_9 = 0x0109 |
| |
| |
| # HF supported features (normative). |
| # Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements |
| class HfSdpFeature(enum.IntFlag): |
| 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 |
| |
| |
| # AG supported features (normative). |
| # Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements |
| class AgSdpFeature(enum.IntFlag): |
| 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]: |
| """Generate 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), |
| ), |
| ] |
