platypush/platypush
platypush/platypush
2
2
Fork 4
Code Issues 30 Pull requests Projects 4 Releases Packages Wiki Activity

Implemented parent/child support for zigbee.mqtt entities

Browse source
This commit is contained in:
Fabio Manganiello 2022-11-30 00:55:04 +01:00
parent abaeabea22
commit 2b532c1947
Signed by: blacklight
GPG key ID: D90FBA7F76362774
1 changed files with 38 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
platypush/plugins/zigbee/mqtt/__init__.py
View file

@ -7,6 +7,7 @@ from typing import Optional, List, Any, Dict, Union, Tuple
from platypush.entities import manages from platypush.entities import manages
from platypush.entities.batteries import Battery from platypush.entities.batteries import Battery
from platypush.entities.devices import Device
from platypush.entities.dimmers import Dimmer from platypush.entities.dimmers import Dimmer
from platypush.entities.electricity import ( from platypush.entities.electricity import (
CurrentSensor, CurrentSensor,
@ -30,7 +31,7 @@ from platypush.message.response import Response
from platypush.plugins.mqtt import MqttPlugin, action from platypush.plugins.mqtt import MqttPlugin, action
@manages(Battery, Dimmer, Light, LinkQuality, Sensor, Switch) @manages(Battery, Device, Dimmer, Light, LinkQuality, Sensor, Switch)
class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init] class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init]
""" """
This plugin allows you to interact with Zigbee devices over MQTT through any Zigbee sniffer and This plugin allows you to interact with Zigbee devices over MQTT through any Zigbee sniffer and
@ -184,30 +185,34 @@ class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init]
if not dev: if not dev:
continue continue
dev_def = dev.get("definition") or {} dev_def = dev.get('definition') or {}
dev_info = { dev_info = {
"type": dev.get("type"), attr: dev.get(attr)
"date_code": dev.get("date_code"), for attr in (
"ieee_address": dev.get("ieee_address"), 'type',
"network_address": dev.get("network_address"), 'date_code',
"power_source": dev.get("power_source"), 'ieee_address',
"software_build_id": dev.get("software_build_id"), 'network_address',
"model_id": dev.get("model_id"), 'power_source',
"model": dev_def.get("model"), 'software_build_id',
"vendor": dev_def.get("vendor"), 'model_id',
"supported": dev.get("supported"), 'model',
'vendor',
'supported',
)
} }
reachable = dev.get('supported', False)
light_info = self._get_light_meta(dev) light_info = self._get_light_meta(dev)
switch_info = self._get_switch_meta(dev) switch_info = self._get_switch_meta(dev)
compatible_entities += ( dev_entities = [
self._get_sensors(dev) *self._get_sensors(dev),
+ self._get_dimmers(dev) *self._get_dimmers(dev),
+ self._get_enum_switches(dev) *self._get_enum_switches(dev),
) ]
if light_info: if light_info:
compatible_entities.append( dev_entities.append(
Light( Light(
id=f'{dev["ieee_address"]}:light', id=f'{dev["ieee_address"]}:light',
name=dev.get('friendly_name'), name=dev.get('friendly_name'),
@ -256,7 +261,7 @@ class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init]
) )
) )
elif switch_info and dev.get('state', {}).get('state') is not None: elif switch_info and dev.get('state', {}).get('state') is not None:
compatible_entities.append( dev_entities.append(
Switch( Switch(
id=f'{dev["ieee_address"]}:switch', id=f'{dev["ieee_address"]}:switch',
name=dev.get('friendly_name'), name=dev.get('friendly_name'),
@ -270,6 +275,20 @@ class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init]
) )
) )
if dev_entities:
parent = Device(
id=dev['ieee_address'],
name=dev.get('friendly_name'),
description=dev_def.get('description'),
reachable=reachable,
)
for entity in dev_entities:
entity.parent = parent
dev_entities.append(parent)
compatible_entities += dev_entities
return super().transform_entities(compatible_entities) # type: ignore return super().transform_entities(compatible_entities) # type: ignore
def _get_network_info(self, **kwargs) -> dict: def _get_network_info(self, **kwargs) -> dict: