From c1660316f9a4475b267dcb783bf5922f61b71606 Mon Sep 17 00:00:00 2001 From: Krisjanis Lejejs Date: Mon, 7 Oct 2024 01:23:39 +0300 Subject: [PATCH] Refactor operational and power statuses --- README.md | 14 +- ThermiaOnlineAPI/const.py | 2 + ThermiaOnlineAPI/model/HeatPump.py | 200 +++++++++++++++++------------ example.py | 24 ++-- 4 files changed, 134 insertions(+), 106 deletions(-) diff --git a/README.md b/README.md index bf6a79d..b05f1a8 100644 --- a/README.md +++ b/README.md @@ -54,17 +54,9 @@ To execute the example file, first run `pip install -r requirements.txt` to inst | `running_operational_statuses` | List of running operational statuses of the Heat Pump | | `available_operational_statuses` | List of available operational statuses | | `available_operational_statuses_map` | Dictionary mapping operational status names to their values | -| `operational_status_auxiliary_heater_3kw` | Auxiliary heater status for 3kw (returns `None` if unavailable) | -| `operational_status_auxiliary_heater_6kw` | Auxiliary heater status for 6kw (returns `None` if unavailable) | -| `operational_status_auxiliary_heater_9kw` | Auxiliary heater status for 9kw (returns `None` if unavailable) | -| `operational_status_auxiliary_heater_12kw` | Auxiliary heater status for 12kw (returns `None` if unavailable) | -| `operational_status_auxiliary_heater_15kw` | Auxiliary heater status for 15kw (returns `None` if unavailable) | -| `operational_status_compressor_status` | Compressor status | -| `operational_status_brine_pump_status` | Brine pump status | -| `operational_status_radiator_pump_status` | Radiator pump status | -| `operational_status_cooling_status` | Cooling status | -| `operational_status_hot_water_status` | Hot water status | -| `operational_status_heating_status` | Heating status | +| `running_power_statuses` | List of running power statuses of the Heat Pump | +| `available_power_statuses` | List of available power statuses | +| `available_power_statuses_map` | Dictionary mapping power status names to their values | | `operational_status_integral` | Integral | | `operational_status_pid` | PID | | --- | --- | diff --git a/ThermiaOnlineAPI/const.py b/ThermiaOnlineAPI/const.py index 3803827..04ab877 100644 --- a/ThermiaOnlineAPI/const.py +++ b/ThermiaOnlineAPI/const.py @@ -77,6 +77,8 @@ REG_INTEGRAL_LSD = "REG_INTEGRAL_LSD" REG_PID = "REG_PID" +COMP_POWER_STATUS = "COMP_POWER_STATUS" + ############################################################################### # Hot water registers ############################################################################### diff --git a/ThermiaOnlineAPI/model/HeatPump.py b/ThermiaOnlineAPI/model/HeatPump.py index c43872a..8bbe2b5 100644 --- a/ThermiaOnlineAPI/model/HeatPump.py +++ b/ThermiaOnlineAPI/model/HeatPump.py @@ -28,6 +28,7 @@ REG_OPER_TIME_IMM3, REG_PID, REG_RETURN_LINE, + COMP_POWER_STATUS, COMP_STATUS, COMP_STATUS_ITEC, REG_SUPPLY_LINE, @@ -87,6 +88,11 @@ def __init__(self, device_data: dict, api_interface: "ThermiaAPI"): self.__visible_operational_statuses_map = None self.__running_operational_statuses = None + self.__power_statuses = None + self.__all_power_statuses_map = None + self.__visible_power_statuses_map = None + self.__running_power_statuses = None + self.update_data() def update_data(self): @@ -126,6 +132,15 @@ def update_data(self): ) self.__running_operational_statuses = self.__get_running_operational_statuses() + self.__power_statuses = self.__get_power_statuses_from_operational_status() + self.__all_power_statuses_map = ( + self.__get_all_power_statuses_from_power_status() + ) + self.__visible_power_statuses_map = ( + self.__get_all_visible_power_statuses_from_power_status() + ) + self.__running_power_statuses = self.__get_running_power_statuses() + def get_register_indexes(self): return self.__register_indexes @@ -517,6 +532,99 @@ def __get_running_operational_statuses( return [] + def __get_power_statuses_from_operational_status(self) -> Optional[Dict]: + data = self.__get_register_from_operational_status(COMP_POWER_STATUS) + + if data is None: + return None + + return data.get("valueNames", []) + + def __get_all_power_statuses_from_power_status( + self, + ) -> Optional[ChainMap]: + data = self.__power_statuses + + if data is None: + return ChainMap() + + power_modes_map = map( + lambda values: { + values.get("value"): { + "name": values.get("name").split("COMP_VALUE_STEP_")[1], + "visible": values.get("visible"), + } + }, + data, + ) + + power_modes_list = list(power_modes_map) + return ChainMap(*power_modes_list) + + def __get_all_visible_power_statuses_from_power_status( + self, + ) -> Optional[ChainMap]: + data = self.__all_power_statuses_map + + if data is None: + return ChainMap() + + power_modes_map = map( + lambda item: ( + {item[0]: item[1].get("name")} if item[1].get("visible") else {} + ), + data.items(), + ) + + power_modes_list = list(filter(lambda x: x != {}, power_modes_map)) + return ChainMap(*power_modes_list) + + def __get_running_power_statuses( + self, + ) -> List[str]: + data = self.__get_register_from_operational_status(COMP_POWER_STATUS) + + if data is None: + return [] + + current_register_value = get_dict_value_or_none(data, "registerValue") + + data = self.__all_power_statuses_map + + if data is None: + return [] + + data_items_list = list(data.items()) + + current_power_status = [ + value.get("name") + for key, value in data_items_list + if key == current_register_value + ] + + if len(current_power_status) == 1: + return current_power_status + + if ( + len(current_power_status) != 1 + and current_register_value > 0 + and len(data_items_list) > 1 + ): + # Attempt to get multiple statuses by binary sum of the values + data_items_list.sort(key=lambda x: x[0], reverse=True) + list_of_current_power_statuses = [] + + for key, value in data_items_list: + if key <= current_register_value: + current_register_value -= key + if value.get("visible"): + list_of_current_power_statuses.append(value.get("name")) + + if current_register_value == 0: + return list_of_current_power_statuses + + return [] + @property def name(self): return get_dict_value_or_none(self.__info, "name") @@ -699,94 +807,26 @@ def available_operational_statuses_map(self) -> Optional[ChainMap]: return self.__visible_operational_statuses_map @property - def operational_status_auxiliary_heater_3kw(self): - return self.__get_value_by_key_and_register_name_from_operational_status( - "COMP_POWER_STATUS", "COMP_VALUE_STEP_3KW" - ) - - @property - def operational_status_auxiliary_heater_6kw(self): - return self.__get_value_by_key_and_register_name_from_operational_status( - "COMP_POWER_STATUS", "COMP_VALUE_STEP_6KW" - ) - - @property - def operational_status_auxiliary_heater_9kw(self): - return self.__get_value_by_key_and_register_name_from_operational_status( - "COMP_POWER_STATUS", "COMP_VALUE_STEP_9KW" - ) - - @property - def operational_status_auxiliary_heater_12kw(self): - return self.__get_value_by_key_and_register_name_from_operational_status( - "COMP_POWER_STATUS", "COMP_VALUE_STEP_12KW" - ) - - @property - def operational_status_auxiliary_heater_15kw(self): - return self.__get_value_by_key_and_register_name_from_operational_status( - "COMP_POWER_STATUS", "COMP_VALUE_STEP_15KW" - ) - - @property - def operational_status_compressor_status(self) -> Optional[bool]: - if ( - self.__visible_operational_statuses_map is None - or "COMPR" not in self.__visible_operational_statuses_map.values() - ): - return None - - return "COMPR" in self.running_operational_statuses - - @property - def operational_status_brine_pump_status(self) -> Optional[bool]: - if ( - self.__visible_operational_statuses_map is None - or "BRINEPUMP" not in self.__visible_operational_statuses_map.values() - ): - return None - - return "BRINEPUMP" in self.running_operational_statuses + def running_power_statuses(self) -> List[str]: + data = self.__running_power_statuses - @property - def operational_status_radiator_pump_status(self) -> Optional[bool]: - if ( - self.__visible_operational_statuses_map is None - or "RADIATORPUMP" not in self.__visible_operational_statuses_map.values() - ): - return None + if data is None: + return [] - return "RADIATORPUMP" in self.running_operational_statuses + return data @property - def operational_status_cooling_status(self) -> Optional[bool]: - if ( - self.__visible_operational_statuses_map is None - or "COOLING" not in self.__visible_operational_statuses_map.values() - ): - return None - - return "COOLING" in self.running_operational_statuses + def available_power_statuses(self) -> Optional[List[str]]: + data = self.__visible_power_statuses_map - @property - def operational_status_hot_water_status(self) -> Optional[bool]: - if ( - self.__visible_operational_statuses_map is None - or "HOT_WATER" not in self.__visible_operational_statuses_map.values() - ): - return None + if data is None: + return [] - return "HOT_WATER" in self.running_operational_statuses + return list(data.values()) @property - def operational_status_heating_status(self) -> Optional[bool]: - if ( - self.__visible_operational_statuses_map is None - or "HEATING" not in self.__visible_operational_statuses_map.values() - ): - return None - - return "HEATING" in self.running_operational_statuses + def available_power_statuses_map(self) -> Optional[ChainMap]: + return self.__visible_power_statuses_map @property def operational_status_integral(self): diff --git a/example.py b/example.py index c8f9f80..3164590 100644 --- a/example.py +++ b/example.py @@ -63,22 +63,16 @@ "Available operational statuses map: " + str(heat_pump.available_operational_statuses_map) ) -print("Auxiliary heater 3KW: " + str(heat_pump.operational_status_auxiliary_heater_3kw)) -print("Auxiliary heater 6KW: " + str(heat_pump.operational_status_auxiliary_heater_6kw)) -print("Auxiliary heater 9KW: " + str(heat_pump.operational_status_auxiliary_heater_9kw)) -print( - "Auxiliary heater 12KW: " + str(heat_pump.operational_status_auxiliary_heater_12kw) -) -print( - "Auxiliary heater 15KW: " + str(heat_pump.operational_status_auxiliary_heater_15kw) -) -print("Compressor status: " + str(heat_pump.operational_status_compressor_status)) -print("Brine pump status: " + str(heat_pump.operational_status_brine_pump_status)) -print("Radiator pump status: " + str(heat_pump.operational_status_radiator_pump_status)) -print("Cooling status: " + str(heat_pump.operational_status_cooling_status)) -print("Hot water status: " + str(heat_pump.operational_status_hot_water_status)) -print("Heating status: " + str(heat_pump.operational_status_heating_status)) +print("\n") + +print("Power status") +print("Running power statuses: " + str(heat_pump.running_power_statuses)) +print("Available power statuses: " + str(heat_pump.available_power_statuses)) +print("Available power statuses map: " + str(heat_pump.available_power_statuses_map)) + +print("\n") + print("Integral: " + str(heat_pump.operational_status_integral)) print("Pid: " + str(heat_pump.operational_status_pid))