Use EntityDescription - efergy (#54210)

This commit is contained in:
Robert Hillis 2021-09-21 17:53:35 -04:00 committed by GitHub
parent b7a758bd0c
commit a653da137c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -1,18 +1,31 @@
"""Support for Efergy sensors.""" """Support for Efergy sensors."""
from __future__ import annotations
import logging import logging
from typing import Any
import requests import requests
import voluptuous as vol import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity from homeassistant.components.sensor import (
PLATFORM_SCHEMA,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import ( from homeassistant.const import (
CONF_CURRENCY, CONF_CURRENCY,
CONF_MONITORED_VARIABLES, CONF_MONITORED_VARIABLES,
CONF_TYPE, CONF_TYPE,
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_MONETARY,
DEVICE_CLASS_POWER,
ENERGY_KILO_WATT_HOUR, ENERGY_KILO_WATT_HOUR,
POWER_WATT, POWER_WATT,
) )
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
_RESOURCE = "https://engage.efergy.com/mobile_proxy/" _RESOURCE = "https://engage.efergy.com/mobile_proxy/"
@ -31,12 +44,34 @@ CONF_CURRENT_VALUES = "current_values"
DEFAULT_PERIOD = "year" DEFAULT_PERIOD = "year"
DEFAULT_UTC_OFFSET = "0" DEFAULT_UTC_OFFSET = "0"
SENSOR_TYPES = { SENSOR_TYPES: dict[str, SensorEntityDescription] = {
CONF_INSTANT: ["Energy Usage", POWER_WATT], CONF_INSTANT: SensorEntityDescription(
CONF_AMOUNT: ["Energy Consumed", ENERGY_KILO_WATT_HOUR], key=CONF_INSTANT,
CONF_BUDGET: ["Energy Budget", None], name="Energy Usage",
CONF_COST: ["Energy Cost", None], device_class=DEVICE_CLASS_POWER,
CONF_CURRENT_VALUES: ["Per-Device Usage", POWER_WATT], native_unit_of_measurement=POWER_WATT,
),
CONF_AMOUNT: SensorEntityDescription(
key=CONF_AMOUNT,
name="Energy Consumed",
device_class=DEVICE_CLASS_ENERGY,
native_unit_of_measurement=ENERGY_KILO_WATT_HOUR,
),
CONF_BUDGET: SensorEntityDescription(
key=CONF_BUDGET,
name="Energy Budget",
),
CONF_COST: SensorEntityDescription(
key=CONF_COST,
name="Energy Cost",
device_class=DEVICE_CLASS_MONETARY,
),
CONF_CURRENT_VALUES: SensorEntityDescription(
key=CONF_CURRENT_VALUES,
name="Per-Device Usage",
device_class=DEVICE_CLASS_POWER,
native_unit_of_measurement=POWER_WATT,
),
} }
TYPES_SCHEMA = vol.In(SENSOR_TYPES) TYPES_SCHEMA = vol.In(SENSOR_TYPES)
@ -58,7 +93,12 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
) )
def setup_platform(hass, config, add_entities, discovery_info=None): def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType = None,
) -> None:
"""Set up the Efergy sensor.""" """Set up the Efergy sensor."""
app_token = config.get(CONF_APPTOKEN) app_token = config.get(CONF_APPTOKEN)
utc_offset = str(config.get(CONF_UTC_OFFSET)) utc_offset = str(config.get(CONF_UTC_OFFSET))
@ -72,21 +112,21 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
sid = sensor["sid"] sid = sensor["sid"]
dev.append( dev.append(
EfergySensor( EfergySensor(
variable[CONF_TYPE],
app_token, app_token,
utc_offset, utc_offset,
variable[CONF_PERIOD], variable[CONF_PERIOD],
variable[CONF_CURRENCY], variable[CONF_CURRENCY],
sid, SENSOR_TYPES[variable[CONF_TYPE]],
sid=sid,
) )
) )
dev.append( dev.append(
EfergySensor( EfergySensor(
variable[CONF_TYPE],
app_token, app_token,
utc_offset, utc_offset,
variable[CONF_PERIOD], variable[CONF_PERIOD],
variable[CONF_CURRENCY], variable[CONF_CURRENCY],
SENSOR_TYPES[variable[CONF_TYPE]],
) )
) )
@ -96,59 +136,46 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
class EfergySensor(SensorEntity): class EfergySensor(SensorEntity):
"""Implementation of an Efergy sensor.""" """Implementation of an Efergy sensor."""
def __init__(self, sensor_type, app_token, utc_offset, period, currency, sid=None): def __init__(
self,
app_token: Any,
utc_offset: str,
period: str,
currency: str,
description: SensorEntityDescription,
sid: str = None,
) -> None:
"""Initialize the sensor.""" """Initialize the sensor."""
self.entity_description = description
self.sid = sid self.sid = sid
if sid: if sid:
self._name = f"efergy_{sid}" self._attr_name = f"efergy_{sid}"
else:
self._name = SENSOR_TYPES[sensor_type][0]
self.type = sensor_type
self.app_token = app_token self.app_token = app_token
self.utc_offset = utc_offset self.utc_offset = utc_offset
self._state = None
self.period = period self.period = period
self.currency = currency if description.key == CONF_COST:
if self.type == "cost": self._attr_native_unit_of_measurement = f"{currency}/{period}"
self._unit_of_measurement = f"{self.currency}/{self.period}"
else:
self._unit_of_measurement = SENSOR_TYPES[sensor_type][1]
@property def update(self) -> None:
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def native_value(self):
"""Return the state of the sensor."""
return self._state
@property
def native_unit_of_measurement(self):
"""Return the unit of measurement of this entity, if any."""
return self._unit_of_measurement
def update(self):
"""Get the Efergy monitor data from the web service.""" """Get the Efergy monitor data from the web service."""
try: try:
if self.type == "instant_readings": if self.entity_description.key == CONF_INSTANT:
url_string = f"{_RESOURCE}getInstant?token={self.app_token}" url_string = f"{_RESOURCE}getInstant?token={self.app_token}"
response = requests.get(url_string, timeout=10) response = requests.get(url_string, timeout=10)
self._state = response.json()["reading"] self._attr_native_value = response.json()["reading"]
elif self.type == "amount": elif self.entity_description.key == CONF_AMOUNT:
url_string = f"{_RESOURCE}getEnergy?token={self.app_token}&offset={self.utc_offset}&period={self.period}" url_string = f"{_RESOURCE}getEnergy?token={self.app_token}&offset={self.utc_offset}&period={self.period}"
response = requests.get(url_string, timeout=10) response = requests.get(url_string, timeout=10)
self._state = response.json()["sum"] self._attr_native_value = response.json()["sum"]
elif self.type == "budget": elif self.entity_description.key == CONF_BUDGET:
url_string = f"{_RESOURCE}getBudget?token={self.app_token}" url_string = f"{_RESOURCE}getBudget?token={self.app_token}"
response = requests.get(url_string, timeout=10) response = requests.get(url_string, timeout=10)
self._state = response.json()["status"] self._attr_native_value = response.json()["status"]
elif self.type == "cost": elif self.entity_description.key == CONF_COST:
url_string = f"{_RESOURCE}getCost?token={self.app_token}&offset={self.utc_offset}&period={self.period}" url_string = f"{_RESOURCE}getCost?token={self.app_token}&offset={self.utc_offset}&period={self.period}"
response = requests.get(url_string, timeout=10) response = requests.get(url_string, timeout=10)
self._state = response.json()["sum"] self._attr_native_value = response.json()["sum"]
elif self.type == "current_values": elif self.entity_description.key == CONF_CURRENT_VALUES:
url_string = ( url_string = (
f"{_RESOURCE}getCurrentValuesSummary?token={self.app_token}" f"{_RESOURCE}getCurrentValuesSummary?token={self.app_token}"
) )
@ -156,8 +183,6 @@ class EfergySensor(SensorEntity):
for sensor in response.json(): for sensor in response.json():
if self.sid == sensor["sid"]: if self.sid == sensor["sid"]:
measurement = next(iter(sensor["data"][0].values())) measurement = next(iter(sensor["data"][0].values()))
self._state = measurement self._attr_native_value = measurement
else:
self._state = None
except (requests.RequestException, ValueError, KeyError): except (requests.RequestException, ValueError, KeyError):
_LOGGER.warning("Could not update status for %s", self.name) _LOGGER.warning("Could not update status for %s", self.name)