Fix Shelly uptime sensor ()

Fix sensor to include time zone
Report new value only if delta > 5 seconds
Modify REST sensors class to use callable attributes
This commit is contained in:
Shay Levy 2020-11-27 10:40:06 +02:00 committed by GitHub
parent 5e3f4954f7
commit 2498340e1f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 49 additions and 59 deletions
homeassistant/components/shelly

View file

@ -75,19 +75,19 @@ SENSORS = {
REST_SENSORS = {
"cloud": RestAttributeDescription(
name="Cloud",
value=lambda status, _: status["cloud"]["connected"],
device_class=DEVICE_CLASS_CONNECTIVITY,
default_enabled=False,
path="cloud/connected",
),
"fwupdate": RestAttributeDescription(
name="Firmware update",
icon="mdi:update",
value=lambda status, _: status["update"]["has_update"],
default_enabled=False,
path="update/has_update",
attributes=[
{"description": "latest_stable_version", "path": "update/new_version"},
{"description": "installed_version", "path": "update/old_version"},
],
device_state_attributes=lambda status: {
"latest_stable_version": status["update"]["new_version"],
"installed_version": status["update"]["old_version"],
},
),
}

View file

@ -9,7 +9,7 @@ from homeassistant.helpers import device_registry, entity, update_coordinator
from . import ShellyDeviceRestWrapper, ShellyDeviceWrapper
from .const import COAP, DATA_CONFIG_ENTRY, DOMAIN, REST
from .utils import async_remove_shelly_entity, get_entity_name, get_rest_value_from_path
from .utils import async_remove_shelly_entity, get_entity_name
async def async_setup_entry_attribute_entities(
@ -64,15 +64,20 @@ async def async_setup_entry_rest(
entities = []
for sensor_id in sensors:
_desc = sensors.get(sensor_id)
description = sensors.get(sensor_id)
if not wrapper.device.settings.get("sleep_mode"):
entities.append(_desc)
entities.append((sensor_id, description))
if not entities:
return
async_add_entities([sensor_class(wrapper, description) for description in entities])
async_add_entities(
[
sensor_class(wrapper, sensor_id, description)
for sensor_id, description in entities
]
)
@dataclass
@ -98,15 +103,13 @@ class BlockAttributeDescription:
class RestAttributeDescription:
"""Class to describe a REST sensor."""
path: str
name: str
# Callable = lambda attr_info: unit
icon: Optional[str] = None
unit: Union[None, str, Callable[[dict], str]] = None
value: Callable[[Any], Any] = lambda val: val
unit: Optional[str] = None
value: Callable[[dict, Any], Any] = None
device_class: Optional[str] = None
default_enabled: bool = True
attributes: Optional[dict] = None
device_state_attributes: Optional[Callable[[dict], Optional[dict]]] = None
class ShellyBlockEntity(entity.Entity):
@ -247,17 +250,18 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
"""Class to load info from REST."""
def __init__(
self, wrapper: ShellyDeviceWrapper, description: RestAttributeDescription
self,
wrapper: ShellyDeviceWrapper,
attribute: str,
description: RestAttributeDescription,
) -> None:
"""Initialize sensor."""
super().__init__(wrapper)
self.wrapper = wrapper
self.attribute = attribute
self.description = description
self._unit = self.description.unit
self._name = get_entity_name(wrapper.device, None, self.description.name)
self.path = self.description.path
self._attributes = self.description.attributes
self._last_value = None
@property
def name(self):
@ -283,10 +287,11 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
@property
def attribute_value(self):
"""Attribute."""
return get_rest_value_from_path(
self.wrapper.device.status, self.description.device_class, self.path
"""Value of sensor."""
self._last_value = self.description.value(
self.wrapper.device.status, self._last_value
)
return self._last_value
@property
def unit_of_measurement(self):
@ -306,23 +311,12 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
@property
def unique_id(self):
"""Return unique ID of entity."""
return f"{self.wrapper.mac}-{self.description.path}"
return f"{self.wrapper.mac}-{self.attribute}"
@property
def device_state_attributes(self) -> dict:
"""Return the state attributes."""
if self._attributes is None:
if self.description.device_state_attributes is None:
return None
attributes = dict()
for attrib in self._attributes:
description = attrib.get("description")
attribute_value = get_rest_value_from_path(
self.wrapper.device.status,
self.description.device_class,
attrib.get("path"),
)
attributes[description] = attribute_value
return attributes
return self.description.device_state_attributes(self.wrapper.device.status)

View file

@ -21,7 +21,7 @@ from .entity import (
async_setup_entry_attribute_entities,
async_setup_entry_rest,
)
from .utils import temperature_unit
from .utils import get_device_uptime, temperature_unit
SENSORS = {
("device", "battery"): BlockAttributeDescription(
@ -170,15 +170,15 @@ REST_SENSORS = {
"rssi": RestAttributeDescription(
name="RSSI",
unit=SIGNAL_STRENGTH_DECIBELS,
value=lambda status, _: status["wifi_sta"]["rssi"],
device_class=sensor.DEVICE_CLASS_SIGNAL_STRENGTH,
default_enabled=False,
path="wifi_sta/rssi",
),
"uptime": RestAttributeDescription(
name="Uptime",
value=get_device_uptime,
device_class=sensor.DEVICE_CLASS_TIMESTAMP,
default_enabled=False,
path="uptime",
),
}

View file

@ -1,13 +1,13 @@
"""Shelly helpers functions."""
from datetime import datetime, timedelta
from datetime import timedelta
import logging
from typing import Optional
import aioshelly
from homeassistant.components.sensor import DEVICE_CLASS_TIMESTAMP
from homeassistant.const import TEMP_CELSIUS, TEMP_FAHRENHEIT
from homeassistant.util.dt import parse_datetime, utcnow
from .const import DOMAIN
@ -81,23 +81,6 @@ def get_entity_name(
return entity_name
def get_rest_value_from_path(status, device_class, path: str):
"""Parser for REST path from device status."""
if "/" not in path:
attribute_value = status[path]
else:
attribute_value = status[path.split("/")[0]][path.split("/")[1]]
if device_class == DEVICE_CLASS_TIMESTAMP:
last_boot = datetime.utcnow() - timedelta(seconds=attribute_value)
attribute_value = last_boot.replace(microsecond=0).isoformat()
if "new_version" in path:
attribute_value = attribute_value.split("/")[1].split("@")[0]
return attribute_value
def is_momentary_input(settings: dict, block: aioshelly.Block) -> bool:
"""Return true if input button settings is set to a momentary type."""
button = settings.get("relays") or settings.get("lights") or settings.get("inputs")
@ -112,3 +95,16 @@ def is_momentary_input(settings: dict, block: aioshelly.Block) -> bool:
button_type = button[channel].get("btn_type")
return button_type in ["momentary", "momentary_on_release"]
def get_device_uptime(status: dict, last_uptime: str) -> str:
"""Return device uptime string, tolerate up to 5 seconds deviation."""
uptime = utcnow() - timedelta(seconds=status["uptime"])
if not last_uptime:
return uptime.replace(microsecond=0).isoformat()
if abs((uptime - parse_datetime(last_uptime)).total_seconds()) > 5:
return uptime.replace(microsecond=0).isoformat()
return last_uptime