From 0cebcf4f9b804c0bfc5fb02620ce16f1c1d5b67e Mon Sep 17 00:00:00 2001 From: Fabio Manganiello Date: Thu, 23 Mar 2023 17:46:54 +0100 Subject: [PATCH] `switchbot.bluetooth` integration migrated to a `bluetooth` plugin. --- .../bluetooth/_ble/_plugins/switchbot.py | 69 +++++++++ .../bluetooth/_model/_service/_directory.py | 7 +- .../bluetooth/_model/_service/_directory.pyi | 2 + .../plugins/switchbot/bluetooth/__init__.py | 145 ------------------ .../plugins/switchbot/bluetooth/manifest.yaml | 7 - 5 files changed, 77 insertions(+), 153 deletions(-) create mode 100644 platypush/plugins/bluetooth/_ble/_plugins/switchbot.py delete mode 100644 platypush/plugins/switchbot/bluetooth/__init__.py delete mode 100644 platypush/plugins/switchbot/bluetooth/manifest.yaml diff --git a/platypush/plugins/bluetooth/_ble/_plugins/switchbot.py b/platypush/plugins/bluetooth/_ble/_plugins/switchbot.py new file mode 100644 index 000000000..84741468a --- /dev/null +++ b/platypush/plugins/bluetooth/_ble/_plugins/switchbot.py @@ -0,0 +1,69 @@ +from enum import Enum +from typing import Iterable +from uuid import UUID + +from typing_extensions import override + +from platypush.entities import Entity +from platypush.entities.bluetooth import BluetoothDevice +from platypush.entities.switches import EnumSwitch +from platypush.plugins.bluetooth.model import ServiceClass +from platypush.plugins.bluetooth._plugins import BaseBluetoothPlugin + + +def _make_uuid(prefix: str) -> UUID: + """ + Utility method to create a Switchbot characteristic UUID given a hex + prefix. + """ + return UUID(f'cba20{prefix}-224d-11e6-9fb8-0002a5d5c51b') + + +class Command(Enum): + """ + Supported commands. + """ + + PRESS = b'\x57\x01\x00' + ON = b'\x57\x01\x01' + OFF = b'\x57\x01\x02' + + +class Characteristic(Enum): + """ + GATT characteristic UUIDs supported by Switchbot devices. + """ + + TX = _make_uuid('002') + RX = _make_uuid('003') + SRV = _make_uuid('d00') + + +# pylint: disable=too-few-public-methods +class SwitchbotPlugin(BaseBluetoothPlugin): + """ + Implements support for Switchbot devices. + """ + + @override + def supports_device(self, device: BluetoothDevice) -> bool: + return any( + srv.service_class == ServiceClass.SWITCHBOT for srv in device.services + ) + + @override + def _extract_entities(self, device: BluetoothDevice) -> Iterable[Entity]: + return [ + EnumSwitch( + id=f'{device.address}::switchbot', + name='Switchbot', + values=[cmd.name.lower() for cmd in Command], + is_write_only=True, + ) + ] + + def set(self, device: BluetoothDevice, value: str, **_) -> None: + value = value.upper() + cmd = getattr(Command, value, None) + assert cmd, f'No such command: {value}. Available commands: {list(Command)}.' + self._manager.write(device.address, cmd.value, Characteristic.TX.value) diff --git a/platypush/plugins/bluetooth/_model/_service/_directory.py b/platypush/plugins/bluetooth/_model/_service/_directory.py index 6696fef14..dbad3afd6 100644 --- a/platypush/plugins/bluetooth/_model/_service/_directory.py +++ b/platypush/plugins/bluetooth/_model/_service/_directory.py @@ -131,14 +131,19 @@ https://btprodspecificationrefs.blob.core.windows.net/assigned-numbers/Assigned% Section 3.3. """ +_custom_service_classes: Dict[RawServiceClass, str] = { + UUID("cba20d00-224d-11e6-9fb8-0002a5d5c51b"): "Switchbot", +} + # Update the base services with the GATT service UUIDs defined in ``bluetooth_numbers``. See # https://btprodspecificationrefs.blob.core.windows.net/assigned-numbers/Assigned%20Number%20Types/Assigned%20Numbers.pdf, # Section 3.4 _service_classes.update(bluetooth_numbers.service) # Extend the service classes with the GATT service UUIDs defined in Bleak -_service_classes.update(uuid16_dict) # type: ignore +_service_classes.update(_custom_service_classes) _service_classes.update({UUID(uuid): name for uuid, name in uuid128_dict.items()}) +_service_classes.update(uuid16_dict) # type: ignore _service_classes_by_name: Dict[str, RawServiceClass] = { name: cls for cls, name in _service_classes.items() diff --git a/platypush/plugins/bluetooth/_model/_service/_directory.pyi b/platypush/plugins/bluetooth/_model/_service/_directory.pyi index ee2186327..8cf2106da 100644 --- a/platypush/plugins/bluetooth/_model/_service/_directory.pyi +++ b/platypush/plugins/bluetooth/_model/_service/_directory.pyi @@ -16,6 +16,8 @@ class ServiceClass(Enum): """ A class for unknown services. """ OBEX_OBJECT_PUSH = ... """ Class for the OBEX Object Push service. """ + SWITCHBOT = ... + """ Class for Switchbot devices services. """ @classmethod def get(cls, value: RawServiceClass) -> "ServiceClass": diff --git a/platypush/plugins/switchbot/bluetooth/__init__.py b/platypush/plugins/switchbot/bluetooth/__init__.py deleted file mode 100644 index de4e5a965..000000000 --- a/platypush/plugins/switchbot/bluetooth/__init__.py +++ /dev/null @@ -1,145 +0,0 @@ -import enum -from typing import Any, Collection, Optional -from uuid import UUID - -from bleak.backends.device import BLEDevice -from typing_extensions import override - -from platypush.context import get_or_create_event_loop -from platypush.entities import EnumSwitchEntityManager -from platypush.entities.switches import EnumSwitch -from platypush.plugins import action -from platypush.plugins.bluetooth.ble import BluetoothBlePlugin, UUIDType - - -class Command(enum.Enum): - """ - Supported commands. - """ - - PRESS = b'\x57\x01\x00' - ON = b'\x57\x01\x01' - OFF = b'\x57\x01\x02' - - -# pylint: disable=too-many-ancestors -class SwitchbotBluetoothPlugin(BluetoothBlePlugin, EnumSwitchEntityManager): - """ - Plugin to interact with a Switchbot (https://www.switch-bot.com/) device and - programmatically control switches over a Bluetooth interface. - - Note that this plugin currently only supports Switchbot "bot" devices - (mechanical switch pressers). For support for other devices, you may want - the :class:`platypush.plugins.switchbot.SwitchbotPlugin` integration - (which requires a Switchbot hub). - - Requires: - - * **bleak** (``pip install bleak``) - - """ - - # Map of service names -> UUID prefixes exposed by SwitchBot devices - _uuid_prefixes = { - 'tx': '002', - 'rx': '003', - 'service': 'd00', - } - - # Static list of Bluetooth service UUIDs commonly exposed by SwitchBot - # devices. - _service_uuids = { - service: UUID(f'cba20{prefix}-224d-11e6-9fb8-0002a5d5c51b') - for service, prefix in _uuid_prefixes.items() - } - - def __init__(self, *args, **kwargs): - super().__init__(*args, uuids=self._service_uuids.values(), **kwargs) - - async def _run( - self, - device: str, - command: Command, - service_uuid: UUIDType = _service_uuids['tx'], - ): - await self._write(device, command.value, service_uuid) - - @action - def press(self, device: str): - """ - Send a press button command to a device - - :param device: Device name or address - """ - loop = get_or_create_event_loop() - return loop.run_until_complete(self._run(device, Command.PRESS)) - - @action - def toggle(self, device, **_): - return self.press(device) - - @action - def on(self, device: str, **_): - """ - Send a press-on button command to a device - - :param device: Device name or address - """ - loop = get_or_create_event_loop() - return loop.run_until_complete(self._run(device, Command.ON)) - - @action - def off(self, device: str, **_): - """ - Send a press-off button command to a device - - :param device: Device name or address - """ - loop = get_or_create_event_loop() - return loop.run_until_complete(self._run(device, Command.OFF)) - - @action - def set_value(self, device: Optional[str] = None, value: Optional[str] = None, **_): - """ - Send a command to a device as a value. - - :param entity: Device name or address - :param value: Command to send. Possible values are: - - - ``on``: Press the button and remain in the pressed state. - - ``off``: Release a previously pressed button. - - ``press``: Press and release the button. - - """ - assert device, 'No device specified' - if value == 'on': - self.on(device) - if value == 'off': - self.off(device) - if value == 'press': - self.press(device) - - self.logger.warning('Unknown command for SwitchBot "%s": "%s"', device, value) - - @override - def set(self, entity: str, value: Any, **kwargs): - return self.set_value(entity, value, **kwargs) - - @override - def transform_entities( - self, entities: Collection[BLEDevice] - ) -> Collection[EnumSwitch]: - devices = super().transform_entities(entities) - return [ - EnumSwitch( - id=dev.id, - name=dev.name, - value=None, - values=['on', 'off', 'press'], - is_write_only=True, - ) - for dev in devices - ] - - -# vim:sw=4:ts=4:et: diff --git a/platypush/plugins/switchbot/bluetooth/manifest.yaml b/platypush/plugins/switchbot/bluetooth/manifest.yaml deleted file mode 100644 index 7e4c018c2..000000000 --- a/platypush/plugins/switchbot/bluetooth/manifest.yaml +++ /dev/null @@ -1,7 +0,0 @@ -manifest: - events: {} - install: - pip: - - bleak - package: platypush.plugins.switchbot.bluetooth - type: plugin