Compare commits
10 Commits
Author | SHA1 | Date | |
---|---|---|---|
5aa94a3c97 | |||
23140f672f | |||
00b161945f | |||
89fc6ea73c | |||
f6f46fba8f | |||
e01842a118 | |||
077ad19255 | |||
c36c75ac48 | |||
431450010e | |||
076af5add5 |
@ -2,8 +2,7 @@ ARG BUILD_FROM
|
||||
FROM $BUILD_FROM
|
||||
|
||||
# Install requirements for add-on
|
||||
RUN apk update && apk add --no-cache python3 py-pip
|
||||
RUN python3 -m pip install websockets requests beautifulsoup4
|
||||
RUN apk update && apk add --no-cache python3 py3-pip py3-websockets py3-requests
|
||||
|
||||
WORKDIR /data
|
||||
|
||||
|
@ -1,6 +1,6 @@
|
||||
name: "Fritz!Box Temperature Sync Dev"
|
||||
description: "Sync Fritz!DECT thermostat temperatures with other sensors in Home Assistant"
|
||||
version: "0.5.0"
|
||||
version: "0.4.3"
|
||||
startup: "application"
|
||||
stage: "stable"
|
||||
slug: "fritz_temp_sync_dev"
|
||||
@ -16,8 +16,6 @@ options:
|
||||
fritzbox:
|
||||
url: "http://fritz.box"
|
||||
password: null
|
||||
verify_ssl: true
|
||||
old_fb: false
|
||||
mappings:
|
||||
- sensor: null
|
||||
thermostate: null
|
||||
@ -27,9 +25,8 @@ schema:
|
||||
url: url
|
||||
username: "str?"
|
||||
password: str
|
||||
verify_ssl: bool
|
||||
old_fb: bool
|
||||
mappings:
|
||||
- sensor: str
|
||||
thermostate: str
|
||||
update_timeout: int
|
||||
log_level: "str?"
|
||||
|
@ -8,22 +8,18 @@ import re
|
||||
import xml.etree.ElementTree as ET
|
||||
from asyncio import Task
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional, Dict, List, Union
|
||||
from typing import Optional, Dict, List
|
||||
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
from device import Device
|
||||
|
||||
|
||||
class FritzBox:
|
||||
def __init__(self, url: str, password: str, update_timeout: int, user: str = None, dry_run: bool = False,
|
||||
verify_ssl: bool = True, old_fb: bool = False) -> None:
|
||||
def __init__(self, url: str, password: str, update_timeout: int, user: str = None, dry_run: bool = False) -> None:
|
||||
self._endpoints = {
|
||||
"login": "login_sid.lua?version=2",
|
||||
"logout": "index.lua",
|
||||
"data": "data.lua",
|
||||
"device_details": "net/home_auto_hkr_edit.lua"
|
||||
"data": "data.lua"
|
||||
}
|
||||
self.url: str = url
|
||||
self.dry_run: bool = dry_run
|
||||
@ -34,12 +30,6 @@ class FritzBox:
|
||||
self.update_timeout: int = update_timeout
|
||||
self.update_time: Dict[str, datetime] = {}
|
||||
self.hold_connection: Optional[Task] = None
|
||||
self.verify_ssl = verify_ssl
|
||||
self.old_fb = old_fb
|
||||
|
||||
if not verify_ssl:
|
||||
import urllib3
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
|
||||
async def hold_connection_alive(self) -> None:
|
||||
while True:
|
||||
@ -52,7 +42,7 @@ class FritzBox:
|
||||
|
||||
logging.debug(f"Calculate v2 challenge: {challenge}")
|
||||
chall_regex = re.compile(
|
||||
"2\$(?P<iter1>[0-9a-zA-Z]+)\$(?P<salt1>[0-9a-zA-Z]+)\$(?P<iter2>[0-9a-zA-Z]+)\$(?P<salt2>[0-9a-zA-Z]+)")
|
||||
r"2\$(?P<iter1>[0-9a-zA-Z]+)\$(?P<salt1>[0-9a-zA-Z]+)\$(?P<iter2>[0-9a-zA-Z]+)\$(?P<salt2>[0-9a-zA-Z]+)")
|
||||
|
||||
chall_parts = chall_regex.match(challenge).groupdict()
|
||||
salt1: bytes = bytes.fromhex(chall_parts["salt1"])
|
||||
@ -82,7 +72,7 @@ class FritzBox:
|
||||
"xhrId": "first",
|
||||
"noMenuRef": 1
|
||||
}
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data)
|
||||
if len(r.history) > 0:
|
||||
if not self.login():
|
||||
logging.error("Failed to login to Fritz!Box")
|
||||
@ -92,7 +82,7 @@ class FritzBox:
|
||||
def login(self) -> bool:
|
||||
logging.info(f"Login user {self.user} to Fritz!Box")
|
||||
challenge = None
|
||||
r = self.session.get(f"{self.url}/{self._endpoints['login']}", verify=self.verify_ssl)
|
||||
r = self.session.get(f"{self.url}/{self._endpoints['login']}")
|
||||
xml = ET.fromstring(r.text)
|
||||
for elem in xml:
|
||||
if elem.tag == "SID":
|
||||
@ -116,7 +106,7 @@ class FritzBox:
|
||||
"response": response
|
||||
}
|
||||
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['login']}", data=data, verify=self.verify_ssl)
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['login']}", data=data)
|
||||
logging.debug(r.text)
|
||||
xml = ET.fromstring(r.text)
|
||||
for elem in xml:
|
||||
@ -135,69 +125,12 @@ class FritzBox:
|
||||
"sid": self.sid,
|
||||
"logout": 1,
|
||||
"no_sidrenew": ""}
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['logout']}", data=data, verify=self.verify_ssl)
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['logout']}", data=data)
|
||||
if self.hold_connection is not None:
|
||||
self.hold_connection.cancel()
|
||||
|
||||
return r.status_code == 200
|
||||
|
||||
def list_devices_old(self) -> Optional[List[Union[Device, Dict]]]:
|
||||
data = {
|
||||
"xhr": 1,
|
||||
"sid": self.sid,
|
||||
"lang": "de",
|
||||
"page": "sh",
|
||||
"xhrId": "all"
|
||||
}
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
logging.debug(r.text[:100])
|
||||
if len(r.history) > 0:
|
||||
if self.login():
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
else:
|
||||
return None
|
||||
|
||||
devices_overview_raw = BeautifulSoup(r.text, "html.parser")
|
||||
table_rows = devices_overview_raw.find(id="uiSmarthomeTables").table
|
||||
|
||||
devices_raw = []
|
||||
|
||||
for r in table_rows.find_all("tr"):
|
||||
name = r.findAll("td", {'class': ['name', 'cut_overflow']})
|
||||
button = r.findAll("td", {'class': 'btncolumn'})
|
||||
temperature = r.findAll("td", {'class': 'temperature'})
|
||||
|
||||
if name is None or len(name) < 1:
|
||||
continue
|
||||
if button is None or len(button) < 1:
|
||||
continue
|
||||
if temperature is None or len(temperature) < 1:
|
||||
continue
|
||||
|
||||
name = name[0].string
|
||||
id = int(button[0].button["value"])
|
||||
temperature = float(temperature[0].string.split(" ")[0].replace(",", "."))
|
||||
|
||||
request_data = {
|
||||
"device": id,
|
||||
"sid": self.sid,
|
||||
"xhr": 1
|
||||
}
|
||||
|
||||
r = self.session.get(f"{self.url}/{self._endpoints['device_details']}", params=request_data,
|
||||
verify=self.verify_ssl)
|
||||
device_content_raw = BeautifulSoup(r.text, "html.parser")
|
||||
offset = float(device_content_raw.find("input", {"type": "hidden", "name": "Offset"})["value"])
|
||||
devices_raw.append({
|
||||
"name": name,
|
||||
"display_name": name,
|
||||
"id": id,
|
||||
"temperature": temperature,
|
||||
"offset": offset
|
||||
})
|
||||
|
||||
return devices_raw
|
||||
|
||||
def list_devices(self) -> Optional[List[Device]]:
|
||||
data = {
|
||||
"xhr": 1,
|
||||
@ -206,11 +139,11 @@ class FritzBox:
|
||||
"page": "sh_dev",
|
||||
"xhrId": "all"
|
||||
}
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data)
|
||||
logging.debug(r.text[:100])
|
||||
if len(r.history) > 0:
|
||||
if self.login():
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data)
|
||||
else:
|
||||
return None
|
||||
devices: List[Device] = []
|
||||
@ -223,16 +156,10 @@ class FritzBox:
|
||||
if idx is None and name is None:
|
||||
logging.debug("No id or name given")
|
||||
return None
|
||||
if self.old_fb:
|
||||
devices = self.list_devices_old()
|
||||
else:
|
||||
|
||||
devices = self.list_devices()
|
||||
device = None
|
||||
for device in devices:
|
||||
if isinstance(device, dict):
|
||||
if device["id"] == idx or device["display_name"] == name:
|
||||
break
|
||||
else:
|
||||
if device.id == idx or device.display_name == name:
|
||||
break
|
||||
device = None
|
||||
@ -243,26 +170,18 @@ class FritzBox:
|
||||
|
||||
return device
|
||||
|
||||
def set_offset(self, device: Union[Device, Dict]) -> None:
|
||||
def set_offset(self, device: Device) -> None:
|
||||
if self.dry_run:
|
||||
logging.warning("No updates in dry-run-mode")
|
||||
return
|
||||
|
||||
old_type = isinstance(device, dict)
|
||||
if old_type:
|
||||
device_id = device["id"]
|
||||
else:
|
||||
device_id = device.id
|
||||
|
||||
if not old_type:
|
||||
data = {
|
||||
"xhr": 1,
|
||||
"sid": self.sid,
|
||||
"lang": "de",
|
||||
"device": device_id,
|
||||
"device": device.id,
|
||||
"page": "home_auto_hkr_edit"
|
||||
}
|
||||
self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
self.session.post(f"{self.url}/{self._endpoints['data']}", data=data)
|
||||
|
||||
data = {
|
||||
"xhr": 1,
|
||||
@ -275,21 +194,7 @@ class FritzBox:
|
||||
}
|
||||
data.update(device.to_web_data())
|
||||
|
||||
else:
|
||||
data = {
|
||||
"xhr": 1,
|
||||
"sid": self.sid,
|
||||
"lang": "de",
|
||||
"no_sidrenew": "",
|
||||
"device": device_id,
|
||||
"view": "",
|
||||
"back_to_page": "/net/home_auto_overview.lua",
|
||||
"ule_device_name": device["name"],
|
||||
"Offset": device["offset"],
|
||||
"apply": "",
|
||||
"oldpage": "/net/home_auto_hkr_edit.lua",
|
||||
}
|
||||
self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
self.session.post(f"{self.url}/{self._endpoints['data']}", data=data)
|
||||
|
||||
def correct_offset(self, device_name: str, real_temp: float):
|
||||
elapsed = None
|
||||
@ -298,21 +203,11 @@ class FritzBox:
|
||||
logging.info(f"Last update for {device_name} {elapsed} ago")
|
||||
delta = timedelta(minutes=self.update_timeout)
|
||||
if device_name not in self.update_time.keys() or elapsed > delta:
|
||||
device: Optional[Union[Dict, Device]] = self.get_device_data(name=device_name)
|
||||
logging.info(f"device: {device}")
|
||||
device: Optional[Device] = self.get_device_data(name=device_name)
|
||||
if device is None:
|
||||
return
|
||||
|
||||
if self.old_fb:
|
||||
new_offset = device["offset"] + real_temp - device["temperature"]
|
||||
logging.info(f"Update offset from {device['offset']} to {new_offset}")
|
||||
else:
|
||||
new_offset = device.get_offset() + real_temp - device.get_temperature()
|
||||
logging.info(f"Update offset from {device.get_offset()} to {new_offset}")
|
||||
|
||||
if self.old_fb:
|
||||
device["offset"] = new_offset
|
||||
else:
|
||||
device.set_offset(new_offset)
|
||||
self.set_offset(device)
|
||||
self.update_time[device_name] = datetime.now()
|
||||
self.update_time[device.display_name] = datetime.now()
|
||||
|
@ -1,58 +1,73 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
from fritzbox import FritzBox
|
||||
from homeassistant import HomeAssistantAPI
|
||||
import logging
|
||||
import json
|
||||
|
||||
sensor_mappings = {}
|
||||
thermostate_mappings = {}
|
||||
|
||||
|
||||
async def handle_event(idx: int):
|
||||
logging.info(f"Wait for events for {idx}")
|
||||
global ha, fb
|
||||
logging.debug(f"Wait for events for {idx}")
|
||||
|
||||
while event := await ha.events[idx].get():
|
||||
try:
|
||||
entity_id = event["data"]["entity_id"]
|
||||
if entity_id in sensor_mappings.keys() or entity_id in thermostate_mappings.keys():
|
||||
if (
|
||||
entity_id in sensor_mappings.keys()
|
||||
or entity_id in thermostate_mappings.keys()
|
||||
):
|
||||
state = await ha.get_device_state(entity_id)
|
||||
new_state = event["data"]["new_state"]
|
||||
logging.debug(f"received changed state from {entity_id}")
|
||||
if entity_id in thermostate_mappings.keys() and state["state"] != "unavailable":
|
||||
logging.info(
|
||||
f"received changed state from {entity_id} {entity_id in thermostate_mappings.keys()} {state['state']} {entity_id in sensor_mappings.keys()}"
|
||||
)
|
||||
if (
|
||||
entity_id in thermostate_mappings.keys()
|
||||
and state["state"] != "unavailable"
|
||||
):
|
||||
therm_temp = new_state["attributes"]["current_temperature"]
|
||||
therm_name = new_state["attributes"]["friendly_name"]
|
||||
sensor = thermostate_mappings[entity_id]
|
||||
sensor_state = await ha.get_device_state(sensor)
|
||||
sensor_state_temp = sensor_state["attributes"]["temperature"] if "temperatur" in sensor_state[
|
||||
"attributes"] else sensor_state["state"]
|
||||
sensor_temp = round(float(sensor_state_temp) * 2) / 2
|
||||
sensor_temp = round(float(sensor_state["state"]) * 2) / 2
|
||||
logging.info(f"temps: {therm_temp} {sensor_temp}")
|
||||
if therm_temp != sensor_temp:
|
||||
logging.info(
|
||||
f"{therm_name}: {therm_temp}\n{sensor}: {sensor_state_temp} ({sensor_temp})")
|
||||
f"{therm_name}: {therm_temp}\n{sensor}: {sensor_state['state']} ({sensor_temp})"
|
||||
)
|
||||
fb.correct_offset(therm_name, sensor_temp)
|
||||
|
||||
elif entity_id in sensor_mappings.keys():
|
||||
new_state_temp = new_state["attributes"]["temperature"] if "temperatur" in new_state[
|
||||
"attributes"] else new_state["state"]
|
||||
sensor_temp = round(float(new_state_temp) * 2) / 2
|
||||
logging.info(sensor_temp)
|
||||
logging.info(f"here {sensor_mappings} {entity_id}")
|
||||
logging.info(f"{new_state}")
|
||||
sensor_temp = round(float(new_state["state"]) * 2) / 2
|
||||
logging.info(f"entry: {sensor_mappings[entity_id]}")
|
||||
for thermostate in sensor_mappings[entity_id]:
|
||||
logging.info(thermostate)
|
||||
therm_state = await ha.get_device_state(thermostate)
|
||||
logging.info(f"{thermostate} {therm_state}")
|
||||
if therm_state["state"] == "unavailable":
|
||||
continue
|
||||
therm_temp = float(therm_state["attributes"]["current_temperature"])
|
||||
therm_temp = float(
|
||||
therm_state["attributes"]["current_temperature"]
|
||||
)
|
||||
therm_name = therm_state["attributes"]["friendly_name"]
|
||||
if therm_temp != sensor_temp or True:
|
||||
logging.info(f"Temps: {therm_temp} {sensor_temp}")
|
||||
if therm_temp != sensor_temp:
|
||||
logging.info(
|
||||
f"{therm_name}: {therm_temp}\n{entity_id}: {new_state_temp} ({sensor_temp})")
|
||||
f"{therm_name}: {therm_temp}\n{entity_id}: {new_state['state']} ({sensor_temp})"
|
||||
)
|
||||
fb.correct_offset(therm_name, sensor_temp)
|
||||
except KeyError as e:
|
||||
logging.error(e)
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
|
||||
async def init(ha: HomeAssistantAPI, fb: FritzBox):
|
||||
@ -72,21 +87,26 @@ async def main():
|
||||
config = json.load(open(config_path))
|
||||
level = logging.INFO
|
||||
if "log_level" in config:
|
||||
print(f"Setting log_level {config['log_level']}")
|
||||
if config["log_level"] == "DEBUG":
|
||||
level = logging.DEBUG
|
||||
logging.basicConfig(level=level, format="[%(asctime)s] [%(levelname)s] %(message)s")
|
||||
logging.basicConfig(
|
||||
level=level, format="[%(asctime)s] [%(levelname)s] %(message)s"
|
||||
)
|
||||
logging.debug(config)
|
||||
|
||||
global fb
|
||||
fb = FritzBox(url=config["fritzbox"]["url"],
|
||||
fb = FritzBox(
|
||||
url=config["fritzbox"]["url"],
|
||||
user=config["fritzbox"]["username"],
|
||||
password=config["fritzbox"]["password"],
|
||||
update_timeout=config["update_timeout"],
|
||||
dry_run=False,
|
||||
verify_ssl=config["fritzbox"]["verify_ssl"],
|
||||
old_fb=config["fritzbox"]["old_fb"])
|
||||
supervisor_url = "ws://192.168.176.2:8123/api/websocket"
|
||||
supervisor_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiIzZWZhZThhYjBhNTE0MDBhYmNjNzE4Yzc3YjE5MzNkNiIsImlhdCI6MTY3MDE1MTQxMCwiZXhwIjoxOTg1NTExNDEwfQ.0_PlrqiUtEjKhsKzID7xpyQGunWlbO5cr1EPtnzO5tE"
|
||||
)
|
||||
supervisor_url = "ws://supervisor/core/websocket"
|
||||
if "SUPERVISOR_URL" in os.environ:
|
||||
supervisor_url = os.environ["SUPERVISOR_URL"]
|
||||
supervisor_token = os.environ["SUPERVISOR_TOKEN"]
|
||||
global ha
|
||||
ha = HomeAssistantAPI(supervisor_token, supervisor_url)
|
||||
|
||||
@ -95,7 +115,7 @@ async def main():
|
||||
sensor_mappings[mapping["sensor"]] = []
|
||||
sensor_mappings[mapping["sensor"]].append(mapping["thermostate"])
|
||||
thermostate_mappings[mapping["thermostate"]] = mapping["sensor"]
|
||||
|
||||
logging.debug(f"Mappings: {sensor_mappings} {thermostate_mappings}")
|
||||
try:
|
||||
await init(ha, fb)
|
||||
except KeyboardInterrupt:
|
||||
|
Loading…
x
Reference in New Issue
Block a user