Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

_ensure_connected() doesn't seem to work. #132

Open
d-walsh opened this issue Oct 16, 2022 · 3 comments
Open

_ensure_connected() doesn't seem to work. #132

d-walsh opened this issue Oct 16, 2022 · 3 comments

Comments

@d-walsh
Copy link

d-walsh commented Oct 16, 2022

If I run:

ble_device = await BleakScanner.find_device_by_address(MAC, timeout=20)
device = Switchbot(ble_device)
device.turn_on()

Then after 60 seconds if I try to run

device.turn_off()

It says:
bleak_retry_connector.BleakNotFoundError: D0-A7-8B-16-46-4E (D0:A7:8B:16:46:4E) - /org/bluez/hci1/dev_D0_A7_8B_16_46_4E: Failed to connect: Device with address D0:A7:8B:16:46:4E was not found. It may have been removed from BlueZ when scanning stopped.

@bdraco
Copy link
Member

bdraco commented Oct 16, 2022

You need to keep the scanner running and pass in new BLE Device objects to the lib via bleak's advertisement callbacks otherwise bluez will remove the object from the bus

@d-walsh
Copy link
Author

d-walsh commented Oct 16, 2022

Thank you for explaining this! Do you have an example of how this would be / has been implemented? Thanks!

@djmaze
Copy link

djmaze commented Feb 9, 2023

@d-walsh I have this working in a (not yet open-source) application of myself.

You need to initialize the bot & scanner like this (taken from my codebase):

class MyBot:
    def __init__(mac, password) 
        self.mac = mac
        self.password = password

    async def press(self):
        return await self._try_command(lambda: self.bot.press())

    async def _init(self):
        if self.bot: return True

        self.scanner = BleakScanner(self._detection_callback)
        device = await self.scanner.find_device_by_address(self.mac)

        if not device:
            logger.error("Could not find device")
            return False

        self.bot = Switchbot(device=device, password=self.password)
        await self.scanner.start()

    def _detection_callback(device, adv_data):
        if device.address == self.mac:
            result = parse_advertisement_data(device, adv_data, SwitchbotModel.BOT)
            self.bot.update_from_advertisement(result)

    async def _try_command(self, block):
        if not await self._init(): return

        try:
            return await block()
        except (BleakNotFoundError, BleakDeviceNotFoundError):
            await self.scanner.start()
            await asyncio.sleep(5)
            return await block()

This makes sure the advertisement data is continously updated. If you have multiple devices, you need to refactor this so you are only using a single BleakScanner instance in your app.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

3 participants