revil-O/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

The zwave and zwave.mqtt plugins should extend a common abstract class instead of having a zwave.mqtt -> zwave functional dependency that introduces the PyOWZ dependency into zwave.mqtt

Browse source
This commit is contained in:
Fabio Manganiello 2021-04-16 20:54:07 +02:00
parent b9bc4b5fe0
commit 02a22d4a88
3 changed files with 701 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -9,9 +9,10 @@ from platypush.backend.zwave import ZwaveBackend
from platypush.context import get_backend
from platypush.plugins import action
from platypush.plugins.switch import SwitchPlugin
from platypush.plugins.zwave._base import ZwaveBasePlugin
class ZwavePlugin(SwitchPlugin):
class ZwavePlugin(ZwaveBasePlugin, SwitchPlugin):
"""
This plugin interacts with the devices on a Z-Wave network started through the
:class:`platypush.backend.zwave.ZwaveBackend` backend.
@ -156,7 +157,8 @@ class ZwavePlugin(SwitchPlugin):
return {}
return {
'command_class': value.node.get_command_class_as_string(value.command_class) if value.command_class else None,
'command_class': value.node.get_command_class_as_string(value.command_class)
if value.command_class else None,
'data': value.data,
'data_as_string': value.data_as_string,
'data_items': list(value.data_items) if isinstance(value.data_items, set) else value.data_items,
@ -218,7 +220,8 @@ class ZwavePlugin(SwitchPlugin):
'is_awake': node.is_awake if hasattr(node, 'is_awake') else False,
'is_failed': node.is_failed if hasattr(node, 'is_failed') else False,
'is_beaming_device': node.is_beaming_device if hasattr(node, 'is_beaming_device') else False,
'is_frequent_listening_device': node.is_frequent_listening_device if hasattr(node, 'is_frequent_listening_device') else False,
'is_frequent_listening_device': node.is_frequent_listening_device
if hasattr(node, 'is_frequent_listening_device') else False,
'is_info_received': node.is_info_received if hasattr(node, 'is_info_received') else False,
'is_listening_device': node.is_listening_device if hasattr(node, 'is_listening_device') else False,
'is_locked': node.is_locked if hasattr(node, 'is_locked') else False,

688
platypush/plugins/zwave/_base.py Normal file
View file

@ -0,0 +1,688 @@
from abc import ABC
from typing import Any, Dict, Optional, List, Union
from platypush.plugins import action
from platypush.plugins.switch import SwitchPlugin
class ZwaveBasePlugin(ABC, SwitchPlugin):
"""
Base class for Z-Wave plugins.
"""
@action
def start_network(self):
raise NotImplementedError
@action
def stop_network(self):
raise NotImplementedError
@action
def status(self) -> Dict[str, Any]:
"""
Get the status of the controller.
"""
raise NotImplementedError
@action
def add_node(self, *args, **kwargs):
"""
Start the inclusion process to add a node to the network.
"""
raise NotImplementedError
@action
def remove_node(self):
"""
Remove a node from the network.
"""
raise NotImplementedError
@action
def remove_failed_node(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Remove a failed node from the network.
:param node_id: Filter by node_id.
:param node_name: Filter by node name.
"""
raise NotImplementedError
@action
def replace_failed_node(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Replace a failed node on the network.
:param node_id: Filter by node_id.
:param node_name: Filter by node name.
"""
raise NotImplementedError
@action
def replication_send(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Send node information from the primary to the secondary controller.
:param node_id: Filter by node_id.
:param node_name: Filter by node name.
"""
raise NotImplementedError
@action
def request_network_update(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Request a network update to a node.
:param node_id: Filter by node_id.
:param node_name: Filter by node name.
"""
raise NotImplementedError
@action
def request_node_neighbour_update(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Request a neighbours list update to a node.
:param node_id: Filter by node_id.
:param node_name: Filter by node name.
"""
raise NotImplementedError
@action
def get_nodes(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) -> Dict[str, Any]:
"""
Get the nodes associated to the network.
:param node_id: Filter by node_id.
:param node_name: Filter by node name.
"""
raise NotImplementedError
@action
def get_node_stats(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) \
-> Dict[str, Any]:
"""
Get the statistics of a node on the network.
:param node_id: Filter by node_id.
:param node_name: Filter by node name.
"""
raise NotImplementedError
@action
def set_node_name(self, new_name: str, node_id: Optional[int] = None, node_name: Optional[str] = None):
"""
Rename a node on the network.
:param new_name: New name for the node.
:param node_id: Filter by node_id.
:param node_name: Filter by current node name.
"""
raise NotImplementedError
@action
def set_node_product_name(self, product_name: str, node_id: Optional[int] = None, node_name: Optional[str] = None,
**kwargs):
"""
Set the product name of a node.
:param product_name: Product name.
:param node_id: Filter by node_id.
:param node_name: Filter by current node name.
"""
raise NotImplementedError
@action
def set_node_manufacturer_name(self, manufacturer_name: str, node_id: Optional[int] = None,
node_name: Optional[str] = None, **kwargs):
"""
Set the manufacturer name of a node.
:param manufacturer_name: Manufacturer name.
:param node_id: Filter by node_id.
:param node_name: Filter by current node name.
"""
raise NotImplementedError
@action
def set_node_location(self, location: str, node_id: Optional[int] = None, node_name: Optional[str] = None,
**kwargs):
"""
Set the location of a node.
:param location: Node location.
:param node_id: Filter by node_id.
:param node_name: Filter by current node name.
"""
raise NotImplementedError
@action
def cancel_command(self):
"""
Cancel the current running command.
"""
raise NotImplementedError
@action
def kill_command(self):
"""
Immediately terminate any running command on the controller and release the lock.
"""
raise NotImplementedError
@action
def set_controller_name(self, name: str, **kwargs):
"""
Set the name of the controller on the network.
:param name: New controller name.
"""
raise NotImplementedError
@action
def get_capabilities(self, **kwargs) -> List[str]:
"""
Get the capabilities of the controller.
"""
raise NotImplementedError
@action
def receive_configuration(self, **kwargs):
"""
Receive the configuration from the primary controller on the network. Requires a primary controller active.
"""
raise NotImplementedError
@action
def transfer_primary_role(self, **kwargs):
"""
Add a new controller to the network and make it the primary.
The existing primary will become a secondary controller.
"""
raise NotImplementedError
@action
def heal(self, refresh_routes: bool = False, **kwargs):
"""
Heal network by requesting nodes rediscover their neighbors.
:param refresh_routes: Whether to perform return routes initialization (default: ``False``).
"""
raise NotImplementedError
@action
def switch_all(self, state: bool, **kwargs):
"""
Switch all the connected devices on/off.
:param state: True (switch on) or False (switch off).
"""
raise NotImplementedError
@action
def test(self, count: int = 1, **kwargs):
"""
Send a number of test messages to every node and record results.
:param count: The number of test messages to send.
"""
raise NotImplementedError
@action
def get_value(self, value_id: Optional[int] = None, id_on_network: Optional[str] = None,
value_label: Optional[str] = None, node_id: Optional[int] = None, node_name: Optional[str] = None,
**kwargs) -> Dict[str, Any]:
"""
Get a value on the network.
:param value_id: Select by value_id.
:param id_on_network: Select value by id_on_network.
:param value_label: Select value by [node_id/node_name, value_label]
:param node_id: Select value by [node_id/node_name, value_label]
:param node_name: Select value by [node_id/node_name, value_label]
"""
raise NotImplementedError
@action
def set_value(self, data, value_id: Optional[int] = None, id_on_network: Optional[str] = None,
value_label: Optional[str] = None, node_id: Optional[int] = None, node_name: Optional[str] = None,
**kwargs):
"""
Set a value.
:param data: Data to set for the value.
:param value_id: Select value by value_id.
:param id_on_network: Select value by id_on_network.
:param value_label: Select value by [node_id/node_name, value_label]
:param node_id: Select value by [node_id/node_name, value_label]
:param node_name: Select value by [node_id/node_name, value_label]
"""
raise NotImplementedError
@action
def set_value_label(self, new_label: str, value_id: Optional[int] = None, id_on_network: Optional[str] = None,
value_label: Optional[str] = None, node_id: Optional[int] = None,
node_name: Optional[str] = None, **kwargs):
"""
Change the label/name of a value.
:param new_label: New value label.
:param value_id: Select value by value_id.
:param id_on_network: Select value by id_on_network.
:param value_label: Select value by [node_id/node_name, value_label]
:param node_id: Select value by [node_id/node_name, value_label]
:param node_name: Select value by [node_id/node_name, value_label]
"""
raise NotImplementedError
@action
def node_add_value(self, value_id: Optional[int] = None, id_on_network: Optional[str] = None,
value_label: Optional[str] = None, node_id: Optional[int] = None,
node_name: Optional[str] = None, **kwargs):
"""
Add a value to a node.
:param value_id: Select value by value_id.
:param id_on_network: Select value by id_on_network.
:param value_label: Select value by label.
:param node_id: Select node by node_id.
:param node_name: Select node by label.
"""
raise NotImplementedError
@action
def node_remove_value(self, value_id: Optional[int] = None, id_on_network: Optional[str] = None,
value_label: Optional[str] = None, node_id: Optional[int] = None,
node_name: Optional[str] = None, **kwargs):
"""
Remove a value from a node.
:param value_id: Select value by value_id.
:param id_on_network: Select value by id_on_network.
:param value_label: Select value by [node_id/node_name, value_label]
:param node_id: Select node by node_id.
:param node_name: Select node by label.
"""
raise NotImplementedError
@action
def node_heal(self, node_id: Optional[int] = None, node_name: Optional[str] = None, refresh_routes: bool = False,
**kwargs):
"""
Heal network node by requesting the node to rediscover their neighbours.
:param node_id: Select node by node_id.
:param node_name: Select node by label.
:param refresh_routes: Whether to perform return routes initialization. (default: ``False``).
"""
raise NotImplementedError
@action
def node_update_neighbours(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Ask a node to update its neighbours table.
:param node_id: Select node by node_id.
:param node_name: Select node by label.
"""
raise NotImplementedError
@action
def node_network_update(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Update the controller with network information.
:param node_id: Select node by node_id.
:param node_name: Select node by label.
"""
raise NotImplementedError
@action
def node_refresh_info(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Fetch up-to-date information about the node.
:param node_id: Select node by node_id.
:param node_name: Select node by label.
"""
raise NotImplementedError
@action
def get_dimmers(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) -> Dict[int, Any]:
"""
Get the dimmers on the network or associated to a node.
:param node_id: Select node by node_id.
:param node_name: Select node by label.
"""
raise NotImplementedError
@action
def get_node_config(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) \
-> Dict[int, Any]:
"""
Get the configuration values of a node or of all the nodes on the network.
:param node_id: Select node by node_id.
:param node_name: Select node by label.
"""
raise NotImplementedError
@action
def get_battery_levels(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) \
-> Dict[int, Any]:
"""
Get the battery levels of a node or of all the nodes on the network.
:param node_id: Select node by node_id.
:param node_name: Select node by name.
"""
raise NotImplementedError
@action
def get_power_levels(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) \
-> Dict[int, Any]:
"""
Get the power levels of this node.
:param node_id: Select node by node_id.
:param node_name: Select node by name.
"""
raise NotImplementedError
@action
def get_bulbs(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) -> Dict[int, Any]:
"""
Get the bulbs/LEDs on the network or associated to a node.
:param node_id: Select node by node_id.
:param node_name: Select node by name.
"""
raise NotImplementedError
@action
def get_switches(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) -> Dict[int, Any]:
"""
Get the switches on the network or associated to a node.
:param node_id: Select node by node_id.
:param node_name: Select node by name.
"""
raise NotImplementedError
@action
def get_sensors(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) -> Dict[int, Any]:
"""
Get the sensors on the network or associated to a node.
:param node_id: Select node by node_id.
:param node_name: Select node by name.
"""
raise NotImplementedError
@action
def get_doorlocks(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) -> Dict[int, Any]:
"""
Get the doorlocks on the network or associated to a node.
:param node_id: Select node by node_id.
:param node_name: Select node by name.
"""
raise NotImplementedError
@action
def get_usercodes(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) -> Dict[int, Any]:
"""
Get the usercodes on the network or associated to a node.
:param node_id: Select node by node_id.
:param node_name: Select node by name.
"""
raise NotImplementedError
@action
def get_thermostats(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) \
-> Dict[int, Any]:
"""
Get the thermostats on the network or associated to a node.
:param node_id: Select node by node_id.
:param node_name: Select node by name.
"""
raise NotImplementedError
@action
def get_protections(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) \
-> Dict[int, Any]:
"""
Get the protection-compatible devices on the network or associated to a node.
:param node_id: Select node by node_id.
:param node_name: Select node by name.
"""
raise NotImplementedError
@action
def get_groups(self, **kwargs) -> Dict[int, Any]:
"""
Get the groups on the network.
"""
raise NotImplementedError
@action
def get_scenes(self, **kwargs) -> Dict[str, Any]:
"""
Get the scenes configured on the network.
"""
raise NotImplementedError
@action
def create_scene(self, label: str, **kwargs):
"""
Create a new scene.
:param label: Scene label.
"""
raise NotImplementedError
@action
def remove_scene(self, scene_id: Optional[int] = None, scene_label: Optional[str] = None, **kwargs):
"""
Remove a scene.
:param scene_id: Select by scene_id.
:param scene_label: Select by scene label.
"""
raise NotImplementedError
@action
def activate_scene(self, scene_id: Optional[int] = None, scene_label: Optional[str] = None, **kwargs):
"""
Activate a scene.
:param scene_id: Select by scene_id.
:param scene_label: Select by scene label.
"""
raise NotImplementedError
@action
def set_scene_label(self, new_label: str, scene_id: Optional[int] = None, scene_label: Optional[str] = None,
**kwargs):
"""
Rename a scene/set the scene label.
:param new_label: New label.
:param scene_id: Select by scene_id.
:param scene_label: Select by current scene label.
"""
raise NotImplementedError
@action
def scene_add_value(self, data: Optional[Any] = None,
value_id: Optional[int] = None, id_on_network: Optional[str] = None,
value_label: Optional[str] = None, scene_id: Optional[int] = None,
scene_label: Optional[str] = None, node_id: Optional[int] = None,
node_name: Optional[str] = None, **kwargs):
"""
Add a value to a scene.
:param data: Data to set for the value (default: current value data).
:param value_id: Select value by value_id.
:param id_on_network: Select value by id_on_network.
:param value_label: Select value by [node_id/node_name, value_label]
:param node_id: Select value by [node_id/node_name, value_label]
:param node_name: Select value by [node_id/node_name, value_label]
:param scene_id: Select scene by scene_id.
:param scene_label: Select scene by scene label.
"""
raise NotImplementedError
@action
def scene_remove_value(self, value_id: Optional[int] = None, id_on_network: Optional[str] = None,
value_label: Optional[str] = None, scene_id: Optional[int] = None,
scene_label: Optional[str] = None, node_id: Optional[int] = None,
node_name: Optional[str] = None, **kwargs):
"""
Remove a value from a scene.
:param value_id: Select value by value_id.
:param id_on_network: Select value by id_on_network.
:param value_label: Select value by [node_id/node_name, value_label]
:param node_id: Select value by [node_id/node_name, value_label]
:param node_name: Select value by [node_id/node_name, value_label]
:param scene_id: Select scene by scene_id.
:param scene_label: Select scene by scene label.
"""
raise NotImplementedError
@action
def get_scene_values(self, scene_id: Optional[int] = None, scene_label: Optional[str] = None, **kwargs) -> dict:
"""
Get the values associated to a scene.
:param scene_id: Select by scene_id.
:param scene_label: Select by scene label.
:return: value_id -> value (as a dict) mapping.
"""
raise NotImplementedError
@action
def get_scene_values(self, scene_id: Optional[int] = None, scene_label: Optional[str] = None, **kwargs) -> dict:
"""
Get the values associated to a scene.
:param scene_id: Select by scene_id.
:param scene_label: Select by scene label.
"""
raise NotImplementedError
@action
def create_button(self, button_id: Union[int, str], node_id: Optional[int] = None, node_name: Optional[str] = None,
**kwargs):
"""
Create a handheld button on a device. Only intended for bridge firmware controllers.
:param button_id: The ID of the button.
:param node_id: Filter by node_id.
:param node_name: Filter by current node name.
"""
raise NotImplementedError
@action
def delete_button(self, button_id: Union[int, str], node_id: Optional[int] = None, node_name: Optional[str] = None,
**kwargs):
"""
Delete a button association from a device. Only intended for bridge firmware controllers.
:param button_id: The ID of the button.
:param node_id: Filter by node_id.
:param node_name: Filter by current node name.
"""
raise NotImplementedError
@action
def add_node_to_group(self, group_index: Optional[int] = None, group_label: Optional[str] = None,
node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Add a node to a group.
:param group_index: Select group by group index.
:param group_label: Select group by group label.
:param node_id: Select node by node_id.
:param node_name: Select node by node name.
"""
raise NotImplementedError
@action
def remove_node_from_group(self, group_index: Optional[int] = None, group_label: Optional[str] = None,
node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs):
"""
Remove a node from a group.
:param group_index: Select group by group index.
:param group_label: Select group by group label.
:param node_id: Select node by node_id.
:param node_name: Select node by node name.
"""
raise NotImplementedError
@action
def create_new_primary(self, **kwargs):
"""
Create a new primary controller on the network when the previous primary fails.
"""
raise NotImplementedError
@action
def hard_reset(self, **kwargs):
"""
Perform a hard reset of the controller. It erases its network configuration settings.
The controller becomes a primary controller ready to add devices to a new network.
"""
raise NotImplementedError
@action
def soft_reset(self, **kwargs):
"""
Perform a soft reset of the controller.
Resets a controller without erasing its network configuration settings.
"""
raise NotImplementedError
@action
def write_config(self, **kwargs):
"""
Store the current configuration of the network to the user directory.
"""
raise NotImplementedError
@action
def on(self, device: str, *args, **kwargs):
"""
Turn on a switch on a device.
:param device: ``id_on_network`` of the value to be switched on.
"""
raise NotImplementedError
@action
def off(self, device: str, *args, **kwargs):
"""
Turn off a switch on a device.
:param device: ``id_on_network`` of the value to be switched off.
"""
raise NotImplementedError
@action
def toggle(self, device: str, *args, **kwargs):
"""
Toggle a switch on a device.
:param device: ``id_on_network`` of the value to be toggled.
"""
raise NotImplementedError
# vim:sw=4:ts=4:et:

14
platypush/plugins/zwave/mqtt.py
View file

@ -10,13 +10,13 @@ from platypush.message.event.zwave import ZwaveNodeRenamedEvent, ZwaveNodeEvent
from platypush.context import get_backend, get_bus
from platypush.message.response import Response
from platypush.plugins.mqtt import MqttPlugin, action
from platypush.plugins.zwave import ZwavePlugin
from platypush.plugins.zwave._base import ZwaveBasePlugin
from platypush.plugins.zwave._constants import command_class_by_name
_NOT_IMPLEMENTED_ERR = NotImplementedError('Not implemented by zwave.mqtt')
class ZwaveMqttPlugin(MqttPlugin, ZwavePlugin):
class ZwaveMqttPlugin(MqttPlugin, ZwaveBasePlugin):
"""
This plugin allows you to manage a Z-Wave network over MQTT through
`zwavejs2mqtt <https://github.com/zwave-js/zwavejs2mqtt>`_.
@ -335,7 +335,7 @@ class ZwaveMqttPlugin(MqttPlugin, ZwavePlugin):
def _get_value(self, value_id: Optional[int] = None, id_on_network: Optional[str] = None,
value_label: Optional[str] = None, node_id: Optional[int] = None, node_name: Optional[str] = None,
use_cache: bool = True, **kwargs) -> Dict[str, Any]:
use_cache: bool = True, **_) -> Dict[str, Any]:
# Unlike python-openzwave, value_id and id_on_network are the same on zwavejs2mqtt
value_id = value_id or id_on_network
assert value_id or value_label, 'Please provide either value_id, id_on_network or value_label'
@ -1140,8 +1140,8 @@ class ZwaveMqttPlugin(MqttPlugin, ZwavePlugin):
return self._filter_values(['user_code'], node_id=node_id, node_name=node_name, **kwargs)
@action
def get_thermostats(self, node_id: Optional[int] = None, node_name: Optional[str] = None,
**kwargs) -> Dict[str, Any]:
def get_thermostats(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) \
-> Dict[str, Any]:
"""
Get the thermostats on the network or associated to a node.
@ -1155,8 +1155,8 @@ class ZwaveMqttPlugin(MqttPlugin, ZwavePlugin):
'thermostat_setback'], node_id=node_id, node_name=node_name, **kwargs)
@action
def get_protections(self, node_id: Optional[int] = None, node_name: Optional[str] = None,
**kwargs) -> Dict[str, Any]:
def get_protections(self, node_id: Optional[int] = None, node_name: Optional[str] = None, **kwargs) \
-> Dict[str, Any]:
"""
Get the protection-compatible devices on the network or associated to a node.