From 28b3672432407c62c3fcaee0ad0e18141c08f62f Mon Sep 17 00:00:00 2001 From: Fabio Manganiello Date: Tue, 5 Apr 2022 00:07:55 +0200 Subject: [PATCH] Added native support for switch entities to the zigbee.mqtt plugin. --- platypush/plugins/zigbee/mqtt/__init__.py | 490 +++++++++++++++------- 1 file changed, 343 insertions(+), 147 deletions(-) diff --git a/platypush/plugins/zigbee/mqtt/__init__.py b/platypush/plugins/zigbee/mqtt/__init__.py index 65cb2734b..f785305d0 100644 --- a/platypush/plugins/zigbee/mqtt/__init__.py +++ b/platypush/plugins/zigbee/mqtt/__init__.py @@ -9,7 +9,7 @@ from platypush.plugins.mqtt import MqttPlugin, action from platypush.plugins.switch import SwitchPlugin -class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-init] +class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-init] """ This plugin allows you to interact with Zigbee devices over MQTT through any Zigbee sniffer and `zigbee2mqtt `_. @@ -35,7 +35,8 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i .. code-block:: shell - wget https://github.com/Koenkk/Z-Stack-firmware/raw/master/coordinator/Z-Stack_Home_1.2/bin/default/CC2531_DEFAULT_20201127.zip + wget https://github.com/Koenkk/Z-Stack-firmware/raw/master\ + /coordinator/Z-Stack_Home_1.2/bin/default/CC2531_DEFAULT_20201127.zip unzip CC2531_DEFAULT_20201127.zip [sudo] cc-tool -e -w CC2531ZNP-Prod.hex @@ -78,7 +79,8 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i configured your network, to prevent accidental/malignant joins from outer Zigbee devices. - Start the ``zigbee2mqtt`` daemon on your device (the - `official documentation `_ + `official documentation `_ also contains instructions on how to configure it as a ``systemd`` service: .. code-block:: shell @@ -103,10 +105,20 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i """ - def __init__(self, host: str = 'localhost', port: int = 1883, base_topic: str = 'zigbee2mqtt', timeout: int = 10, - tls_certfile: Optional[str] = None, tls_keyfile: Optional[str] = None, - tls_version: Optional[str] = None, tls_ciphers: Optional[str] = None, - username: Optional[str] = None, password: Optional[str] = None, **kwargs): + def __init__( + self, + host: str = 'localhost', + port: int = 1883, + base_topic: str = 'zigbee2mqtt', + timeout: int = 10, + tls_certfile: Optional[str] = None, + tls_keyfile: Optional[str] = None, + tls_version: Optional[str] = None, + tls_ciphers: Optional[str] = None, + username: Optional[str] = None, + password: Optional[str] = None, + **kwargs, + ): """ :param host: Default MQTT broker where ``zigbee2mqtt`` publishes its messages (default: ``localhost``). :param port: Broker listen port (default: 1883). @@ -124,9 +136,17 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i :param username: If the connection requires user authentication, specify the username (default: None) :param password: If the connection requires user authentication, specify the password (default: None) """ - super().__init__(host=host, port=port, tls_certfile=tls_certfile, tls_keyfile=tls_keyfile, - tls_version=tls_version, tls_ciphers=tls_ciphers, username=username, - password=password, **kwargs) + super().__init__( + host=host, + port=port, + tls_certfile=tls_certfile, + tls_keyfile=tls_keyfile, + tls_version=tls_version, + tls_ciphers=tls_ciphers, + username=username, + password=password, + **kwargs, + ) self.base_topic = base_topic self.timeout = timeout @@ -135,6 +155,38 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i 'groups': {}, } + def transform_entities(self, devices): + from platypush.entities.switches import Switch + + compatible_entities = [] + for dev in devices: + dev_info = { + "type": dev.get("type"), + "date_code": dev.get("date_code"), + "ieee_address": dev.get("ieee_address"), + "network_address": dev.get("network_address"), + "power_source": dev.get("power_source"), + "software_build_id": dev.get("software_build_id"), + "model_id": dev.get("model_id"), + "model": dev.get("definition", {}).get("model"), + "vendor": dev.get("definition", {}).get("vendor"), + "supported": dev.get("supported"), + "description": dev.get("definition", {}).get("description"), + } + + switch_info = self._get_switch_info(dev) + if switch_info: + compatible_entities.append( + Switch( + id=dev['ieee_address'], + name=dev.get('friendly_name'), + state=switch_info['property'] == switch_info['value_on'], + data=dev_info, + ) + ) + + return compatible_entities + def _get_network_info(self, **kwargs): self.logger.info('Fetching Zigbee network information') client = None @@ -157,7 +209,11 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i def callback(_, __, msg): topic = msg.topic.split('/')[-1] if topic in info: - info[topic] = msg.payload.decode() if topic == 'state' else json.loads(msg.payload.decode()) + info[topic] = ( + msg.payload.decode() + if topic == 'state' + else json.loads(msg.payload.decode()) + ) info_ready_events[topic].set() return callback @@ -174,7 +230,9 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i for event in info_ready_events.values(): info_ready = event.wait(timeout=timeout) if not info_ready: - raise TimeoutError('A timeout occurred while fetching the Zigbee network information') + raise TimeoutError( + 'A timeout occurred while fetching the Zigbee network information' + ) # Cache the new results self._info['devices'] = { @@ -183,10 +241,10 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i } self._info['groups'] = { - group.get('name'): group - for group in info.get('groups', []) + group.get('name'): group for group in info.get('groups', []) } + self.publish_entities(self._info['devices'].values()) # type: ignore self.logger.info('Zigbee network configuration updated') return info finally: @@ -194,7 +252,9 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i client.loop_stop() client.disconnect() except Exception as e: - self.logger.warning('Error on MQTT client disconnection: {}'.format(str(e))) + self.logger.warning( + 'Error on MQTT client disconnection: {}'.format(str(e)) + ) def _topic(self, topic): return self.base_topic + '/' + topic @@ -204,7 +264,9 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i if isinstance(response, Response): response = response.output - assert response.get('status') != 'error', response.get('error', 'zigbee2mqtt error') + assert response.get('status') != 'error', response.get( + 'error', 'zigbee2mqtt error' + ) return response @action @@ -291,7 +353,7 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i "value_min": 150 }, { - "description": "Color of this light in the CIE 1931 color space (x/y)", + "description": "Color of this light in the XY space", "features": [ { "access": 7, @@ -315,7 +377,7 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i }, { "access": 2, - "description": "Triggers an effect on the light (e.g. make light blink for a few seconds)", + "description": "Triggers an effect on the light", "name": "effect", "property": "effect", "type": "enum", @@ -382,7 +444,9 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i return self._get_network_info(**kwargs).get('devices') @action - def permit_join(self, permit: bool = True, timeout: Optional[float] = None, **kwargs): + def permit_join( + self, permit: bool = True, timeout: Optional[float] = None, **kwargs + ): """ Enable/disable devices from joining the network. This is not persistent (will not be saved to ``configuration.yaml``). @@ -394,14 +458,19 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i """ if timeout: return self._parse_response( - self.publish(topic=self._topic('bridge/request/permit_join'), - msg={'value': permit, 'time': timeout}, - reply_topic=self._topic('bridge/response/permit_join'), - **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/permit_join'), + msg={'value': permit, 'time': timeout}, + reply_topic=self._topic('bridge/response/permit_join'), + **self._mqtt_args(**kwargs), + ) + ) - return self.publish(topic=self._topic('bridge/request/permit_join'), - msg={'value': permit}, - **self._mqtt_args(**kwargs)) + return self.publish( + topic=self._topic('bridge/request/permit_join'), + msg={'value': permit}, + **self._mqtt_args(**kwargs), + ) @action def factory_reset(self, **kwargs): @@ -413,7 +482,11 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i :param kwargs: Extra arguments to be passed to :meth:`platypush.plugins.mqtt.MqttPlugin.publish`` (default: query the default configured device). """ - self.publish(topic=self._topic('bridge/request/touchlink/factory_reset'), msg='', **self._mqtt_args(**kwargs)) + self.publish( + topic=self._topic('bridge/request/touchlink/factory_reset'), + msg='', + **self._mqtt_args(**kwargs), + ) @action def log_level(self, level: str, **kwargs): @@ -425,9 +498,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/config/log_level'), msg={'value': level}, - reply_topic=self._topic('bridge/response/config/log_level'), - **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/config/log_level'), + msg={'value': level}, + reply_topic=self._topic('bridge/response/config/log_level'), + **self._mqtt_args(**kwargs), + ) + ) @action def device_set_option(self, device: str, option: str, value: Any, **kwargs): @@ -441,14 +518,18 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/device/options'), - reply_topic=self._topic('bridge/response/device/options'), - msg={ - 'id': device, - 'options': { - option: value, - } - }, **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/device/options'), + reply_topic=self._topic('bridge/response/device/options'), + msg={ + 'id': device, + 'options': { + option: value, + }, + }, + **self._mqtt_args(**kwargs), + ) + ) @action def device_remove(self, device: str, force: bool = False, **kwargs): @@ -463,10 +544,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/device/remove'), - msg={'id': device, 'force': force}, - reply_topic=self._topic('bridge/response/device/remove'), - **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/device/remove'), + msg={'id': device, 'force': force}, + reply_topic=self._topic('bridge/response/device/remove'), + **self._mqtt_args(**kwargs), + ) + ) @action def device_ban(self, device: str, **kwargs): @@ -478,10 +562,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/device/ban'), - reply_topic=self._topic('bridge/response/device/ban'), - msg={'id': device}, - **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/device/ban'), + reply_topic=self._topic('bridge/response/device/ban'), + msg={'id': device}, + **self._mqtt_args(**kwargs), + ) + ) @action def device_whitelist(self, device: str, **kwargs): @@ -494,10 +581,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/device/whitelist'), - reply_topic=self._topic('bridge/response/device/whitelist'), - msg={'id': device}, - **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/device/whitelist'), + reply_topic=self._topic('bridge/response/device/whitelist'), + msg={'id': device}, + **self._mqtt_args(**kwargs), + ) + ) @action def device_rename(self, name: str, device: Optional[str] = None, **kwargs): @@ -516,8 +606,9 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i # noinspection PyUnresolvedReferences devices = self.devices().output - assert not [dev for dev in devices if dev.get('friendly_name') == name], \ - 'A device named {} already exists on the network'.format(name) + assert not [ + dev for dev in devices if dev.get('friendly_name') == name + ], 'A device named {} already exists on the network'.format(name) if device: req = { @@ -531,10 +622,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i } return self._parse_response( - self.publish(topic=self._topic('bridge/request/device/rename'), - msg=req, - reply_topic=self._topic('bridge/response/device/rename'), - **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/device/rename'), + msg=req, + reply_topic=self._topic('bridge/response/device/rename'), + **self._mqtt_args(**kwargs), + ) + ) @staticmethod def build_device_get_request(values: List[Dict[str, Any]]) -> dict: @@ -563,7 +657,9 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i # noinspection PyShadowingBuiltins @action - def device_get(self, device: str, property: Optional[str] = None, **kwargs) -> Dict[str, Any]: + def device_get( + self, device: str, property: Optional[str] = None, **kwargs + ) -> Dict[str, Any]: """ Get the properties of a device. The returned keys vary depending on the device. For example, a light bulb may have the "``state``" and "``brightness``" properties, while an environment sensor may have the @@ -578,26 +674,45 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i kwargs = self._mqtt_args(**kwargs) if property: - properties = self.publish(topic=self._topic(device) + '/get/' + property, reply_topic=self._topic(device), - msg={property: ''}, **kwargs).output + properties = self.publish( + topic=self._topic(device) + f'/get/{property}', + reply_topic=self._topic(device), + msg={property: ''}, + **kwargs, + ).output - assert property in properties, 'No such property: ' + property + assert property in properties, f'No such property: {property}' return {property: properties[property]} + refreshed = False if device not in self._info.get('devices', {}): # Refresh devices info self._get_network_info(**kwargs) + refreshed = True - assert self._info.get('devices', {}).get(device), 'No such device: ' + device - exposes = (self._info.get('devices', {}).get(device, {}).get('definition', {}) or {}).get('exposes', []) + assert self._info.get('devices', {}).get(device), f'No such device: {device}' + exposes = ( + self._info.get('devices', {}).get(device, {}).get('definition', {}) or {} + ).get('exposes', []) if not exposes: return {} - return self.publish(topic=self._topic(device) + '/get', reply_topic=self._topic(device), - msg=self.build_device_get_request(exposes), **kwargs) + device_info = self.publish( + topic=self._topic(device) + '/get', + reply_topic=self._topic(device), + msg=self.build_device_get_request(exposes), + **kwargs, + ) + + if not refreshed: + self.publish_entities([device_info]) # type: ignore + + return device_info @action - def devices_get(self, devices: Optional[List[str]] = None, **kwargs) -> Dict[str, dict]: + def devices_get( + self, devices: Optional[List[str]] = None, **kwargs + ) -> Dict[str, dict]: """ Get the properties of the devices connected to the network. @@ -622,14 +737,14 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i kwargs = self._mqtt_args(**kwargs) if not devices: - # noinspection PyUnresolvedReferences - devices = set([ - device['friendly_name'] or device['ieee_address'] - for device in self.devices(**kwargs).output - ]) + devices = { + [ + device['friendly_name'] or device['ieee_address'] + for device in self.devices(**kwargs).output + ] + } def worker(device: str, q: Queue): - # noinspection PyUnresolvedReferences q.put(self.device_get(device, **kwargs).output) queues = {} @@ -638,7 +753,9 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i for device in devices: queues[device] = Queue() - workers[device] = threading.Thread(target=worker, args=(device, queues[device])) + workers[device] = threading.Thread( + target=worker, args=(device, queues[device]) + ) workers[device].start() for device in devices: @@ -646,8 +763,11 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i response[device] = queues[device].get(timeout=kwargs.get('timeout')) workers[device].join(timeout=kwargs.get('timeout')) except Exception as e: - self.logger.warning('An error while getting the status of the device {}: {}'.format( - device, str(e))) + self.logger.warning( + 'An error while getting the status of the device {}: {}'.format( + device, str(e) + ) + ) return response @@ -658,7 +778,7 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i :param device: Device friendly name (default: get all devices). """ - return self.devices_get([device], *args, **kwargs) + return self.devices_get([device] if device else None, *args, **kwargs) # noinspection PyShadowingBuiltins,DuplicatedCode @action @@ -674,9 +794,12 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i :param kwargs: Extra arguments to be passed to :meth:`platypush.plugins.mqtt.MqttPlugin.publish`` (default: query the default configured device). """ - properties = self.publish(topic=self._topic(device + '/set'), - reply_topic=self._topic(device), - msg={property: value}, **self._mqtt_args(**kwargs)).output + properties = self.publish( + topic=self._topic(device + '/set'), + reply_topic=self._topic(device), + msg={property: value}, + **self._mqtt_args(**kwargs), + ).output if property: assert property in properties, 'No such property: ' + property @@ -705,9 +828,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i """ ret = self._parse_response( - self.publish(topic=self._topic('bridge/request/device/ota_update/check'), - reply_topic=self._topic('bridge/response/device/ota_update/check'), - msg={'id': device}, **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/device/ota_update/check'), + reply_topic=self._topic('bridge/response/device/ota_update/check'), + msg={'id': device}, + **self._mqtt_args(**kwargs), + ) + ) return { 'status': ret['status'], @@ -725,9 +852,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/device/ota_update/update'), - reply_topic=self._topic('bridge/response/device/ota_update/update'), - msg={'id': device}, **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/device/ota_update/update'), + reply_topic=self._topic('bridge/response/device/ota_update/update'), + msg={'id': device}, + **self._mqtt_args(**kwargs), + ) + ) @action def groups(self, **kwargs) -> List[dict]: @@ -883,16 +1014,22 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i :param kwargs: Extra arguments to be passed to :meth:`platypush.plugins.mqtt.MqttPlugin.publish`` (default: query the default configured device). """ - payload = name if id is None else { - 'id': id, - 'friendly_name': name, - } + payload = ( + name + if id is None + else { + 'id': id, + 'friendly_name': name, + } + ) return self._parse_response( - self.publish(topic=self._topic('bridge/request/group/add'), - reply_topic=self._topic('bridge/response/group/add'), - msg=payload, - **self._mqtt_args(**kwargs)) + self.publish( + topic=self._topic('bridge/request/group/add'), + reply_topic=self._topic('bridge/response/group/add'), + msg=payload, + **self._mqtt_args(**kwargs), + ) ) @action @@ -911,9 +1048,12 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i if property: msg = {property: ''} - properties = self.publish(topic=self._topic(group + '/get'), - reply_topic=self._topic(group), - msg=msg, **self._mqtt_args(**kwargs)).output + properties = self.publish( + topic=self._topic(group + '/get'), + reply_topic=self._topic(group), + msg=msg, + **self._mqtt_args(**kwargs), + ).output if property: assert property in properties, 'No such property: ' + property @@ -935,9 +1075,12 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i :param kwargs: Extra arguments to be passed to :meth:`platypush.plugins.mqtt.MqttPlugin.publish`` (default: query the default configured device). """ - properties = self.publish(topic=self._topic(group + '/set'), - reply_topic=self._topic(group), - msg={property: value}, **self._mqtt_args(**kwargs)).output + properties = self.publish( + topic=self._topic(group + '/set'), + reply_topic=self._topic(group), + msg={property: value}, + **self._mqtt_args(**kwargs), + ).output if property: assert property in properties, 'No such property: ' + property @@ -961,13 +1104,18 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i # noinspection PyUnresolvedReferences groups = {group.get('friendly_name'): group for group in self.groups().output} - assert name not in groups, 'A group named {} already exists on the network'.format(name) + assert ( + name not in groups + ), 'A group named {} already exists on the network'.format(name) return self._parse_response( - self.publish(topic=self._topic('bridge/request/group/rename'), - reply_topic=self._topic('bridge/response/group/rename'), - msg={'from': group, 'to': name} if group else name, - **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/group/rename'), + reply_topic=self._topic('bridge/response/group/rename'), + msg={'from': group, 'to': name} if group else name, + **self._mqtt_args(**kwargs), + ) + ) @action def group_remove(self, name: str, **kwargs): @@ -979,10 +1127,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/group/remove'), - reply_topic=self._topic('bridge/response/group/remove'), - msg=name, - **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/group/remove'), + reply_topic=self._topic('bridge/response/group/remove'), + msg=name, + **self._mqtt_args(**kwargs), + ) + ) @action def group_add_device(self, group: str, device: str, **kwargs): @@ -995,12 +1146,16 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/group/members/add'), - reply_topic=self._topic('bridge/response/group/members/add'), - msg={ - 'group': group, - 'device': device, - }, **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/group/members/add'), + reply_topic=self._topic('bridge/response/group/members/add'), + msg={ + 'group': group, + 'device': device, + }, + **self._mqtt_args(**kwargs), + ) + ) @action def group_remove_device(self, group: str, device: Optional[str] = None, **kwargs): @@ -1015,13 +1170,23 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i """ return self._parse_response( self.publish( - topic=self._topic('bridge/request/group/members/remove{}'.format('_all' if device is None else '')), + topic=self._topic( + 'bridge/request/group/members/remove{}'.format( + '_all' if device is None else '' + ) + ), reply_topic=self._topic( - 'bridge/response/group/members/remove{}'.format('_all' if device is None else '')), + 'bridge/response/group/members/remove{}'.format( + '_all' if device is None else '' + ) + ), msg={ 'group': group, 'device': device, - }, **self._mqtt_args(**kwargs))) + }, + **self._mqtt_args(**kwargs), + ) + ) @action def bind_devices(self, source: str, target: str, **kwargs): @@ -1040,9 +1205,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/device/bind'), - reply_topic=self._topic('bridge/response/device/bind'), - msg={'from': source, 'to': target}, **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/device/bind'), + reply_topic=self._topic('bridge/response/device/bind'), + msg={'from': source, 'to': target}, + **self._mqtt_args(**kwargs), + ) + ) @action def unbind_devices(self, source: str, target: str, **kwargs): @@ -1057,9 +1226,13 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i (default: query the default configured device). """ return self._parse_response( - self.publish(topic=self._topic('bridge/request/device/unbind'), - reply_topic=self._topic('bridge/response/device/unbind'), - msg={'from': source, 'to': target}, **self._mqtt_args(**kwargs))) + self.publish( + topic=self._topic('bridge/request/device/unbind'), + reply_topic=self._topic('bridge/response/device/unbind'), + msg={'from': source, 'to': target}, + **self._mqtt_args(**kwargs), + ) + ) @action def on(self, device, *args, **kwargs) -> dict: @@ -1069,8 +1242,12 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i """ switch_info = self._get_switches_info().get(device) assert switch_info, '{} is not a valid switch'.format(device) - props = self.device_set(device, switch_info['property'], switch_info['value_on']).output - return self._properties_to_switch(device=device, props=props, switch_info=switch_info) + props = self.device_set( + device, switch_info['property'], switch_info['value_on'] + ).output + return self._properties_to_switch( + device=device, props=props, switch_info=switch_info + ) @action def off(self, device, *args, **kwargs) -> dict: @@ -1080,8 +1257,12 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i """ switch_info = self._get_switches_info().get(device) assert switch_info, '{} is not a valid switch'.format(device) - props = self.device_set(device, switch_info['property'], switch_info['value_off']).output - return self._properties_to_switch(device=device, props=props, switch_info=switch_info) + props = self.device_set( + device, switch_info['property'], switch_info['value_off'] + ).output + return self._properties_to_switch( + device=device, props=props, switch_info=switch_info + ) @action def toggle(self, device, *args, **kwargs) -> dict: @@ -1091,8 +1272,12 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i """ switch_info = self._get_switches_info().get(device) assert switch_info, '{} is not a valid switch'.format(device) - props = self.device_set(device, switch_info['property'], switch_info['value_toggle']).output - return self._properties_to_switch(device=device, props=props, switch_info=switch_info) + props = self.device_set( + device, switch_info['property'], switch_info['value_toggle'] + ).output + return self._properties_to_switch( + device=device, props=props, switch_info=switch_info + ) @staticmethod def _properties_to_switch(device: str, props: dict, switch_info: dict) -> dict: @@ -1103,32 +1288,39 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i **props, } + @staticmethod + def _get_switch_info(device_info: dict) -> dict: + exposes = (device_info.get('definition', {}) or {}).get('exposes', []) + for exposed in exposes: + for feature in exposed.get('features', []): + if ( + feature.get('type') == 'binary' + and 'value_on' in feature + and 'value_off' in feature + and feature.get('access', 0) & 2 + ): + return { + 'property': feature['property'], + 'value_on': feature['value_on'], + 'value_off': feature['value_off'], + 'value_toggle': feature.get('value_toggle', None), + } + + return {} + def _get_switches_info(self) -> dict: - def switch_info(device_info: dict) -> dict: - exposes = (device_info.get('definition', {}) or {}).get('exposes', []) - for exposed in exposes: - for feature in exposed.get('features', []): - if feature.get('type') == 'binary' and 'value_on' in feature and 'value_off' in feature and \ - feature.get('access', 0) & 2: - return { - 'property': feature['property'], - 'value_on': feature['value_on'], - 'value_off': feature['value_off'], - 'value_toggle': feature.get('value_toggle', None), - } - - return {} - # noinspection PyUnresolvedReferences devices = self.devices().output switches_info = {} for device in devices: - info = switch_info(device) + info = self._get_switch_info(device) if not info: continue - switches_info[device.get('friendly_name', device.get('ieee_address'))] = info + switches_info[ + device.get('friendly_name', device.get('ieee_address')) + ] = info return switches_info @@ -1142,8 +1334,12 @@ class ZigbeeMqttPlugin(MqttPlugin, SwitchPlugin): # lgtm [py/missing-call-to-i switches_info = self._get_switches_info() # noinspection PyUnresolvedReferences return [ - self._properties_to_switch(device=name, props=switch, switch_info=switches_info[name]) - for name, switch in self.devices_get(list(switches_info.keys())).output.items() + self._properties_to_switch( + device=name, props=switch, switch_info=switches_info[name] + ) + for name, switch in self.devices_get( + list(switches_info.keys()) + ).output.items() ]