Added support for link_quality entities to zigbee.mqtt

This commit is contained in:
Fabio Manganiello 2022-10-30 11:03:22 +01:00
parent 78dc8416fb
commit a1cf671334
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
6 changed files with 158 additions and 17 deletions

View file

@ -0,0 +1,64 @@
<template>
<div class="entity link-quality-container">
<div class="head">
<div class="col-1 icon">
<EntityIcon
:icon="value.meta?.icon || {}"
:loading="loading"
:error="error" />
</div>
<div class="col-s-8 col-m-9 label">
<div class="name" v-text="value.name" />
</div>
<div class="col-s-3 col-m-2 buttons pull-right">
<span class="value-percent"
v-text="valuePercent + '%'"
v-if="valuePercent != null" />
</div>
</div>
</div>
</template>
<script>
import EntityMixin from "./EntityMixin"
import EntityIcon from "./EntityIcon"
export default {
name: 'LinkQuality',
components: {EntityIcon},
mixins: [EntityMixin],
data() {
return {
expanded: false,
}
},
computed: {
valuePercent() {
if (this.value?.value == null)
return null
const min = this.value.min || 0
const max = this.value.max || 100
return ((100 * this.value.value) / (max - min)).toFixed(0)
},
},
}
</script>
<style lang="scss" scoped>
@import "common";
.link-quality-container {
.head {
.value-percent {
font-size: 1.1em;
font-weight: bold;
opacity: 0.7;
}
}
}
</style>

View file

@ -39,6 +39,14 @@
} }
}, },
"link_quality": {
"name": "Link Quality",
"name_plural": "Link Qualities",
"icon": {
"class": "fas fa-signal"
}
},
"switch": { "switch": {
"name": "Switch", "name": "Switch",
"name_plural": "Switches", "name_plural": "Switches",

View file

@ -10,19 +10,9 @@ if not entity_types_registry.get('Battery'):
__tablename__ = 'battery' __tablename__ = 'battery'
def __init__( def __init__(
self, self, *args, unit: str = '%', min: float = 0, max: float = 100, **kwargs
*args,
value,
unit: str = '%',
min: float = 0,
max: float = 100,
**kwargs
): ):
super().__init__(*args, **kwargs) super().__init__(*args, min=min, max=max, unit=unit, **kwargs)
self.value = float(value)
self.unit = unit
self.min = min
self.max = max
id = Column( id = Column(
Integer, ForeignKey(NumericSensor.id, ondelete='CASCADE'), primary_key=True Integer, ForeignKey(NumericSensor.id, ondelete='CASCADE'), primary_key=True

View file

@ -0,0 +1,27 @@
from sqlalchemy import Column, Integer, ForeignKey
from .devices import entity_types_registry
from .sensors import NumericSensor
if not entity_types_registry.get('LinkQuality'):
class LinkQuality(NumericSensor):
__tablename__ = 'link_quality'
def __init__(
self, *args, unit: str = '%', min: float = 0, max: float = 100, **kwargs
):
super().__init__(*args, min=min, max=max, unit=unit, **kwargs)
id = Column(
Integer, ForeignKey(NumericSensor.id, ondelete='CASCADE'), primary_key=True
)
__mapper_args__ = {
'polymorphic_identity': __tablename__,
}
entity_types_registry['LinkQuality'] = LinkQuality
else:
LinkQuality = entity_types_registry['LinkQuality']

View file

@ -33,6 +33,7 @@ if not entity_types_registry.get('NumericSensor'):
value = Column(Numeric) value = Column(Numeric)
min = Column(Numeric) min = Column(Numeric)
max = Column(Numeric) max = Column(Numeric)
unit = Column(String)
__mapper_args__ = { __mapper_args__ = {
'polymorphic_identity': __tablename__, 'polymorphic_identity': __tablename__,

View file

@ -7,13 +7,14 @@ from typing import Optional, List, Any, Dict, Union
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.lights import Light from platypush.entities.lights import Light
from platypush.entities.linkquality import LinkQuality
from platypush.entities.switches import Switch from platypush.entities.switches import Switch
from platypush.message import Mapping from platypush.message import Mapping
from platypush.message.response import Response from platypush.message.response import Response
from platypush.plugins.mqtt import MqttPlugin, action from platypush.plugins.mqtt import MqttPlugin, action
@manages(Light, Switch, Battery) @manages(Light, Switch, LinkQuality, Battery)
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
@ -187,6 +188,7 @@ class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init]
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)
battery_info = self._get_battery_meta(dev) battery_info = self._get_battery_meta(dev)
link_quality_info = self._get_link_quality_meta(dev)
if light_info: if light_info:
converted_entities.append( converted_entities.append(
@ -246,6 +248,8 @@ class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init]
== switch_info['value_on'], == switch_info['value_on'],
description=dev_def.get("description"), description=dev_def.get("description"),
data=dev_info, data=dev_info,
is_read_only=link_quality_info['is_read_only'],
is_write_only=link_quality_info['is_write_only'],
) )
) )
@ -254,10 +258,28 @@ class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init]
Battery( Battery(
id=dev['ieee_address'], id=dev['ieee_address'],
name=battery_info.get('friendly_name'), name=battery_info.get('friendly_name'),
value=dev.get('battery'), value=dev.get('state', {}).get('battery'),
description=battery_info.get('description'), description=battery_info.get('description'),
min=battery_info['min'], min=battery_info['min'],
max=battery_info['max'], max=battery_info['max'],
is_read_only=battery_info['is_read_only'],
is_write_only=battery_info['is_write_only'],
data=dev_info,
)
)
if link_quality_info:
converted_entities.append(
LinkQuality(
id=dev['ieee_address'],
name=link_quality_info.get('friendly_name'),
value=dev.get('state', {}).get('linkquality'),
description=link_quality_info.get('description'),
min=link_quality_info['min'],
max=link_quality_info['max'],
unit=link_quality_info['unit'],
is_read_only=link_quality_info['is_read_only'],
is_write_only=link_quality_info['is_write_only'],
data=dev_info, data=dev_info,
) )
) )
@ -1455,13 +1477,15 @@ class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init]
return { return {
'friendly_name': ( 'friendly_name': (
device_info.get('friendly_name', '[Unnamed device]') device_info.get('friendly_name', '[Unnamed device]')
+ ' [Battery]' + ' ['
+ feature.get('description', 'Battery')
+ ']'
), ),
'ieee_address': device_info.get('friendly_name'), 'ieee_address': device_info.get('friendly_name'),
'property': feature['property'], 'property': feature['property'],
'description': feature.get('description'), 'description': feature.get('description'),
'value_min': feature.get('value_min', 0), 'min': feature.get('value_min', 0),
'value_max': feature.get('value_max', 100), 'max': feature.get('value_max', 100),
'unit': feature.get('unit', '%'), 'unit': feature.get('unit', '%'),
'is_read_only': not bool(feature.get('access', 0) & 2), 'is_read_only': not bool(feature.get('access', 0) & 2),
'is_write_only': not bool(feature.get('access', 0) & 1), 'is_write_only': not bool(feature.get('access', 0) & 1),
@ -1469,6 +1493,33 @@ class ZigbeeMqttPlugin(MqttPlugin): # lgtm [py/missing-call-to-init]
return {} return {}
@staticmethod
def _get_link_quality_meta(device_info: dict) -> dict:
exposes = (device_info.get('definition', {}) or {}).get('exposes', [])
for exposed in exposes:
if (
exposed.get('property') == 'linkquality'
and exposed.get('type') == 'numeric'
):
return {
'friendly_name': (
device_info.get('friendly_name', '[Unnamed device]')
+ ' ['
+ exposed.get('description', 'Link Quality')
+ ']'
),
'ieee_address': device_info.get('friendly_name'),
'property': exposed['property'],
'description': exposed.get('description'),
'min': exposed.get('value_min', 0),
'max': exposed.get('value_max', 100),
'unit': exposed.get('unit', '%'),
'is_read_only': not bool(exposed.get('access', 0) & 2),
'is_write_only': not bool(exposed.get('access', 0) & 1),
}
return {}
@staticmethod @staticmethod
def _get_light_meta(device_info: dict) -> dict: def _get_light_meta(device_info: dict) -> dict:
exposes = (device_info.get('definition', {}) or {}).get('exposes', []) exposes = (device_info.get('definition', {}) or {}).get('exposes', [])