From e8bf6496afcef7459be2de1004c3972f0e0c263b Mon Sep 17 00:00:00 2001 From: davefrooney Date: Mon, 13 May 2024 00:33:25 +0100 Subject: [PATCH 1/6] Update climate.py fix deprecating units --- custom_components/smartbox/climate.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/custom_components/smartbox/climate.py b/custom_components/smartbox/climate.py index 56ea541..5e7efcd 100644 --- a/custom_components/smartbox/climate.py +++ b/custom_components/smartbox/climate.py @@ -11,7 +11,7 @@ from homeassistant.const import ( ATTR_LOCKED, ATTR_TEMPERATURE, - TEMP_CELSIUS, + UnitOfTemperature, ) from homeassistant.core import HomeAssistant import logging @@ -75,11 +75,19 @@ def __init__(self, node: Union[MagicMock, SmartboxNode]) -> None: self._node = node self._status: Dict[str, Any] = {} self._available = False # unavailable until we get an update + self._enable_turn_on_off_backwards_compatibility = False self._supported_features = ( - ClimateEntityFeature.TARGET_TEMPERATURE | ClimateEntityFeature.PRESET_MODE + ClimateEntityFeature.TARGET_TEMPERATURE | ClimateEntityFeature.PRESET_MODE | ClimateEntityFeature.TURN_OFF | ClimateEntityFeature.TURN_ON ) _LOGGER.debug(f"Created node {self.name} unique_id={self.unique_id}") + async def async_turn_off(self) -> None: + await self.async_set_hvac_mode(HVACMode.OFF) + + async def async_turn_on(self) -> None: + await self.async_set_hvac_mode(HVACMode.AUTO) + + @property def unique_id(self) -> str: """Return Unique ID string.""" @@ -108,7 +116,7 @@ def temperature_unit(self) -> str: return unit else: return ( - TEMP_CELSIUS # climate sensors need a temperature unit on construction + UnitOfTemperature.CELSIUS # climate sensors need a temperature unit on construction ) @property From 23cc846c952d5c4ce9c527a7418e32da368c7c67 Mon Sep 17 00:00:00 2001 From: davefrooney Date: Mon, 13 May 2024 00:33:50 +0100 Subject: [PATCH 2/6] Update model.py fix deprecating units --- custom_components/smartbox/model.py | 44 +++++++++++++++-------------- 1 file changed, 23 insertions(+), 21 deletions(-) diff --git a/custom_components/smartbox/model.py b/custom_components/smartbox/model.py index 99eb59a..69e8f55 100644 --- a/custom_components/smartbox/model.py +++ b/custom_components/smartbox/model.py @@ -2,19 +2,21 @@ import logging from homeassistant.const import ( - TEMP_CELSIUS, - TEMP_FAHRENHEIT, + UnitOfTemperature, +) + +from homeassistant.components.climate import ( + HVACMode, ) from homeassistant.components.climate.const import ( - HVAC_MODE_AUTO, - HVAC_MODE_HEAT, - HVAC_MODE_OFF, PRESET_ACTIVITY, PRESET_AWAY, PRESET_COMFORT, PRESET_ECO, PRESET_HOME, ) + + from homeassistant.core import HomeAssistant from smartbox import Session, UpdateManager from typing import Any, cast, Dict, List, Union @@ -246,9 +248,9 @@ def get_temperature_unit(status): return None unit = status["units"] if unit == "C": - return TEMP_CELSIUS + return UnitOfTemperature.CELSIUS elif unit == "F": - return TEMP_FAHRENHEIT + return UnitOfTemperature.FAHRENHEIT else: raise ValueError(f"Unknown temp unit {unit}") @@ -382,21 +384,21 @@ def set_temperature_args( def get_hvac_mode(node_type: str, status: Dict[str, Any]) -> str: _check_status_key("mode", node_type, status) if status["mode"] == "off": - return HVAC_MODE_OFF + return HVACMode.OFF elif node_type == HEATER_NODE_TYPE_HTR_MOD and not status["on"]: - return HVAC_MODE_OFF + return HVACMode.OFF elif status["mode"] == "manual": - return HVAC_MODE_HEAT + return HVACMode.HEAT elif status["mode"] == "auto": - return HVAC_MODE_AUTO + return HVACMode.AUTO elif status["mode"] == "modified_auto": # This occurs when the temperature is modified while in auto mode. # Mapping it to auto seems to make this most sense - return HVAC_MODE_AUTO + return HVACMode.AUTO elif status["mode"] == "self_learn": - return HVAC_MODE_AUTO + return HVACMode.AUTO elif status["mode"] == "presence": - return HVAC_MODE_AUTO + return HVACMode.AUTO else: _LOGGER.error(f"Unknown smartbox node mode {status['mode']}") raise ValueError(f"Unknown smartbox node mode {status['mode']}") @@ -406,9 +408,9 @@ def set_hvac_mode_args( node_type: str, status: Dict[str, Any], hvac_mode: str ) -> Dict[str, Any]: if node_type == HEATER_NODE_TYPE_HTR_MOD: - if hvac_mode == HVAC_MODE_OFF: + if hvac_mode == HVACMode.OFF: return {"on": False} - elif hvac_mode == HVAC_MODE_HEAT: + elif hvac_mode == HVACMode.HEAT: # We need to pass these status keys on when setting the mode required_status_keys = ["selected_temp"] for key in required_status_keys: @@ -417,16 +419,16 @@ def set_hvac_mode_args( hvac_mode_args["on"] = True hvac_mode_args["mode"] = "manual" return hvac_mode_args - elif hvac_mode == HVAC_MODE_AUTO: + elif hvac_mode == HVACMode.AUTO: return {"on": True, "mode": "auto"} else: raise ValueError(f"Unsupported hvac mode {hvac_mode}") else: - if hvac_mode == HVAC_MODE_OFF: + if hvac_mode == HVACMode.OFF: return {"mode": "off"} - elif hvac_mode == HVAC_MODE_HEAT: + elif hvac_mode == HVACMode.HEAT: return {"mode": "manual"} - elif hvac_mode == HVAC_MODE_AUTO: + elif hvac_mode == HVACMode.AUTO: return {"mode": "auto"} else: raise ValueError(f"Unsupported hvac mode {hvac_mode}") @@ -492,7 +494,7 @@ def set_preset_mode_status_update( assert preset_mode != PRESET_HOME and preset_mode != PRESET_AWAY if preset_mode == PRESET_SCHEDULE: - return set_hvac_mode_args(node_type, status, HVAC_MODE_AUTO) + return set_hvac_mode_args(node_type, status, HVACMode.AUTO) elif preset_mode == PRESET_SELF_LEARN: return {"on": True, "mode": "self_learn"} elif preset_mode == PRESET_ACTIVITY: From c32f693bd3805b8daf979b8ed9b89272c7acd5ed Mon Sep 17 00:00:00 2001 From: davefrooney Date: Mon, 13 May 2024 00:34:14 +0100 Subject: [PATCH 3/6] Update model.py fix deprecating units --- custom_components/smartbox/model.py | 669 +++++++++------------------- 1 file changed, 209 insertions(+), 460 deletions(-) diff --git a/custom_components/smartbox/model.py b/custom_components/smartbox/model.py index 69e8f55..5133a15 100644 --- a/custom_components/smartbox/model.py +++ b/custom_components/smartbox/model.py @@ -1,525 +1,274 @@ -import asyncio -import logging - +from datetime import datetime, timedelta from homeassistant.const import ( - UnitOfTemperature, + ATTR_LOCKED, + UnitOfEnergy, + PERCENTAGE, + UnitOfPower, ) - -from homeassistant.components.climate import ( - HVACMode, +from homeassistant.components.sensor import ( + SensorEntity, + SensorStateClass, + SensorDeviceClass, ) -from homeassistant.components.climate.const import ( - PRESET_ACTIVITY, - PRESET_AWAY, - PRESET_COMFORT, - PRESET_ECO, - PRESET_HOME, -) - - from homeassistant.core import HomeAssistant -from smartbox import Session, UpdateManager -from typing import Any, cast, Dict, List, Union +import logging +from typing import Any, Callable, Dict, Optional, Union from unittest.mock import MagicMock from .const import ( - GITHUB_ISSUES_URL, + DOMAIN, HEATER_NODE_TYPE_ACM, + HEATER_NODE_TYPE_HTR, HEATER_NODE_TYPE_HTR_MOD, - HEATER_NODE_TYPES, - PRESET_FROST, - PRESET_SCHEDULE, - PRESET_SELF_LEARN, + SMARTBOX_NODES, ) -from .types import FactoryOptionsDict, SetupDict, StatusDict +from .model import get_temperature_unit, is_heater_node, is_heating, SmartboxNode _LOGGER = logging.getLogger(__name__) -class SmartboxDevice(object): - def __init__( - self, - dev_id: str, - name: str, - session: Union[Session, MagicMock], - socket_reconnect_attempts: int, - socket_backoff_factor: float, - ) -> None: - self._dev_id = dev_id - self._name = name - self._session = session - self._socket_reconnect_attempts = socket_reconnect_attempts - self._socket_backoff_factor = socket_backoff_factor - self._away = False - self._power_limit: int = 0 - - async def initialise_nodes(self, hass: HomeAssistant) -> None: - # Would do in __init__, but needs to be a coroutine - session_nodes = await hass.async_add_executor_job( - self._session.get_nodes, self.dev_id - ) - self._nodes = {} - for node_info in session_nodes: - status = await hass.async_add_executor_job( - self._session.get_status, self._dev_id, node_info - ) - setup = await hass.async_add_executor_job( - self._session.get_setup, self._dev_id, node_info - ) - node = SmartboxNode(self, node_info, self._session, status, setup) - self._nodes[(node.node_type, node.addr)] = node - - _LOGGER.debug(f"Creating SocketSession for device {self._dev_id}") - self._update_manager = UpdateManager( - self._session, - self._dev_id, - reconnect_attempts=self._socket_reconnect_attempts, - backoff_factor=self._socket_backoff_factor, - ) +async def async_setup_platform( + hass: HomeAssistant, + config: Dict[Any, Any], + async_add_entities: Callable, + discovery_info: Optional[Dict[Any, Any]] = None, +) -> None: + """Set up platform.""" + _LOGGER.debug("Setting up Smartbox sensor platform") + if discovery_info is None: + return + + # Temperature + async_add_entities( + [ + TemperatureSensor(node) + for node in hass.data[DOMAIN][SMARTBOX_NODES] + if is_heater_node(node) + ], + True, + ) + # Power + async_add_entities( + [ + PowerSensor(node) + for node in hass.data[DOMAIN][SMARTBOX_NODES] + if is_heater_node(node) and node.node_type != HEATER_NODE_TYPE_HTR_MOD + ], + True, + ) + # Duty Cycle and Energy + # Only nodes of type 'htr' seem to report the duty cycle, which is needed + # to compute energy consumption + async_add_entities( + [ + DutyCycleSensor(node) + for node in hass.data[DOMAIN][SMARTBOX_NODES] + if node.node_type == HEATER_NODE_TYPE_HTR + ], + True, + ) + async_add_entities( + [ + EnergySensor(node) + for node in hass.data[DOMAIN][SMARTBOX_NODES] + if node.node_type == HEATER_NODE_TYPE_HTR + ], + True, + ) + # Charge Level + async_add_entities( + [ + ChargeLevelSensor(node) + for node in hass.data[DOMAIN][SMARTBOX_NODES] + if is_heater_node(node) and node.node_type == HEATER_NODE_TYPE_ACM + ], + True, + ) - self._update_manager.subscribe_to_device_away_status(self._away_status_update) - self._update_manager.subscribe_to_device_power_limit(self._power_limit_update) - self._update_manager.subscribe_to_node_status(self._node_status_update) - self._update_manager.subscribe_to_node_setup(self._node_setup_update) - - _LOGGER.debug(f"Starting UpdateManager task for device {self._dev_id}") - asyncio.create_task(self._update_manager.run()) - - def _away_status_update(self, away_status: Dict[str, bool]) -> None: - _LOGGER.debug(f"Away status update: {away_status}") - self._away = away_status["away"] - - def _power_limit_update(self, power_limit: int) -> None: - _LOGGER.debug(f"power_limit update: {power_limit}") - self._power_limit = power_limit - - def _node_status_update( - self, node_type: str, addr: int, node_status: StatusDict - ) -> None: - _LOGGER.debug(f"Node status update: {node_status}") - node = self._nodes.get((node_type, addr), None) - if node is not None: - node.update_status(node_status) - else: - _LOGGER.error(f"Received status update for unknown node {node_type} {addr}") - - def _node_setup_update( - self, node_type: str, addr: int, node_setup: SetupDict - ) -> None: - _LOGGER.debug(f"Node setup update: {node_setup}") - node = self._nodes.get((node_type, addr), None) - if node is not None: - node.update_setup(node_setup) - else: - _LOGGER.error(f"Received setup update for unknown node {node_type} {addr}") + _LOGGER.debug("Finished setting up Smartbox sensor platform") - @property - def dev_id(self) -> str: - return self._dev_id - def get_nodes(self): - return self._nodes.values() +class SmartboxSensorBase(SensorEntity): + def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: + self._node = node + self._status: Dict[str, Any] = {} + self._available = False # unavailable until we get an update + self._last_update: Optional[datetime] = None + self._time_since_last_update: Optional[timedelta] = None + _LOGGER.debug(f"Created node {self.name} unique_id={self.unique_id}") @property - def name(self) -> str: - return self._name + def extra_state_attributes(self) -> Dict[str, bool]: + return { + ATTR_LOCKED: self._status["locked"], + } @property - def away(self) -> bool: - return self._away - - def set_away_status(self, away: bool): - self._session.set_device_away_status(self.dev_id, {"away": away}) - self._away = away + def available(self) -> bool: + return self._available + + async def async_update(self) -> None: + new_status = await self._node.async_update(self.hass) + if new_status["sync_status"] == "ok": + # update our status + self._status = new_status + self._available = True + update_time = datetime.now() + if self._last_update is not None: + self._time_since_last_update = update_time - self._last_update + self._last_update = update_time + else: + self._available = False + self._last_update = None + self._time_since_last_update = None @property - def power_limit(self) -> int: - return self._power_limit - - def set_power_limit(self, power_limit: int) -> None: - self._session.set_device_power_limit(self.dev_id, power_limit) - self._power_limit = power_limit - - -class SmartboxNode(object): - def __init__( - self, - device: Union[SmartboxDevice, MagicMock], - node_info: Dict[str, Any], - session: Union[Session, MagicMock], - status: Dict[str, Any], - setup: Dict[str, Any], - ) -> None: - self._device = device - self._node_info = node_info - self._session = session - self._status = status - self._setup = setup + def time_since_last_update(self) -> Optional[timedelta]: + return self._time_since_last_update - @property - def node_id(self) -> str: - # TODO: are addrs only unique among node types, or for the whole device? - return f"{self._device.dev_id}-{self._node_info['addr']}" + +class TemperatureSensor(SmartboxSensorBase): + """Smartbox heater temperature sensor""" + + device_class = SensorDeviceClass.TEMPERATURE + state_class = SensorStateClass.MEASUREMENT + + def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: + super().__init__(node) @property def name(self) -> str: - return self._node_info["name"] + return f"{self._node.name} Temperature" @property - def node_type(self) -> str: - """Return node type, e.g. 'htr' for heaters""" - return self._node_info["type"] + def unique_id(self) -> str: + return f"{self._node.node_id}_temperature" @property - def addr(self) -> int: - return self._node_info["addr"] + def native_value(self) -> float: + return self._status["mtemp"] @property - def status(self) -> StatusDict: - return self._status + def native_unit_of_measurement(self) -> str: + return get_temperature_unit(self._status) - def update_status(self, status: StatusDict) -> None: - _LOGGER.debug(f"Updating node {self.name} status: {status}") - self._status = status - @property - def setup(self) -> SetupDict: - return self._setup +class PowerSensor(SmartboxSensorBase): + """Smartbox heater power sensor - def update_setup(self, setup: SetupDict) -> None: - _LOGGER.debug(f"Updating node {self.name} setup: {setup}") - self._setup = setup + Note: this represents the power the heater is drawing *when heating*; the + heater is not always active over the entire period since the last update, + even when 'active' is true. The duty cycle sensor indicates how much it + was active. To measure energy consumption, use the corresponding energy + sensor. + """ - def set_status(self, **status_args) -> StatusDict: - self._session.set_status(self._device.dev_id, self._node_info, status_args) - # update our status locally until we get an update - self._status |= {**status_args} - return self._status + device_class = SensorDeviceClass.POWER + native_unit_of_measurement = UnitOfPower.WATT + state_class = SensorStateClass.MEASUREMENT - @property - def away(self): - return self._device.away + def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: + super().__init__(node) - def update_device_away_status(self, away: bool): - self._device.set_away_status(away) - - async def async_update(self, hass: HomeAssistant) -> StatusDict: - return self.status + @property + def name(self) -> str: + return f"{self._node.name} Power" @property - def window_mode(self) -> bool: - if "window_mode_enabled" not in self._setup: - raise KeyError( - "window_mode_enabled not present in setup for node {self.name}" - ) - return self._setup["window_mode_enabled"] + def unique_id(self) -> str: + return f"{self._node.node_id}_power" - def set_window_mode(self, window_mode: bool): - self._session.set_setup( - self._device.dev_id, self._node_info, {"window_mode_enabled": window_mode} + @property + def native_value(self) -> float: + return ( + self._status["power"] + if is_heating(self._node.node_type, self._status) + else 0 ) - self._setup["window_mode_enabled"] = window_mode - @property - def true_radiant(self) -> bool: - if "true_radiant_enabled" not in self._setup: - raise KeyError( - "true_radiant_enabled not present in setup for node {self.name}" - ) - return self._setup["true_radiant_enabled"] - def set_true_radiant(self, true_radiant: bool): - self._session.set_setup( - self._device.dev_id, self._node_info, {"true_radiant_enabled": true_radiant} - ) - self._setup["true_radiant_enabled"] = true_radiant +class DutyCycleSensor(SmartboxSensorBase): + """Smartbox heater duty cycle sensor + Represents the duty cycle for the heater. + """ -def is_heater_node(node: Union[SmartboxNode, MagicMock]) -> bool: - return node.node_type in HEATER_NODE_TYPES + device_class = SensorDeviceClass.POWER_FACTOR + native_unit_of_measurement = PERCENTAGE + state_class = SensorStateClass.MEASUREMENT + def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: + super().__init__(node) -def is_supported_node(node: Union[SmartboxNode, MagicMock]) -> bool: - return is_heater_node(node) + @property + def name(self) -> str: + return f"{self._node.name} Duty Cycle" + @property + def unique_id(self) -> str: + return f"{self._node.node_id}_duty_cycle" -def get_temperature_unit(status): - if "units" not in status: - return None - unit = status["units"] - if unit == "C": - return UnitOfTemperature.CELSIUS - elif unit == "F": - return UnitOfTemperature.FAHRENHEIT - else: - raise ValueError(f"Unknown temp unit {unit}") + @property + def native_value(self) -> float: + return self._status["duty"] -async def get_devices( - hass: HomeAssistant, - api_name: str, - basic_auth_creds: str, - username: str, - password: str, - session_retry_attempts: int, - session_backoff_factor: float, - socket_reconnect_attempts: int, - socket_backoff_factor: float, -) -> List[SmartboxDevice]: - _LOGGER.info( - f"Creating Smartbox session for {api_name}" - f"(session_retry_attempts={session_retry_attempts}" - f", session_backoff_factor={session_backoff_factor}" - f", socket_reconnect_attempts={socket_reconnect_attempts}" - f", socket_backoff_factor={session_backoff_factor})" - ) - session = await hass.async_add_executor_job( - Session, - api_name, - basic_auth_creds, - username, - password, - session_retry_attempts, - session_backoff_factor, - ) - session_devices = await hass.async_add_executor_job(session.get_devices) - # TODO: gather? - devices = [ - await create_smartbox_device( - hass, - session_device["dev_id"], - session_device["name"], - session, - socket_reconnect_attempts, - socket_backoff_factor, - ) - for session_device in session_devices - ] - return devices +class EnergySensor(SmartboxSensorBase): + """Smartbox heater energy sensor + Represents the energy consumed by the heater. + """ -async def create_smartbox_device( - hass: HomeAssistant, - dev_id: str, - name: str, - session: Union[Session, MagicMock], - socket_reconnect_attempts: int, - socket_backoff_factor: float, -) -> Union[SmartboxDevice, MagicMock]: - """Factory function for SmartboxDevices""" - device = SmartboxDevice( - dev_id, name, session, socket_reconnect_attempts, socket_backoff_factor - ) - await device.initialise_nodes(hass) - return device + device_class = SensorDeviceClass.ENERGY + native_unit_of_measurement = UnitOfEnergy.WATT_HOUR + state_class = SensorStateClass.TOTAL + def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: + super().__init__(node) -def _check_status_key(key: str, node_type: str, status: Dict[str, Any]): - if key not in status: - raise KeyError( - f"'{key}' not found in {node_type} - please report to {GITHUB_ISSUES_URL}. " - f"status: {status}" - ) + @property + def name(self) -> str: + return f"{self._node.name} Energy" + @property + def unique_id(self) -> str: + return f"{self._node.node_id}_energy" -def get_target_temperature(node_type: str, status: Dict[str, Any]) -> float: - if node_type == HEATER_NODE_TYPE_HTR_MOD: - _check_status_key("selected_temp", node_type, status) - if status["selected_temp"] == "comfort": - _check_status_key("comfort_temp", node_type, status) - return float(status["comfort_temp"]) - elif status["selected_temp"] == "eco": - _check_status_key("comfort_temp", node_type, status) - _check_status_key("eco_offset", node_type, status) - return float(status["comfort_temp"]) - float(status["eco_offset"]) - elif status["selected_temp"] == "ice": - _check_status_key("ice_temp", node_type, status) - return float(status["ice_temp"]) - else: - raise KeyError( - f"'Unexpected 'selected_temp' value {status['selected_temp']}" - f" found for {node_type} - please report to" - f" {GITHUB_ISSUES_URL}. status: {status}" - ) - else: - _check_status_key("stemp", node_type, status) - return float(status["stemp"]) - - -def set_temperature_args( - node_type: str, status: Dict[str, Any], temp: float -) -> Dict[str, Any]: - _check_status_key("units", node_type, status) - if node_type == HEATER_NODE_TYPE_HTR_MOD: - if status["selected_temp"] == "comfort": - target_temp = temp - elif status["selected_temp"] == "eco": - _check_status_key("eco_offset", node_type, status) - target_temp = temp + float(status["eco_offset"]) - elif status["selected_temp"] == "ice": - raise ValueError( - "Can't set temperature for htr_mod devices when ice mode is selected" + @property + def native_value(self) -> float | None: + time_since_last_update = self.time_since_last_update + if time_since_last_update is not None: + return ( + float(self._status["power"]) + * float(self._status["duty"]) + / 100 + * time_since_last_update.seconds + / 60 + / 60 ) else: - raise KeyError( - f"'Unexpected 'selected_temp' value {status['selected_temp']}" - f" found for {node_type} - please report to " - f"{GITHUB_ISSUES_URL}. status: {status}" - ) - return { - "on": True, - "mode": status["mode"], - "selected_temp": status["selected_temp"], - "comfort_temp": str(target_temp), - "eco_offset": status["eco_offset"], - "units": status["units"], - } - else: - return { - "stemp": str(temp), - "units": status["units"], - } + return None -def get_hvac_mode(node_type: str, status: Dict[str, Any]) -> str: - _check_status_key("mode", node_type, status) - if status["mode"] == "off": - return HVACMode.OFF - elif node_type == HEATER_NODE_TYPE_HTR_MOD and not status["on"]: - return HVACMode.OFF - elif status["mode"] == "manual": - return HVACMode.HEAT - elif status["mode"] == "auto": - return HVACMode.AUTO - elif status["mode"] == "modified_auto": - # This occurs when the temperature is modified while in auto mode. - # Mapping it to auto seems to make this most sense - return HVACMode.AUTO - elif status["mode"] == "self_learn": - return HVACMode.AUTO - elif status["mode"] == "presence": - return HVACMode.AUTO - else: - _LOGGER.error(f"Unknown smartbox node mode {status['mode']}") - raise ValueError(f"Unknown smartbox node mode {status['mode']}") - - -def set_hvac_mode_args( - node_type: str, status: Dict[str, Any], hvac_mode: str -) -> Dict[str, Any]: - if node_type == HEATER_NODE_TYPE_HTR_MOD: - if hvac_mode == HVACMode.OFF: - return {"on": False} - elif hvac_mode == HVACMode.HEAT: - # We need to pass these status keys on when setting the mode - required_status_keys = ["selected_temp"] - for key in required_status_keys: - _check_status_key(key, node_type, status) - hvac_mode_args = {k: status[k] for k in required_status_keys} - hvac_mode_args["on"] = True - hvac_mode_args["mode"] = "manual" - return hvac_mode_args - elif hvac_mode == HVACMode.AUTO: - return {"on": True, "mode": "auto"} - else: - raise ValueError(f"Unsupported hvac mode {hvac_mode}") - else: - if hvac_mode == HVACMode.OFF: - return {"mode": "off"} - elif hvac_mode == HVACMode.HEAT: - return {"mode": "manual"} - elif hvac_mode == HVACMode.AUTO: - return {"mode": "auto"} - else: - raise ValueError(f"Unsupported hvac mode {hvac_mode}") +class ChargeLevelSensor(SmartboxSensorBase): + """Smartbox storage heater charge level sensor""" + device_class = SensorDeviceClass.BATTERY + native_unit_of_measurement = PERCENTAGE + state_class = SensorStateClass.MEASUREMENT + + def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: + super().__init__(node) + + @property + def name(self) -> str: + return f"{self._node.name} Charge Level" + + @property + def unique_id(self) -> str: + return f"{self._node.node_id}_charge_level" + + @property + def native_value(self) -> int: + return self._status["charge_level"] -def _get_htr_mod_preset_mode(node_type: str, mode: str, selected_temp: str) -> str: - if mode == "manual": - if selected_temp == "comfort": - return PRESET_COMFORT - elif selected_temp == "eco": - return PRESET_ECO - elif selected_temp == "ice": - return PRESET_FROST - else: - raise ValueError( - f"'Unexpected 'selected_temp' value {'selected_temp'} found for " - f"{node_type} - please report to {GITHUB_ISSUES_URL}." - ) - elif mode == "auto": - return PRESET_SCHEDULE - elif mode == "presence": - return PRESET_ACTIVITY - elif mode == "self_learn": - return PRESET_SELF_LEARN - else: - raise ValueError(f"Unknown smartbox node mode {mode}") - - -def get_preset_mode(node_type: str, status: Dict[str, Any], away: bool) -> str: - if away: - return PRESET_AWAY - if node_type == HEATER_NODE_TYPE_HTR_MOD: - _check_status_key("mode", node_type, status) - _check_status_key("selected_temp", node_type, status) - return _get_htr_mod_preset_mode( - node_type, status["mode"], status["selected_temp"] - ) - else: - return PRESET_HOME - - -def get_preset_modes(node_type: str) -> List[str]: - if node_type == HEATER_NODE_TYPE_HTR_MOD: - return [ - PRESET_ACTIVITY, - PRESET_AWAY, - PRESET_COMFORT, - PRESET_ECO, - PRESET_FROST, - PRESET_SCHEDULE, - PRESET_SELF_LEARN, - ] - else: - return [PRESET_AWAY, PRESET_HOME] - - -def set_preset_mode_status_update( - node_type: str, status: Dict[str, Any], preset_mode: str -) -> Dict[str, Any]: - if node_type != HEATER_NODE_TYPE_HTR_MOD: - raise ValueError(f"{node_type} nodes do not support preset {preset_mode}") - # PRESET_HOME and PRESET_AWAY are not handled via status updates - assert preset_mode != PRESET_HOME and preset_mode != PRESET_AWAY - - if preset_mode == PRESET_SCHEDULE: - return set_hvac_mode_args(node_type, status, HVACMode.AUTO) - elif preset_mode == PRESET_SELF_LEARN: - return {"on": True, "mode": "self_learn"} - elif preset_mode == PRESET_ACTIVITY: - return {"on": True, "mode": "presence"} - elif preset_mode == PRESET_COMFORT: - return {"on": True, "mode": "manual", "selected_temp": "comfort"} - elif preset_mode == PRESET_ECO: - return {"on": True, "mode": "manual", "selected_temp": "eco"} - elif preset_mode == PRESET_FROST: - return {"on": True, "mode": "manual", "selected_temp": "ice"} - else: - raise ValueError(f"Unsupported preset {preset_mode} for node type {node_type}") - - -def is_heating(node_type: str, status: Dict[str, Any]) -> str: - return status["charging"] if node_type == HEATER_NODE_TYPE_ACM else status["active"] - - -def get_factory_options(node: Union[SmartboxNode, MagicMock]) -> FactoryOptionsDict: - return cast(FactoryOptionsDict, node.setup.get("factory_options", {})) - - -def window_mode_available(node: Union[SmartboxNode, MagicMock]) -> bool: - return get_factory_options(node).get("window_mode_available", False) - - -def true_radiant_available(node: Union[SmartboxNode, MagicMock]) -> bool: - return get_factory_options(node).get("true_radiant_available", False) From e40a6ad21120746d6b6e60c3914e292c39e28885 Mon Sep 17 00:00:00 2001 From: davefrooney Date: Mon, 13 May 2024 00:34:30 +0100 Subject: [PATCH 4/6] Update model.py --- custom_components/smartbox/model.py | 669 +++++++++++++++++++--------- 1 file changed, 460 insertions(+), 209 deletions(-) diff --git a/custom_components/smartbox/model.py b/custom_components/smartbox/model.py index 5133a15..69e8f55 100644 --- a/custom_components/smartbox/model.py +++ b/custom_components/smartbox/model.py @@ -1,274 +1,525 @@ -from datetime import datetime, timedelta +import asyncio +import logging + from homeassistant.const import ( - ATTR_LOCKED, - UnitOfEnergy, - PERCENTAGE, - UnitOfPower, + UnitOfTemperature, ) -from homeassistant.components.sensor import ( - SensorEntity, - SensorStateClass, - SensorDeviceClass, + +from homeassistant.components.climate import ( + HVACMode, ) +from homeassistant.components.climate.const import ( + PRESET_ACTIVITY, + PRESET_AWAY, + PRESET_COMFORT, + PRESET_ECO, + PRESET_HOME, +) + + from homeassistant.core import HomeAssistant -import logging -from typing import Any, Callable, Dict, Optional, Union +from smartbox import Session, UpdateManager +from typing import Any, cast, Dict, List, Union from unittest.mock import MagicMock from .const import ( - DOMAIN, + GITHUB_ISSUES_URL, HEATER_NODE_TYPE_ACM, - HEATER_NODE_TYPE_HTR, HEATER_NODE_TYPE_HTR_MOD, - SMARTBOX_NODES, + HEATER_NODE_TYPES, + PRESET_FROST, + PRESET_SCHEDULE, + PRESET_SELF_LEARN, ) -from .model import get_temperature_unit, is_heater_node, is_heating, SmartboxNode +from .types import FactoryOptionsDict, SetupDict, StatusDict _LOGGER = logging.getLogger(__name__) -async def async_setup_platform( - hass: HomeAssistant, - config: Dict[Any, Any], - async_add_entities: Callable, - discovery_info: Optional[Dict[Any, Any]] = None, -) -> None: - """Set up platform.""" - _LOGGER.debug("Setting up Smartbox sensor platform") - if discovery_info is None: - return - - # Temperature - async_add_entities( - [ - TemperatureSensor(node) - for node in hass.data[DOMAIN][SMARTBOX_NODES] - if is_heater_node(node) - ], - True, - ) - # Power - async_add_entities( - [ - PowerSensor(node) - for node in hass.data[DOMAIN][SMARTBOX_NODES] - if is_heater_node(node) and node.node_type != HEATER_NODE_TYPE_HTR_MOD - ], - True, - ) - # Duty Cycle and Energy - # Only nodes of type 'htr' seem to report the duty cycle, which is needed - # to compute energy consumption - async_add_entities( - [ - DutyCycleSensor(node) - for node in hass.data[DOMAIN][SMARTBOX_NODES] - if node.node_type == HEATER_NODE_TYPE_HTR - ], - True, - ) - async_add_entities( - [ - EnergySensor(node) - for node in hass.data[DOMAIN][SMARTBOX_NODES] - if node.node_type == HEATER_NODE_TYPE_HTR - ], - True, - ) - # Charge Level - async_add_entities( - [ - ChargeLevelSensor(node) - for node in hass.data[DOMAIN][SMARTBOX_NODES] - if is_heater_node(node) and node.node_type == HEATER_NODE_TYPE_ACM - ], - True, - ) - - _LOGGER.debug("Finished setting up Smartbox sensor platform") - +class SmartboxDevice(object): + def __init__( + self, + dev_id: str, + name: str, + session: Union[Session, MagicMock], + socket_reconnect_attempts: int, + socket_backoff_factor: float, + ) -> None: + self._dev_id = dev_id + self._name = name + self._session = session + self._socket_reconnect_attempts = socket_reconnect_attempts + self._socket_backoff_factor = socket_backoff_factor + self._away = False + self._power_limit: int = 0 + + async def initialise_nodes(self, hass: HomeAssistant) -> None: + # Would do in __init__, but needs to be a coroutine + session_nodes = await hass.async_add_executor_job( + self._session.get_nodes, self.dev_id + ) + self._nodes = {} + for node_info in session_nodes: + status = await hass.async_add_executor_job( + self._session.get_status, self._dev_id, node_info + ) + setup = await hass.async_add_executor_job( + self._session.get_setup, self._dev_id, node_info + ) + node = SmartboxNode(self, node_info, self._session, status, setup) + self._nodes[(node.node_type, node.addr)] = node + + _LOGGER.debug(f"Creating SocketSession for device {self._dev_id}") + self._update_manager = UpdateManager( + self._session, + self._dev_id, + reconnect_attempts=self._socket_reconnect_attempts, + backoff_factor=self._socket_backoff_factor, + ) -class SmartboxSensorBase(SensorEntity): - def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: - self._node = node - self._status: Dict[str, Any] = {} - self._available = False # unavailable until we get an update - self._last_update: Optional[datetime] = None - self._time_since_last_update: Optional[timedelta] = None - _LOGGER.debug(f"Created node {self.name} unique_id={self.unique_id}") + self._update_manager.subscribe_to_device_away_status(self._away_status_update) + self._update_manager.subscribe_to_device_power_limit(self._power_limit_update) + self._update_manager.subscribe_to_node_status(self._node_status_update) + self._update_manager.subscribe_to_node_setup(self._node_setup_update) + + _LOGGER.debug(f"Starting UpdateManager task for device {self._dev_id}") + asyncio.create_task(self._update_manager.run()) + + def _away_status_update(self, away_status: Dict[str, bool]) -> None: + _LOGGER.debug(f"Away status update: {away_status}") + self._away = away_status["away"] + + def _power_limit_update(self, power_limit: int) -> None: + _LOGGER.debug(f"power_limit update: {power_limit}") + self._power_limit = power_limit + + def _node_status_update( + self, node_type: str, addr: int, node_status: StatusDict + ) -> None: + _LOGGER.debug(f"Node status update: {node_status}") + node = self._nodes.get((node_type, addr), None) + if node is not None: + node.update_status(node_status) + else: + _LOGGER.error(f"Received status update for unknown node {node_type} {addr}") + + def _node_setup_update( + self, node_type: str, addr: int, node_setup: SetupDict + ) -> None: + _LOGGER.debug(f"Node setup update: {node_setup}") + node = self._nodes.get((node_type, addr), None) + if node is not None: + node.update_setup(node_setup) + else: + _LOGGER.error(f"Received setup update for unknown node {node_type} {addr}") @property - def extra_state_attributes(self) -> Dict[str, bool]: - return { - ATTR_LOCKED: self._status["locked"], - } + def dev_id(self) -> str: + return self._dev_id - @property - def available(self) -> bool: - return self._available - - async def async_update(self) -> None: - new_status = await self._node.async_update(self.hass) - if new_status["sync_status"] == "ok": - # update our status - self._status = new_status - self._available = True - update_time = datetime.now() - if self._last_update is not None: - self._time_since_last_update = update_time - self._last_update - self._last_update = update_time - else: - self._available = False - self._last_update = None - self._time_since_last_update = None + def get_nodes(self): + return self._nodes.values() @property - def time_since_last_update(self) -> Optional[timedelta]: - return self._time_since_last_update + def name(self) -> str: + return self._name + @property + def away(self) -> bool: + return self._away -class TemperatureSensor(SmartboxSensorBase): - """Smartbox heater temperature sensor""" + def set_away_status(self, away: bool): + self._session.set_device_away_status(self.dev_id, {"away": away}) + self._away = away - device_class = SensorDeviceClass.TEMPERATURE - state_class = SensorStateClass.MEASUREMENT + @property + def power_limit(self) -> int: + return self._power_limit + + def set_power_limit(self, power_limit: int) -> None: + self._session.set_device_power_limit(self.dev_id, power_limit) + self._power_limit = power_limit + + +class SmartboxNode(object): + def __init__( + self, + device: Union[SmartboxDevice, MagicMock], + node_info: Dict[str, Any], + session: Union[Session, MagicMock], + status: Dict[str, Any], + setup: Dict[str, Any], + ) -> None: + self._device = device + self._node_info = node_info + self._session = session + self._status = status + self._setup = setup - def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: - super().__init__(node) + @property + def node_id(self) -> str: + # TODO: are addrs only unique among node types, or for the whole device? + return f"{self._device.dev_id}-{self._node_info['addr']}" @property def name(self) -> str: - return f"{self._node.name} Temperature" + return self._node_info["name"] @property - def unique_id(self) -> str: - return f"{self._node.node_id}_temperature" + def node_type(self) -> str: + """Return node type, e.g. 'htr' for heaters""" + return self._node_info["type"] @property - def native_value(self) -> float: - return self._status["mtemp"] + def addr(self) -> int: + return self._node_info["addr"] @property - def native_unit_of_measurement(self) -> str: - return get_temperature_unit(self._status) - + def status(self) -> StatusDict: + return self._status -class PowerSensor(SmartboxSensorBase): - """Smartbox heater power sensor + def update_status(self, status: StatusDict) -> None: + _LOGGER.debug(f"Updating node {self.name} status: {status}") + self._status = status - Note: this represents the power the heater is drawing *when heating*; the - heater is not always active over the entire period since the last update, - even when 'active' is true. The duty cycle sensor indicates how much it - was active. To measure energy consumption, use the corresponding energy - sensor. - """ + @property + def setup(self) -> SetupDict: + return self._setup - device_class = SensorDeviceClass.POWER - native_unit_of_measurement = UnitOfPower.WATT - state_class = SensorStateClass.MEASUREMENT + def update_setup(self, setup: SetupDict) -> None: + _LOGGER.debug(f"Updating node {self.name} setup: {setup}") + self._setup = setup - def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: - super().__init__(node) + def set_status(self, **status_args) -> StatusDict: + self._session.set_status(self._device.dev_id, self._node_info, status_args) + # update our status locally until we get an update + self._status |= {**status_args} + return self._status @property - def name(self) -> str: - return f"{self._node.name} Power" + def away(self): + return self._device.away - @property - def unique_id(self) -> str: - return f"{self._node.node_id}_power" + def update_device_away_status(self, away: bool): + self._device.set_away_status(away) + + async def async_update(self, hass: HomeAssistant) -> StatusDict: + return self.status @property - def native_value(self) -> float: - return ( - self._status["power"] - if is_heating(self._node.node_type, self._status) - else 0 + def window_mode(self) -> bool: + if "window_mode_enabled" not in self._setup: + raise KeyError( + "window_mode_enabled not present in setup for node {self.name}" + ) + return self._setup["window_mode_enabled"] + + def set_window_mode(self, window_mode: bool): + self._session.set_setup( + self._device.dev_id, self._node_info, {"window_mode_enabled": window_mode} ) + self._setup["window_mode_enabled"] = window_mode + @property + def true_radiant(self) -> bool: + if "true_radiant_enabled" not in self._setup: + raise KeyError( + "true_radiant_enabled not present in setup for node {self.name}" + ) + return self._setup["true_radiant_enabled"] -class DutyCycleSensor(SmartboxSensorBase): - """Smartbox heater duty cycle sensor + def set_true_radiant(self, true_radiant: bool): + self._session.set_setup( + self._device.dev_id, self._node_info, {"true_radiant_enabled": true_radiant} + ) + self._setup["true_radiant_enabled"] = true_radiant - Represents the duty cycle for the heater. - """ - device_class = SensorDeviceClass.POWER_FACTOR - native_unit_of_measurement = PERCENTAGE - state_class = SensorStateClass.MEASUREMENT +def is_heater_node(node: Union[SmartboxNode, MagicMock]) -> bool: + return node.node_type in HEATER_NODE_TYPES - def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: - super().__init__(node) - @property - def name(self) -> str: - return f"{self._node.name} Duty Cycle" +def is_supported_node(node: Union[SmartboxNode, MagicMock]) -> bool: + return is_heater_node(node) - @property - def unique_id(self) -> str: - return f"{self._node.node_id}_duty_cycle" - @property - def native_value(self) -> float: - return self._status["duty"] +def get_temperature_unit(status): + if "units" not in status: + return None + unit = status["units"] + if unit == "C": + return UnitOfTemperature.CELSIUS + elif unit == "F": + return UnitOfTemperature.FAHRENHEIT + else: + raise ValueError(f"Unknown temp unit {unit}") -class EnergySensor(SmartboxSensorBase): - """Smartbox heater energy sensor +async def get_devices( + hass: HomeAssistant, + api_name: str, + basic_auth_creds: str, + username: str, + password: str, + session_retry_attempts: int, + session_backoff_factor: float, + socket_reconnect_attempts: int, + socket_backoff_factor: float, +) -> List[SmartboxDevice]: + _LOGGER.info( + f"Creating Smartbox session for {api_name}" + f"(session_retry_attempts={session_retry_attempts}" + f", session_backoff_factor={session_backoff_factor}" + f", socket_reconnect_attempts={socket_reconnect_attempts}" + f", socket_backoff_factor={session_backoff_factor})" + ) + session = await hass.async_add_executor_job( + Session, + api_name, + basic_auth_creds, + username, + password, + session_retry_attempts, + session_backoff_factor, + ) + session_devices = await hass.async_add_executor_job(session.get_devices) + # TODO: gather? + devices = [ + await create_smartbox_device( + hass, + session_device["dev_id"], + session_device["name"], + session, + socket_reconnect_attempts, + socket_backoff_factor, + ) + for session_device in session_devices + ] + return devices - Represents the energy consumed by the heater. - """ - device_class = SensorDeviceClass.ENERGY - native_unit_of_measurement = UnitOfEnergy.WATT_HOUR - state_class = SensorStateClass.TOTAL +async def create_smartbox_device( + hass: HomeAssistant, + dev_id: str, + name: str, + session: Union[Session, MagicMock], + socket_reconnect_attempts: int, + socket_backoff_factor: float, +) -> Union[SmartboxDevice, MagicMock]: + """Factory function for SmartboxDevices""" + device = SmartboxDevice( + dev_id, name, session, socket_reconnect_attempts, socket_backoff_factor + ) + await device.initialise_nodes(hass) + return device - def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: - super().__init__(node) - @property - def name(self) -> str: - return f"{self._node.name} Energy" +def _check_status_key(key: str, node_type: str, status: Dict[str, Any]): + if key not in status: + raise KeyError( + f"'{key}' not found in {node_type} - please report to {GITHUB_ISSUES_URL}. " + f"status: {status}" + ) - @property - def unique_id(self) -> str: - return f"{self._node.node_id}_energy" - @property - def native_value(self) -> float | None: - time_since_last_update = self.time_since_last_update - if time_since_last_update is not None: - return ( - float(self._status["power"]) - * float(self._status["duty"]) - / 100 - * time_since_last_update.seconds - / 60 - / 60 +def get_target_temperature(node_type: str, status: Dict[str, Any]) -> float: + if node_type == HEATER_NODE_TYPE_HTR_MOD: + _check_status_key("selected_temp", node_type, status) + if status["selected_temp"] == "comfort": + _check_status_key("comfort_temp", node_type, status) + return float(status["comfort_temp"]) + elif status["selected_temp"] == "eco": + _check_status_key("comfort_temp", node_type, status) + _check_status_key("eco_offset", node_type, status) + return float(status["comfort_temp"]) - float(status["eco_offset"]) + elif status["selected_temp"] == "ice": + _check_status_key("ice_temp", node_type, status) + return float(status["ice_temp"]) + else: + raise KeyError( + f"'Unexpected 'selected_temp' value {status['selected_temp']}" + f" found for {node_type} - please report to" + f" {GITHUB_ISSUES_URL}. status: {status}" + ) + else: + _check_status_key("stemp", node_type, status) + return float(status["stemp"]) + + +def set_temperature_args( + node_type: str, status: Dict[str, Any], temp: float +) -> Dict[str, Any]: + _check_status_key("units", node_type, status) + if node_type == HEATER_NODE_TYPE_HTR_MOD: + if status["selected_temp"] == "comfort": + target_temp = temp + elif status["selected_temp"] == "eco": + _check_status_key("eco_offset", node_type, status) + target_temp = temp + float(status["eco_offset"]) + elif status["selected_temp"] == "ice": + raise ValueError( + "Can't set temperature for htr_mod devices when ice mode is selected" ) else: - return None - - -class ChargeLevelSensor(SmartboxSensorBase): - """Smartbox storage heater charge level sensor""" - - device_class = SensorDeviceClass.BATTERY - native_unit_of_measurement = PERCENTAGE - state_class = SensorStateClass.MEASUREMENT - - def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: - super().__init__(node) + raise KeyError( + f"'Unexpected 'selected_temp' value {status['selected_temp']}" + f" found for {node_type} - please report to " + f"{GITHUB_ISSUES_URL}. status: {status}" + ) + return { + "on": True, + "mode": status["mode"], + "selected_temp": status["selected_temp"], + "comfort_temp": str(target_temp), + "eco_offset": status["eco_offset"], + "units": status["units"], + } + else: + return { + "stemp": str(temp), + "units": status["units"], + } - @property - def name(self) -> str: - return f"{self._node.name} Charge Level" - @property - def unique_id(self) -> str: - return f"{self._node.node_id}_charge_level" +def get_hvac_mode(node_type: str, status: Dict[str, Any]) -> str: + _check_status_key("mode", node_type, status) + if status["mode"] == "off": + return HVACMode.OFF + elif node_type == HEATER_NODE_TYPE_HTR_MOD and not status["on"]: + return HVACMode.OFF + elif status["mode"] == "manual": + return HVACMode.HEAT + elif status["mode"] == "auto": + return HVACMode.AUTO + elif status["mode"] == "modified_auto": + # This occurs when the temperature is modified while in auto mode. + # Mapping it to auto seems to make this most sense + return HVACMode.AUTO + elif status["mode"] == "self_learn": + return HVACMode.AUTO + elif status["mode"] == "presence": + return HVACMode.AUTO + else: + _LOGGER.error(f"Unknown smartbox node mode {status['mode']}") + raise ValueError(f"Unknown smartbox node mode {status['mode']}") + + +def set_hvac_mode_args( + node_type: str, status: Dict[str, Any], hvac_mode: str +) -> Dict[str, Any]: + if node_type == HEATER_NODE_TYPE_HTR_MOD: + if hvac_mode == HVACMode.OFF: + return {"on": False} + elif hvac_mode == HVACMode.HEAT: + # We need to pass these status keys on when setting the mode + required_status_keys = ["selected_temp"] + for key in required_status_keys: + _check_status_key(key, node_type, status) + hvac_mode_args = {k: status[k] for k in required_status_keys} + hvac_mode_args["on"] = True + hvac_mode_args["mode"] = "manual" + return hvac_mode_args + elif hvac_mode == HVACMode.AUTO: + return {"on": True, "mode": "auto"} + else: + raise ValueError(f"Unsupported hvac mode {hvac_mode}") + else: + if hvac_mode == HVACMode.OFF: + return {"mode": "off"} + elif hvac_mode == HVACMode.HEAT: + return {"mode": "manual"} + elif hvac_mode == HVACMode.AUTO: + return {"mode": "auto"} + else: + raise ValueError(f"Unsupported hvac mode {hvac_mode}") - @property - def native_value(self) -> int: - return self._status["charge_level"] +def _get_htr_mod_preset_mode(node_type: str, mode: str, selected_temp: str) -> str: + if mode == "manual": + if selected_temp == "comfort": + return PRESET_COMFORT + elif selected_temp == "eco": + return PRESET_ECO + elif selected_temp == "ice": + return PRESET_FROST + else: + raise ValueError( + f"'Unexpected 'selected_temp' value {'selected_temp'} found for " + f"{node_type} - please report to {GITHUB_ISSUES_URL}." + ) + elif mode == "auto": + return PRESET_SCHEDULE + elif mode == "presence": + return PRESET_ACTIVITY + elif mode == "self_learn": + return PRESET_SELF_LEARN + else: + raise ValueError(f"Unknown smartbox node mode {mode}") + + +def get_preset_mode(node_type: str, status: Dict[str, Any], away: bool) -> str: + if away: + return PRESET_AWAY + if node_type == HEATER_NODE_TYPE_HTR_MOD: + _check_status_key("mode", node_type, status) + _check_status_key("selected_temp", node_type, status) + return _get_htr_mod_preset_mode( + node_type, status["mode"], status["selected_temp"] + ) + else: + return PRESET_HOME + + +def get_preset_modes(node_type: str) -> List[str]: + if node_type == HEATER_NODE_TYPE_HTR_MOD: + return [ + PRESET_ACTIVITY, + PRESET_AWAY, + PRESET_COMFORT, + PRESET_ECO, + PRESET_FROST, + PRESET_SCHEDULE, + PRESET_SELF_LEARN, + ] + else: + return [PRESET_AWAY, PRESET_HOME] + + +def set_preset_mode_status_update( + node_type: str, status: Dict[str, Any], preset_mode: str +) -> Dict[str, Any]: + if node_type != HEATER_NODE_TYPE_HTR_MOD: + raise ValueError(f"{node_type} nodes do not support preset {preset_mode}") + # PRESET_HOME and PRESET_AWAY are not handled via status updates + assert preset_mode != PRESET_HOME and preset_mode != PRESET_AWAY + + if preset_mode == PRESET_SCHEDULE: + return set_hvac_mode_args(node_type, status, HVACMode.AUTO) + elif preset_mode == PRESET_SELF_LEARN: + return {"on": True, "mode": "self_learn"} + elif preset_mode == PRESET_ACTIVITY: + return {"on": True, "mode": "presence"} + elif preset_mode == PRESET_COMFORT: + return {"on": True, "mode": "manual", "selected_temp": "comfort"} + elif preset_mode == PRESET_ECO: + return {"on": True, "mode": "manual", "selected_temp": "eco"} + elif preset_mode == PRESET_FROST: + return {"on": True, "mode": "manual", "selected_temp": "ice"} + else: + raise ValueError(f"Unsupported preset {preset_mode} for node type {node_type}") + + +def is_heating(node_type: str, status: Dict[str, Any]) -> str: + return status["charging"] if node_type == HEATER_NODE_TYPE_ACM else status["active"] + + +def get_factory_options(node: Union[SmartboxNode, MagicMock]) -> FactoryOptionsDict: + return cast(FactoryOptionsDict, node.setup.get("factory_options", {})) + + +def window_mode_available(node: Union[SmartboxNode, MagicMock]) -> bool: + return get_factory_options(node).get("window_mode_available", False) + + +def true_radiant_available(node: Union[SmartboxNode, MagicMock]) -> bool: + return get_factory_options(node).get("true_radiant_available", False) From c82dd36f21daaeee81981d6841cc9e3e47be6317 Mon Sep 17 00:00:00 2001 From: davefrooney Date: Mon, 13 May 2024 00:34:50 +0100 Subject: [PATCH 5/6] Update sensor.py --- custom_components/smartbox/sensor.py | 25 +++++++++++-------------- 1 file changed, 11 insertions(+), 14 deletions(-) diff --git a/custom_components/smartbox/sensor.py b/custom_components/smartbox/sensor.py index 1353a8c..5133a15 100644 --- a/custom_components/smartbox/sensor.py +++ b/custom_components/smartbox/sensor.py @@ -1,18 +1,14 @@ from datetime import datetime, timedelta from homeassistant.const import ( ATTR_LOCKED, - DEVICE_CLASS_BATTERY, - DEVICE_CLASS_ENERGY, - DEVICE_CLASS_POWER, - DEVICE_CLASS_POWER_FACTOR, - DEVICE_CLASS_TEMPERATURE, - ENERGY_WATT_HOUR, + UnitOfEnergy, PERCENTAGE, - POWER_WATT, + UnitOfPower, ) from homeassistant.components.sensor import ( SensorEntity, SensorStateClass, + SensorDeviceClass, ) from homeassistant.core import HomeAssistant import logging @@ -134,7 +130,7 @@ def time_since_last_update(self) -> Optional[timedelta]: class TemperatureSensor(SmartboxSensorBase): """Smartbox heater temperature sensor""" - device_class = DEVICE_CLASS_TEMPERATURE + device_class = SensorDeviceClass.TEMPERATURE state_class = SensorStateClass.MEASUREMENT def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: @@ -167,8 +163,8 @@ class PowerSensor(SmartboxSensorBase): sensor. """ - device_class = DEVICE_CLASS_POWER - native_unit_of_measurement = POWER_WATT + device_class = SensorDeviceClass.POWER + native_unit_of_measurement = UnitOfPower.WATT state_class = SensorStateClass.MEASUREMENT def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: @@ -197,7 +193,7 @@ class DutyCycleSensor(SmartboxSensorBase): Represents the duty cycle for the heater. """ - device_class = DEVICE_CLASS_POWER_FACTOR + device_class = SensorDeviceClass.POWER_FACTOR native_unit_of_measurement = PERCENTAGE state_class = SensorStateClass.MEASUREMENT @@ -223,8 +219,8 @@ class EnergySensor(SmartboxSensorBase): Represents the energy consumed by the heater. """ - device_class = DEVICE_CLASS_ENERGY - native_unit_of_measurement = ENERGY_WATT_HOUR + device_class = SensorDeviceClass.ENERGY + native_unit_of_measurement = UnitOfEnergy.WATT_HOUR state_class = SensorStateClass.TOTAL def __init__(self, node: Union[SmartboxNode, MagicMock]) -> None: @@ -257,7 +253,7 @@ def native_value(self) -> float | None: class ChargeLevelSensor(SmartboxSensorBase): """Smartbox storage heater charge level sensor""" - device_class = DEVICE_CLASS_BATTERY + device_class = SensorDeviceClass.BATTERY native_unit_of_measurement = PERCENTAGE state_class = SensorStateClass.MEASUREMENT @@ -275,3 +271,4 @@ def unique_id(self) -> str: @property def native_value(self) -> int: return self._status["charge_level"] + From e5d190d3a55c11c5d0a3e8360ac65dd37cd2aa39 Mon Sep 17 00:00:00 2001 From: davefrooney Date: Mon, 13 May 2024 01:34:36 +0100 Subject: [PATCH 6/6] Update __init__.py fix helpers deprecation warning --- custom_components/smartbox/__init__.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/custom_components/smartbox/__init__.py b/custom_components/smartbox/__init__.py index 5850ce0..1600953 100644 --- a/custom_components/smartbox/__init__.py +++ b/custom_components/smartbox/__init__.py @@ -6,6 +6,7 @@ import homeassistant.helpers.config_validation as cv from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant +from homeassistant.helpers.discovery import async_load_platform from smartbox import __version__ as SMARTBOX_VERSION @@ -133,9 +134,7 @@ async def async_setup(hass: HomeAssistant, config: dict) -> bool: if hass.data[DOMAIN][SMARTBOX_DEVICES]: for component in PLATFORMS: - await hass.helpers.discovery.async_load_platform( - component, DOMAIN, {}, config - ) + await async_load_platform(hass, component, DOMAIN, {}, config) _LOGGER.debug("Finished setting up Smartbox integration")