evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Support local push updates for most ScreenLogic entities (#87438)

Browse source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Kevin Worrel 2023-02-06 18:13:36 -08:00 • committed by GitHub
parent 4fbb14ecc7
commit 687d326bb4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 352 additions and 252 deletions
Download patch file Download diff file Expand all files Collapse all files

101
homeassistant/components/screenlogic/sensor.py
View file

@ -1,9 +1,13 @@
"""Support for a ScreenLogic Sensor."""
from typing import Any
from screenlogicpy.const import (
CHEM_DOSING_STATE,
CODE,
DATA as SL_DATA,
DEVICE_TYPE,
EQUIPMENT,
STATE_TYPE,
UNIT,
)
@ -26,8 +30,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ScreenlogicEntity
from . import ScreenlogicDataUpdateCoordinator
from .const import DOMAIN
from .entity import ScreenlogicEntity, ScreenLogicPushEntity
SUPPORTED_BASIC_SENSORS = (
"air_temperature",
@ -68,12 +73,18 @@ SUPPORTED_PUMP_SENSORS = ("currentWatts", "currentRPM", "currentGPM")
SL_DEVICE_TYPE_TO_HA_DEVICE_CLASS = {
DEVICE_TYPE.DURATION: SensorDeviceClass.DURATION,
DEVICE_TYPE.ENUM: SensorDeviceClass.ENUM,
DEVICE_TYPE.ENERGY: SensorDeviceClass.POWER,
DEVICE_TYPE.POWER: SensorDeviceClass.POWER,
DEVICE_TYPE.TEMPERATURE: SensorDeviceClass.TEMPERATURE,
DEVICE_TYPE.VOLUME: SensorDeviceClass.VOLUME,
}
SL_STATE_TYPE_TO_HA_STATE_CLASS = {
STATE_TYPE.MEASUREMENT: SensorStateClass.MEASUREMENT,
STATE_TYPE.TOTAL_INCREASING: SensorStateClass.TOTAL_INCREASING,
}
SL_UNIT_TO_HA_UNIT = {
UNIT.CELSIUS: UnitOfTemperature.CELSIUS,
UNIT.FAHRENHEIT: UnitOfTemperature.FAHRENHEIT,
@ -93,14 +104,18 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up entry."""
entities = []
coordinator = hass.data[DOMAIN][config_entry.entry_id]
equipment_flags = coordinator.data[SL_DATA.KEY_CONFIG]["equipment_flags"]
entities: list[ScreenLogicSensorEntity] = []
coordinator: ScreenlogicDataUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
]
equipment_flags = coordinator.gateway_data[SL_DATA.KEY_CONFIG]["equipment_flags"]
# Generic sensors
for sensor_name in coordinator.data[SL_DATA.KEY_SENSORS]:
# Generic push sensors
for sensor_name in coordinator.gateway_data[SL_DATA.KEY_SENSORS]:
if sensor_name in SUPPORTED_BASIC_SENSORS:
entities.append(ScreenLogicSensor(coordinator, sensor_name))
entities.append(
ScreenLogicStatusSensor(coordinator, sensor_name, CODE.STATUS_CHANGED)
)
# While these values exist in the chemistry data, their last value doesn't
# persist there when the pump is off/there is no flow. Pulling them from
@ -109,10 +124,12 @@ async def async_setup_entry(
equipment_flags & EQUIPMENT.FLAG_INTELLICHEM
and sensor_name in SUPPORTED_BASIC_CHEM_SENSORS
):
entities.append(ScreenLogicSensor(coordinator, sensor_name))
entities.append(
ScreenLogicStatusSensor(coordinator, sensor_name, CODE.STATUS_CHANGED)
)
# Pump sensors
for pump_num, pump_data in coordinator.data[SL_DATA.KEY_PUMPS].items():
for pump_num, pump_data in coordinator.gateway_data[SL_DATA.KEY_PUMPS].items():
if pump_data["data"] != 0 and "currentWatts" in pump_data:
for pump_key in pump_data:
enabled = True
@ -129,14 +146,16 @@ async def async_setup_entry(
# IntelliChem sensors
if equipment_flags & EQUIPMENT.FLAG_INTELLICHEM:
for chem_sensor_name in coordinator.data[SL_DATA.KEY_CHEMISTRY]:
for chem_sensor_name in coordinator.gateway_data[SL_DATA.KEY_CHEMISTRY]:
enabled = True
if equipment_flags & EQUIPMENT.FLAG_CHLORINATOR:
if chem_sensor_name in ("salt_tds_ppm",):
enabled = False
if chem_sensor_name in SUPPORTED_CHEM_SENSORS:
entities.append(
ScreenLogicChemistrySensor(coordinator, chem_sensor_name, enabled)
ScreenLogicChemistrySensor(
coordinator, chem_sensor_name, CODE.CHEMISTRY_CHANGED, enabled
)
)
# SCG sensors
@ -144,7 +163,7 @@ async def async_setup_entry(
entities.extend(
[
ScreenLogicSCGSensor(coordinator, scg_sensor)
for scg_sensor in coordinator.data[SL_DATA.KEY_SCG]
for scg_sensor in coordinator.gateway_data[SL_DATA.KEY_SCG]
if scg_sensor in SUPPORTED_SCG_SENSORS
]
)
@ -152,54 +171,66 @@ async def async_setup_entry(
async_add_entities(entities)
class ScreenLogicSensor(ScreenlogicEntity, SensorEntity):
"""Representation of the basic ScreenLogic sensor entity."""
class ScreenLogicSensorEntity(ScreenlogicEntity, SensorEntity):
"""Base class for all ScreenLogic sensor entities."""
_attr_has_entity_name = True
@property
def name(self):
def name(self) -> str | None:
"""Name of the sensor."""
return self.sensor["name"]
@property
def native_unit_of_measurement(self):
def native_unit_of_measurement(self) -> str | None:
"""Return the unit of measurement."""
sl_unit = self.sensor.get("unit")
return SL_UNIT_TO_HA_UNIT.get(sl_unit, sl_unit)
@property
def device_class(self):
def device_class(self) -> SensorDeviceClass | None:
"""Device class of the sensor."""
device_type = self.sensor.get("device_type")
return SL_DEVICE_TYPE_TO_HA_DEVICE_CLASS.get(device_type)
@property
def entity_category(self):
def entity_category(self) -> EntityCategory | None:
"""Entity Category of the sensor."""
return (
None if self._data_key == "air_temperature" else EntityCategory.DIAGNOSTIC
)
@property
def state_class(self):
def state_class(self) -> SensorStateClass | None:
"""Return the state class of the sensor."""
state_type = self.sensor.get("state_type")
if self._data_key == "scg_super_chlor_timer":
return None
return SensorStateClass.MEASUREMENT
return SL_STATE_TYPE_TO_HA_STATE_CLASS.get(
state_type, SensorStateClass.MEASUREMENT
)
@property
def native_value(self):
def options(self) -> list[str] | None:
"""Return a set of possible options."""
return self.sensor.get("enum_options")
@property
def native_value(self) -> str | int | float:
"""State of the sensor."""
return self.sensor["value"]
@property
def sensor(self):
def sensor(self) -> dict[str | int, Any]:
"""Shortcut to access the sensor data."""
return self.coordinator.data[SL_DATA.KEY_SENSORS][self._data_key]
return self.gateway_data[SL_DATA.KEY_SENSORS][self._data_key]
class ScreenLogicPumpSensor(ScreenLogicSensor):
class ScreenLogicStatusSensor(ScreenLogicSensorEntity, ScreenLogicPushEntity):
"""Representation of a basic ScreenLogic sensor entity."""
class ScreenLogicPumpSensor(ScreenLogicSensorEntity):
"""Representation of a ScreenLogic pump sensor entity."""
def __init__(self, coordinator, pump, key, enabled=True):
@ -209,21 +240,21 @@ class ScreenLogicPumpSensor(ScreenLogicSensor):
self._key = key
@property
def sensor(self):
def sensor(self) -> dict[str | int, Any]:
"""Shortcut to access the pump sensor data."""
return self.coordinator.data[SL_DATA.KEY_PUMPS][self._pump_id][self._key]
return self.gateway_data[SL_DATA.KEY_PUMPS][self._pump_id][self._key]
class ScreenLogicChemistrySensor(ScreenLogicSensor):
class ScreenLogicChemistrySensor(ScreenLogicSensorEntity, ScreenLogicPushEntity):
"""Representation of a ScreenLogic IntelliChem sensor entity."""
def __init__(self, coordinator, key, enabled=True):
def __init__(self, coordinator, key, message_code, enabled=True):
"""Initialize of the pump sensor."""
super().__init__(coordinator, f"chem_{key}", enabled)
super().__init__(coordinator, f"chem_{key}", message_code, enabled)
self._key = key
@property
def native_value(self):
def native_value(self) -> str | int | float:
"""State of the sensor."""
value = self.sensor["value"]
if "dosing_state" in self._key:
@ -231,15 +262,15 @@ class ScreenLogicChemistrySensor(ScreenLogicSensor):
return (value - 1) if "supply" in self._data_key else value
@property
def sensor(self):
def sensor(self) -> dict[str | int, Any]:
"""Shortcut to access the pump sensor data."""
return self.coordinator.data[SL_DATA.KEY_CHEMISTRY][self._key]
return self.gateway_data[SL_DATA.KEY_CHEMISTRY][self._key]
class ScreenLogicSCGSensor(ScreenLogicSensor):
class ScreenLogicSCGSensor(ScreenLogicSensorEntity):
"""Representation of ScreenLogic SCG sensor entity."""
@property
def sensor(self):
def sensor(self) -> dict[str | int, Any]:
"""Shortcut to access the pump sensor data."""
return self.coordinator.data[SL_DATA.KEY_SCG][self._data_key]
return self.gateway_data[SL_DATA.KEY_SCG][self._data_key]