345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""Support for a ScreenLogic Sensor."""
|
|
from collections.abc import Callable
|
|
from copy import copy
|
|
from dataclasses import dataclass
|
|
import logging
|
|
|
|
from screenlogicpy.const.data import ATTR, DEVICE, GROUP, VALUE
|
|
from screenlogicpy.const.msg import CODE
|
|
from screenlogicpy.device_const.chemistry import DOSE_STATE
|
|
from screenlogicpy.device_const.pump import PUMP_TYPE
|
|
from screenlogicpy.device_const.system import EQUIPMENT_FLAG
|
|
|
|
from homeassistant.components.sensor import (
|
|
DOMAIN,
|
|
SensorDeviceClass,
|
|
SensorEntity,
|
|
SensorEntityDescription,
|
|
SensorStateClass,
|
|
)
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import EntityCategory
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
|
|
from .const import DOMAIN as SL_DOMAIN
|
|
from .coordinator import ScreenlogicDataUpdateCoordinator
|
|
from .entity import (
|
|
ScreenlogicEntity,
|
|
ScreenLogicEntityDescription,
|
|
ScreenLogicPushEntity,
|
|
ScreenLogicPushEntityDescription,
|
|
)
|
|
from .util import cleanup_excluded_entity, get_ha_unit
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ScreenLogicSensorMixin:
|
|
"""Mixin for SecreenLogic sensor entity."""
|
|
|
|
value_mod: Callable[[int | str], int | str] | None = None
|
|
|
|
|
|
@dataclass
|
|
class ScreenLogicSensorDescription(
|
|
ScreenLogicSensorMixin, SensorEntityDescription, ScreenLogicEntityDescription
|
|
):
|
|
"""Describes a ScreenLogic sensor."""
|
|
|
|
|
|
@dataclass
|
|
class ScreenLogicPushSensorDescription(
|
|
ScreenLogicSensorDescription, ScreenLogicPushEntityDescription
|
|
):
|
|
"""Describes a ScreenLogic push sensor."""
|
|
|
|
|
|
SUPPORTED_CORE_SENSORS = [
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.STATUS_CHANGED,
|
|
data_root=(DEVICE.CONTROLLER, GROUP.SENSOR),
|
|
key=VALUE.AIR_TEMPERATURE,
|
|
device_class=SensorDeviceClass.TEMPERATURE,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
),
|
|
]
|
|
|
|
SUPPORTED_PUMP_SENSORS = [
|
|
ScreenLogicSensorDescription(
|
|
data_root=(DEVICE.PUMP,),
|
|
key=VALUE.WATTS_NOW,
|
|
device_class=SensorDeviceClass.POWER,
|
|
),
|
|
ScreenLogicSensorDescription(
|
|
data_root=(DEVICE.PUMP,),
|
|
key=VALUE.GPM_NOW,
|
|
enabled_lambda=lambda type: type != PUMP_TYPE.INTELLIFLO_VS,
|
|
),
|
|
ScreenLogicSensorDescription(
|
|
data_root=(DEVICE.PUMP,),
|
|
key=VALUE.RPM_NOW,
|
|
enabled_lambda=lambda type: type != PUMP_TYPE.INTELLIFLO_VF,
|
|
),
|
|
]
|
|
|
|
SUPPORTED_INTELLICHEM_SENSORS = [
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.STATUS_CHANGED,
|
|
data_root=(DEVICE.CONTROLLER, GROUP.SENSOR),
|
|
key=VALUE.ORP,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.STATUS_CHANGED,
|
|
data_root=(DEVICE.CONTROLLER, GROUP.SENSOR),
|
|
key=VALUE.PH,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.SENSOR),
|
|
key=VALUE.ORP_NOW,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.SENSOR),
|
|
key=VALUE.PH_NOW,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.SENSOR),
|
|
key=VALUE.ORP_SUPPLY_LEVEL,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
value_mod=lambda val: int(val) - 1,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.SENSOR),
|
|
key=VALUE.PH_SUPPLY_LEVEL,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
value_mod=lambda val: int(val) - 1,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.SENSOR),
|
|
key=VALUE.PH_PROBE_WATER_TEMP,
|
|
device_class=SensorDeviceClass.TEMPERATURE,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.SENSOR),
|
|
key=VALUE.SATURATION,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.CONFIGURATION),
|
|
key=VALUE.CALCIUM_HARNESS,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.CONFIGURATION),
|
|
key=VALUE.CYA,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.CONFIGURATION),
|
|
key=VALUE.ORP_SETPOINT,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.CONFIGURATION),
|
|
key=VALUE.PH_SETPOINT,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.CONFIGURATION),
|
|
key=VALUE.TOTAL_ALKALINITY,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.CONFIGURATION),
|
|
key=VALUE.SALT_TDS_PPM,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.DOSE_STATUS),
|
|
key=VALUE.ORP_DOSING_STATE,
|
|
device_class=SensorDeviceClass.ENUM,
|
|
options=["Dosing", "Mixing", "Monitoring"],
|
|
value_mod=lambda val: DOSE_STATE(val).title,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.DOSE_STATUS),
|
|
key=VALUE.ORP_LAST_DOSE_TIME,
|
|
device_class=SensorDeviceClass.DURATION,
|
|
state_class=SensorStateClass.TOTAL_INCREASING,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.DOSE_STATUS),
|
|
key=VALUE.ORP_LAST_DOSE_VOLUME,
|
|
device_class=SensorDeviceClass.VOLUME,
|
|
state_class=SensorStateClass.TOTAL_INCREASING,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.DOSE_STATUS),
|
|
key=VALUE.PH_DOSING_STATE,
|
|
device_class=SensorDeviceClass.ENUM,
|
|
options=["Dosing", "Mixing", "Monitoring"],
|
|
value_mod=lambda val: DOSE_STATE(val).title,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.DOSE_STATUS),
|
|
key=VALUE.PH_LAST_DOSE_TIME,
|
|
device_class=SensorDeviceClass.DURATION,
|
|
state_class=SensorStateClass.TOTAL_INCREASING,
|
|
),
|
|
ScreenLogicPushSensorDescription(
|
|
subscription_code=CODE.CHEMISTRY_CHANGED,
|
|
data_root=(DEVICE.INTELLICHEM, GROUP.DOSE_STATUS),
|
|
key=VALUE.PH_LAST_DOSE_VOLUME,
|
|
device_class=SensorDeviceClass.VOLUME,
|
|
state_class=SensorStateClass.TOTAL_INCREASING,
|
|
),
|
|
]
|
|
|
|
SUPPORTED_SCG_SENSORS = [
|
|
ScreenLogicSensorDescription(
|
|
data_root=(DEVICE.SCG, GROUP.SENSOR),
|
|
key=VALUE.SALT_PPM,
|
|
state_class=SensorStateClass.MEASUREMENT,
|
|
),
|
|
ScreenLogicSensorDescription(
|
|
data_root=(DEVICE.SCG, GROUP.CONFIGURATION),
|
|
key=VALUE.SUPER_CHLOR_TIMER,
|
|
),
|
|
]
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
config_entry: ConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up entry."""
|
|
entities: list[ScreenLogicSensor] = []
|
|
coordinator: ScreenlogicDataUpdateCoordinator = hass.data[SL_DOMAIN][
|
|
config_entry.entry_id
|
|
]
|
|
gateway = coordinator.gateway
|
|
|
|
for core_sensor_description in SUPPORTED_CORE_SENSORS:
|
|
if (
|
|
gateway.get_data(
|
|
*core_sensor_description.data_root, core_sensor_description.key
|
|
)
|
|
is not None
|
|
):
|
|
entities.append(ScreenLogicPushSensor(coordinator, core_sensor_description))
|
|
|
|
for pump_index, pump_data in gateway.get_data(DEVICE.PUMP).items():
|
|
if not pump_data or not pump_data.get(VALUE.DATA):
|
|
continue
|
|
pump_type = pump_data[VALUE.TYPE]
|
|
for proto_pump_sensor_description in SUPPORTED_PUMP_SENSORS:
|
|
if not pump_data.get(proto_pump_sensor_description.key):
|
|
continue
|
|
entities.append(
|
|
ScreenLogicPumpSensor(
|
|
coordinator,
|
|
copy(proto_pump_sensor_description),
|
|
pump_index,
|
|
pump_type,
|
|
)
|
|
)
|
|
|
|
chem_sensor_description: ScreenLogicPushSensorDescription
|
|
for chem_sensor_description in SUPPORTED_INTELLICHEM_SENSORS:
|
|
chem_sensor_data_path = (
|
|
*chem_sensor_description.data_root,
|
|
chem_sensor_description.key,
|
|
)
|
|
if EQUIPMENT_FLAG.INTELLICHEM not in gateway.equipment_flags:
|
|
cleanup_excluded_entity(coordinator, DOMAIN, chem_sensor_data_path)
|
|
continue
|
|
if gateway.get_data(*chem_sensor_data_path):
|
|
chem_sensor_description.entity_category = EntityCategory.DIAGNOSTIC
|
|
entities.append(ScreenLogicPushSensor(coordinator, chem_sensor_description))
|
|
|
|
scg_sensor_description: ScreenLogicSensorDescription
|
|
for scg_sensor_description in SUPPORTED_SCG_SENSORS:
|
|
scg_sensor_data_path = (
|
|
*scg_sensor_description.data_root,
|
|
scg_sensor_description.key,
|
|
)
|
|
if EQUIPMENT_FLAG.CHLORINATOR not in gateway.equipment_flags:
|
|
cleanup_excluded_entity(coordinator, DOMAIN, scg_sensor_data_path)
|
|
continue
|
|
if gateway.get_data(*scg_sensor_data_path):
|
|
scg_sensor_description.entity_category = EntityCategory.DIAGNOSTIC
|
|
entities.append(ScreenLogicSensor(coordinator, scg_sensor_description))
|
|
|
|
async_add_entities(entities)
|
|
|
|
|
|
class ScreenLogicSensor(ScreenlogicEntity, SensorEntity):
|
|
"""Representation of a ScreenLogic sensor entity."""
|
|
|
|
entity_description: ScreenLogicSensorDescription
|
|
_attr_has_entity_name = True
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: ScreenlogicDataUpdateCoordinator,
|
|
entity_description: ScreenLogicSensorDescription,
|
|
) -> None:
|
|
"""Initialize of the entity."""
|
|
super().__init__(coordinator, entity_description)
|
|
self._attr_native_unit_of_measurement = get_ha_unit(
|
|
self.entity_data.get(ATTR.UNIT)
|
|
)
|
|
|
|
@property
|
|
def native_value(self) -> str | int | float:
|
|
"""State of the sensor."""
|
|
val = self.entity_data[ATTR.VALUE]
|
|
value_mod = self.entity_description.value_mod
|
|
return value_mod(val) if value_mod else val
|
|
|
|
|
|
class ScreenLogicPushSensor(ScreenLogicSensor, ScreenLogicPushEntity):
|
|
"""Representation of a ScreenLogic push sensor entity."""
|
|
|
|
entity_description: ScreenLogicPushSensorDescription
|
|
|
|
|
|
class ScreenLogicPumpSensor(ScreenLogicSensor):
|
|
"""Representation of a ScreenLogic pump sensor."""
|
|
|
|
_attr_entity_category = EntityCategory.DIAGNOSTIC
|
|
_attr_state_class = SensorStateClass.MEASUREMENT
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: ScreenlogicDataUpdateCoordinator,
|
|
entity_description: ScreenLogicSensorDescription,
|
|
pump_index: int,
|
|
pump_type: int,
|
|
) -> None:
|
|
"""Initialize of the entity."""
|
|
entity_description.data_root = (DEVICE.PUMP, pump_index)
|
|
super().__init__(coordinator, entity_description)
|
|
if entity_description.enabled_lambda:
|
|
self._attr_entity_registry_enabled_default = (
|
|
entity_description.enabled_lambda(pump_type)
|
|
)
|