lda
Verified
038e152bbc
Checks / tests (ubuntu-latest, 3.11) (push) Canceled after 0s
Checks / tests (ubuntu-latest, 3.14) (push) Canceled after 0s
Checks / tests (windows-latest, 3.11) (push) Canceled after 0s
Checks / tests (windows-latest, 3.14) (push) Canceled after 0s
Checks / package (push) Canceled after 0s
44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
"""Optional WinRT initialization matching the successful service probe."""
|
|
|
|
import asyncio
|
|
import warnings
|
|
|
|
|
|
async def prepare_device(address, report):
|
|
"""Return a live Windows device handle after uncached service discovery.
|
|
|
|
Keeping this handle alive is an experiment based on the difference between
|
|
the successful probe and a failing bare socket connect. It is not evidence
|
|
that every RFCOMM device requires this initialization.
|
|
"""
|
|
from winrt.windows.devices.bluetooth import BluetoothCacheMode, BluetoothDevice
|
|
|
|
report("Opening speaker through Windows Bluetooth API…")
|
|
device = await asyncio.wait_for(
|
|
BluetoothDevice.from_bluetooth_address_async(int(address.replace(":", ""), 16)), 20
|
|
)
|
|
if device is None:
|
|
raise ConnectionError("Windows Bluetooth API could not open the speaker.")
|
|
try:
|
|
report("Querying speaker services before direct connection…")
|
|
with warnings.catch_warnings():
|
|
warnings.filterwarnings(
|
|
"ignore",
|
|
message=r"BluetoothDevice\.get_rfcomm_services_with_cache_mode_async\(\).*",
|
|
category=DeprecationWarning,
|
|
)
|
|
result = await asyncio.wait_for(
|
|
device.get_rfcomm_services_with_cache_mode_async(BluetoothCacheMode.UNCACHED), 30
|
|
)
|
|
report(
|
|
f"Windows service query: status {result.error}, {len(result.services)} RFCOMM services"
|
|
)
|
|
for service in result.services:
|
|
service.close()
|
|
# Empty SDP results were also observed in the successful direct probe;
|
|
# they must not prevent the explicit channel-1 connection attempt.
|
|
return device
|
|
except BaseException:
|
|
device.close()
|
|
raise
|