563 lines
19 KiB
Python
563 lines
19 KiB
Python
"""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
|