fabaff/platypush
1
0
Fork 0
forked from platypush/platypush
Code Issues Pull requests Projects Releases Packages Wiki Activity

Added Switchbot plugin

Browse source
This commit is contained in:
Fabio Manganiello 2021-05-10 18:43:00 +02:00
parent 69583d2e15
commit 664ce4050d
14 changed files with 407 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
CHANGELOG.md
View file

@ -3,6 +3,22 @@
All notable changes to this project will be documented in this file.
Given the high speed of development in the first phase, changes are being reported only starting from v0.20.2.
## [Unreleased]
### Added
- Added `switchbot` plugin to interact with Switchbot devices over the cloud API instead of
directly accessing the device's Bluetooth interface.
- Added `marshmallow` dependency - it will be used from now own to dump and document schemas
and responses instead of the currently mixed approach with `Response` objects and plain
dictionaries and lists.
### Changed
- `switch.switchbot` plugin renamed to `switchbot.bluetooth` plugin, while the new plugin
that uses the Switchbot API is simply named `switchbot`.
## [0.21.0] - 2021-05-06
### Added

6
docs/source/platypush/plugins/switch.switchbot.rst
View file

@ -1,6 +0,0 @@
``platypush.plugins.switch.switchbot``
======================================
.. automodule:: platypush.plugins.switch.switchbot
:members:

5
docs/source/platypush/plugins/switchbot.bluetooth.rst Normal file
View file

@ -0,0 +1,5 @@
``platypush.plugins.switchbot.bluetooth``
=========================================
.. automodule:: platypush.plugins.switchbot.bluetooth
:members:

5
docs/source/platypush/plugins/switchbot.rst Normal file
View file

@ -0,0 +1,5 @@
``platypush.plugins.switchbot``
===============================
.. automodule:: platypush.plugins.switchbot
:members:

5
docs/source/platypush/plugins/zwave._base.rst Normal file
View file

@ -0,0 +1,5 @@
``platypush.plugins.zwave._base``
=================================
.. automodule:: platypush.plugins.zwave._base
:members:

4
docs/source/plugins.rst
View file

@ -118,9 +118,10 @@ Plugins
platypush/plugins/stt.picovoice.hotword.rst
platypush/plugins/stt.picovoice.speech.rst
platypush/plugins/switch.rst
platypush/plugins/switch.switchbot.rst
platypush/plugins/switch.tplink.rst
platypush/plugins/switch.wemo.rst
platypush/plugins/switchbot.rst
platypush/plugins/switchbot.bluetooth.rst
platypush/plugins/system.rst
platypush/plugins/tcp.rst
platypush/plugins/tensorflow.rst
@ -145,4 +146,5 @@ Plugins
platypush/plugins/zeroconf.rst
platypush/plugins/zigbee.mqtt.rst
platypush/plugins/zwave.rst
platypush/plugins/zwave._base.rst
platypush/plugins/zwave.mqtt.rst

1
platypush/backend/http/webapp/src/components/panels/Switches/Switchbot Symbolic link
View file

@ -0,0 +1 @@
SwitchbotBluetooth

2
platypush/backend/http/webapp/src/components/panels/Switches/SwitchSwitchbot/Index.vue → platypush/backend/http/webapp/src/components/panels/Switches/SwitchbotBluetooth/Index.vue
View file

@ -35,7 +35,7 @@ import Switch from "@/components/panels/Switches/Switch";
import Modal from "@/components/Modal";
export default {
name: "SwitchSwitchbot",
name: "SwitchbotBluetooth",
components: {Modal, Switch, Loading},
mixins: [SwitchMixin],
}

42
platypush/plugins/switch/__init__.py
View file

@ -1,4 +1,4 @@
from typing import List
from typing import List, Union
from platypush.plugins import Plugin, action
@ -27,46 +27,34 @@ class SwitchPlugin(Plugin):
raise NotImplementedError()
@action
def switch_status(self, device=None):
""" Get the status of a specified device or of all the configured devices (default)"""
def switch_status(self, device=None) -> Union[dict, List[dict]]:
"""
Get the status of a specified device or of all the configured devices (default).
:param device: Filter by device name or ID.
:return: .. schema:: switch.SwitchStatusSchema(many=True)
"""
devices = self.switches
if device:
devices = [d for d in self.switches if d.get('id') == device or d.get('name') == device]
if devices:
return devices[0]
else:
return None
return devices[0] if devices else []
return devices
@action
def status(self, device=None, *args, **kwargs):
def status(self, device=None, *args, **kwargs) -> Union[dict, List[dict]]:
"""
Status function - if not overridden it calls :meth:`.switch_status`. You may want to override it if your plugin
does not handle only switches.
Get the status of all the devices, or filter by device name or ID (alias for :meth:`.switch_status`).
:param device: Filter by device name or ID.
:return: .. schema:: switch.SwitchStatusSchema(many=True)
"""
return self.switch_status(device)
@property
def switches(self) -> List[dict]:
"""
This property must be implemented by the derived classes and must return a dictionary in the following format:
.. code-block:: json
[
{
"name": "switch_1",
"on": true
},
{
"name": "switch_2",
"on": false
},
]
``name`` and ``on`` are the minimum set of attributes that should be returned for a switch, but more attributes
can also be added.
:return: .. schema:: switch.SwitchStatusSchema(many=True)
"""
raise NotImplementedError()

264
platypush/plugins/switchbot/__init__.py Normal file
View file

@ -0,0 +1,264 @@
import queue
import requests
import threading
from typing import List, Optional, Union
from platypush.plugins import action
from platypush.plugins.switch import SwitchPlugin
from platypush.schemas.switchbot import DeviceSchema, DeviceStatusSchema, SceneSchema
class SwitchbotPlugin(SwitchPlugin):
"""
Plugin to interact with the devices registered to a Switchbot (https://www.switch-bot.com/) account/hub.
The difference between this plugin and :class:`platypush.plugins.switchbot.bluetooth.SwitchbotBluetoothPlugin` is
that the latter acts like a Bluetooth hub/bridge that interacts directly with your Switchbot devices, while this
plugin requires the devices to be connected to a Switchbot Hub and it controls them through your cloud account.
In order to use this plugin:
- Set up a Switchbot Hub and configure your devices through the Switchbot app.
- Follow the steps on the `Switchbot API repo <https://github.com/OpenWonderLabs/SwitchBotAPI#getting-started>`_
to get an API token from the app.
"""
def __init__(self, api_token: str, **kwargs):
"""
:param api_token: API token (see
`Getting started with the Switchbot API <https://github.com/OpenWonderLabs/SwitchBotAPI#getting-started>`_).
"""
super().__init__(**kwargs)
self._api_token = api_token
self._devices_by_id = {}
self._devices_by_name = {}
@staticmethod
def _url_for(*args, device=None):
url = 'https://api.switch-bot.com/v1.0/'
if device:
url += f'devices/{device["id"]}/'
url += '/'.join(args)
return url
def _run(self, method: str = 'get', *args, device=None, **kwargs):
response = getattr(requests, method)(self._url_for(*args, device=device), headers={
'Authorization': self._api_token,
'Accept': 'application/json',
'Content-Type': 'application/json; charset=utf-8',
}, **kwargs)
response.raise_for_status()
response = response.json()
assert response.get('statusCode') == 100, \
f'Switchbot API request failed: {response.get("statusCode")}: {response.get("message")}'
return response.get('body')
def _get_device(self, device: str, use_cache=True):
if not use_cache:
self.devices()
if device in self._devices_by_id:
return self._devices_by_id[device]
if device in self._devices_by_name:
return self._devices_by_name[device]
assert use_cache, f'Device not found: {device}'
return self._get_device(device, use_cache=False)
@action
def devices(self) -> List[dict]:
"""
Get the list of devices associated to the specified Switchbot API account.
:return: .. schema:: switchbot.DeviceSchema(many=True)
"""
devices = DeviceSchema().dump(self._run('get', 'devices').get('deviceList', []), many=True)
for device in devices:
self._devices_by_id[device['id']] = device
self._devices_by_name[device['name']] = device
return devices
def _worker(self, q: queue.Queue, method: str = 'get', *args, device=None, **kwargs):
try:
res = self._run(method, *args, device=device, **kwargs)
q.put(DeviceStatusSchema().dump(res))
except Exception as e:
self.logger.exception(e)
q.put(e)
@action
def status(self, device: Optional[str] = None) -> Union[dict, List[dict]]:
"""
Get the status of all the registered devices or of a specific device.
:param device: Filter by device ID or name.
:return: .. schema:: switchbot.DeviceStatusSchema(many=True)
"""
# noinspection PyUnresolvedReferences
devices = self.devices().output
if device:
return {
**device,
**self._run('get', 'status', device=self._get_device(device)),
}
devices_by_id = {dev['id']: dev for dev in devices}
queues = [queue.Queue()] * len(devices)
workers = [
threading.Thread(target=self._worker, args=(queues[i], 'get', 'status'), kwargs={'device': dev})
for i, dev in enumerate(devices)
]
results = []
for worker in workers:
worker.start()
for q in queues:
response = q.get()
if not response:
continue
assert not isinstance(response, Exception), str(response)
results.append({
**devices_by_id.get(response.get('id'), {}),
**response,
})
for worker in workers:
worker.join()
return results
@action
def press(self, device: str):
"""
Send a press-button command to a device.
:param device: Device name or ID.
"""
device = self._get_device(device)
return self._run('post', 'commands', device=device, json={'command': 'press'})
@action
def toggle(self, device: str, **kwargs):
"""
Shortcut for :meth:`.press`.
:param device: Device name or ID.
"""
return self.press(device)
@action
def on(self, device: str, **kwargs):
"""
Send a turn-on command to a device
:param device: Device name or ID.
"""
device = self._get_device(device)
return self._run('post', 'commands', device=device, json={'command': 'turnOn'})
@action
def off(self, device: str, **kwargs):
"""
Send a turn-off command to a device
:param device: Device name or ID.
"""
device = self._get_device(device)
return self._run('post', 'commands', device=device, json={'command': 'turnOff'})
@property
def switches(self) -> List[dict]:
# noinspection PyUnresolvedReferences
return [
dev for dev in self.status().output if 'on' in dev
]
@action
def set_curtain_position(self, device: str, position: int):
"""
Set the position of a curtain device.
:param device: Device name or ID.
:param position: An integer between 0 (open) and 100 (closed).
"""
device = self._get_device(device)
return self._run('post', 'commands', device=device, json={
'command': 'setPosition',
'commandType': 'command',
'parameter': f'0,ff,{position}',
})
@action
def set_humidifier_efficiency(self, device: str, efficiency: Union[int, str]):
"""
Set the nebulization efficiency of a humidifier device.
:param device: Device name or ID.
:param efficiency: An integer between 0 (open) and 100 (closed) or `auto`.
"""
device = self._get_device(device)
return self._run('post', 'commands', device=device, json={
'command': 'setMode',
'commandType': 'command',
'parameter': efficiency,
})
@action
def set_fan(self, device: str, speed: Optional[int] = None, swing_range: Optional[int] = None,
mode: Optional[int] = None):
"""
Set properties of a smart fan device.
:param device: Device name or ID.
:param speed: Speed between 1 and 4.
:param swing_range: Swing range angle, between 0 and 120.
:param mode: Fan mode (1 or 2).
"""
# noinspection PyUnresolvedReferences
status = self.status(device=device).output
if speed is None:
speed = status.get('speed')
if mode is None:
mode = status.get('mode')
if swing_range is None:
swing_range = status.get('swing_range')
return self._run('post', 'commands', device=device, json={
'command': 'set',
'commandType': 'command',
'parameter': ','.join(['on', str(mode), str(speed), str(swing_range)]),
})
@action
def scenes(self) -> List[dict]:
"""
Get the list of registered scenes.
:return: .. schema:: switchbot.SceneSchema(many=True)
"""
return SceneSchema().dump(self._run('get', 'scenes'), many=True)
@action
def run_scene(self, scene: str):
"""
Execute a scene.
:param scene: Scene ID or name.
"""
# noinspection PyUnresolvedReferences
scenes = [
s for s in self.scenes().output
if s.get('id') == scene or s.get('name') == scene
]
assert scenes, f'No such scene: {scene}'
return self._run('post', 'scenes', scenes[0]['id'], 'execute')
# vim:sw=4:ts=4:et:

5
platypush/plugins/switch/switchbot/__init__.py → platypush/plugins/switchbot/bluetooth.py
View file

@ -8,10 +8,10 @@ from platypush.plugins.bluetooth.ble import BluetoothBlePlugin
from platypush.plugins.switch import SwitchPlugin
class SwitchSwitchbotPlugin(SwitchPlugin, BluetoothBlePlugin):
class SwitchbotBluetoothPlugin(SwitchPlugin, BluetoothBlePlugin):
"""
Plugin to interact with a Switchbot (https://www.switch-bot.com/) device and
programmatically control buttons.
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).
@ -21,6 +21,7 @@ class SwitchSwitchbotPlugin(SwitchPlugin, BluetoothBlePlugin):
* **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'

1
platypush/plugins/zwave/__init__.py
View file

@ -53,6 +53,7 @@ class ZwavePlugin(ZwaveBasePlugin, SwitchPlugin):
def status(self) -> Dict[str, Any]:
"""
Get the status of the controller.
:return: dict
"""
backend = self._get_backend()

8
platypush/schemas/switch.py Normal file
View file

@ -0,0 +1,8 @@
from marshmallow import fields
from marshmallow.schema import Schema
class SwitchStatusSchema(Schema):
id = fields.Raw(metadata=dict(description='Device unique ID'))
name = fields.String(required=True, metadata=dict(description='Device name'))
on = fields.Boolean(required=True, metadata=dict(description='True if the device is on, False otherwise'))

80
platypush/schemas/switchbot.py Normal file
View file

@ -0,0 +1,80 @@
from marshmallow import fields
from marshmallow.schema import Schema
from marshmallow.validate import OneOf
device_types = [
'Hub',
'Hub Plus',
'Hub Mini',
'Bot',
'Curtain',
'Plug',
'Meter',
'Humidifier',
'Smart Fan',
'Air Conditioner',
'TV',
'Light',
'IPTV / Streamer',
'Set Top Box',
'DVD',
'Fan',
'Projector',
'Camera',
'Air Purifier',
'Speaker',
'Water Heater',
'Vacuum Cleaner',
'Others',
]
class DeviceSchema(Schema):
id = fields.String(attribute='deviceId', required=True, metadata=dict(description='Device unique ID'))
name = fields.String(attribute='deviceName', metadata=dict(description='Device name'))
type = fields.String(attribute='deviceType', required=True, validate=OneOf(device_types),
metadata=dict(description=f'Supported types: [{", ".join(device_types)}]'))
hub_id = fields.String(attribute='hubDeviceId', metadata=dict(description='Parent hub device unique ID'))
cloud_service_enabled = fields.Boolean(attribute='enableCloudService',
metadata=dict(description='True if cloud access is enabled on this device,'
'False otherwise. Only cloud-enabled devices can be '
'controlled from the switchbot plugin.'))
calibrated = fields.Boolean(attribute='calibrate',
metadata=dict(description='[Curtain devices only] Set to True if the device has '
'been calibrated, False otherwise'))
open_direction = fields.String(attribute='openDirection',
metadata=dict(description='[Curtain devices only] Direction where the curtains will '
'be opened ("left" or "right")'))
class DeviceStatusSchema(DeviceSchema):
on = fields.Boolean(attribute='power', metadata=dict(description='True if the device is on, False otherwise'))
moving = fields.Boolean(metadata=dict(
description='[Curtain devices only] True if the device is moving, False otherwise'))
position = fields.Int(attribute='slidePosition', metadata=dict(
description='[Curtain devices only] Position of the device on the curtain rail, between '
'0 (open) and 1 (closed)'))
temperature = fields.Float(metadata=dict(description='[Meter/humidifier devices only] Temperature in Celsius'))
humidity = fields.Float(metadata=dict(description='[Meter/humidifier devices only] Humidity in %'))
nebulization_efficiency = fields.Float(attribute='nebulizationEfficiency',
metadata=dict(description='[Humidifier devices only] Nebulization '
'efficiency in %'))
auto = fields.Boolean(metadata=dict(description='[Humidifier devices only] True if auto mode is on'))
child_lock = fields.Boolean(attribute='childLock',
metadata=dict(description='[Humidifier devices only] True if safety lock is on'))
sound = fields.Boolean(metadata=dict(description='[Humidifier devices only] True if sound is muted'))
mode = fields.Int(metadata=dict(description='[Smart fan devices only] Fan mode'))
speed = fields.Float(metadata=dict(description='[Smart fan devices only] Fan speed, between 1 and 4'))
swinging = fields.Boolean(attribute='shaking',
metadata=dict(description='[Smart fan devices only] True if the device is swinging'))
swing_direction = fields.Int(attribute='shakeCenter',
metadata=dict(description='[Smart fan devices only] Swing direction'))
swing_range = fields.Float(attribute='shakeRange',
metadata=dict(description='[Smart fan devices only] Swing range angle, '
'between 0 and 120'))
class SceneSchema(Schema):
id = fields.String(attribute='sceneId', required=True, metadata=dict(description='Scene ID'))
name = fields.String(attribute='sceneName', metadata=dict(description='Scene name'))