"""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()