platypush/platypush/plugins/switchbot/bluetooth/__init__.py

import enum
import time
from typing import List, Optional

from platypush.message.response.bluetooth import BluetoothScanResponse
from platypush.plugins import action
from platypush.plugins.bluetooth.ble import BluetoothBlePlugin
from platypush.plugins.switch import SwitchPlugin


class SwitchbotBluetoothPlugin(  # lgtm [py/missing-call-to-init]
    SwitchPlugin, BluetoothBlePlugin
):
    """
    Plugin to interact with a Switchbot (https://www.switch-bot.com/) device and
    programmatically control switches over a Bluetooth interface.

    See :class:`platypush.plugins.bluetooth.ble.BluetoothBlePlugin` for how to enable BLE permissions for
    the platypush user (a simple solution may be to run it as root, but that's usually NOT a good idea).

    Requires:

        * **pybluez** (``pip install pybluez``)
        * **gattlib** (``pip install gattlib``)
        * **libboost** (on Debian ```apt-get install libboost-python-dev libboost-thread-dev``)

    """

    uuid = 'cba20002-224d-11e6-9fb8-0002a5d5c51b'
    handle = 0x16

    class Command(enum.Enum):
        """
        Base64 encoded commands
        """

        # \x57\x01\x00
        PRESS = 'VwEA'
        # # \x57\x01\x01
        ON = 'VwEB'
        # # \x57\x01\x02
        OFF = 'VwEC'

    def __init__(
        self,
        interface=None,
        connect_timeout=None,
        scan_timeout=2,
        devices=None,
        **kwargs
    ):
        """
        :param interface: Bluetooth interface to use (e.g. hci0) default: first available one
        :type interface: str

        :param connect_timeout: Timeout for the connection to the Switchbot device - default: None
        :type connect_timeout: float

        :param scan_timeout: Timeout for the scan operations
        :type scan_timeout: float

        :param devices: Devices to control, as a MAC address -> name map
        :type devices: dict
        """
        super().__init__(interface=interface, **kwargs)

        self.connect_timeout = connect_timeout if connect_timeout else 5
        self.scan_timeout = scan_timeout if scan_timeout else 2
        self.configured_devices = devices or {}
        self.configured_devices_by_name = {
            name: addr for addr, name in self.configured_devices.items()
        }

    def _run(self, device: str, command: Command):
        device = self.configured_devices_by_name.get(device, '')
        n_tries = 1

        try:
            self.write(
                device,
                command.value,
                handle=self.handle,
                channel_type='random',
                binary=True,
            )
        except Exception as e:
            self.logger.exception(e)
            n_tries -= 1

            if n_tries == 0:
                raise e
            time.sleep(5)

        return self.status(device)  # type: ignore

    @action
    def press(self, device):
        """
        Send a press button command to a device

        :param device: Device name or address
        :type device: str
        """
        return self._run(device, self.Command.PRESS)

    @action
    def toggle(self, device, **_):
        return self.press(device)

    @action
    def on(self, device, **_):
        """
        Send a press-on button command to a device

        :param device: Device name or address
        :type device: str
        """
        return self._run(device, self.Command.ON)

    @action
    def off(self, device, **_):
        """
        Send a press-off button command to a device

        :param device: Device name or address
        :type device: str
        """
        return self._run(device, self.Command.OFF)

    @action
    def scan(
        self, interface: Optional[str] = None, duration: int = 10
    ) -> BluetoothScanResponse:
        """
        Scan for available Switchbot devices nearby.

        :param interface: Bluetooth interface to scan (default: default configured interface)
        :param duration: Scan duration in seconds
        """

        devices = super().scan(interface=interface, duration=duration).devices
        compatible_devices = {}

        for dev in devices:
            try:
                characteristics = [
                    chrc
                    for chrc in self.discover_characteristics(
                        dev['addr'],
                        channel_type='random',
                        wait=False,
                        timeout=self.scan_timeout,
                    ).characteristics
                    if chrc.get('uuid') == self.uuid
                ]

                if characteristics:
                    compatible_devices[dev['addr']] = None
            except Exception as e:
                self.logger.warning('Device scan error', e)

        self.publish_entities(compatible_devices)  # type: ignore
        return BluetoothScanResponse(devices=compatible_devices)

    @property
    def switches(self) -> List[dict]:
        self.publish_entities(self.configured_devices)  # type: ignore
        return [
            {
                'address': addr,
                'id': addr,
                'name': name,
                'on': False,
            }
            for addr, name in self.configured_devices.items()
        ]

    def transform_entities(self, devices: dict):
        from platypush.entities.switches import Switch

        return super().transform_entities(  # type: ignore
            [
                Switch(
                    id=addr,
                    name=name,
                    state=False,
                    is_write_only=True,
                )
                for addr, name in devices.items()
            ]
        )


# vim:sw=4:ts=4:et: