Python script for updating the offset of a FRITZ!Box Thermostat

fritz_thermostat_updater.py

import requests, json, urllib
from xml.etree import ElementTree

class FritzHandler:

    def __init__(self, user, pw):
        self.base_url = "http://fritz.box/"
        self.login_app = "login_sid.lua?version=2"
        self.data_app = "data.lua"
        self.update_app = "net/home_auto_hkr_edit.lua"
        self.headers = {"Content-Type": "application/x-www-form-urlencoded"}
        self.user = user
        self.pw = pw
        self.sid = None
        self.devices = None

        self.reqbase = {
                    'sid': '', 
                    'xhr': 1,
                    'device': '',
                    'page': 'home_auto_hkr_edit'}
        self.req1add = {'validate': 'apply', 
                        'useajax': 1,
                        'view': '', 
                        'back_to_page': '/smarthome/devices.lua',
                        'apply': '',
                        'lang':'en', 
                        'tempsensor': 'own', 
                        'graphState': 1}
        self.req1add.update(self.reqbase)
        pass

    def buildResponse(self, challenge, pw):
        iter1, salt1, iter2, salt2 = challenge.split('$')[1:]
        from hashlib import pbkdf2_hmac
        hash1 = pbkdf2_hmac('sha256', str.encode(pw), bytes.fromhex(salt1), int(iter1))
        return f"{salt2}${pbkdf2_hmac('sha256', hash1, bytes.fromhex(salt2), int(iter2)).hex()}"
    
    def login(self):
        r =  requests.get(self.base_url+self.login_app).text
        chal = ElementTree.fromstring(r).find("Challenge").text
        res = self.buildResponse(chal, self.pw)
        post_data_dict = {"username": self.user, "response": res}
        post_data = urllib.parse.urlencode(post_data_dict).encode()
        req = requests.post(self.base_url+self.login_app, data=post_data, headers=self.headers)
        sid = ElementTree.fromstring(req.text).find("SID").text
        self.sid = sid

    def getThermos(self):
        if not self.sid:
            self.login()
        if self.devices:
            return self.devices
        
        getThermoPara = {'sid': self.sid,'xhr': 1, 'lang': 'de', 'page': 'sh_dev', 'xhrId': 'all', 'no_sidrenew': ''}
        payload = urllib.parse.urlencode(getThermoPara).encode()
        resp = requests.post(self.base_url+self.data_app, headers=self.headers, data =  payload)
        devices = resp.json()["data"]["devices"]
        self.devices = devices
        return self.devices
    
    def getThermoParams(self, name):
        device = None
        for d in self.getThermos():
            if d["displayName"] == name:
                device = d
                break
        if not device:
            return {}

        res = {
        "device": device["id"],
        "ule_device_name": device["displayName"],
        "Heiztemp": 23,
        "Absenktemp": 15,
        "Roomtemp": 24.5,
        "Offset": -1,
        "WindowOpenTrigger": 8,
        "WindowOpenTimer": 10}
        thermos = None
        sensor = None
        for x in device["units"]:
            if x["type"] == "THERMOSTAT":
                thermos = x
            elif x["type"] =="TEMPERATURE_SENSOR":
                sensor = x
        for x in sensor["skills"]:
            res["Offset"] = x.get("offset", res["Offset"])
            res["Roomtemp"] = x.get("currentInCelsius", res["Roomtemp"])
        for x in thermos["skills"]:
            res["hkr_adaptheat"] = 1 if x.get('adaptivHeating', {}).get("isEnabled", False) else 0
            for y in x.get("presets", []):
                if y["name"] == "UPPER_TEMPERATURE":
                    res["Heiztemp"] = y["temperature"]
                elif y["name"] == "LOWER_TEMPERATURE":
                    res["Absenktemp"] = y["temperature"]
            window = x.get("temperatureDropDetection", {})
            res["WindowOpenTimer"] = window.get("doNotHeatOffsetInMinutes", res["WindowOpenTimer"])
            res["WindowOpenTrigger"] = window.get("sensitivity", res["WindowOpenTimer"])
            tcontrol = x.get("timeControl", {})
            schedules = tcontrol.get("timeSchedules", {})
            holidays = [y for y in schedules if y["name"] == "HOLIDAYS"]
            summer = [y for y in schedules if y["name"] == "SUMMER_TIME"]
            weekdays = [y for y in schedules if y["kind"] == "WEEKLY_TIMETABLE"]
            for s in summer:
                for a in s.get("actions", []):
                    start = a['timeSetting']['startDate']
                    end = a['timeSetting']['endDate']
                    en = a['isEnabled']
                    d = {"SummerStartDay": int(start.split("-")[-1]),
                            "SummerStartMonth": int(start.split("-")[-2]),
                            "SummerEndDay": int(end.split("-")[-1]),
                            "SummerEndMonth": int(end.split("-")[-2]),
                            "SummerEnabled": int(en)}
                    res.update(d)
            for h in holidays:
                for i, a in enumerate(h.get("actions", [])):
                    id = i+1
                    start = a['timeSetting']['startDate']
                    end = a['timeSetting']['endDate']
                    starth = a['timeSetting']['startTime']
                    endh = a['timeSetting']['endTime']
                    d = {"Holiday1StartDay": int(start.split("-")[-1]),
                            f"Holiday{id}StartMonth": int(start.split("-")[-2]),
                            f"Holiday{id}StartHour": int(starth.split(":")[0]),
                            f"Holiday{id}EndDay": int(end.split("-")[-1]),
                            f"Holiday{id}EndMonth": int(end.split("-")[-2]),
                            f"Holiday{id}EndHour": int(endh.split(":")[0]),
                            f"Holiday{id}Enabled": a["isEnabled"],
                            f"Holiday{id}ID": id}
                    res.update(d)
            days = {k:v for v, k in enumerate(["MON", "TUE", "WED", "THU", "FRI", "SAT", "SUN"])}
            times = {}
            for w in weekdays:
                for i, a in enumerate(w.get("actions", [])):
                    start = a['timeSetting']['startTime']
                    day = a['timeSetting']['dayOfWeek']
                    high = a["description"]["presetTemperature"]["name"] == "UPPER_TEMPERATURE"
                    timing = f'{"".join(start.split(":")[:2])};{int(high)}'
                    t = times.get(timing, 0)
                    t+= 0b1<<days[day]
                    times[timing] = t
            timerdic = {}
            for i, t in enumerate(times.items()):
                timerdic[f"timer_item_{i}"] = f"{t[0]};{str(t[1])}"
            res.update(timerdic)
        return res
    
    def updateOffset(self, devicename, temp):
        params = self.getThermoParams(devicename)
        measured_temp = params["Roomtemp"] - params["Offset"]
        new_offset = temp - measured_temp
        params["Roomtemp"] = round(temp*2)/2
        params["Offset"] = round(new_offset*2)/2
        self.req1add['sid'] = self.sid
        self.reqbase['sid'] = self.sid
        self.req1add['device'] = params['device']
        self.reqbase['device'] = params['device']
        params.update(self.req1add)
        post_data = urllib.parse.urlencode(params)
        req1 = requests.post(self.base_url+self.data_app, data=post_data, headers=self.headers)
        post_data = urllib.parse.urlencode(self.reqbase)
        req2 = requests.post(self.base_url+self.data_app, data=post_data, headers=self.headers)