"""Support for 1-Wire environment sensors.""" from __future__ import annotations import asyncio import copy from dataclasses import dataclass import logging import os from types import MappingProxyType from typing import Any from pi1wire import InvalidCRCException, OneWireInterface, UnsupportResponseException from homeassistant.components.sensor import ( STATE_CLASS_MEASUREMENT, STATE_CLASS_TOTAL_INCREASING, SensorEntity, SensorEntityDescription, ) from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( ATTR_IDENTIFIERS, ATTR_MANUFACTURER, ATTR_MODEL, ATTR_NAME, CONF_TYPE, DEVICE_CLASS_CURRENT, DEVICE_CLASS_HUMIDITY, DEVICE_CLASS_ILLUMINANCE, DEVICE_CLASS_PRESSURE, DEVICE_CLASS_TEMPERATURE, DEVICE_CLASS_VOLTAGE, ELECTRIC_CURRENT_AMPERE, ELECTRIC_POTENTIAL_VOLT, LIGHT_LUX, PERCENTAGE, PRESSURE_MBAR, TEMP_CELSIUS, ) from homeassistant.core import HomeAssistant from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.typing import StateType from .const import ( CONF_MOUNT_DIR, CONF_NAMES, CONF_TYPE_OWSERVER, CONF_TYPE_SYSBUS, DOMAIN, PRESSURE_CBAR, READ_MODE_FLOAT, READ_MODE_INT, ) from .onewire_entities import ( OneWireBaseEntity, OneWireEntityDescription, OneWireProxyEntity, ) from .onewirehub import OneWireHub @dataclass class OneWireSensorEntityDescription(OneWireEntityDescription, SensorEntityDescription): """Class describing OneWire sensor entities.""" SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION = OneWireSensorEntityDescription( key="temperature", device_class=DEVICE_CLASS_TEMPERATURE, name="Temperature", native_unit_of_measurement=TEMP_CELSIUS, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ) _LOGGER = logging.getLogger(__name__) DEVICE_SENSORS: dict[str, tuple[OneWireSensorEntityDescription, ...]] = { "10": (SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION,), "12": ( OneWireSensorEntityDescription( key="TAI8570/temperature", device_class=DEVICE_CLASS_TEMPERATURE, entity_registry_enabled_default=False, name="Temperature", native_unit_of_measurement=TEMP_CELSIUS, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="TAI8570/pressure", device_class=DEVICE_CLASS_PRESSURE, entity_registry_enabled_default=False, name="Pressure", native_unit_of_measurement=PRESSURE_MBAR, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), ), "22": (SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION,), "26": ( SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION, OneWireSensorEntityDescription( key="humidity", device_class=DEVICE_CLASS_HUMIDITY, entity_registry_enabled_default=False, name="Humidity", native_unit_of_measurement=PERCENTAGE, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="HIH3600/humidity", device_class=DEVICE_CLASS_HUMIDITY, entity_registry_enabled_default=False, name="Humidity HIH3600", native_unit_of_measurement=PERCENTAGE, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="HIH4000/humidity", device_class=DEVICE_CLASS_HUMIDITY, entity_registry_enabled_default=False, name="Humidity HIH4000", native_unit_of_measurement=PERCENTAGE, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="HIH5030/humidity", device_class=DEVICE_CLASS_HUMIDITY, entity_registry_enabled_default=False, name="Humidity HIH5030", native_unit_of_measurement=PERCENTAGE, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="HTM1735/humidity", device_class=DEVICE_CLASS_HUMIDITY, entity_registry_enabled_default=False, name="Humidity HTM1735", native_unit_of_measurement=PERCENTAGE, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="B1-R1-A/pressure", device_class=DEVICE_CLASS_PRESSURE, entity_registry_enabled_default=False, name="Pressure", native_unit_of_measurement=PRESSURE_MBAR, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="S3-R1-A/illuminance", device_class=DEVICE_CLASS_ILLUMINANCE, entity_registry_enabled_default=False, name="Illuminance", native_unit_of_measurement=LIGHT_LUX, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="VAD", device_class=DEVICE_CLASS_VOLTAGE, entity_registry_enabled_default=False, name="Voltage VAD", native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="VDD", device_class=DEVICE_CLASS_VOLTAGE, entity_registry_enabled_default=False, name="Voltage VDD", native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="IAD", device_class=DEVICE_CLASS_CURRENT, entity_registry_enabled_default=False, name="Current", native_unit_of_measurement=ELECTRIC_CURRENT_AMPERE, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), ), "28": (SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION,), "3B": (SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION,), "42": (SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION,), "1D": ( OneWireSensorEntityDescription( key="counter.A", name="Counter A", native_unit_of_measurement="count", read_mode=READ_MODE_INT, state_class=STATE_CLASS_TOTAL_INCREASING, ), OneWireSensorEntityDescription( key="counter.B", name="Counter B", native_unit_of_measurement="count", read_mode=READ_MODE_INT, state_class=STATE_CLASS_TOTAL_INCREASING, ), ), "EF": (), # "HobbyBoard": special "7E": (), # "EDS": special } DEVICE_SUPPORT_SYSBUS = ["10", "22", "28", "3B", "42"] # EF sensors are usually hobbyboards specialized sensors. # These can only be read by OWFS. Currently this driver only supports them # via owserver (network protocol) HOBBYBOARD_EF: dict[str, tuple[OneWireSensorEntityDescription, ...]] = { "HobbyBoards_EF": ( OneWireSensorEntityDescription( key="humidity/humidity_corrected", device_class=DEVICE_CLASS_HUMIDITY, name="Humidity", native_unit_of_measurement=PERCENTAGE, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="humidity/humidity_raw", device_class=DEVICE_CLASS_HUMIDITY, name="Humidity Raw", native_unit_of_measurement=PERCENTAGE, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="humidity/temperature", device_class=DEVICE_CLASS_TEMPERATURE, name="Temperature", native_unit_of_measurement=TEMP_CELSIUS, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), ), "HB_MOISTURE_METER": ( OneWireSensorEntityDescription( key="moisture/sensor.0", device_class=DEVICE_CLASS_PRESSURE, name="Moisture 0", native_unit_of_measurement=PRESSURE_CBAR, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="moisture/sensor.1", device_class=DEVICE_CLASS_PRESSURE, name="Moisture 1", native_unit_of_measurement=PRESSURE_CBAR, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="moisture/sensor.2", device_class=DEVICE_CLASS_PRESSURE, name="Moisture 2", native_unit_of_measurement=PRESSURE_CBAR, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="moisture/sensor.3", device_class=DEVICE_CLASS_PRESSURE, name="Moisture 3", native_unit_of_measurement=PRESSURE_CBAR, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), ), } # 7E sensors are special sensors by Embedded Data Systems EDS_SENSORS: dict[str, tuple[OneWireSensorEntityDescription, ...]] = { "EDS0066": ( OneWireSensorEntityDescription( key="EDS0066/temperature", device_class=DEVICE_CLASS_TEMPERATURE, name="Temperature", native_unit_of_measurement=TEMP_CELSIUS, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="EDS0066/pressure", device_class=DEVICE_CLASS_PRESSURE, name="Pressure", native_unit_of_measurement=PRESSURE_MBAR, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), ), "EDS0068": ( OneWireSensorEntityDescription( key="EDS0068/temperature", device_class=DEVICE_CLASS_TEMPERATURE, name="Temperature", native_unit_of_measurement=TEMP_CELSIUS, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="EDS0068/pressure", device_class=DEVICE_CLASS_PRESSURE, name="Pressure", native_unit_of_measurement=PRESSURE_MBAR, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="EDS0068/light", device_class=DEVICE_CLASS_ILLUMINANCE, name="Illuminance", native_unit_of_measurement=LIGHT_LUX, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), OneWireSensorEntityDescription( key="EDS0068/humidity", device_class=DEVICE_CLASS_HUMIDITY, name="Humidity", native_unit_of_measurement=PERCENTAGE, read_mode=READ_MODE_FLOAT, state_class=STATE_CLASS_MEASUREMENT, ), ), } def get_sensor_types(device_sub_type: str) -> dict[str, Any]: """Return the proper info array for the device type.""" if "HobbyBoard" in device_sub_type: return HOBBYBOARD_EF if "EDS" in device_sub_type: return EDS_SENSORS return DEVICE_SENSORS async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: """Set up 1-Wire platform.""" onewirehub = hass.data[DOMAIN][config_entry.entry_id] entities = await hass.async_add_executor_job( get_entities, onewirehub, config_entry.data ) async_add_entities(entities, True) def get_entities( onewirehub: OneWireHub, config: MappingProxyType[str, Any] ) -> list[SensorEntity]: """Get a list of entities.""" if not onewirehub.devices: return [] entities: list[SensorEntity] = [] device_names = {} if CONF_NAMES in config and isinstance(config[CONF_NAMES], dict): device_names = config[CONF_NAMES] conf_type = config[CONF_TYPE] # We have an owserver on a remote(or local) host/port if conf_type == CONF_TYPE_OWSERVER: assert onewirehub.owproxy for device in onewirehub.devices: family = device["family"] device_type = device["type"] device_id = os.path.split(os.path.split(device["path"])[0])[1] device_sub_type = "std" device_path = device["path"] if "EF" in family: device_sub_type = "HobbyBoard" family = device_type elif "7E" in family: device_sub_type = "EDS" family = onewirehub.owproxy.read(f"{device_path}device_type").decode() if family not in get_sensor_types(device_sub_type): _LOGGER.warning( "Ignoring unknown family (%s) of sensor found for device: %s", family, device_id, ) continue device_info: DeviceInfo = { ATTR_IDENTIFIERS: {(DOMAIN, device_id)}, ATTR_MANUFACTURER: "Maxim Integrated", ATTR_MODEL: device_type, ATTR_NAME: device_id, } for description in get_sensor_types(device_sub_type)[family]: if description.key.startswith("moisture/"): s_id = description.key.split(".")[1] is_leaf = int( onewirehub.owproxy.read( f"{device_path}moisture/is_leaf.{s_id}" ).decode() ) if is_leaf: description = copy.deepcopy(description) description.device_class = DEVICE_CLASS_HUMIDITY description.native_unit_of_measurement = PERCENTAGE description.name = f"Wetness {s_id}" device_file = os.path.join( os.path.split(device["path"])[0], description.key ) name = f"{device_names.get(device_id, device_id)} {description.name}" entities.append( OneWireProxySensor( description=description, device_id=device_id, device_file=device_file, device_info=device_info, name=name, owproxy=onewirehub.owproxy, ) ) # We have a raw GPIO ow sensor on a Pi elif conf_type == CONF_TYPE_SYSBUS: base_dir = config[CONF_MOUNT_DIR] _LOGGER.debug("Initializing using SysBus %s", base_dir) for p1sensor in onewirehub.devices: family = p1sensor.mac_address[:2] device_id = f"{family}-{p1sensor.mac_address[2:]}" if family not in DEVICE_SUPPORT_SYSBUS: _LOGGER.warning( "Ignoring unknown family (%s) of sensor found for device: %s", family, device_id, ) continue device_info = { ATTR_IDENTIFIERS: {(DOMAIN, device_id)}, ATTR_MANUFACTURER: "Maxim Integrated", ATTR_MODEL: family, ATTR_NAME: device_id, } description = SIMPLE_TEMPERATURE_SENSOR_DESCRIPTION device_file = f"/sys/bus/w1/devices/{device_id}/w1_slave" name = f"{device_names.get(device_id, device_id)} {description.name}" entities.append( OneWireDirectSensor( description=description, device_id=device_id, device_file=device_file, device_info=device_info, name=name, owsensor=p1sensor, ) ) if not entities: _LOGGER.error( "No onewire sensor found. Check if dtoverlay=w1-gpio " "is in your /boot/config.txt. " "Check the mount_dir parameter if it's defined" ) return entities class OneWireSensor(OneWireBaseEntity, SensorEntity): """Mixin for sensor specific attributes.""" entity_description: OneWireSensorEntityDescription class OneWireProxySensor(OneWireProxyEntity, OneWireSensor): """Implementation of a 1-Wire sensor connected through owserver.""" entity_description: OneWireSensorEntityDescription @property def native_value(self) -> StateType: """Return the state of the entity.""" return self._state class OneWireDirectSensor(OneWireSensor): """Implementation of a 1-Wire sensor directly connected to RPI GPIO.""" def __init__( self, description: OneWireSensorEntityDescription, device_id: str, device_info: DeviceInfo, device_file: str, name: str, owsensor: OneWireInterface, ) -> None: """Initialize the sensor.""" super().__init__( description=description, device_id=device_id, device_info=device_info, device_file=device_file, name=name, ) self._attr_unique_id = device_file self._owsensor = owsensor @property def native_value(self) -> StateType: """Return the state of the entity.""" return self._state async def get_temperature(self) -> float: """Get the latest data from the device.""" attempts = 1 while True: try: return await self.hass.async_add_executor_job( self._owsensor.get_temperature ) except UnsupportResponseException as ex: _LOGGER.debug( "Cannot read from sensor %s (retry attempt %s): %s", self._device_file, attempts, ex, ) await asyncio.sleep(0.2) attempts += 1 if attempts > 10: raise async def async_update(self) -> None: """Get the latest data from the device.""" try: self._value_raw = await self.get_temperature() self._state = round(self._value_raw, 1) except ( FileNotFoundError, InvalidCRCException, UnsupportResponseException, ) as ex: _LOGGER.warning( "Cannot read from sensor %s: %s", self._device_file, ex, ) self._state = None