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