Last active
June 2, 2023 12:02
-
-
Save Langerz82/5a5508ce5ce5b43ce5b709b4e6659246 to your computer and use it in GitHub Desktop.
EmuELEC dev - emuelec-bluetooth - v3 - issues - see comments.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
from subprocess import Popen, PIPE | |
from dataclasses import dataclass | |
import time | |
DEBUG = True | |
@dataclass | |
class BluetoothDevice: | |
mac: str | |
name: str | |
alias: str | |
clas: str | |
icon: str | |
paired: bool | |
bounded: bool | |
trusted: bool | |
blocked: bool | |
connected: bool | |
wake_allowed: bool | |
legacy_pairing: bool | |
rssi: int | |
class BluetoothCTL: | |
def _execute(self, cmd: list[str]) -> list[str]: | |
if DEBUG: | |
print("> " + " ".join(x for x in cmd)) | |
out: list[str] = [] | |
with Popen(cmd, stdout=PIPE, bufsize=1, universal_newlines=True) as p: | |
for line in p.stdout: | |
out.append(line) | |
if DEBUG: | |
print(line) | |
return out | |
def _execute_lookup(self, cmd: list[str], match: str) -> str: | |
out = self._execute(cmd) | |
for o in out: | |
if match in o: | |
return o | |
return "" | |
def _execute_async(self, cmd: list[str]): | |
Popen(cmd, stdout=PIPE, bufsize=1, universal_newlines=True) | |
def _parse_device_info(self, mac: str): | |
info = self._execute(["bluetoothctl", "info", mac]) | |
dev = BluetoothDevice( | |
mac=mac, | |
name="", | |
alias="", | |
clas="", | |
icon="", | |
paired=False, | |
bounded=False, | |
trusted=False, | |
blocked=False, | |
connected=False, | |
wake_allowed=False, | |
legacy_pairing=False, | |
rssi=0, | |
) | |
print(info) | |
for i in info: | |
if "Name:" in i: | |
dev.name = i.split(":")[1].strip("\n ") # bug - default strip make name and alias blank. | |
if "Alias:" in i: | |
dev.alias = i.split(":")[1].strip("\n ") | |
if "Class:" in i: | |
dev.clas = i.split(":")[1].strip() | |
if "Icon:" in i: | |
dev.icon = i.split(":")[1].strip() | |
if "Paired:" in i: | |
dev.paired = i.split(":")[1].strip() == "yes" | |
if "Bounded:" in i: | |
dev.bounded = i.split(":")[1].strip() == "yes" | |
if "Trusted:" in i: | |
dev.trusted = i.split(":")[1].strip() == "yes" | |
if "Blocked:" in i: | |
dev.blocked = i.split(":")[1].strip() == "yes" | |
if "Connected:" in i: | |
dev.connected = i.split(":")[1].strip() == "yes" | |
if "WakeAllowed:" in i: | |
dev.wake_allowed = i.split(":")[1].strip() == "yes" | |
if "LegacyPairing:" in i: | |
dev.legacy_pairing = i.split(":")[1].strip() == "yes" | |
if "RSSI:" in i: | |
dev.rssi = int(i.split(":")[1].strip()) # bug - my PS3 Gamepad does not have this value. | |
print(dev.name) | |
print(dev.alias) | |
return dev | |
@property | |
def power(self) -> bool: | |
return "yes" in self._execute_lookup(["bluetoothctl", "show"], "Pairable") | |
@property | |
def discoverable(self) -> bool: | |
return "yes" in self._execute_lookup(["bluetoothctl", "show"], "Discoverable") | |
@property | |
def pairable(self) -> bool: | |
return "yes" in self._execute_lookup(["bluetoothctl", "show"], "Pairable") | |
@power.setter | |
def power(self, value): | |
self._execute(["bluetoothctl", "power", "on" if value else "off"]) | |
@discoverable.setter | |
def discoverable(self, value): | |
self._execute(["bluetoothctl", "discoverable", "on" if value else "off"]) | |
@pairable.setter | |
def pairable(self, value): | |
self._execute(["bluetoothctl", "pairable", "on" if value else "off"]) | |
@pairable.setter | |
def agent(self, value): | |
self._execute(["bluetoothctl", "agent", "on" if value else "off"]) | |
def scan(self, timeout=10): | |
self._execute(["bluetoothctl", "--timeout", str(timeout), "scan", "on"]) | |
def scan_async(self, timeout=10): | |
self._execute_async(["bluetoothctl", "--timeout", str(timeout), "scan", "on"]) | |
def devices(self) -> list[BluetoothDevice]: | |
devs = self._execute(["bluetoothctl", "devices"]) | |
return [self._parse_device_info(dev.split()[1]) for dev in devs] | |
def trust(self, dev: BluetoothDevice) -> bool: | |
print("trusting.") # debugging | |
return "succeeded" in self._execute_lookup( | |
["bluetoothctl", "trust", dev.mac], "trust succeeded" | |
) | |
def untrust(self, dev: BluetoothDevice) -> bool: | |
print("untrusting.") # debugging | |
return "succeeded" in self._execute_lookup( | |
["bluetoothctl", "untrust", dev.mac], "untrust succeeded" | |
) | |
def pair(self, dev: BluetoothDevice) -> bool: | |
print("pairing.") # debugging | |
return "successful" in self._execute_lookup( | |
["bluetoothctl", "pair", dev.mac], "Pairing successful" | |
) | |
def connect(self, dev: BluetoothDevice) -> bool: | |
print("connecting") # debugging | |
return "successful" in self._execute_lookup( | |
["bluetoothctl", "connect", dev.mac], "Connection successful" | |
) | |
def forget(self, dev: BluetoothDevice) -> bool: | |
print("forgeting") # debugging | |
return "removed" in self._execute_lookup( | |
["bluetoothctl", "remove", dev.mac], "Device has been removed" | |
) | |
SCAN_TIME = 150 | |
SCAN_INTERVAL = 15 | |
if __name__ == "__main__": | |
bt = BluetoothCTL() | |
bt.power = True | |
bt.agent = True | |
bt.discoverable = True | |
bt.pairable = True | |
print( | |
"Scanning available devices for {} seconds, with interval {}, please wait...".format( | |
SCAN_TIME, SCAN_INTERVAL | |
) | |
) | |
bt.scan_async(timeout=SCAN_TIME) | |
for i in range(SCAN_TIME // SCAN_INTERVAL): | |
print("Starting scan {}, please wait...".format(i + 1)) | |
time.sleep(SCAN_INTERVAL) | |
devices = bt.devices() | |
for dev in devices: | |
print( | |
"detecting device {}, mac: {}, icon: {}, paired: {}".format( | |
dev.name, dev.mac, dev.icon, dev.paired | |
) | |
) | |
if ( | |
not (dev.paired and dev.connected) # think this will work instead | |
#and dev.rssi != 0 # exclude inactive devices (saved) # bug - rssi for my gamepad is 0 so needed to comment too to bypass issue. | |
and "input-" in dev.icon # exclude any non-input device | |
and len(dev.name.strip()) > 0 | |
): | |
print( | |
"found device {}, mac: {}, icon: {}, paired: {}".format( | |
dev.name, dev.mac, dev.icon, dev.paired | |
) | |
) | |
# handle if a device is trying to pair while it's already paired | |
if dev.paired: | |
print("pairing a paired device {}".format(dev.name)) | |
bt.forget(dev) | |
trusting = bt.trust(dev) # new code | |
if trusting: | |
pairing = bt.pair(dev) # new code | |
# for devices that need plugging into machine for security: | |
# they will trust and pair, but not connect so this instance needs to be handled. | |
if not (trusting and pairing): # new code | |
continue | |
# trust then pair and finally connect | |
if bt.connect(dev): # new code bug - see comments up. | |
print("successfully paired {}".format(dev.name)) | |
else: | |
# if for any reason one step fails, forget the device so next time it can be paired | |
bt.forget(dev) | |
print("failed to pair {}, try again".format(dev.name)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment