Skip to content

Commit d821ff6

Browse files
mietzenmietzen
committed
Split up code in multiple classes
1 parent 030d3f5 commit d821ff6

File tree

3 files changed

+275
-277
lines changed

3 files changed

+275
-277
lines changed

‎fritz_advanced_thermostat.py‎ renamed to ‎fritz_advanced_thermostat/__init__.py‎

Lines changed: 16 additions & 277 deletions
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,6 @@
1010
Classes:
1111
FritzAdvancedThermostat: Main class for interacting with Fritz!DECT thermostats.
1212
13-
Functions:
14-
get_logger: Creates and configures a logger instance.
15-
1613
Usage:
1714
from fritz_advanced_thermostat import FritzAdvancedThermostat
1815
@@ -31,7 +28,6 @@
3128
3229
Dependencies:
3330
- requests
34-
- urllib3
3531
- packaging
3632
3733
"""
@@ -50,25 +46,8 @@
5046
import urllib3
5147
from packaging import version
5248

53-
54-
class FritzAdvancedThermostatExecutionError(Exception):
55-
"""Unknown error while executing."""
56-
57-
58-
class FritzAdvancedThermostatCompatibilityError(Exception):
59-
"""Fritz!BOX is not compatible with this module."""
60-
61-
62-
class FritzAdvancedThermostatKeyError(KeyError):
63-
"""Error while obtaining a key from the Fritz!BOX."""
64-
65-
66-
class FritzAdvancedThermostatConnectionError(ConnectionError):
67-
"""Error while connecting to the Fritz!BOX."""
68-
69-
70-
FritzAdvancedThermostatError = (FritzAdvancedThermostatExecutionError, FritzAdvancedThermostatCompatibilityError, FritzAdvancedThermostatKeyError, FritzAdvancedThermostatConnectionError)
71-
49+
from utils import FritzRequests, ThermostatDataGenerator, Logger
50+
from errors import FritzAdvancedThermostatCompatibilityError, FritzAdvancedThermostatConnectionError, FritzAdvancedThermostatExecutionError, FritzAdvancedThermostatKeyError
7251

7352
# Silence annoying urllib3 Unverified HTTPS warnings, even so if we have checked verify ssl false in requests
7453
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
@@ -77,29 +56,6 @@ class FritzAdvancedThermostatConnectionError(ConnectionError):
7756
PYTHON_VERSION = ".".join([str(x) for x in sys.version_info[0:3]])
7857

7958

80-
def get_logger(name: str, level: str = "warning") -> logging.Logger:
81-
"""Create a logger with the given name and logging level.
82-
83-
Args:
84-
name (str): Name of the logger.
85-
level (str, optional): Logging level (e.g., "INFO", "DEBUG").
86-
Defaults to "warning".
87-
88-
Returns:
89-
logging.Logger: Configured logger instance.
90-
91-
"""
92-
logger = logging.getLogger(name)
93-
logger.setLevel(level.upper())
94-
handler = logging.StreamHandler(sys.stdout)
95-
handler.setLevel(logger.level)
96-
formatter = logging.Formatter(
97-
"%(asctime)s - %(levelname)s - %(message)s", "%Y-%m-%d %H:%M:%S")
98-
handler.setFormatter(formatter)
99-
logger.addHandler(handler)
100-
return logger
101-
102-
10359
class FritzAdvancedThermostat:
10460
"""A class to manage and control Fritz!DECT thermostats connected to a Fritz!Box.
10561
@@ -143,7 +99,6 @@ def __init__(
14399
host: str,
144100
user: str,
145101
password: str,
146-
log_level: str = "warning",
147102
timeout: int = 60,
148103
retries: int = 3,
149104
ssl_verify: bool = False,
@@ -170,8 +125,7 @@ def __init__(
170125
FritzAdvancedThermostatKeyError: If trying to set a thermostat value with an unsupported key.
171126
172127
"""
173-
self._logger = get_logger(
174-
"FritzAdvancedThermostatLogger", level=log_level)
128+
self._logger = logging.getLogger("FritzAdvancedThermostatLogger")
175129

176130
if experimental:
177131
self._logger.warning("Experimental mode! All checks disabled!")
@@ -187,7 +141,6 @@ def __init__(
187141
self._experimental = experimental
188142
self._ssl_verify = ssl_verify
189143
self._timeout = timeout
190-
self._retries = retries
191144
# Set data structures
192145
self._thermostat_data = {}
193146
self._raw_device_data = {}
@@ -221,6 +174,10 @@ def __init__(
221174

222175
self._check_fritzos()
223176

177+
# Setup utils objects
178+
self._fritz_req = FritzRequests(self._prefixed_host, retries, timeout, ssl_verify)
179+
self._thermostat_data_generator = ThermostatDataGenerator(self._sid, self._fritz_req.post)
180+
224181
def _login(self, user: str, password: str) -> str:
225182
url = "/".join([self._prefixed_host, f"login_sid.lua?version=2&user={user}"])
226183
response = requests.get(
@@ -256,54 +213,14 @@ def _login(self, user: str, password: str) -> str:
256213
raise FritzAdvancedThermostatConnectionError(err)
257214
return sid
258215

259-
def _generate_headers(self, data: dict) -> dict:
260-
return {
261-
"Accept": "*/*",
262-
"Content-Type": "application/x-www-form-urlencoded",
263-
"Origin": self._prefixed_host,
264-
"Content-Length": str(len(data)),
265-
"Accept-Language": "en-GB,en;q=0.9",
266-
"Host": self._prefixed_host.split("://")[1],
267-
"User-Agent":
268-
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.2 Safari/605.1.15",
269-
"Referer": self._prefixed_host,
270-
"Accept-Encoding": "gzip, deflate",
271-
"Connection": "keep-alive",
272-
}
273-
274-
def _fritz_post_req(self, payload: dict, site: str) -> dict:
275-
url = f"{self._prefixed_host}/{site}"
276-
retries = 0
277-
while retries <= self._retries:
278-
try:
279-
response = requests.post(
280-
url,
281-
headers=self._generate_headers(payload),
282-
data=payload,
283-
verify=self._ssl_verify, timeout=self._timeout)
284-
break
285-
except ConnectionError as e:
286-
self._logger.warning("Connection Error on loading data")
287-
retries += 1
288-
if retries > self._retries:
289-
err = "Tried 3 times, got Connection Error on loading raw thermostat data"
290-
raise FritzAdvancedThermostatConnectionError(err) from e
291-
self._logger.warning("Retry %s of %s", str(
292-
retries), str(self._retries))
293-
if response.status_code != requests.codes.ok:
294-
err = "Error: " + str(response.status_code)
295-
self._logger.error(err)
296-
raise FritzAdvancedThermostatConnectionError(err)
297-
return response.text
298-
299216
def _get_fritz_os_version(self) -> str:
300217
payload = {
301218
"sid": self._sid,
302219
"xhr": "1",
303220
"page": "overview",
304221
"xhrId": "first",
305222
"noMenuRef": "1"}
306-
response = self._fritz_post_req(
223+
response = self._fritz_req.post(
307224
payload, "data.lua")
308225

309226
try:
@@ -329,7 +246,7 @@ def _load_raw_device_data(self, force_reload: bool = False) -> None:
329246
"xhrId": "all",
330247
"useajax": "1"}
331248

332-
response = self._fritz_post_req(
249+
response = self._fritz_req.post(
333250
payload, "data.lua")
334251

335252
try:
@@ -345,6 +262,11 @@ def _load_raw_device_data(self, force_reload: bool = False) -> None:
345262
raise FritzAdvancedThermostatExecutionError(err)
346263
self._raw_device_data = req_data["data"]
347264

265+
def _generate_thermostat_data(self, force_reload: bool = False) -> None:
266+
self._load_raw_device_data(force_reload)
267+
if not self._thermostat_data or force_reload:
268+
self._thermostat_data = self._thermostat_data_generator.generate(self._raw_device_data)
269+
348270
def _check_fritzos(self) -> None:
349271
if self._fritzos not in self._supported_firmware:
350272
if self._experimental:
@@ -383,189 +305,6 @@ def _set_thermostat_values(self, device_name: str, **kwargs: any) -> None:
383305
self._logger.error(err)
384306
raise FritzAdvancedThermostatKeyError(err)
385307

386-
def _generate_thermostat_data(self, force_reload: bool = False) -> None:
387-
def __get_object(device: dict, unit_name: str, skill_type: str, skill_name: str | None = None) -> any:
388-
thermostat_obj = None
389-
for unit in device["units"]:
390-
if unit["type"] == unit_name:
391-
if skill_name:
392-
for skill in unit["skills"]:
393-
if skill["type"] == skill_type:
394-
thermostat_obj = skill[skill_name]
395-
else:
396-
thermostat_obj = unit
397-
return thermostat_obj
398-
399-
def __get_schedule(schedules: list, schedule_name: str) -> dict | None:
400-
schedule = [x for x in schedules if x["name"] == schedule_name]
401-
return schedule[0] if schedule else None
402-
403-
def __get_temperature(presets: list, target: str) -> str:
404-
temp = "7.5" # Represents Off / AUS
405-
for preset in presets:
406-
if preset["name"] == target:
407-
temp = str(preset["temperature"])
408-
return temp
409-
410-
def __get_lock(locks: list, target: str) -> bool:
411-
locked = False
412-
for lock in locks:
413-
if lock["devControlName"] == target and lock["isLocked"]:
414-
locked = True
415-
return locked
416-
417-
def __get_holiday_temp(device_id: int) -> str:
418-
# I found no other way then to parse the HTML with a regex, I don't know where I can find this.
419-
payload = {
420-
"sid": self._sid,
421-
"xhr": "1",
422-
"device": device_id,
423-
"page": "home_auto_hkr_edit"}
424-
response = self._fritz_post_req(payload, "data.lua")
425-
regex = r'(?<=<input type="hidden" name="Holidaytemp" value=")\d+\.?\d?(?=" id="uiNum:Holidaytemp">)'
426-
return re.findall(regex, response)[0]
427-
428-
def __first_day_in_bitmask(bitmask: int) -> int:
429-
for i in range(7):
430-
if bitmask & (1 << i):
431-
return i
432-
return -1
433-
434-
def __generate_weekly_timers(raw_timers: dict) -> dict:
435-
"""
436-
Week Binary Conversion (reversed)
437-
Mo Tu We Th Fr Sa Su
438-
1 1 1 1 1 1 1 = 127
439-
1 0 0 0 0 1 0 = 33
440-
0 0 1 1 0 1 0 = 44
441-
442-
timer_item_x=${TIME};${STATE};${DAYS}
443-
timer_item_0= 0530 ; 1 ; 127
444-
This means turn the device on at 5:30 on all days of the week
445-
"""
446-
447-
weekly_timers = {}
448-
# day - bitmask mapping
449-
day_to_bit = {
450-
'MON': 1 << 0, # Monday -> 1
451-
'TUE': 1 << 1, # Tuesday -> 2
452-
'WED': 1 << 2, # Wednesday -> 4
453-
'THU': 1 << 3, # Thursday -> 8
454-
'FRI': 1 << 4, # Friday -> 16
455-
'SAT': 1 << 5, # Saturday -> 32
456-
'SUN': 1 << 6 # Sunday -> 64
457-
}
458-
459-
# action states mapping
460-
set_action = {
461-
'UPPER_TEMPERATURE': 1,
462-
'LOWER_TEMPERATURE': 0,
463-
'SET_OFF': 0
464-
}
465-
combined_times = {}
466-
for action in raw_timers['actions']:
467-
day = action['timeSetting']['dayOfWeek']
468-
start_time = action['timeSetting']['startTime']
469-
470-
if 'presetTemperature' in action['description']:
471-
state = action['description']['presetTemperature']['name']
472-
elif action['description']['action'] == 'SET_OFF':
473-
state = 'SET_OFF'
474-
475-
# Get bitmask and category for the action
476-
if day in day_to_bit:
477-
bitmask = day_to_bit[day]
478-
category = set_action[state]
479-
time_str = start_time.replace(':', '')[:4] # Format time to HHMM
480-
key = (time_str, category)
481-
482-
# Initialize bitmask if not present
483-
if key not in combined_times:
484-
combined_times[key] = 0
485-
486-
# Update the bitmask for the day
487-
combined_times[key] |= bitmask
488-
489-
sorted_times = sorted(combined_times.items(), key=lambda x: (__first_day_in_bitmask(x[1]), x[0][0]))
490-
491-
for i, ((time_str, category), bitmask) in enumerate(sorted_times):
492-
weekly_timers[f"timer_item_{i}"] = "{time_str};{category};{bitmask}"
493-
494-
return weekly_timers
495-
496-
def __generate_holiday_schedule(raw_holidays: dict) -> dict:
497-
holiday_schedule = {}
498-
if holidays["isEnabled"]:
499-
holiday_id_count = 0
500-
for i, holiday in enumerate(raw_holidays["actions"], 1):
501-
if holiday["isEnabled"]:
502-
holiday_id_count += 1
503-
holiday_schedule[f"Holiday{i}Enabled"] = "1"
504-
holiday_schedule[f"Holiday{holiday_id_count!s}ID"] = holiday_id_count
505-
holiday_schedule[f"Holiday{i}EndDay"] = str(int(holiday["timeSetting"]["endDate"].split("-")[2]))
506-
holiday_schedule[f"Holiday{i}EndHour"] = str(int(holiday["timeSetting"]["startTime"].split(":")[1]))
507-
holiday_schedule[f"Holiday{i}EndMonth"] = str(int(holiday["timeSetting"]["endDate"].split("-")[1]))
508-
holiday_schedule[f"Holiday{i}StartDay"] = str(int(holiday["timeSetting"]["startDate"].split("-")[2]))
509-
holiday_schedule[f"Holiday{i}StartHour"] = str(int(holiday["timeSetting"]["startTime"].split(":")[1]))
510-
holiday_schedule[f"Holiday{i}StartMonth"] = str(int(holiday["timeSetting"]["startDate"].split("-")[1]))
511-
holiday_schedule["HolidayEnabledCount"] = str(holiday_id_count - 1)
512-
holiday_schedule["Holidaytemp"] = __get_holiday_temp(device["id"])
513-
514-
return holiday_schedule
515-
516-
def __generate_summer_time_schedule(raw_summer_time: dict) -> dict:
517-
summer_time_schedule = {}
518-
if raw_summer_time["isEnabled"]:
519-
summer_time_schedule["SummerEnabled"] = "1"
520-
summer_time_schedule["SummerEndDay"] = str(int(raw_summer_time["actions"][0]["timeSetting"]["endDate"].split("-")[2]))
521-
summer_time_schedule["SummerEndMonth"] = str(int(raw_summer_time["actions"][0]["timeSetting"]["endDate"].split("-")[1]))
522-
summer_time_schedule["SummerStartDay"] = str(int(raw_summer_time["actions"][0]["timeSetting"]["startDate"].split("-")[2]))
523-
summer_time_schedule["SummerStartMonth"] = str(int(raw_summer_time["actions"][0]["timeSetting"]["startDate"].split("-")[1]))
524-
else:
525-
summer_time_schedule["SummerEnabled"] = "0"
526-
return summer_time_schedule
527-
528-
if not self._thermostat_data or force_reload:
529-
self._load_raw_device_data(force_reload)
530-
for device in self._raw_device_data["devices"]:
531-
name = device["displayName"]
532-
grouped = name in [i["displayName"] for i in [i["members"] for i in self._raw_device_data["groups"]][0]]
533-
if device["category"] == "THERMOSTAT":
534-
self._thermostat_data[name] = {}
535-
self._thermostat_data[name]["Offset"] = str(
536-
__get_object(device, "TEMPERATURE_SENSOR", "SmartHomeTemperatureSensor", "offset"))
537-
self._thermostat_data[name]["WindowOpenTimer"] = str(
538-
__get_object(device, "THERMOSTAT", "SmartHomeThermostat", "temperatureDropDetection")["doNotHeatOffsetInMinutes"])
539-
# WindowOpenTrigger musst always be + 3
540-
# xhr - json GUI
541-
# 4 (01) 1 -> niedrig
542-
# 8 (10) 5 -> mittel
543-
# 12 (11) 9 -> hoch
544-
self._thermostat_data[name]["WindowOpenTrigger"] = str(
545-
__get_object(device, "THERMOSTAT", "SmartHomeThermostat", "temperatureDropDetection")["sensitivity"] + 3)
546-
547-
locks = __get_object(device, "THERMOSTAT", "SmartHomeThermostat")["interactionControls"]
548-
self._thermostat_data[name]["locklocal"] = __get_lock(locks, "BUTTON")
549-
self._thermostat_data[name]["lockuiapp"] = __get_lock(locks, "EXTERNAL")
550-
551-
adaptiv_heating = __get_object(device, "THERMOSTAT", "SmartHomeThermostat", "adaptivHeating")
552-
self._thermostat_data[name]["hkr_adaptheat"] = adaptiv_heating['isEnabled'] and adaptiv_heating['supported']
553-
554-
if not grouped:
555-
self._thermostat_data[name]['graphState'] = "1"
556-
temperatures = __get_object(device, "THERMOSTAT", "SmartHomeThermostat", "presets")
557-
self._thermostat_data[name]["Absenktemp"] = __get_temperature(temperatures, "LOWER_TEMPERATURE")
558-
self._thermostat_data[name]["Heiztemp"] = __get_temperature(temperatures, "UPPER_TEMPERATURE")
559-
560-
summer_time = __get_schedule(__get_object(device, "THERMOSTAT", "SmartHomeThermostat", "timeControl")["timeSchedules"], "SUMMER_TIME")
561-
self._thermostat_data[name] |= __generate_summer_time_schedule(summer_time)
562-
563-
holidays = __get_schedule(__get_object(device, "THERMOSTAT", "SmartHomeThermostat", "timeControl")["timeSchedules"], "HOLIDAYS")
564-
self._thermostat_data[name] |= __generate_holiday_schedule(holidays)
565-
566-
raw_weekly_timetable = __get_schedule(__get_object(device, "THERMOSTAT", "SmartHomeThermostat", "timeControl")["timeSchedules"], "TEMPERATURE")
567-
self._thermostat_data[name] |= __generate_weekly_timers(raw_weekly_timetable)
568-
569308
def _get_device_id_by_name(self, device_name: str) -> int:
570309
self._load_raw_device_data()
571310
return [device["id"] for device in self._raw_device_data["devices"] if device["displayName"] == device_name][0]
@@ -629,7 +368,7 @@ def commit(self) -> None:
629368
site = "net/home_auto_hkr_edit.lua"
630369
payload = self._generate_data_pkg(
631370
thermostat_name, dry_run=True)
632-
dry_run_response = self._fritz_post_req(payload, site)
371+
dry_run_response = self._fritz_req.post(payload, site)
633372
try:
634373
dry_run_check = json.loads(dry_run_response)
635374
if not dry_run_check["ok"]:
@@ -650,7 +389,7 @@ def commit(self) -> None:
650389
err) from e
651390

652391
payload = self._generate_data_pkg(thermostat_name, dry_run=False)
653-
response = self._fritz_post_req(payload, "data.lua")
392+
response = self._fritz_req.post(payload, "data.lua")
654393
try:
655394
check = json.loads(response)
656395
if version.parse("7.0") < version.parse(self._fritzos) <= version.parse("7.31") and check["pid"] != "sh_dev":

0 commit comments

Comments
 (0)