Skip to content
Permalink

Comparing changes

Choose two branches to see what’s changed or to start a new pull request. If you need to, you can also or learn more about diff comparisons.

Open a pull request

Create a new pull request by comparing changes across two branches. If you need to, you can also . Learn more about diff comparisons here.
base repository: klejejs/python-thermia-online-api
Failed to load repositories. Confirm that selected base ref is valid, then try again.
Loading
base: 6.0.0
Choose a base ref
...
head repository: klejejs/python-thermia-online-api
Failed to load repositories. Confirm that selected head ref is valid, then try again.
Loading
compare: 6.1.0
Choose a head ref
  • 6 commits
  • 14 files changed
  • 1 contributor

Commits on Oct 13, 2024

  1. Copy the full SHA
    7b0de93 View commit details
  2. Minor naming fixes

    klejejs committed Oct 13, 2024
    Copy the full SHA
    ca0589d View commit details
  3. Add basic tests

    klejejs committed Oct 13, 2024
    Copy the full SHA
    59b6c0a View commit details
  4. Verified

    This commit was created on GitHub.com and signed with GitHub’s verified signature. The key has expired.
    Copy the full SHA
    82b85e4 View commit details
  5. Copy the full SHA
    b3b629a View commit details

Commits on Oct 15, 2024

  1. Return API values even if they are hidden, as they are usable

    klejejs committed Oct 15, 2024
    Copy the full SHA
    67096d5 View commit details
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
# Python
__pycache__/
.pytest_cache/

# IDE
.vscode
37 changes: 28 additions & 9 deletions ThermiaOnlineAPI/api/ThermiaAPI.py
Original file line number Diff line number Diff line change
@@ -45,6 +45,9 @@
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"
}

# Fix for multiple operation modes with the same value
REG_OPERATIONMODE_SKIP_VALUES = ["REG_VALUE_OPERATION_MODE_SERVICE"]


class ThermiaAPI:
def __init__(self, email, password):
@@ -267,10 +270,20 @@ def get__group_operational_time(self, device_id: str):
return self.__get_register_group(device_id, REG_GROUP_OPERATIONAL_TIME)

def get_group_operational_operation(self, device: ThermiaHeatPump):
register_data = self.__get_register_group(
device.id, REG_GROUP_OPERATIONAL_OPERATION
return self.__get_group_operational_operation_from_register_group(
device, REG_GROUP_OPERATIONAL_OPERATION
)

def get_group_operational_operation_from_status(self, device: ThermiaHeatPump):
return self.__get_group_operational_operation_from_register_group(
device, REG_GROUP_OPERATIONAL_STATUS
)

def __get_group_operational_operation_from_register_group(
self, device: ThermiaHeatPump, register_group: str
):
register_data = self.__get_register_group(device.id, register_group)

data = [d for d in register_data if d["registerName"] == REG_OPERATIONMODE]

if len(data) != 1:
@@ -286,14 +299,18 @@ def get_group_operational_operation(self, device: ThermiaHeatPump):

if operation_modes_data is not None:
operation_modes_map = map(
lambda values: {
values.get("value"): values.get("name").split(
"REG_VALUE_OPERATION_MODE_"
)[1],
},
lambda values: (
{
values.get("value"): values.get("name").split(
"REG_VALUE_OPERATION_MODE_"
)[1],
}
if values.get("name") not in REG_OPERATIONMODE_SKIP_VALUES
else {}
),
operation_modes_data,
)
operation_modes_list = list(operation_modes_map)
operation_modes_list = list(filter(lambda x: x != {}, operation_modes_map))
operation_modes = ChainMap(*operation_modes_list)

current_operation_mode = [
@@ -677,7 +694,9 @@ def __authenticate(self) -> bool:
"client_id": THERMIA_AZURE_AUTH_CLIENT_ID_AND_SCOPE,
"redirect_uri": THERMIA_AZURE_AUTH_REDIRECT_URI,
"scope": THERMIA_AZURE_AUTH_CLIENT_ID_AND_SCOPE,
"code": request_confirmed.url.split("code=")[1],
"code": utils.get_list_value_or_default(
request_confirmed.url.split("code="), 1, ""
),
"code_verifier": code_challenge,
"grant_type": "authorization_code",
}
157 changes: 53 additions & 104 deletions ThermiaOnlineAPI/model/HeatPump.py
Original file line number Diff line number Diff line change
@@ -71,6 +71,7 @@ def __init__(self, device_data: dict, api_interface: "ThermiaAPI"):
self.__group_operational_status = None
self.__group_operational_time = None
self.__group_operational_operation = None
self.__group_operational_operation_read_only = None
self.__group_hot_water: Dict[str, Optional[int]] = {
"hot_water_switch": None,
"hot_water_boost_switch": None,
@@ -85,12 +86,10 @@ def __init__(self, device_data: dict, api_interface: "ThermiaAPI"):

self.__operational_statuses = None
self.__all_operational_statuses_map = None
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()
@@ -116,6 +115,9 @@ def update_data(self):
self.__group_operational_operation = (
self.__api_interface.get_group_operational_operation(self)
)
self.__group_operational_operation_read_only = (
self.__api_interface.get_group_operational_operation_from_status(self)
)
self.__group_hot_water = self.__api_interface.get_group_hot_water(self)

self.__alarms = self.__api_interface.get_all_alarms(self.__device_id)
@@ -127,18 +129,12 @@ def update_data(self):
self.__all_operational_statuses_map = (
self.__get_all_operational_statuses_from_operational_status()
)
self.__visible_operational_statuses_map = (
self.__get_all_visible_operational_statuses_from_operational_status()
)
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):
@@ -365,27 +361,6 @@ def __get_register_from_operational_status(

return data[0]

def __get_value_by_key_and_register_name_from_operational_status(
self, register_name: str, value_key: str
):
data_from_register = self.__get_register_from_operational_status(register_name)

if data_from_register is None:
return None

register_values_list = data_from_register.get("valueNames", [])
values_list: list = [d for d in register_values_list if d["name"] == value_key]

if len(values_list) != 1:
return None

value: dict = values_list[0]

if value.get("visible"):
return value.get("value")

return None

def __get_operational_statuses_from_operational_status(self) -> Optional[Dict]:
if self.__device_config["operational_status_register"] is not None:
data = self.__get_register_from_operational_status(
@@ -443,38 +418,17 @@ def __get_all_operational_statuses_from_operational_status(
if data is None:
return ChainMap()

operation_modes_map = map(
operation_statuses_map = map(
lambda values: {
values.get("value"): {
"name": values.get("name").split(
self.__device_config["operational_status_valueNamePrefix"]
)[1],
"visible": values.get("visible"),
}
values.get("value"): values.get("name").split(
self.__device_config["operational_status_valueNamePrefix"]
)[1],
},
data,
)

operation_modes_list = list(operation_modes_map)
return ChainMap(*operation_modes_list)

def __get_all_visible_operational_statuses_from_operational_status(
self,
) -> Optional[ChainMap]:
data = self.__all_operational_statuses_map

if data is None:
return ChainMap()

operation_modes_map = map(
lambda item: (
{item[0]: item[1].get("name")} if item[1].get("visible") else {}
),
data.items(),
)

operation_modes_list = list(filter(lambda x: x != {}, operation_modes_map))
return ChainMap(*operation_modes_list)
operation_statuses_list = list(operation_statuses_map)
return ChainMap(*operation_statuses_list)

def __get_running_operational_statuses(
self,
@@ -499,9 +453,7 @@ def __get_running_operational_statuses(
data_items_list = list(data.items())

current_operation_mode = [
value.get("name")
for key, value in data_items_list
if key == current_register_value
value for key, value in data_items_list if key == current_register_value
]

if len(current_operation_mode) == 1:
@@ -514,7 +466,7 @@ def __get_running_operational_statuses(
):
# 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_operation_modes = []
list_of_current_operation_statuses = []

if self.__device_config["operational_status_minRegisterValue"] is not None:
current_register_value -= int(
@@ -524,11 +476,10 @@ def __get_running_operational_statuses(
for key, value in data_items_list:
if key <= current_register_value:
current_register_value -= key
if value.get("visible"):
list_of_current_operation_modes.append(value.get("name"))
list_of_current_operation_statuses.append(value)

if current_register_value == 0:
return list_of_current_operation_modes
return list_of_current_operation_statuses

return []

@@ -548,36 +499,15 @@ def __get_all_power_statuses_from_power_status(
if data is None:
return ChainMap()

power_modes_map = map(
power_statuses_map = map(
lambda values: {
values.get("value"): {
"name": values.get("name").split("COMP_VALUE_STEP_")[1],
"visible": values.get("visible"),
}
values.get("value"): values.get("name").split("COMP_VALUE_STEP_")[1],
},
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)
power_statuses_list = list(power_statuses_map)
return ChainMap(*power_statuses_list)

def __get_running_power_statuses(
self,
@@ -597,9 +527,7 @@ def __get_running_power_statuses(
data_items_list = list(data.items())

current_power_status = [
value.get("name")
for key, value in data_items_list
if key == current_register_value
value for key, value in data_items_list if key == current_register_value
]

if len(current_power_status) == 1:
@@ -617,8 +545,7 @@ def __get_running_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"))
list_of_current_power_statuses.append(value)

if current_register_value == 0:
return list_of_current_power_statuses
@@ -795,7 +722,7 @@ def running_operational_statuses(self) -> List[str]:

@property
def available_operational_statuses(self) -> Optional[List[str]]:
data = self.__visible_operational_statuses_map
data = self.__all_operational_statuses_map

if data is None:
return []
@@ -804,7 +731,7 @@ def available_operational_statuses(self) -> Optional[List[str]]:

@property
def available_operational_statuses_map(self) -> Optional[ChainMap]:
return self.__visible_operational_statuses_map
return self.__all_operational_statuses_map

@property
def running_power_statuses(self) -> List[str]:
@@ -817,7 +744,7 @@ def running_power_statuses(self) -> List[str]:

@property
def available_power_statuses(self) -> Optional[List[str]]:
data = self.__visible_power_statuses_map
data = self.__all_power_statuses_map

if data is None:
return []
@@ -826,7 +753,7 @@ def available_power_statuses(self) -> Optional[List[str]]:

@property
def available_power_statuses_map(self) -> Optional[ChainMap]:
return self.__visible_power_statuses_map
return self.__all_power_statuses_map

@property
def operational_status_integral(self):
@@ -890,25 +817,50 @@ def auxiliary_heater_3_operational_time(self):

@property
def operation_mode(self):
return get_dict_value_or_none(self.__group_operational_operation, "current")
if self.__group_operational_operation is not None:
return get_dict_value_or_none(self.__group_operational_operation, "current")

return get_dict_value_or_none(
self.__group_operational_operation_read_only, "current"
)

@property
def available_operation_modes(self):
if self.__group_operational_operation is not None:
return list(
get_dict_value_or_default(
self.__group_operational_operation, "available", {}
).values()
)

return list(
get_dict_value_or_default(
self.__group_operational_operation, "available", {}
self.__group_operational_operation_read_only, "available", {}
).values()
)

@property
def available_operation_mode_map(self):
if self.__group_operational_operation is not None:
return get_dict_value_or_default(
self.__group_operational_operation, "available", {}
)

return get_dict_value_or_default(
self.__group_operational_operation, "available", {}
self.__group_operational_operation_read_only, "available", {}
)

@property
def is_operation_mode_read_only(self):
return get_dict_value_or_none(self.__group_operational_operation, "isReadOnly")
if self.__group_operational_operation is not None:
return get_dict_value_or_none(
self.__group_operational_operation, "isReadOnly"
)

if self.__group_operational_operation_read_only is not None:
return True

return None

###########################################################################
# Hot water data
@@ -1021,9 +973,6 @@ def debug(self):
["macAddress", "owner", "retailerAccess", "retailerId", "id", "status"],
)

print("self.__group_temperatures:")
pretty_print_except(self.__group_temperatures)

installation_profile_id = get_dict_value_or_none(
self.__info, "installationProfileId"
)
Empty file.
Loading