Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 17 additions & 7 deletions src/BTDevicesManager/bluetoothctl.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ class Bluetoothctl:
max_start_attempts = 3
retry_delay = 2
retry_cooldown = 60
prompt = compile(r"\[[^\]\r\n]+\][#>]\s*(?:\x1b\[[0-9;]*m)?")
ansi_escape = compile(r'\x1B\[[0-?]*[ -/]*[@-~]')

def __init__(self):
self.process = None
Expand Down Expand Up @@ -93,8 +95,13 @@ def _start_bluetoothctl(self):
print(f"Trying to start bluetoothctl ({attempt}/{self.max_start_attempts})...")
try:
process = spawnu("bluetoothctl", echo=False, timeout=10)
result = process.expect(["Agent registered", r"\[bluetooth\]#", EOF, TIMEOUT], timeout=10)
result = process.expect(["Agent registered", self.prompt, EOF, TIMEOUT], timeout=10)
if result in (0, 1):
# BlueZ 5.87 emits its first prompt only after input; consume
# it so command output is not shifted by one request.
process.send("\n")
if process.expect([self.prompt, EOF, TIMEOUT], timeout=10):
raise RuntimeError("bluetoothctl prompt not available")
self.process = process
self.isReady = True
self.start_error = None
Expand Down Expand Up @@ -137,9 +144,9 @@ def send(self, command, pause=0):
try:
self.process.send(f"{command}\n")
sleep(pause)
result = self.process.expect(["#", EOF, TIMEOUT], timeout=10)
result = self.process.expect([self.prompt, EOF, TIMEOUT], timeout=10)
if result:
self.isReady = True
self.isReady = False
if result == 1:
self._close_process()
raise Exception(f"bluetoothctl failed after {command}")
Expand Down Expand Up @@ -187,6 +194,9 @@ def parse_device_info(self, info_string):
device = {}
block_list = ["[\x1b[0;", "removed"]
if not any(keyword in info_string for keyword in block_list):
# BlueZ 5.87 wraps devices that advertise as non-discoverable in
# COLOR_BOLDGRAY, which would otherwise end up in the name.
info_string = self.ansi_escape.sub("", info_string)
try:
device_position = info_string.index("Device")
except ValueError:
Expand Down Expand Up @@ -277,8 +287,7 @@ def pair(self, mac_address):
res = self.process.expect(["Request confirmation", EOF])
return res == 0
elif res in [2, 3]:
ansi_escape = compile(r'\x1B\[[0-?]*[ -/]*[@-~]')
self.passkey = ansi_escape.sub('', str(self.process.buffer))
self.passkey = self.ansi_escape.sub('', str(self.process.buffer))
return False
else:
print(f"Failed to pair. Res = {res}")
Expand Down Expand Up @@ -332,9 +341,10 @@ def disconnect(self, mac_address):
return False
else:
res = self.process.expect(
["Failed to disconnect", "Successful disconnected", EOF]
# BlueZ renamed the message after 5.70; keep both spellings.
["Failed to disconnect", "Disconnection successful", "Successful disconnected", EOF]
)
return res == 1
return res in (1, 2)

def agent_noinputnooutput(self):
"""Start agent"""
Expand Down