286 lines
9.2 KiB
Python
286 lines
9.2 KiB
Python
"""Support for monitoring emoncms feeds."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
from http import HTTPStatus
|
|
import logging
|
|
from typing import Any
|
|
|
|
import requests
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.sensor import (
|
|
PLATFORM_SCHEMA,
|
|
SensorDeviceClass,
|
|
SensorEntity,
|
|
SensorStateClass,
|
|
)
|
|
from homeassistant.const import (
|
|
CONF_API_KEY,
|
|
CONF_ID,
|
|
CONF_UNIT_OF_MEASUREMENT,
|
|
CONF_URL,
|
|
CONF_VALUE_TEMPLATE,
|
|
STATE_UNKNOWN,
|
|
UnitOfPower,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers import template
|
|
import homeassistant.helpers.config_validation as cv
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
|
from homeassistant.util import Throttle
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
ATTR_FEEDID = "FeedId"
|
|
ATTR_FEEDNAME = "FeedName"
|
|
ATTR_LASTUPDATETIME = "LastUpdated"
|
|
ATTR_LASTUPDATETIMESTR = "LastUpdatedStr"
|
|
ATTR_SIZE = "Size"
|
|
ATTR_TAG = "Tag"
|
|
ATTR_USERID = "UserId"
|
|
|
|
CONF_EXCLUDE_FEEDID = "exclude_feed_id"
|
|
CONF_ONLY_INCLUDE_FEEDID = "include_only_feed_id"
|
|
CONF_SENSOR_NAMES = "sensor_names"
|
|
|
|
DECIMALS = 2
|
|
DEFAULT_UNIT = UnitOfPower.WATT
|
|
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=5)
|
|
|
|
ONLY_INCL_EXCL_NONE = "only_include_exclude_or_none"
|
|
|
|
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|
{
|
|
vol.Required(CONF_API_KEY): cv.string,
|
|
vol.Required(CONF_URL): cv.string,
|
|
vol.Required(CONF_ID): cv.positive_int,
|
|
vol.Exclusive(CONF_ONLY_INCLUDE_FEEDID, ONLY_INCL_EXCL_NONE): vol.All(
|
|
cv.ensure_list, [cv.positive_int]
|
|
),
|
|
vol.Exclusive(CONF_EXCLUDE_FEEDID, ONLY_INCL_EXCL_NONE): vol.All(
|
|
cv.ensure_list, [cv.positive_int]
|
|
),
|
|
vol.Optional(CONF_SENSOR_NAMES): vol.All(
|
|
{cv.positive_int: vol.All(cv.string, vol.Length(min=1))}
|
|
),
|
|
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
|
|
vol.Optional(CONF_UNIT_OF_MEASUREMENT, default=DEFAULT_UNIT): cv.string,
|
|
}
|
|
)
|
|
|
|
|
|
def get_id(
|
|
sensorid: str, feedtag: str, feedname: str, feedid: str, feeduserid: str
|
|
) -> str:
|
|
"""Return unique identifier for feed / sensor."""
|
|
return f"emoncms{sensorid}_{feedtag}_{feedname}_{feedid}_{feeduserid}"
|
|
|
|
|
|
def setup_platform(
|
|
hass: HomeAssistant,
|
|
config: ConfigType,
|
|
add_entities: AddEntitiesCallback,
|
|
discovery_info: DiscoveryInfoType | None = None,
|
|
) -> None:
|
|
"""Set up the Emoncms sensor."""
|
|
apikey = config[CONF_API_KEY]
|
|
url = config[CONF_URL]
|
|
sensorid = config[CONF_ID]
|
|
value_template = config.get(CONF_VALUE_TEMPLATE)
|
|
config_unit = config.get(CONF_UNIT_OF_MEASUREMENT)
|
|
exclude_feeds = config.get(CONF_EXCLUDE_FEEDID)
|
|
include_only_feeds = config.get(CONF_ONLY_INCLUDE_FEEDID)
|
|
sensor_names = config.get(CONF_SENSOR_NAMES)
|
|
|
|
if value_template is not None:
|
|
value_template.hass = hass
|
|
|
|
data = EmonCmsData(hass, url, apikey)
|
|
|
|
data.update()
|
|
|
|
if data.data is None:
|
|
return
|
|
|
|
sensors = []
|
|
|
|
for elem in data.data:
|
|
if exclude_feeds is not None and int(elem["id"]) in exclude_feeds:
|
|
continue
|
|
|
|
if include_only_feeds is not None and int(elem["id"]) not in include_only_feeds:
|
|
continue
|
|
|
|
name = None
|
|
if sensor_names is not None:
|
|
name = sensor_names.get(int(elem["id"]), None)
|
|
|
|
if unit := elem.get("unit"):
|
|
unit_of_measurement = unit
|
|
else:
|
|
unit_of_measurement = config_unit
|
|
|
|
sensors.append(
|
|
EmonCmsSensor(
|
|
hass,
|
|
data,
|
|
name,
|
|
value_template,
|
|
unit_of_measurement,
|
|
str(sensorid),
|
|
elem,
|
|
)
|
|
)
|
|
add_entities(sensors)
|
|
|
|
|
|
class EmonCmsSensor(SensorEntity):
|
|
"""Implementation of an Emoncms sensor."""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
data: EmonCmsData,
|
|
name: str | None,
|
|
value_template: template.Template | None,
|
|
unit_of_measurement: str | None,
|
|
sensorid: str,
|
|
elem: dict[str, Any],
|
|
) -> None:
|
|
"""Initialize the sensor."""
|
|
if name is None:
|
|
# Suppress ID in sensor name if it's 1, since most people won't
|
|
# have more than one EmonCMS source and it's redundant to show the
|
|
# ID if there's only one.
|
|
id_for_name = "" if str(sensorid) == "1" else sensorid
|
|
# Use the feed name assigned in EmonCMS or fall back to the feed ID
|
|
feed_name = elem.get("name") or f"Feed {elem['id']}"
|
|
self._attr_name = f"EmonCMS{id_for_name} {feed_name}"
|
|
else:
|
|
self._attr_name = name
|
|
self._identifier = get_id(
|
|
sensorid, elem["tag"], elem["name"], elem["id"], elem["userid"]
|
|
)
|
|
self._hass = hass
|
|
self._data = data
|
|
self._value_template = value_template
|
|
self._attr_native_unit_of_measurement = unit_of_measurement
|
|
self._sensorid = sensorid
|
|
|
|
if unit_of_measurement in ("kWh", "Wh"):
|
|
self._attr_device_class = SensorDeviceClass.ENERGY
|
|
self._attr_state_class = SensorStateClass.TOTAL_INCREASING
|
|
elif unit_of_measurement == "W":
|
|
self._attr_device_class = SensorDeviceClass.POWER
|
|
self._attr_state_class = SensorStateClass.MEASUREMENT
|
|
elif unit_of_measurement == "V":
|
|
self._attr_device_class = SensorDeviceClass.VOLTAGE
|
|
self._attr_state_class = SensorStateClass.MEASUREMENT
|
|
elif unit_of_measurement == "A":
|
|
self._attr_device_class = SensorDeviceClass.CURRENT
|
|
self._attr_state_class = SensorStateClass.MEASUREMENT
|
|
elif unit_of_measurement == "VA":
|
|
self._attr_device_class = SensorDeviceClass.APPARENT_POWER
|
|
self._attr_state_class = SensorStateClass.MEASUREMENT
|
|
elif unit_of_measurement in ("°C", "°F", "K"):
|
|
self._attr_device_class = SensorDeviceClass.TEMPERATURE
|
|
self._attr_state_class = SensorStateClass.MEASUREMENT
|
|
elif unit_of_measurement == "Hz":
|
|
self._attr_device_class = SensorDeviceClass.FREQUENCY
|
|
self._attr_state_class = SensorStateClass.MEASUREMENT
|
|
elif unit_of_measurement == "hPa":
|
|
self._attr_device_class = SensorDeviceClass.PRESSURE
|
|
self._attr_state_class = SensorStateClass.MEASUREMENT
|
|
self._update_attributes(elem)
|
|
|
|
def _update_attributes(self, elem: dict[str, Any]) -> None:
|
|
"""Update entity attributes."""
|
|
self._attr_extra_state_attributes = {
|
|
ATTR_FEEDID: elem["id"],
|
|
ATTR_TAG: elem["tag"],
|
|
ATTR_FEEDNAME: elem["name"],
|
|
}
|
|
if elem["value"] is not None:
|
|
self._attr_extra_state_attributes[ATTR_SIZE] = elem["size"]
|
|
self._attr_extra_state_attributes[ATTR_USERID] = elem["userid"]
|
|
self._attr_extra_state_attributes[ATTR_LASTUPDATETIME] = elem["time"]
|
|
self._attr_extra_state_attributes[ATTR_LASTUPDATETIMESTR] = (
|
|
template.timestamp_local(float(elem["time"]))
|
|
)
|
|
|
|
self._attr_native_value = None
|
|
if self._value_template is not None:
|
|
self._attr_native_value = (
|
|
self._value_template.render_with_possible_json_value(
|
|
elem["value"], STATE_UNKNOWN
|
|
)
|
|
)
|
|
elif elem["value"] is not None:
|
|
self._attr_native_value = round(float(elem["value"]), DECIMALS)
|
|
|
|
def update(self) -> None:
|
|
"""Get the latest data and updates the state."""
|
|
self._data.update()
|
|
|
|
if self._data.data is None:
|
|
return
|
|
|
|
elem = next(
|
|
(
|
|
elem
|
|
for elem in self._data.data
|
|
if get_id(
|
|
self._sensorid,
|
|
elem["tag"],
|
|
elem["name"],
|
|
elem["id"],
|
|
elem["userid"],
|
|
)
|
|
== self._identifier
|
|
),
|
|
None,
|
|
)
|
|
|
|
if elem is None:
|
|
return
|
|
|
|
self._update_attributes(elem)
|
|
|
|
|
|
class EmonCmsData:
|
|
"""The class for handling the data retrieval."""
|
|
|
|
def __init__(self, hass: HomeAssistant, url: str, apikey: str) -> None:
|
|
"""Initialize the data object."""
|
|
self._apikey = apikey
|
|
self._url = f"{url}/feed/list.json"
|
|
self._hass = hass
|
|
self.data: list[dict[str, Any]] | None = None
|
|
|
|
@Throttle(MIN_TIME_BETWEEN_UPDATES)
|
|
def update(self) -> None:
|
|
"""Get the latest data from Emoncms."""
|
|
try:
|
|
parameters = {"apikey": self._apikey}
|
|
req = requests.get(
|
|
self._url, params=parameters, allow_redirects=True, timeout=5
|
|
)
|
|
except requests.exceptions.RequestException as exception:
|
|
_LOGGER.error(exception)
|
|
return
|
|
|
|
if req.status_code == HTTPStatus.OK:
|
|
self.data = req.json()
|
|
else:
|
|
_LOGGER.error(
|
|
(
|
|
"Please verify if the specified configuration value "
|
|
"'%s' is correct! (HTTP Status_code = %d)"
|
|
),
|
|
CONF_URL,
|
|
req.status_code,
|
|
)
|