Initial commit
This commit is contained in:
gerviba committed
commit
e1e03ba73f
38 files changed
+2968
No files matched your search
@@ -0,0 +1,293 @@
|
||||
"""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
|
||||
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 . import HaProbeConfigEntry
|
||||
from .const import DOMAIN, signal_entities
|
||||
|
||||
_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
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
sensor = HaProbeSensor(
|
||||
entry_id=entry.entry_id,
|
||||
entity_key=entity_id,
|
||||
device_ha_id=device_ha_id or "probe",
|
||||
device_name=device_name,
|
||||
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:
|
||||
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_ha_id: str,
|
||||
device_name: str,
|
||||
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._attr_native_unit_of_measurement = unit
|
||||
self._attr_native_value = initial_state
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, f"{entry_id}_{device_ha_id}")},
|
||||
name=device_name,
|
||||
manufacturer="gerviba/systemmonitor",
|
||||
model="Probe",
|
||||
)
|
||||
|
||||
@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.
|
||||
"""
|
||||
self._attr_native_value = state
|
||||
|
||||
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._attr_native_unit_of_measurement = _resolve_unit(unit_name)
|
||||
|
||||
if self.hass is not None:
|
||||
self.async_write_ha_state()
|
||||
Reference in new issue
Block a user