364 lines
12 KiB
Python
364 lines
12 KiB
Python
"""Sensor platform for the HA Probe integration.
|
|
|
|
Entities are created and updated dynamically from whatever the probe sends -
|
|
there is no static sensor list here. The probe only sends full metadata
|
|
(name/deviceClass/unit/device) once per entity, at startup; every later push
|
|
is just {"id": ..., "value"/"textValue": ...}, so an already-known entity is
|
|
updated in place and never re-created.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from typing import Any
|
|
|
|
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
|
|
from homeassistant.const import PERCENTAGE
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers.device_registry import DeviceInfo
|
|
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
|
from homeassistant.util import dt as dt_util
|
|
|
|
from . import HaProbeConfigEntry
|
|
from .const import signal_entities
|
|
from .device import build_device_info
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
# hu.gerviba.probe.util.UnitOfMeasurement, by Kotlin constant name. kotlinx.serialization
|
|
# encodes enums by their declared constant name (e.g. "MEGABYTES"), not the overridden
|
|
# toString() display symbol (e.g. "MB") - verified against the probe's actual webhook
|
|
# payload, so this is a name -> unit-symbol table, not the symbols themselves.
|
|
UNIT_MAP: dict[str, str] = {
|
|
"VOLT_AMPERE": "VA",
|
|
"WATT": "W",
|
|
"KILO_WATT": "kW",
|
|
"BTU_PER_HOUR": "BTU/h",
|
|
"VOLT_AMPERE_REACTIVE": "var",
|
|
"GIGA_JOULE": "GJ",
|
|
"KILO_WATT_HOUR": "kWh",
|
|
"MEGA_JOULE": "MJ",
|
|
"MEGA_WATT_HOUR": "MWh",
|
|
"WATT_HOUR": "Wh",
|
|
"MILLIAMPERE": "mA",
|
|
"AMPERE": "A",
|
|
"MILLIVOLT": "mV",
|
|
"VOLT": "V",
|
|
"CELSIUS": "°C",
|
|
"FAHRENHEIT": "°F",
|
|
"KELVIN": "K",
|
|
"MICROSECONDS": "μs",
|
|
"MILLISECONDS": "ms",
|
|
"SECONDS": "s",
|
|
"MINUTES": "min",
|
|
"HOURS": "h",
|
|
"DAYS": "d",
|
|
"WEEKS": "w",
|
|
"MONTHS": "m",
|
|
"YEARS": "y",
|
|
"MILLIMETERS": "mm",
|
|
"CENTIMETERS": "cm",
|
|
"METERS": "m",
|
|
"KILOMETERS": "km",
|
|
"INCHES": "in",
|
|
"FEET": "ft",
|
|
"YARDS": "yd",
|
|
"MILES": "mi",
|
|
"HERTZ": "Hz",
|
|
"KILOHERTZ": "kHz",
|
|
"MEGAHERTZ": "MHz",
|
|
"GIGAHERTZ": "GHz",
|
|
"PA": "Pa",
|
|
"HPA": "hPa",
|
|
"KPA": "kPa",
|
|
"BAR": "bar",
|
|
"CBAR": "cbar",
|
|
"MBAR": "mbar",
|
|
"MMHG": "mmHg",
|
|
"INHG": "inHg",
|
|
"PSI": "psi",
|
|
"DECIBEL": "dB",
|
|
"WEIGHTED_DECIBEL_A": "dBA",
|
|
"CUBIC_FEET": "ft³",
|
|
"CENTUM_CUBIC_FEET": "CCF",
|
|
"CUBIC_METERS": "m³",
|
|
"LITERS": "L",
|
|
"MILLILITERS": "mL",
|
|
"GALLONS": "gal",
|
|
"FLUID_OUNCES": "fl. oz.",
|
|
"CUBIC_METERS_PER_HOUR": "m³/h",
|
|
"CUBIC_FEET_PER_MINUTE": "ft³/m",
|
|
"GRAMS": "g",
|
|
"KILOGRAMS": "kg",
|
|
"MILLIGRAMS": "mg",
|
|
"MICROGRAMS": "µg",
|
|
"OUNCES": "oz",
|
|
"POUNDS": "lb",
|
|
"STONES": "st",
|
|
"WATTS_PER_SQUARE_METER": "W/m²",
|
|
"BTUS_PER_HOUR_SQUARE_FOOT": "BTU/(h⋅ft²)",
|
|
"INCHES_PER_DAY": "in/d",
|
|
"INCHES_PER_HOUR": "in/h",
|
|
"MILLIMETERS_PER_DAY": "mm/d",
|
|
"MILLIMETERS_PER_HOUR": "mm/h",
|
|
"PRECIPITATION_INCHES": "in",
|
|
"PRECIPITATION_MILLIMETERS": "mm",
|
|
"PRECIPITATION_CENTIMETERS": "cm",
|
|
"FEET_PER_SECOND": "ft/s",
|
|
"METERS_PER_SECOND": "m/s",
|
|
"KILOMETERS_PER_HOUR": "km/h",
|
|
"KNOTS": "kn",
|
|
"MILES_PER_HOUR": "mph",
|
|
"BITS": "bit",
|
|
"KILOBITS": "kbit",
|
|
"MEGABITS": "Mbit",
|
|
"GIGABITS": "Gbit",
|
|
"BYTES": "B",
|
|
"KILOBYTES": "kB",
|
|
"MEGABYTES": "MB",
|
|
"GIGABYTES": "GB",
|
|
"TERABYTES": "TB",
|
|
"PETABYTES": "PB",
|
|
"EXABYTES": "EB",
|
|
"ZETTABYTES": "ZB",
|
|
"YOTTABYTES": "YB",
|
|
"KIBIBYTES": "KiB",
|
|
"MEBIBYTES": "MiB",
|
|
"GIBIBYTES": "GiB",
|
|
"TEBIBYTES": "TiB",
|
|
"PEBIBYTES": "PiB",
|
|
"EXBIBYTES": "EiB",
|
|
"ZEBIBYTES": "ZiB",
|
|
"YOBIBYTES": "YiB",
|
|
"BITS_PER_SECOND": "bit/s",
|
|
"KILOBITS_PER_SECOND": "kbit/s",
|
|
"MEGABITS_PER_SECOND": "Mbit/s",
|
|
"GIGABITS_PER_SECOND": "Gbit/s",
|
|
"BYTES_PER_SECOND": "B/s",
|
|
"KILOBYTES_PER_SECOND": "kB/s",
|
|
"MEGABYTES_PER_SECOND": "MB/s",
|
|
"GIGABYTES_PER_SECOND": "GB/s",
|
|
"KIBIBYTES_PER_SECOND": "KiB/s",
|
|
"MEBIBYTES_PER_SECOND": "MiB/s",
|
|
"GIBIBYTES_PER_SECOND": "GiB/s",
|
|
"DEGREE": "°",
|
|
"CURRENCY_EURO": "€",
|
|
"CURRENCY_DOLLAR": "$",
|
|
"CURRENCY_CENT": "¢",
|
|
"PERCENTAGE": PERCENTAGE,
|
|
"AREA_SQUARE_METERS": "m²",
|
|
"CONDUCTIVITY": "µS/cm",
|
|
"LIGHT_LUX": "lx",
|
|
"UV_INDEX": "UV index",
|
|
"REVOLUTIONS_PER_MINUTE": "rpm",
|
|
"SIGNAL_STRENGTH_DECIBELS": "dB",
|
|
"SIGNAL_STRENGTH_DECIBELS_MILLIWATT": "dBm",
|
|
"CONCENTRATION_MICROGRAMS_PER_CUBIC_METER": "µg/m³",
|
|
"CONCENTRATION_PARTS_PER_MILLION": "ppm",
|
|
"CONCENTRATION_PARTS_PER_BILLION": "ppb",
|
|
}
|
|
|
|
|
|
def _resolve_unit(unit_name: str | None) -> str | None:
|
|
"""Map a probe UnitOfMeasurement constant name to its HA unit symbol."""
|
|
if unit_name is None:
|
|
return None
|
|
unit = UNIT_MAP.get(unit_name)
|
|
if unit is None:
|
|
_LOGGER.warning("Unknown HA Probe unit '%s'", unit_name)
|
|
return unit
|
|
|
|
|
|
def _resolve_device_class(device_class_name: str | None) -> SensorDeviceClass | None:
|
|
"""Map a probe SensorDeviceClass constant name to HA's enum member."""
|
|
if device_class_name is None:
|
|
return None
|
|
if device_class_name == "ENUM":
|
|
# HA's enum device class demands a fixed `options` list up front, which
|
|
# the probe can't know (container states, health values, ...). Without
|
|
# it the state write raises and the entity shows "unknown", so expose
|
|
# these as plain text sensors instead.
|
|
return None
|
|
device_class = SensorDeviceClass.__members__.get(device_class_name)
|
|
if device_class is None:
|
|
_LOGGER.warning("Unknown HA Probe device class '%s'", device_class_name)
|
|
return device_class
|
|
|
|
|
|
# Python's datetime parsing only copes with up to 6 fractional digits, but the
|
|
# probe sends nanosecond-precision timestamps (e.g. Docker's StartedAt).
|
|
_EXTRA_FRACTION = re.compile(r"(\.\d{6})\d+")
|
|
|
|
|
|
def _coerce_state(device_class: SensorDeviceClass | None, state: Any) -> Any:
|
|
"""Turn the probe's text timestamps into the datetime HA requires."""
|
|
if device_class != SensorDeviceClass.TIMESTAMP or not isinstance(state, str):
|
|
return state
|
|
parsed = dt_util.parse_datetime(_EXTRA_FRACTION.sub(r"\1", state))
|
|
if parsed is None:
|
|
_LOGGER.warning("Unparseable HA Probe timestamp '%s'", state)
|
|
return None
|
|
return dt_util.as_utc(parsed)
|
|
|
|
|
|
def _display_precision(
|
|
device_class: SensorDeviceClass | None, unit: str | None
|
|
) -> int | None:
|
|
"""Decimals to show: 0 for percentages, at most 2 for sizes, else HA's default."""
|
|
if unit == PERCENTAGE:
|
|
return 0
|
|
if device_class in (SensorDeviceClass.DATA_SIZE, SensorDeviceClass.VOLUME_STORAGE):
|
|
return 2
|
|
return None
|
|
|
|
|
|
# Seconds per unit, largest first, for picking a readable duration unit.
|
|
_DURATION_STEPS: tuple[tuple[str, float], ...] = (("d", 86400), ("h", 3600), ("min", 60))
|
|
|
|
|
|
def _scale_duration(
|
|
device_class: SensorDeviceClass | None, unit: str | None, state: Any
|
|
) -> tuple[Any, str | None]:
|
|
"""Show second-based durations in the largest unit that keeps them >= 1."""
|
|
if (
|
|
device_class != SensorDeviceClass.DURATION
|
|
or unit != "s"
|
|
or isinstance(state, bool)
|
|
or not isinstance(state, (int, float))
|
|
):
|
|
return state, unit
|
|
for symbol, seconds in _DURATION_STEPS:
|
|
if abs(state) >= seconds:
|
|
return round(state / seconds, 2), symbol
|
|
return state, unit
|
|
|
|
|
|
def _fallback_name(entity_id: str) -> str:
|
|
"""Make a readable name out of an id when no schema metadata is known yet."""
|
|
return entity_id.rsplit("__", 1)[-1].replace("_", " ").title()
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: HaProbeConfigEntry,
|
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
|
) -> None:
|
|
"""Set up HA Probe sensors for a config entry."""
|
|
runtime_data = entry.runtime_data
|
|
|
|
@callback
|
|
def handle_entities(entities: list[dict[str, Any]]) -> None:
|
|
new_sensors: list[HaProbeSensor] = []
|
|
|
|
for entity in entities:
|
|
entity_id = entity.get("id")
|
|
if not entity_id:
|
|
continue
|
|
|
|
text_value = entity.get("textValue")
|
|
state = text_value if text_value is not None else entity.get("value")
|
|
|
|
existing = runtime_data.entities.get(entity_id)
|
|
if existing is not None:
|
|
existing.handle_update(entity, state)
|
|
continue
|
|
|
|
device_ha_id = entity.get("device")
|
|
device_name = (
|
|
runtime_data.devices.get(device_ha_id, device_ha_id)
|
|
if device_ha_id
|
|
else entry.title
|
|
)
|
|
|
|
resolved_ha_id = device_ha_id or "probe"
|
|
device_info = build_device_info(
|
|
entry.entry_id,
|
|
resolved_ha_id,
|
|
device_name,
|
|
runtime_data.device_meta.get(resolved_ha_id, {}),
|
|
)
|
|
|
|
sensor = HaProbeSensor(
|
|
entry_id=entry.entry_id,
|
|
entity_key=entity_id,
|
|
device_info=device_info,
|
|
initial_state=state,
|
|
name=entity.get("name") or _fallback_name(entity_id),
|
|
device_class=_resolve_device_class(entity.get("deviceClass")),
|
|
unit=_resolve_unit(entity.get("unit")),
|
|
)
|
|
runtime_data.entities[entity_id] = sensor
|
|
new_sensors.append(sensor)
|
|
|
|
if new_sensors:
|
|
# Register parent devices before the container devices nested under them.
|
|
new_sensors.sort(key=lambda sensor: "via_device" in sensor.device_info)
|
|
async_add_entities(new_sensors)
|
|
|
|
entry.async_on_unload(
|
|
async_dispatcher_connect(hass, signal_entities(entry.entry_id), handle_entities)
|
|
)
|
|
|
|
|
|
class HaProbeSensor(SensorEntity):
|
|
"""A sensor created dynamically from a probe webhook push."""
|
|
|
|
_attr_has_entity_name = True
|
|
_attr_should_poll = False
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
entry_id: str,
|
|
entity_key: str,
|
|
device_info: DeviceInfo,
|
|
initial_state: Any,
|
|
name: str,
|
|
device_class: SensorDeviceClass | None,
|
|
unit: str | None,
|
|
) -> None:
|
|
"""Initialize the sensor from its first-seen webhook entity."""
|
|
self._attr_unique_id = f"{entry_id}_{entity_key}"
|
|
self._attr_name = name
|
|
self._attr_device_class = device_class
|
|
self._probe_unit = unit
|
|
self._apply_state(initial_state)
|
|
self._attr_device_info = device_info
|
|
|
|
def _apply_state(self, state: Any) -> None:
|
|
"""Set value, unit and precision from a raw probe state."""
|
|
value = _coerce_state(self._attr_device_class, state)
|
|
value, unit = _scale_duration(self._attr_device_class, self._probe_unit, value)
|
|
self._attr_native_value = value
|
|
self._attr_native_unit_of_measurement = unit
|
|
self._attr_suggested_display_precision = _display_precision(
|
|
self._attr_device_class, unit
|
|
)
|
|
|
|
@callback
|
|
def handle_update(self, entity: dict[str, Any], state: Any) -> None:
|
|
"""Apply a later webhook push for this already-known entity.
|
|
|
|
Usually this is just a bare value/textValue push, but if the probe
|
|
restarted it will resend full metadata too - pick that back up
|
|
instead of only ever trusting the first value we saw.
|
|
"""
|
|
name = entity.get("name")
|
|
if name:
|
|
self._attr_name = name
|
|
|
|
device_class_name = entity.get("deviceClass")
|
|
if device_class_name:
|
|
self._attr_device_class = _resolve_device_class(device_class_name)
|
|
|
|
unit_name = entity.get("unit")
|
|
if unit_name:
|
|
self._probe_unit = _resolve_unit(unit_name)
|
|
|
|
self._apply_state(state)
|
|
|
|
if self.hass is not None:
|
|
self.async_write_ha_state()
|