Merge pull request 'old_fb' (#1) from old_fb into master
Reviewed-on: SecretMineDE/HOHA-Addons#1
This commit is contained in:
commit
76fcec8112
@ -3,7 +3,7 @@ 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
|
||||
RUN python3 -m pip install websockets requests beautifulsoup4
|
||||
|
||||
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.4.3"
|
||||
version: "0.5.0"
|
||||
startup: "application"
|
||||
stage: "stable"
|
||||
slug: "fritz_temp_sync_dev"
|
||||
@ -16,6 +16,8 @@ options:
|
||||
fritzbox:
|
||||
url: "http://fritz.box"
|
||||
password: null
|
||||
verify_ssl: true
|
||||
old_fb: false
|
||||
mappings:
|
||||
- sensor: null
|
||||
thermostate: null
|
||||
@ -25,6 +27,8 @@ schema:
|
||||
url: url
|
||||
username: "str?"
|
||||
password: str
|
||||
verify_ssl: bool
|
||||
old_fb: bool
|
||||
mappings:
|
||||
- sensor: str
|
||||
thermostate: str
|
||||
|
||||
@ -8,18 +8,22 @@ import re
|
||||
import xml.etree.ElementTree as ET
|
||||
from asyncio import Task
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional, Dict, List
|
||||
from typing import Optional, Dict, List, Union
|
||||
|
||||
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) -> None:
|
||||
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:
|
||||
self._endpoints = {
|
||||
"login": "login_sid.lua?version=2",
|
||||
"logout": "index.lua",
|
||||
"data": "data.lua"
|
||||
"data": "data.lua",
|
||||
"device_details": "net/home_auto_hkr_edit.lua"
|
||||
}
|
||||
self.url: str = url
|
||||
self.dry_run: bool = dry_run
|
||||
@ -30,6 +34,12 @@ 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:
|
||||
@ -72,7 +82,7 @@ class FritzBox:
|
||||
"xhrId": "first",
|
||||
"noMenuRef": 1
|
||||
}
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data)
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
if len(r.history) > 0:
|
||||
if not self.login():
|
||||
logging.error("Failed to login to Fritz!Box")
|
||||
@ -82,7 +92,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']}")
|
||||
r = self.session.get(f"{self.url}/{self._endpoints['login']}", verify=self.verify_ssl)
|
||||
xml = ET.fromstring(r.text)
|
||||
for elem in xml:
|
||||
if elem.tag == "SID":
|
||||
@ -106,7 +116,7 @@ class FritzBox:
|
||||
"response": response
|
||||
}
|
||||
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['login']}", data=data)
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['login']}", data=data, verify=self.verify_ssl)
|
||||
logging.debug(r.text)
|
||||
xml = ET.fromstring(r.text)
|
||||
for elem in xml:
|
||||
@ -125,12 +135,69 @@ class FritzBox:
|
||||
"sid": self.sid,
|
||||
"logout": 1,
|
||||
"no_sidrenew": ""}
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['logout']}", data=data)
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['logout']}", data=data, verify=self.verify_ssl)
|
||||
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,
|
||||
@ -139,11 +206,11 @@ class FritzBox:
|
||||
"page": "sh_dev",
|
||||
"xhrId": "all"
|
||||
}
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data)
|
||||
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)
|
||||
r = self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
else:
|
||||
return None
|
||||
devices: List[Device] = []
|
||||
@ -156,12 +223,18 @@ class FritzBox:
|
||||
if idx is None and name is None:
|
||||
logging.debug("No id or name given")
|
||||
return None
|
||||
|
||||
devices = self.list_devices()
|
||||
if self.old_fb:
|
||||
devices = self.list_devices_old()
|
||||
else:
|
||||
devices = self.list_devices()
|
||||
device = None
|
||||
for device in devices:
|
||||
if device.id == idx or device.display_name == name:
|
||||
break
|
||||
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
|
||||
|
||||
if device is None:
|
||||
@ -170,31 +243,53 @@ class FritzBox:
|
||||
|
||||
return device
|
||||
|
||||
def set_offset(self, device: Device) -> None:
|
||||
def set_offset(self, device: Union[Device, Dict]) -> None:
|
||||
if self.dry_run:
|
||||
logging.warning("No updates in dry-run-mode")
|
||||
return
|
||||
data = {
|
||||
"xhr": 1,
|
||||
"sid": self.sid,
|
||||
"lang": "de",
|
||||
"device": device.id,
|
||||
"page": "home_auto_hkr_edit"
|
||||
}
|
||||
self.session.post(f"{self.url}/{self._endpoints['data']}", data=data)
|
||||
|
||||
data = {
|
||||
"xhr": 1,
|
||||
"sid": self.sid,
|
||||
"lang": "de",
|
||||
"view": "",
|
||||
"back_to_page": "sh_dev",
|
||||
"apply": "",
|
||||
"oldpage": "/net/home_auto_hkr_edit.lua"
|
||||
}
|
||||
data.update(device.to_web_data())
|
||||
old_type = isinstance(device, dict)
|
||||
if old_type:
|
||||
device_id = device["id"]
|
||||
else:
|
||||
device_id = device.id
|
||||
|
||||
self.session.post(f"{self.url}/{self._endpoints['data']}", data=data)
|
||||
if not old_type:
|
||||
data = {
|
||||
"xhr": 1,
|
||||
"sid": self.sid,
|
||||
"lang": "de",
|
||||
"device": device_id,
|
||||
"page": "home_auto_hkr_edit"
|
||||
}
|
||||
self.session.post(f"{self.url}/{self._endpoints['data']}", data=data, verify=self.verify_ssl)
|
||||
|
||||
data = {
|
||||
"xhr": 1,
|
||||
"sid": self.sid,
|
||||
"lang": "de",
|
||||
"view": "",
|
||||
"back_to_page": "sh_dev",
|
||||
"apply": "",
|
||||
"oldpage": "/net/home_auto_hkr_edit.lua"
|
||||
}
|
||||
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)
|
||||
|
||||
def correct_offset(self, device_name: str, real_temp: float):
|
||||
elapsed = None
|
||||
@ -203,11 +298,21 @@ 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[Device] = self.get_device_data(name=device_name)
|
||||
device: Optional[Union[Dict, Device]] = self.get_device_data(name=device_name)
|
||||
logging.info(f"device: {device}")
|
||||
if device is None:
|
||||
return
|
||||
new_offset = device.get_offset() + real_temp - device.get_temperature()
|
||||
logging.info(f"Update offset from {device.get_offset()} to {new_offset}")
|
||||
device.set_offset(new_offset)
|
||||
|
||||
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.display_name] = datetime.now()
|
||||
self.update_time[device_name] = datetime.now()
|
||||
|
||||
@ -14,7 +14,7 @@ thermostate_mappings = {}
|
||||
|
||||
|
||||
async def handle_event(idx: int):
|
||||
logging.debug(f"Wait for events for {idx}")
|
||||
logging.info(f"Wait for events for {idx}")
|
||||
|
||||
while event := await ha.events[idx].get():
|
||||
try:
|
||||
@ -28,24 +28,31 @@ async def handle_event(idx: int):
|
||||
therm_name = new_state["attributes"]["friendly_name"]
|
||||
sensor = thermostate_mappings[entity_id]
|
||||
sensor_state = await ha.get_device_state(sensor)
|
||||
sensor_temp = round(float(sensor_state["attributes"]["temperature"]) * 2) / 2
|
||||
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
|
||||
if therm_temp != sensor_temp:
|
||||
logging.info(f"{therm_name}: {therm_temp}\n{sensor}: {sensor_state['attributes']['temperature']} ({sensor_temp})")
|
||||
logging.info(
|
||||
f"{therm_name}: {therm_temp}\n{sensor}: {sensor_state_temp} ({sensor_temp})")
|
||||
fb.correct_offset(therm_name, sensor_temp)
|
||||
|
||||
elif entity_id in sensor_mappings.keys():
|
||||
sensor_temp = round(float(new_state["attributes"]["temperature"]) * 2) / 2
|
||||
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)
|
||||
for thermostate in sensor_mappings[entity_id]:
|
||||
therm_state = await ha.get_device_state(thermostate)
|
||||
if therm_state["state"] == "unavailable":
|
||||
continue
|
||||
therm_temp = float(therm_state["attributes"]["current_temperature"])
|
||||
therm_name = therm_state["attributes"]["friendly_name"]
|
||||
if therm_temp != sensor_temp:
|
||||
logging.info(f"{therm_name}: {therm_temp}\n{entity_id}: {new_state['attributes']['temperature']} ({sensor_temp})")
|
||||
if therm_temp != sensor_temp or True:
|
||||
logging.info(
|
||||
f"{therm_name}: {therm_temp}\n{entity_id}: {new_state_temp} ({sensor_temp})")
|
||||
fb.correct_offset(therm_name, sensor_temp)
|
||||
except KeyError:
|
||||
pass
|
||||
except KeyError as e:
|
||||
logging.error(e)
|
||||
|
||||
|
||||
async def init(ha: HomeAssistantAPI, fb: FritzBox):
|
||||
@ -59,6 +66,7 @@ async def init(ha: HomeAssistantAPI, fb: FritzBox):
|
||||
await ha.wait_for_close()
|
||||
logging.info("Websocket closed, shutting down..")
|
||||
|
||||
|
||||
async def main():
|
||||
config_path = sys.argv[1]
|
||||
config = json.load(open(config_path))
|
||||
@ -71,12 +79,14 @@ async def main():
|
||||
|
||||
global fb
|
||||
fb = FritzBox(url=config["fritzbox"]["url"],
|
||||
user=config["fritzbox"]["username"],
|
||||
password=config["fritzbox"]["password"],
|
||||
update_timeout=config["update_timeout"],
|
||||
dry_run=False)
|
||||
supervisor_url = "ws://supervisor/core/websocket"
|
||||
supervisor_token = os.environ["SUPERVISOR_TOKEN"]
|
||||
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"
|
||||
global ha
|
||||
ha = HomeAssistantAPI(supervisor_token, supervisor_url)
|
||||
|
||||
@ -91,4 +101,5 @@ async def main():
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
|
||||
|
||||
asyncio.run(main())
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user