jyapayne/nxbt
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bleak support

Browse source
This commit is contained in:
Joey Yakimowich-Payne 2025-11-12 06:41:11 -07:00
commit 1ebc0978cf
No known key found for this signature in database
GPG key ID: 6BFE655FA5ABD1E1
22 changed files with 1655 additions and 1621 deletions
Download patch file Download diff file Expand all files Collapse all files

56
scripts/demo_loop.py
View file

@ -1,7 +1,7 @@
import asyncio
from random import randint
from time import sleep
from nxbt import Nxbt, PRO_CONTROLLER
from nxbt import AsyncNxbtClient, PRO_CONTROLLER
MACRO = """
@ -69,39 +69,39 @@ def random_colour():
]
def demo():
async def demo():
"""Loops over all available Bluetooth adapters
and creates controllers on each. The last available adapter
is used to run a macro.
"""
nx = Nxbt(debug=False)
adapters = nx.get_available_adapters()
if len(adapters) < 1:
raise OSError("Unable to detect any Bluetooth adapters.")
async with AsyncNxbtClient(debug=False) as nx:
adapters = await nx.get_available_adapters()
if len(adapters) < 1:
raise OSError("Unable to detect any Bluetooth adapters.")
controller_idxs = []
for i in range(0, len(adapters)):
index = nx.create_controller(
PRO_CONTROLLER,
adapters[i],
colour_body=random_colour(),
colour_buttons=random_colour())
controller_idxs.append(index)
controller_idxs = []
for adapter in adapters:
index = await nx.create_controller(
PRO_CONTROLLER,
adapter,
colour_body=random_colour(),
colour_buttons=random_colour())
controller_idxs.append(index)
# Run a macro on the last controller
for i in range(100):
print(f"Running Demo: Iteration {i}")
macro_id = nx.macro(controller_idxs[-1], MACRO, block=False)
while macro_id not in nx.state[controller_idxs[-1]]["finished_macros"]:
state = nx.state[controller_idxs[-1]]
if state['state'] == 'crashed':
print("An error occurred while running the demo:")
print(state['errors'])
exit(1)
sleep(1.0)
# Run a macro on the last controller
for i in range(100):
print(f"Running Demo: Iteration {i}")
macro_id = await nx.macro(controller_idxs[-1], MACRO, block=False)
while macro_id not in nx.state[controller_idxs[-1]]["finished_macros"]:
state = nx.state[controller_idxs[-1]]
if state['state'] == 'crashed':
print("An error occurred while running the demo:")
print(state['errors'])
exit(1)
await asyncio.sleep(1.0)
print("Finished!")
print("Finished!")
if __name__ == "__main__":
demo()
asyncio.run(demo())

118
scripts/testbt.py
View file

@ -1,65 +1,73 @@
"""
A quick script to test aspects of the BlueZ API.
A quick script to test the async Bleak-backed adapter helper.
"""
import dbus
from nxbt import BlueZ, find_objects, SERVICE_NAME, ADAPTER_INTERFACE
import asyncio
import os
from nxbt import (
AsyncBleakAdapter,
async_find_objects,
find_objects,
SERVICE_NAME,
ADAPTER_INTERFACE,
)
TARGET_ADDRESS = os.environ.get("NXBT_TEST_DEVICE")
bus = dbus.SystemBus()
adapters = find_objects(bus, SERVICE_NAME, ADAPTER_INTERFACE)
print(adapters)
async def main():
# Prefer the async helper when running inside an event loop.
try:
adapters = await async_find_objects(None, SERVICE_NAME, ADAPTER_INTERFACE)
except RuntimeError:
# Fallback to the legacy synchronous variant if no loop is running.
adapters = find_objects(None, SERVICE_NAME, ADAPTER_INTERFACE)
print(adapters)
bt = BlueZ(device_id=adapters[0].split("/")[-1])
adapter = AsyncBleakAdapter(adapter_path=adapters[0])
# jc_MAC = "XX:XX:XX:XX:XX:XX"
# res = bt.discover_devices(alias="Joy-Con (L)", timeout=10)
# for key in res.keys():
# print(res[key]["Alias"], res[key]["Address"])
# print(bt.find_device_by_address(jc_MAC))
address = await adapter.get_address()
print("Address", address)
print("Name", adapter.name)
print("Alias", adapter.alias)
print("Pairable", adapter.pairable)
# devices = bt.discover_devices(alias="Joy-Con (L)")
# print(devices.keys())
print("Address", bt.address)
print("Name", bt.name)
print("Alias", bt.alias)
print("Pairable", bt.pairable)
print("")
print("Pairable Timeout", bt.pairable_timeout)
bt.set_pairable_timeout(10)
print("Pairable Timeout", bt.pairable_timeout)
bt.set_pairable_timeout(0)
print("Pairable Timeout", bt.pairable_timeout)
print("")
print("Discoverable", bt.discoverable)
bt.set_discoverable(True)
print("Discoverable", bt.discoverable)
bt.set_discoverable(False)
print("Discoverable", bt.discoverable)
print("")
print("Discoverable Timeout", bt.discoverable_timeout)
bt.set_discoverable_timeout(0)
print("Discoverable Timeout", bt.discoverable_timeout)
bt.set_discoverable_timeout(180)
print("Discoverable Timeout", bt.discoverable_timeout)
try:
print("")
print("Device Class", bt.device_class)
bt.set_device_class("0x002058")
print("Device Class", bt.device_class)
bt.set_device_class("0x480000")
print("Device Class", bt.device_class)
except Exception as e:
print(e)
print("Pairable Timeout", adapter.pairable_timeout)
adapter.set_pairable_timeout(10)
print("Pairable Timeout", adapter.pairable_timeout)
adapter.set_pairable_timeout(0)
print("Pairable Timeout", adapter.pairable_timeout)
print("")
print("Powered", bt.powered)
bt.set_powered(False)
print("Powered", bt.powered)
bt.set_powered(True)
print("Powered", bt.powered)
print("")
print("Discoverable", adapter.discoverable)
adapter.set_discoverable(True)
print("Discoverable", adapter.discoverable)
adapter.set_discoverable(False)
print("Discoverable", adapter.discoverable)
print("")
print("Discoverable Timeout", adapter.discoverable_timeout)
adapter.set_discoverable_timeout(0)
print("Discoverable Timeout", adapter.discoverable_timeout)
adapter.set_discoverable_timeout(180)
print("Discoverable Timeout", adapter.discoverable_timeout)
print("\nScanning for nearby devices...")
try:
devices = await adapter.discover_devices(timeout=5)
except Exception as exc:
print(f"Discovery failed: {exc}")
else:
for path, props in devices.items():
print(f"{props['Alias'] or 'UNKNOWN'} -> {props['Address']} ({path})")
if TARGET_ADDRESS:
print(f"\nAttempting a short Bleak connection to {TARGET_ADDRESS}")
await adapter.connect_device(TARGET_ADDRESS)
print("Connection attempt complete.")
if __name__ == "__main__":
asyncio.run(main())