switchbot.bluetooth integration migrated to a bluetooth plugin.

This commit is contained in:
Fabio Manganiello 2023-03-23 17:46:54 +01:00
parent 4fac110bb8
commit 0cebcf4f9b
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
5 changed files with 77 additions and 153 deletions

View file

@ -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)

View file

@ -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()

View file

@ -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":

View file

@ -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:

View file

@ -1,7 +0,0 @@
manifest:
events: {}
install:
pip:
- bleak
package: platypush.plugins.switchbot.bluetooth
type: plugin