277 lines
11 KiB
Python
Executable File
277 lines
11 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, List, Optional, cast
|
|
|
|
|
|
@dataclass
|
|
class TemperatureSensor:
|
|
current_celsius: Optional[float]
|
|
offset: Optional[float]
|
|
|
|
|
|
@dataclass
|
|
class Device:
|
|
id: int
|
|
display_name: str
|
|
temp_sensor: TemperatureSensor
|
|
raw_state: Dict[str, Any]
|
|
|
|
def __init__(self, state: Dict[str, Any]):
|
|
self.raw_state = state
|
|
id_value = state.get("id")
|
|
if isinstance(id_value, int):
|
|
self.id = id_value
|
|
else:
|
|
self.id = -1
|
|
name_value = state.get("displayName")
|
|
if isinstance(name_value, str):
|
|
self.display_name = name_value
|
|
else:
|
|
self.display_name = ""
|
|
self.temp_sensor = TemperatureSensor(
|
|
current_celsius=self._get_temp_value("currentInCelsius"),
|
|
offset=self._get_temp_value("offset"),
|
|
)
|
|
|
|
def to_web_data(self) -> Dict[str, Any]:
|
|
data: Dict[str, Any] = {
|
|
"device": self.id,
|
|
"ule_device_name": self.display_name,
|
|
}
|
|
|
|
units = self._get_units()
|
|
for unit in units:
|
|
data.update(self._unit_to_web_data(unit))
|
|
return data
|
|
|
|
def set_offset(self, offset: float) -> None:
|
|
self.temp_sensor.offset = offset
|
|
self._set_temp_value("offset", offset)
|
|
|
|
def get_offset(self) -> Optional[float]:
|
|
return self.temp_sensor.offset
|
|
|
|
def get_temperature(self) -> Optional[float]:
|
|
return self.temp_sensor.current_celsius
|
|
|
|
def _get_temp_value(self, key: str) -> Optional[float]:
|
|
sensor_skill = self._find_device_temp_skill()
|
|
if sensor_skill is None:
|
|
return None
|
|
value = sensor_skill.get(key)
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
return None
|
|
|
|
def _set_temp_value(self, key: str, value: float) -> None:
|
|
sensor_skill = self._find_device_temp_skill()
|
|
if sensor_skill is None:
|
|
return
|
|
sensor_skill[key] = value
|
|
|
|
def _find_device_temp_skill(self) -> Optional[Dict[str, Any]]:
|
|
for unit in self._as_dict_list(self.raw_state.get("units")):
|
|
if unit.get("type") != "TEMPERATURE_SENSOR":
|
|
continue
|
|
if unit.get("id") != self.id:
|
|
continue
|
|
for skill in self._as_dict_list(unit.get("skills")):
|
|
if skill.get("type") == "SmartHomeTemperatureSensor":
|
|
return skill
|
|
return None
|
|
|
|
def _get_units(self) -> List[Dict[str, Any]]:
|
|
return self._as_dict_list(self.raw_state.get("units"))
|
|
|
|
def _unit_to_web_data(self, unit: Dict[str, Any]) -> Dict[str, Any]:
|
|
data: Dict[str, Any] = {}
|
|
for skill in self._as_dict_list(unit.get("skills")):
|
|
skill_type = skill.get("type")
|
|
if skill_type == "SmartHomeThermostat":
|
|
data.update(self._thermostat_to_web_data(skill))
|
|
elif skill_type == "SmartHomeTemperatureSensor":
|
|
data.update(self._temperature_to_web_data(skill))
|
|
return data
|
|
|
|
def _temperature_to_web_data(self, skill: Dict[str, Any]) -> Dict[str, Any]:
|
|
data: Dict[str, Any] = {}
|
|
current = skill.get("currentInCelsius")
|
|
offset = skill.get("offset")
|
|
if current is not None:
|
|
data["Roomtemp"] = current
|
|
if offset is not None:
|
|
data["Offset"] = offset
|
|
return data
|
|
|
|
def _thermostat_to_web_data(self, skill: Dict[str, Any]) -> Dict[str, Any]:
|
|
data: Dict[str, Any] = {}
|
|
|
|
upper = 0.0
|
|
lower = 0.0
|
|
for preset in self._as_dict_list(skill.get("presets")):
|
|
if preset.get("name") == "UPPER_TEMPERATURE":
|
|
upper = self._as_float(preset.get("temperature"), upper)
|
|
elif preset.get("name") == "LOWER_TEMPERATURE":
|
|
lower = self._as_float(preset.get("temperature"), lower)
|
|
data.update({"Heiztemp": upper, "Absenktemp": lower})
|
|
|
|
used_temp_sensor = self._as_dict(skill.get("usedTempSensor"))
|
|
if used_temp_sensor is not None:
|
|
used_id = used_temp_sensor.get("id")
|
|
if used_id == self.id:
|
|
data.update({"ExtTempsensorID": "tochoose", "tempsensor": "own"})
|
|
else:
|
|
data.update({"ExtTempsensorID": used_id, "tempsensor": "extern"})
|
|
|
|
time_control = self._as_dict(skill.get("timeControl"))
|
|
if time_control is not None:
|
|
data.update(self._time_control_to_web_data(time_control))
|
|
|
|
drop_detection = self._as_dict(skill.get("temperatureDropDetection"))
|
|
if drop_detection is not None:
|
|
data.update(self._drop_detection_to_web_data(drop_detection))
|
|
|
|
return data
|
|
|
|
def _drop_detection_to_web_data(self, drop: Dict[str, Any]) -> Dict[str, Any]:
|
|
sensitivity = drop.get("sensitivity")
|
|
do_not_heat = drop.get("doNotHeatOffsetInMinutes")
|
|
data: Dict[str, Any] = {}
|
|
if sensitivity is not None:
|
|
data["WindowOpenTrigger"] = sensitivity + 3
|
|
if do_not_heat is not None:
|
|
data["WindowOpenTimer"] = do_not_heat
|
|
return data
|
|
|
|
def _time_control_to_web_data(self, time_control: Dict[str, Any]) -> Dict[str, Any]:
|
|
data: Dict[str, Any] = {}
|
|
for schedule in self._as_dict_list(time_control.get("timeSchedules")):
|
|
data.update(self._schedule_to_web_data(schedule))
|
|
return data
|
|
|
|
def _schedule_to_web_data(self, schedule: Dict[str, Any]) -> Dict[str, Any]:
|
|
data: Dict[str, Any] = {}
|
|
kind = schedule.get("kind")
|
|
name = schedule.get("name")
|
|
actions = self._as_dict_list(schedule.get("actions"))
|
|
|
|
if kind == "REPETITIVE" and name == "HOLIDAYS":
|
|
enabled_count = 0
|
|
for num, holiday in enumerate(actions, start=1):
|
|
time_setting = self._as_dict(holiday.get("timeSetting"))
|
|
if time_setting is None:
|
|
continue
|
|
start_date = self._as_str(time_setting.get("startDate"), "0000-00-00")
|
|
end_date = self._as_str(time_setting.get("endDate"), "0000-00-00")
|
|
start_time = self._as_str(time_setting.get("startTime"), "00:00:00")
|
|
end_time = self._as_str(time_setting.get("endTime"), "00:00:00")
|
|
_, start_month, start_day = start_date.split("-")
|
|
_, end_month, end_day = end_date.split("-")
|
|
data.update(
|
|
{
|
|
f"Holiday{num}StartDay": start_day,
|
|
f"Holiday{num}StartMonth": start_month,
|
|
f"Holiday{num}StartHour": start_time.split(":")[0],
|
|
f"Holiday{num}EndDay": end_day,
|
|
f"Holiday{num}EndMonth": end_month,
|
|
f"Holiday{num}EndHour": end_time.split(":")[0],
|
|
f"Holiday{num}Enabled": 1 if holiday.get("isEnabled") else 0,
|
|
f"Holiday{num}ID": num,
|
|
}
|
|
)
|
|
if holiday.get("isEnabled"):
|
|
enabled_count += 1
|
|
data["HolidayEnabledCount"] = enabled_count
|
|
elif kind == "REPETITIVE" and name == "SUMMER_TIME" and actions:
|
|
action = actions[0]
|
|
time_setting = self._as_dict(action.get("timeSetting"))
|
|
if time_setting is not None:
|
|
start_date = self._as_str(time_setting.get("startDate"), "0000-00-00")
|
|
end_date = self._as_str(time_setting.get("endDate"), "0000-00-00")
|
|
_, start_month, start_day = start_date.split("-")
|
|
_, end_month, end_day = end_date.split("-")
|
|
data = {
|
|
"SummerStartDay": start_day,
|
|
"SummerStartMonth": start_month,
|
|
"SummerEndDay": end_day,
|
|
"SummerEndMonth": end_month,
|
|
"SummerEnabled": 1 if schedule.get("isEnabled") else 0,
|
|
}
|
|
elif kind == "WEEKLY_TIMETABLE" and name == "TEMPERATURE":
|
|
timer_items: Dict[str, List[int]] = {}
|
|
for action in actions:
|
|
if not action.get("isEnabled"):
|
|
continue
|
|
time_setting = self._as_dict(action.get("timeSetting"))
|
|
if time_setting is None:
|
|
continue
|
|
start_time = self._as_str(time_setting.get("startTime"), "")
|
|
if not start_time:
|
|
continue
|
|
time = "".join(start_time.split(":")[:-1])
|
|
if time not in timer_items:
|
|
timer_items[time] = [0, 0]
|
|
description = self._as_dict(action.get("description"))
|
|
preset_temperature = None
|
|
if description is not None:
|
|
preset_temperature = self._as_dict(
|
|
description.get("presetTemperature")
|
|
)
|
|
heat = 0
|
|
if (
|
|
preset_temperature is not None
|
|
and preset_temperature.get("name") == "UPPER_TEMPERATURE"
|
|
):
|
|
heat = 1
|
|
day_of_week = self._as_int(time_setting.get("dayOfWeek"), 0)
|
|
timer_items[time][heat] |= day_of_week
|
|
num = 0
|
|
for key, value in timer_items.items():
|
|
if value[0] != 0:
|
|
data.update({f"timer_item_{num}": f"{key};0;{value[0]}"})
|
|
num += 1
|
|
if value[1] != 0:
|
|
data.update({f"timer_item_{num}": f"{key};1;{value[1]}"})
|
|
num += 1
|
|
|
|
return data
|
|
|
|
@staticmethod
|
|
def _as_dict_list(value: Any) -> List[Dict[str, Any]]:
|
|
result: List[Dict[str, Any]] = []
|
|
if not isinstance(value, list):
|
|
return result
|
|
items = cast(List[Any], value)
|
|
for item in items:
|
|
if isinstance(item, dict):
|
|
result.append(cast(Dict[str, Any], item))
|
|
return result
|
|
|
|
@staticmethod
|
|
def _as_dict(value: Any) -> Optional[Dict[str, Any]]:
|
|
if isinstance(value, dict):
|
|
return cast(Dict[str, Any], value)
|
|
return None
|
|
|
|
@staticmethod
|
|
def _as_str(value: Any, default: str) -> str:
|
|
if isinstance(value, str):
|
|
return value
|
|
return default
|
|
|
|
@staticmethod
|
|
def _as_int(value: Any, default: int) -> int:
|
|
if isinstance(value, int):
|
|
return value
|
|
return default
|
|
|
|
@staticmethod
|
|
def _as_float(value: Any, default: float) -> float:
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
return default
|
|
|
|
|
|
__all__ = ["Device", "TemperatureSensor"]
|