HorlabsDE/HOHA-Addons
2
0
Fork 0
Code Issues 5 Pull Requests Projects Releases 3 Wiki Activity

Update for FritzOS 7.50

Browse Source
This commit is contained in:
Micha Horlboge 2022-12-03 21:51:26 +01:00
parent 52c0e90a52
commit ce59adccf0
8 changed files with 793 additions and 816 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -243,4 +243,4 @@ fabric.properties
# local config test
options.json
.DS_Store

1
fritz_temp_sync/Dockerfile
View File

@ -11,6 +11,7 @@ WORKDIR /data
COPY sync_ha_fb.py /srv
COPY fritzbox.py /srv
COPY homeassistant.py /srv
COPY device.py /srv
COPY run.sh /
RUN chmod a+x /run.sh

12
fritz_temp_sync/config.yaml
View File

@ -1,10 +1,11 @@
name: "Fritz!Box Temperature Sync"
name: "Fritz!Box Temperature Sync Dev"
description: "Sync Fritz!DECT thermostat temperatures with other sensors in Home Assistant"
version: "0.3.1"
version: "0.4.3"
startup: "application"
stage: "experimental"
slug: "fritz_temp_sync"
stage: "stable"
slug: "fritz_temp_sync_dev"
homeassistant_api: true
init: false
arch:
- aarch64
- amd64
@ -14,7 +15,6 @@ arch:
options:
fritzbox:
url: "http://fritz.box"
username: null
password: null
mappings:
- sensor: null
@ -28,4 +28,4 @@ schema:
mappings:
- sensor: str
thermostate: str
update_timeout: int
update_timeout: int

714
fritz_temp_sync/device.py Executable file
View File

@ -0,0 +1,714 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from enum import Enum, IntFlag, auto
from typing import Dict, List, Optional, Union
import typing
class WeekDay(IntFlag):
MON = 0b1
TUE = 0b10
WED = 0b100
THU = 0b1000
FRI = 0b10000
SAT = 0b100000
SUN = 0b1000000
class Manufacturer:
def __init__(self, name: str):
self.name: str = name
def __repr__(self):
return f"name: {self.name}"
def to_json(self):
return {"name": self.name}
@staticmethod
def parse_dict(manufacturer: Dict):
return Manufacturer(manufacturer["name"])
class FirmwareVersion:
def __init__(self, search: bool, current: str, update: bool, running: bool):
self.search: bool = search
self.current: str = current
self.update: bool = update
self.running: bool = running
def __repr__(self):
return f"search: {self.search}; current: {self.current}; update: {self.update}; running: {self.running}"
def to_json(self):
return {"search": self.search, "current": self.current, "update": self.update, "running": self.running}
@staticmethod
def parse_dict(firmware_version: dict):
return FirmwareVersion(firmware_version["search"],
firmware_version["current"],
firmware_version["update"],
firmware_version["running"])
class PushService:
def __init__(self, mail_address: str, unit_settings: List, is_enabled: bool):
self.mail_address: str = mail_address
self.unit_settings: List = unit_settings
self.is_enabled: bool = is_enabled
def __repr__(self):
return f"<PushService MailAddress: {self.mail_address}; UnitSettings: {self.unit_settings}; " \
f"isEnabled: {self.is_enabled}>"
def to_json(self):
return {"mailAddress": self.mail_address, "unitSettings": self.unit_settings, "isEnabled": self.is_enabled}
@staticmethod
def parse_dict(push_service: Dict):
return PushService(push_service["mailAddress"], push_service["unitSettings"], push_service["isEnabled"])
class SkillType(Enum):
SmartHomeTemperatureSensor = auto()
SmartHomeThermostat = auto()
SmartHomeBattery = auto()
class Skill(ABC):
def __init__(self, skill_type: SkillType):
self.type: SkillType = skill_type
@abstractmethod
def to_json(self):
pass
@abstractmethod
def to_web_data(self, device_id: int):
pass
@staticmethod
def parse_dict(skill: Dict):
skill_type = SkillType[skill["type"]]
if skill_type == SkillType.SmartHomeTemperatureSensor:
return TemperatureSkill.parse_dict(skill)
elif skill_type == SkillType.SmartHomeThermostat:
return ThermostatSkill.parse_dict(skill)
elif skill_type == SkillType.SmartHomeBattery:
return BatterySkill.parse_dict(skill)
else:
raise NotImplementedError(skill_type)
class PresetName(Enum):
LOWER_TEMPERATURE = auto()
UPPER_TEMPERATURE = auto()
HOLIDAY_TEMPERATURE = auto()
class Preset:
def __init__(self, name: PresetName, temperature: Optional[int]):
self.name: PresetName = name
self.temperature: Optional[int] = temperature
def __repr__(self):
return f"<Preset name: {self.name.name}; temperature: {self.temperature}>"
def to_json(self):
return {"name": self.name.name, "temperature": self.temperature}
class Description:
def __init__(self, action: str, preset_temperature: Optional[Preset]):
self.action: str = action
self.preset_temperature: Optional[Preset] = preset_temperature
def __repr__(self):
desc = f"<Description action: {self.action}"
if self.preset_temperature is not None:
desc += f"; presetTemperaturet: {self.preset_temperature}"
desc += " >"
return desc
def to_json(self):
if self.preset_temperature is not None:
return {"action": self.action, "presetTemperature": self.preset_temperature.to_json()}
return {"action": self.action}
@staticmethod
def parse_dict(description: Dict):
preset: Optional[Preset] = None
if "presetTemperature" in description.keys():
preset = Preset(PresetName[description["presetTemperature"]["name"]],
description["presetTemperature"]["temperature"] if "temperature" in description["presetTemperature"] else None)
return Description(description["action"], preset)
class Repetition(Enum):
NONE = auto()
YEARLY = auto()
class TimeSetting:
def __init__(self, start_date: Optional[str], start_time: Optional[str],
end_date: Optional[str] = None, end_time: Optional[str] = None,
repetition: Repetition = Repetition.NONE, day_of_week: Optional[WeekDay] = None):
self.start_date: Optional[str] = start_date
self.start_time: Optional[str] = start_time
self.end_date: Optional[str] = end_date
self.end_time: Optional[str] = end_time
self.repetition: Repetition = repetition
self.day_of_week: Optional[WeekDay] = day_of_week
def __repr__(self):
desc = f"<TimeSettings "
if self.day_of_week is not None:
desc += f"dayOfWeek: {self.day_of_week.name}; "
if self.start_date is not None:
desc += f"startDate: {self.start_date} "
if self.start_time is not None:
desc += f"startTime: {self.start_time} "
if self.end_date is not None:
desc += f"endDate: {self.end_date} "
if self.end_time is not None:
desc += f"endTime: {self.end_time} "
if self.repetition != Repetition.NONE:
desc += f"repetition: {self.repetition.name} "
desc += ">"
return desc
def to_json(self): # ToDo: Return typehints for all to_json
state = {}
if self.start_date is not None:
state["startDate"] = self.start_date
if self.start_time is not None:
state["startTime"] = self.start_time
if self.end_date is not None:
state["endDate"] = self.end_date
if self.end_time is not None:
state["endTime"] = self.end_time
if self.repetition != Repetition.NONE:
state["repetition"] = self.repetition.name
if self.day_of_week is not None:
state = {"dayOfWeek": self.day_of_week.name,
"time": state}
return state
@staticmethod
def parse_dict(time_setting: Dict):
start_date: Optional[str] = None
start_time: Optional[str] = None
end_date: Optional[str] = None
end_time: Optional[str] = None
day_of_week: Optional[WeekDay] = None
repetition: Repetition = Repetition.NONE
if "dayOfWeek" in time_setting.keys():
day_of_week = WeekDay[time_setting["dayOfWeek"]]
time_setting = time_setting["time"]
if "startDate" in time_setting.keys():
start_date = time_setting["startDate"]
if "startTime" in time_setting.keys():
start_time = time_setting["startTime"]
if "endDate" in time_setting.keys():
end_date = time_setting["endDate"]
if "endTime" in time_setting.keys():
end_time = time_setting["endTime"]
if "repetition" in time_setting.keys():
repetition = Repetition[time_setting["repetition"]]
return TimeSetting(start_date, start_time, end_date, end_time, repetition, day_of_week)
class Change:
def __init__(self, description: Description, time_setting: TimeSetting):
self.description: Description = description
self.time_setting: TimeSetting = time_setting
def __repr__(self):
return f"<Change description: {self.description}; timeSetting: {self.time_setting}>"
def to_json(self):
return {"description": self.description.to_json(), "timeSetting": self.time_setting.to_json()}
@staticmethod
def parse_dict(change: Dict):
return Change(Description.parse_dict(change["description"]), TimeSetting.parse_dict(change["timeSetting"]))
class TemperatureDropDetection:
def __init__(self, do_not_heat_offset_in_minutes: int, sensitivity: int):
self.do_not_heat_offset_in_minutes: int = do_not_heat_offset_in_minutes
self.sensitivity: int = sensitivity
def __repr__(self):
# ToDo
pass
def to_json(self):
return {
"doNotHeatOffsetInMinutes": self.do_not_heat_offset_in_minutes,
"sensitivity": self.sensitivity
}
def to_web_data(self, device_id: int):
return {
"WindowOpenTrigger": self.sensitivity + 3,
"WindowOpenTimer": self.do_not_heat_offset_in_minutes
}
@staticmethod
def parse_dict(drop_detection: Dict):
return TemperatureDropDetection(drop_detection["doNotHeatOffsetInMinutes"],
drop_detection["sensitivity"])
class ScheduleKind(Enum):
REPETITIVE = auto()
WEEKLY_TIMETABLE = auto()
class ThermostatSkillMode(Enum):
TARGET_TEMPERATURE = auto()
class Action:
def __init__(self, is_enabled: bool, time_setting: TimeSetting, description: Description):
self.is_enabled: bool = is_enabled
self.time_setting: TimeSetting = time_setting
self.description: Description = description
def __repr__(self):
return f"<Action isEnabled: {self.is_enabled}; timeSetting: {self.time_setting}; " \
f"description: {self.description} >"
def to_json(self):
return {"isEnabled": self.is_enabled,
"timeSetting": self.time_setting.to_json(),
"description": self.description.to_json()}
@staticmethod
def parse_dict(action: Dict):
return Action(action["isEnabled"], TimeSetting.parse_dict(action["timeSetting"]),
Description.parse_dict(action["description"]))
class Schedule:
def __init__(self, is_enabled: bool, kind: ScheduleKind, name: str, actions: List[Action]):
self.is_enabled: bool = is_enabled
self.kind: ScheduleKind = kind
self.name: str = name
self.actions: List[Action] = actions
def __repr__(self):
return f"<Schedule isEnabled: {self.is_enabled}; kind: {self.kind.name}; " \
f"name: {self.name}; actions: {self.actions}>"
def to_json(self):
return {
"isEnabled": self.is_enabled,
"kind": self.kind.name,
"name": self.name,
"actions": [action.to_json() for action in self.actions]
}
def to_web_data(self, device_id: int):
data = {}
if self.kind == ScheduleKind.REPETITIVE and self.name == "HOLIDAYS": # ToDo: Enum for names?
enabled_count = 0
for num, holiday in enumerate(self.actions):
num += 1
_, start_month, start_day = holiday.time_setting.start_date.split("-")
_, end_month, end_day = holiday.time_setting.end_date.split("-")
data.update({
f"Holiday{num}StartDay": start_day,
f"Holiday{num}StartMonth": start_month,
f"Holiday{num}StartHour": holiday.time_setting.start_time.split(":")[0],
f"Holiday{num}EndDay": end_day,
f"Holiday{num}EndMonth": end_month,
f"Holiday{num}EndHour": holiday.time_setting.end_time.split(":")[0],
f"Holiday{num}Enabled": 1 if holiday.is_enabled else 0,
f"Holiday{num}ID": num
})
if holiday.is_enabled:
enabled_count += 1
data["HolidayEnabledCount"] = enabled_count
elif self.kind == ScheduleKind.REPETITIVE and self.name == "SUMMER_TIME":
_, start_month, start_day = self.actions[0].time_setting.start_date.split("-")
_, end_month, end_day = self.actions[0].time_setting.end_date.split("-")
data = {
"SummerStartDay": start_day,
"SummerStartMonth": start_month,
"SummerEndDay": end_day,
"SummerEndMonth": end_month,
"SummerEnabled": 1 if self.is_enabled else 0
}
elif self.kind == ScheduleKind.WEEKLY_TIMETABLE and self.name == "TEMPERATURE":
timer_items = {}
for action in self.actions:
if not action.is_enabled:
continue
time = "".join(action.time_setting.start_time.split(":")[:-1])
if time not in timer_items.keys():
timer_items[time] = [0, 0]
heat = 1 if action.description.preset_temperature.name == PresetName.UPPER_TEMPERATURE else 0
days = timer_items[time][heat]
days |= action.time_setting.day_of_week
timer_items[time][heat] = days
num = 0
for key in timer_items.keys():
if timer_items[key][0] != 0:
data.update({f"timer_item_{num}": f"{key};0;{timer_items[key][0].as_integer_ratio()[0]}"})
num += 1
if timer_items[key][1] != 0:
data.update({f"timer_item_{num}": f"{key};1;{timer_items[key][1].as_integer_ratio()[0]}"})
num += 1
else:
raise NotImplementedError(self.name)
return data
@staticmethod
def parse_dict(schedule: Dict): # ToDo: Make TypedDicts for all those dicts
return Schedule(schedule["isEnabled"],
ScheduleKind[schedule["kind"]],
schedule["name"],
[Action.parse_dict(action) for action in schedule["actions"]])
class TimeControl:
def __init__(self, is_enabled: bool, time_schedules: List[Schedule]):
self.is_enabled: bool = is_enabled
self.time_schedules: List[Schedule] = time_schedules
def __repr__(self):
return f"<TimeControl isEnabled: {self.is_enabled}; timeSchedules: {self.time_schedules}"
def to_json(self):
return {
"isEnabled": self.is_enabled,
"timeSchedules": [schedule.to_json() for schedule in self.time_schedules]
}
def to_web_data(self, device_id: int):
data = {}
for schedule in self.time_schedules:
data.update(schedule.to_web_data(device_id))
return data
@staticmethod
def parse_dict(time_control: Dict):
return TimeControl(time_control["isEnabled"],
[Schedule.parse_dict(schedule) for schedule in time_control["timeSchedules"]])
class ThermostatSkill(Skill):
def __init__(self, presets: List[Preset], next_change: Optional[Change],
temperature_drop_detection: TemperatureDropDetection,
target_temp: int, time_control: TimeControl,
mode: ThermostatSkillMode, used_temp_sensor: Unit):
super().__init__(SkillType.SmartHomeThermostat)
self.presets: List[Preset] = presets
self.next_change: Optional[Change] = next_change
self.temperature_drop_detection: TemperatureDropDetection = temperature_drop_detection
self.target_temp: int = target_temp
self.time_control: TimeControl = time_control
self.mode: ThermostatSkillMode = mode
self.used_temp_sensor: Unit = used_temp_sensor
def __repr__(self):
# ToDo
pass
def to_web_data(self, device_id: int):
upper = 0.0
lower = 0.0
for preset in self.presets:
if preset.name == PresetName.UPPER_TEMPERATURE:
upper = preset.temperature
elif preset.name == PresetName.LOWER_TEMPERATURE:
lower = preset.temperature
data = {
"Heiztemp": upper,
"Absenktemp": lower
}
if device_id == self.used_temp_sensor.id:
data.update({"ExtTempsensorID": "tochoose", "tempsensor": "own"})
else:
data.update({"ExtTempsensorID": self.used_temp_sensor.id, "tempsensor": "extern"})
data.update(self.time_control.to_web_data(device_id))
data.update(self.temperature_drop_detection.to_web_data(device_id))
return data
def to_json(self):
state = {
"type": self.type.name,
"presets": [preset.to_json() for preset in self.presets],
"temperatureDropDetection": self.temperature_drop_detection.to_json(),
"targetTemp": self.target_temp,
"timeControl": self.time_control.to_json(),
"mode": self.mode.name,
"usedTempSensor": self.used_temp_sensor.to_json()
}
if self.next_change is not None:
state["nextChange"] = self.next_change.to_json()
return state
@staticmethod
def parse_dict(skill: Dict):
return ThermostatSkill(
[Preset(PresetName[preset["name"]], preset["temperature"]) for preset in skill["presets"]],
Change.parse_dict(skill["nextChange"]) if "nextChange" in skill.keys() else None,
TemperatureDropDetection.parse_dict(skill["temperatureDropDetection"]),
skill["targetTemp"],
TimeControl.parse_dict(skill["timeControl"]),
ThermostatSkillMode[skill["mode"]],
Unit.parse_dict(skill["usedTempSensor"], None)
)
class TemperatureSkill(Skill):
def __init__(self, offset: float, current_in_celsius: int):
super().__init__(SkillType.SmartHomeTemperatureSensor)
self.offset: float = offset
self.current_in_celsius: int = current_in_celsius
def __repr__(self):
return f"{self.type.name}: " \
f"offset: {self.offset}; currentInCelsius: {self.current_in_celsius}"
def to_json(self):
return {
"type": self.type.name,
"offset": self.offset,
"currentInCelsius": self.current_in_celsius
}
def to_web_data(self, device_id: int):
data = {
"Roomtemp": self.current_in_celsius,
"Offset": self.offset
}
return data
@staticmethod
def parse_dict(skill: Dict):
return TemperatureSkill(skill["offset"], skill["currentInCelsius"])
class BatterySkill(Skill):
def __init__(self, charge_level_in_percent: int):
super().__init__(SkillType.SmartHomeBattery)
self.charge_level_in_percent: int = charge_level_in_percent
def __repr__(self):
return f"{self.type.name}: chargeLevelInPercent: {self.charge_level_in_percent}"
def to_json(self):
return {"type": self.type.name, "chargeLevelInPercent": self.charge_level_in_percent}
def to_web_data(self, device_id: int):
return {}
@staticmethod
def parse_dict(skill: Dict):
return BatterySkill(skill["chargeLevelInPercent"])
class UnitTypes(Enum):
BATTERY = auto()
TEMPERATURE_SENSOR = auto()
THERMOSTAT = auto()
class Unit:
def __init__(self, unit_type: UnitTypes, idx: int, display_name: str,
device: Optional[Device], skills: List[Union[Dict, Skill]],
interaction_controls: Optional[List] = None):
self.type: UnitTypes = unit_type
self.id: int = idx
self.display_name: str = display_name
self.device: Optional[Device] = device
self.skills: List[Skill] = skills
self.interaction_controls: Optional[List] = interaction_controls # ToDo: Type for this
def __repr__(self):
desc = f"type: {self.type.name}; id: {self.id}; displayName: {self.display_name};"
if self.device is not None:
desc += f"device: {self.device.short_description()}; "
desc += f"skills: {self.skills}; "
if self.interaction_controls is not None:
desc += f"interactionControls: {self.interaction_controls} ;"
return desc
def to_json(self):
state = {
"type": self.type.name,
"id": self.id,
"displayName": self.display_name,
"skills": [skill.to_json() for skill in self.skills]
}
if self.device is not None:
state["device"] = self.device.to_short_json()
if self.interaction_controls is not None:
state["interactionControls"] = self.interaction_controls
return state
def to_web_data(self, device_id: int):
data = {}
for skill in self.skills:
data.update(skill.to_web_data(device_id))
return data
@staticmethod
def parse_dict(unit: Dict, device: Optional[Device]):
return Unit(UnitTypes[unit["type"]],
unit["id"],
unit["displayName"],
device,
[Skill.parse_dict(skill) for skill in unit["skills"]],
unit["interactionControls"] if "interactionControls" in unit.keys() else None)
class DeviceType(Enum):
SmartHomeDevice = auto()
class DeviceCategory(Enum):
THERMOSTAT = auto()
class Device:
def __init__(self, state):
self.type: Optional[DeviceType] = None
self.is_deletable: bool = False
self.id: int = -1
self.master_connection_state = None # ToDo
self.display_name: Optional[str] = None
self.category: Optional[str] = None
self.units = None # ToDo
self.firmware_version: Optional[FirmwareVersion] = None
self.model: Optional[str] = None
self.is_editable: bool = False
self.manufacturer: Optional[Manufacturer] = None
self.push_service: Optional[PushService] = None
self.actor_identification_number: Optional[str] = None
if state is not None:
self.parse_state(state)
def __repr__(self):
# ToDo
pass
def short_description(self):
desc = f"masterConnectionState: {self.master_connection_state}; " \
f"type: {self.type}; model: {self.model}; id: {self.id}; " \
f"manufacturer: {self.manufacturer}; " \
f"actorIdentificationNumber: {self.actor_identification_number}; " \
f"displayName: {self.display_name}"
return desc
def parse_state(self, state):
self.type = DeviceType[state["type"]]
self.is_deletable = state["isDeletable"]
self.id = state["id"]
self.master_connection_state = state["masterConnectionState"]
self.display_name = state["displayName"]
self.category = state["category"]
self.units = [Unit.parse_dict(unit, self) for unit in state["units"]]
self.firmware_version = FirmwareVersion.parse_dict(state["firmwareVersion"])
self.model = state["model"]
self.is_editable = state["isEditable"]
self.manufacturer = Manufacturer.parse_dict(state["manufacturer"])
self.push_service = PushService.parse_dict(state["pushService"])
self.actor_identification_number = state["actorIdentificationNumber"]
def to_web_data(self):
data = {
"device": self.id,
"ule_device_name": self.display_name
}
for unit in self.units:
data.update(unit.to_web_data(self.id))
return data
def to_json(self):
state = {"type": self.type.name,
"isDeletable": self.is_deletable,
"id": self.id,
"masterConnectionState": self.master_connection_state,
"displayName": self.display_name,
"category": self.category,
"units": [unit.to_json() for unit in self.units],
"firmwareVersion": self.firmware_version.to_json(),
"model": self.model,
"isEditable": self.is_editable,
"manufacturer": self.manufacturer.to_json(),
"pushService": self.push_service.to_json(),
"actorIdentificationNumber": self.actor_identification_number}
return state
def set_offset(self, offset: float):
temp_sensor: Optional[Unit] = None
for unit in self.units:
if unit.type == UnitTypes.TEMPERATURE_SENSOR:
temp_sensor = unit
break
temp_skill: Optional[TemperatureSkill] = None
for skill in temp_sensor.skills:
if skill.type == SkillType.SmartHomeTemperatureSensor:
temp_skill = typing.cast(TemperatureSkill, skill)
break
temp_skill.offset = offset
def get_offset(self):
temp_sensor: Optional[Unit] = None
for unit in self.units:
if unit.type == UnitTypes.TEMPERATURE_SENSOR:
temp_sensor = unit
break
temp_skill: Optional[TemperatureSkill] = None
for skill in temp_sensor.skills:
if skill.type == SkillType.SmartHomeTemperatureSensor:
temp_skill = typing.cast(TemperatureSkill, skill)
break
return temp_skill.offset
def get_temperature(self):
temp_sensor: Optional[Unit] = None
for unit in self.units:
if unit.type == UnitTypes.TEMPERATURE_SENSOR:
temp_sensor = unit
break
temp_skill: Optional[TemperatureSkill] = None
for skill in temp_sensor.skills:
if skill.type == SkillType.SmartHomeTemperatureSensor:
temp_skill = typing.cast(TemperatureSkill, skill)
break
return temp_skill.current_in_celsius
def to_short_json(self):
return {
"masterConnectionState": self.master_connection_state,
"type": self.type.name,
"model": self.model,
"id": self.id,
"manufacturer": self.manufacturer.to_json(),
"actorIdentificationNumber": self.actor_identification_number,
"displayName": self.display_name
}

732
fritz_temp_sync/fritzbox.py
View File

@ -5,737 +5,14 @@ import hashlib
import json
import logging
import re
import typing
import xml.etree.ElementTree as ET
from abc import ABC, abstractmethod
from asyncio import Task
from datetime import datetime, timedelta
from enum import IntFlag, Enum, auto
from typing import Optional, Tuple, Dict, List, Union, TypedDict
from urllib.parse import urlencode
from typing import Optional, Dict, List
import requests
class WeekDay(IntFlag):
MON = 0b1
TUE = 0b10
WED = 0b100
THU = 0b1000
FRI = 0b10000
SAT = 0b100000
SUN = 0b1000000
class Manufacturer:
def __init__(self, name: str):
self.name: str = name
def __repr__(self):
return f"name: {self.name}"
def to_json(self):
return {"name": self.name}
@staticmethod
def parse_dict(manufacturer: Dict):
return Manufacturer(manufacturer["name"])
class FirmwareVersion:
def __init__(self, search: bool, current: str, update: bool, running: bool):
self.search: bool = search
self.current: str = current
self.update: bool = update
self.running: bool = running
def __repr__(self):
return f"search: {self.search}; current: {self.current}; update: {self.update}; running: {self.running}"
def to_json(self):
return {"search": self.search, "current": self.current, "update": self.update, "running": self.running}
@staticmethod
def parse_dict(firmware_version: dict):
return FirmwareVersion(firmware_version["search"],
firmware_version["current"],
firmware_version["update"],
firmware_version["running"])
class PushService:
def __init__(self, mail_address: str, unit_settings: List, is_enabled: bool):
self.mail_address: str = mail_address
self.unit_settings: List = unit_settings
self.is_enabled: bool = is_enabled
def __repr__(self):
return f"<PushService MailAddress: {self.mail_address}; UnitSettings: {self.unit_settings}; " \
f"isEnabled: {self.is_enabled}>"
def to_json(self):
return {"mailAddress": self.mail_address, "unitSettings": self.unit_settings, "isEnabled": self.is_enabled}
@staticmethod
def parse_dict(push_service: Dict):
return PushService(push_service["mailAddress"], push_service["unitSettings"], push_service["isEnabled"])
class SkillType(Enum):
SmartHomeTemperatureSensor = auto()
SmartHomeThermostat = auto()
SmartHomeBattery = auto()
class Skill(ABC):
def __init__(self, skill_type: SkillType):
self.type: SkillType = skill_type
@abstractmethod
def to_json(self):
pass
@abstractmethod
def to_web_data(self, device_id: int):
pass
@staticmethod
def parse_dict(skill: Dict):
skill_type = SkillType[skill["type"]]
if skill_type == SkillType.SmartHomeTemperatureSensor:
return TemperatureSkill.parse_dict(skill)
elif skill_type == SkillType.SmartHomeThermostat:
return ThermostatSkill.parse_dict(skill)
elif skill_type == SkillType.SmartHomeBattery:
return BatterySkill.parse_dict(skill)
else:
raise NotImplementedError(skill_type)
class PresetName(Enum):
LOWER_TEMPERATURE = auto()
UPPER_TEMPERATURE = auto()
HOLIDAY_TEMPERATURE = auto()
class Preset:
def __init__(self, name: PresetName, temperature: int):
self.name: PresetName = name
self.temperature: int = temperature
def __repr__(self):
return f"<Preset name: {self.name.name}; temperature: {self.temperature}>"
def to_json(self):
return {"name": self.name.name, "temperature": self.temperature}
class Description:
def __init__(self, action: str, preset_temperature: Optional[Preset]):
self.action: str = action
self.preset_temperature: Optional[Preset] = preset_temperature
def __repr__(self):
desc = f"<Description action: {self.action}"
if self.preset_temperature is not None:
desc += f"; presetTemperaturet: {self.preset_temperature}"
desc += " >"
return desc
def to_json(self):
if self.preset_temperature is not None:
return {"action": self.action, "presetTemperature": self.preset_temperature.to_json()}
return {"action": self.action}
@staticmethod
def parse_dict(description: Dict):
preset: Optional[Preset] = None
if "presetTemperature" in description.keys():
preset = Preset(PresetName[description["presetTemperature"]["name"]],
description["presetTemperature"]["temperature"])
return Description(description["action"], preset)
class Repetition(Enum):
NONE = auto()
YEARLY = auto()
class TimeSetting:
def __init__(self, start_date: Optional[str], start_time: Optional[str],
end_date: Optional[str] = None, end_time: Optional[str] = None,
repetition: Repetition = Repetition.NONE, day_of_week: Optional[WeekDay] = None):
self.start_date: Optional[str] = start_date
self.start_time: Optional[str] = start_time
self.end_date: Optional[str] = end_date
self.end_time: Optional[str] = end_time
self.repetition: Repetition = repetition
self.day_of_week: Optional[WeekDay] = day_of_week
def __repr__(self):
desc = f"<TimeSettings "
if self.day_of_week is not None:
desc += f"dayOfWeek: {self.day_of_week.name}; "
if self.start_date is not None:
desc += f"startDate: {self.start_date} "
if self.start_time is not None:
desc += f"startTime: {self.start_time} "
if self.end_date is not None:
desc += f"endDate: {self.end_date} "
if self.end_time is not None:
desc += f"endTime: {self.end_time} "
if self.repetition != Repetition.NONE:
desc += f"repetition: {self.repetition.name} "
desc += ">"
return desc
def to_json(self): # ToDo: Return typehints for all to_json
state = {}
if self.start_date is not None:
state["startDate"] = self.start_date
if self.start_time is not None:
state["startTime"] = self.start_time
if self.end_date is not None:
state["endDate"] = self.end_date
if self.end_time is not None:
state["endTime"] = self.end_time
if self.repetition != Repetition.NONE:
state["repetition"] = self.repetition.name
if self.day_of_week is not None:
state = {"dayOfWeek": self.day_of_week.name,
"time": state}
return state
@staticmethod
def parse_dict(time_setting: Dict):
start_date: Optional[str] = None
start_time: Optional[str] = None
end_date: Optional[str] = None
end_time: Optional[str] = None
day_of_week: Optional[WeekDay] = None
repetition: Repetition = Repetition.NONE
if "dayOfWeek" in time_setting.keys():
day_of_week = WeekDay[time_setting["dayOfWeek"]]
time_setting = time_setting["time"]
if "startDate" in time_setting.keys():
start_date = time_setting["startDate"]
if "startTime" in time_setting.keys():
start_time = time_setting["startTime"]
if "endDate" in time_setting.keys():
end_date = time_setting["endDate"]
if "endTime" in time_setting.keys():
end_time = time_setting["endTime"]
if "repetition" in time_setting.keys():
repetition = Repetition[time_setting["repetition"]]
return TimeSetting(start_date, start_time, end_date, end_time, repetition, day_of_week)
class Change:
def __init__(self, description: Description, time_setting: TimeSetting):
self.description: Description = description
self.time_setting: TimeSetting = time_setting
def __repr__(self):
return f"<Change description: {self.description}; timeSetting: {self.time_setting}>"
def to_json(self):
return {"description": self.description.to_json(), "timeSetting": self.time_setting.to_json()}
@staticmethod
def parse_dict(change: Dict):
return Change(Description.parse_dict(change["description"]), TimeSetting.parse_dict(change["timeSetting"]))
class TemperatureDropDetection:
def __init__(self, do_not_heat_offset_in_minutes: int, sensitivity: int, is_window_open: bool):
self.do_not_heat_offset_in_minutes: int = do_not_heat_offset_in_minutes
self.sensitivity: int = sensitivity
self.is_window_open: bool = is_window_open
def __repr__(self):
# ToDo
pass
def to_json(self):
return {
"doNotHeatOffsetInMinutes": self.do_not_heat_offset_in_minutes,
"sensitivity": self.sensitivity,
"isWindowOpen": self.is_window_open
}
def to_web_data(self, device_id: int):
return {
"WindowOpenTrigger": self.sensitivity + 3,
"WindowOpenTimer": self.do_not_heat_offset_in_minutes
}
@staticmethod
def parse_dict(drop_detection: Dict):
return TemperatureDropDetection(drop_detection["doNotHeatOffsetInMinutes"],
drop_detection["sensitivity"],
drop_detection["isWindowOpen"])
class ScheduleKind(Enum):
REPETITIVE = auto()
WEEKLY_TIMETABLE = auto()
class ThermostatSkillMode(Enum):
TARGET_TEMPERATURE = auto()
class Action:
def __init__(self, is_enabled: bool, time_setting: TimeSetting, description: Description):
self.is_enabled: bool = is_enabled
self.time_setting: TimeSetting = time_setting
self.description: Description = description
def __repr__(self):
return f"<Action isEnabled: {self.is_enabled}; timeSetting: {self.time_setting}; " \
f"description: {self.description} >"
def to_json(self):
return {"isEnabled": self.is_enabled,
"timeSetting": self.time_setting.to_json(),
"description": self.description.to_json()}
@staticmethod
def parse_dict(action: Dict):
return Action(action["isEnabled"], TimeSetting.parse_dict(action["timeSetting"]),
Description.parse_dict(action["description"]))
class Schedule:
def __init__(self, is_enabled: bool, kind: ScheduleKind, name: str, actions: List[Action]):
self.is_enabled: bool = is_enabled
self.kind: ScheduleKind = kind
self.name: str = name
self.actions: List[Action] = actions
def __repr__(self):
return f"<Schedule isEnabled: {self.is_enabled}; kind: {self.kind.name}; " \
f"name: {self.name}; actions: {self.actions}>"
def to_json(self):
return {
"isEnabled": self.is_enabled,
"kind": self.kind.name,
"name": self.name,
"actions": [action.to_json() for action in self.actions]
}
def to_web_data(self, device_id: int):
data = {}
if self.kind == ScheduleKind.REPETITIVE and self.name == "HOLIDAYS": # ToDo: Enum for names?
enabled_count = 0
for num, holiday in enumerate(self.actions):
num += 1
_, start_month, start_day = holiday.time_setting.start_date.split("-")
_, end_month, end_day = holiday.time_setting.end_date.split("-")
data.update({
f"Holiday{num}StartDay": start_day,
f"Holiday{num}StartMonth": start_month,
f"Holiday{num}StartHour": holiday.time_setting.start_time.split(":")[0],
f"Holiday{num}EndDay": end_day,
f"Holiday{num}EndMonth": end_month,
f"Holiday{num}EndHour": holiday.time_setting.end_time.split(":")[0],
f"Holiday{num}Enabled": 1 if holiday.is_enabled else 0,
f"Holiday{num}ID": num
})
if holiday.is_enabled:
enabled_count += 1
data["HolidayEnabledCount"] = enabled_count
elif self.kind == ScheduleKind.REPETITIVE and self.name == "SUMMER_TIME":
_, start_month, start_day = self.actions[0].time_setting.start_date.split("-")
_, end_month, end_day = self.actions[0].time_setting.end_date.split("-")
data = {
"SummerStartDay": start_day,
"SummerStartMonth": start_month,
"SummerEndDay": end_day,
"SummerEndMonth": end_month,
"SummerEnabled": 1 if self.is_enabled else 0
}
elif self.kind == ScheduleKind.WEEKLY_TIMETABLE and self.name == "TEMPERATURE":
timer_items = {}
for action in self.actions:
if not action.is_enabled:
continue
time = "".join(action.time_setting.start_time.split(":")[:-1])
if time not in timer_items.keys():
timer_items[time] = [0, 0]
heat = 1 if action.description.preset_temperature.name == PresetName.UPPER_TEMPERATURE else 0
days = timer_items[time][heat]
days |= action.time_setting.day_of_week
timer_items[time][heat] = days
num = 0
for key in timer_items.keys():
if timer_items[key][0] != 0:
data.update({f"timer_item_{num}": f"{key};0;{timer_items[key][0].as_integer_ratio()[0]}"})
num += 1
if timer_items[key][1] != 0:
data.update({f"timer_item_{num}": f"{key};1;{timer_items[key][1].as_integer_ratio()[0]}"})
num += 1
else:
raise NotImplementedError(self.name)
return data
@staticmethod
def parse_dict(schedule: Dict): # ToDo: Make TypedDicts for all those dicts
return Schedule(schedule["isEnabled"],
ScheduleKind[schedule["kind"]],
schedule["name"],
[Action.parse_dict(action) for action in schedule["actions"]])
class TimeControl:
def __init__(self, is_enabled: bool, time_schedules: List[Schedule]):
self.is_enabled: bool = is_enabled
self.time_schedules: List[Schedule] = time_schedules
def __repr__(self):
return f"<TimeControl isEnabled: {self.is_enabled}; timeSchedules: {self.time_schedules}"
def to_json(self):
return {
"isEnabled": self.is_enabled,
"timeSchedules": [schedule.to_json() for schedule in self.time_schedules]
}
def to_web_data(self, device_id: int):
data = {}
for schedule in self.time_schedules:
data.update(schedule.to_web_data(device_id))
return data
@staticmethod
def parse_dict(time_control: Dict):
return TimeControl(time_control["isEnabled"],
[Schedule.parse_dict(schedule) for schedule in time_control["timeSchedules"]])
class ThermostatSkill(Skill):
def __init__(self, presets: List[Preset], next_change: Optional[Change],
temperature_drop_detection: TemperatureDropDetection,
target_temp: int, time_control: TimeControl, holiday_active: bool,
mode: ThermostatSkillMode, summer_active: bool, used_temp_sensor: Unit):
super().__init__(SkillType.SmartHomeThermostat)
self.presets: List[Preset] = presets
self.next_change: Optional[Change] = next_change
self.temperature_drop_detection: TemperatureDropDetection = temperature_drop_detection
self.target_temp: int = target_temp
self.time_control: TimeControl = time_control
self.holiday_active: bool = holiday_active
self.mode: ThermostatSkillMode = mode
self.summer_active: bool = summer_active
self.used_temp_sensor: Unit = used_temp_sensor
def __repr__(self):
# ToDo
pass
def to_web_data(self, device_id: int):
upper = 0.0
lower = 0.0
for preset in self.presets:
if preset.name == PresetName.UPPER_TEMPERATURE:
upper = preset.temperature
elif preset.name == PresetName.LOWER_TEMPERATURE:
lower = preset.temperature
data = {
"Heiztemp": upper,
"Absenktemp": lower
}
if device_id == self.used_temp_sensor.id:
data.update({"ExtTempsensorID": "tochoose", "tempsensor": "own"})
else:
data.update({"ExtTempsensorID": self.used_temp_sensor.id, "tempsensor": "extern"})
data.update(self.time_control.to_web_data(device_id))
data.update(self.temperature_drop_detection.to_web_data(device_id))
return data
def to_json(self):
state = {
"type": self.type.name,
"presets": [preset.to_json() for preset in self.presets],
"temperatureDropDetection": self.temperature_drop_detection.to_json(),
"targetTemp": self.target_temp,
"timeControl": self.time_control.to_json(),
"holidayActive": self.holiday_active,
"mode": self.mode.name,
"summerActive": self.summer_active,
"usedTempSensor": self.used_temp_sensor.to_json()
}
if self.next_change is not None:
state["nextChange"] = self.next_change.to_json()
return state
@staticmethod
def parse_dict(skill: Dict):
return ThermostatSkill(
[Preset(PresetName[preset["name"]], preset["temperature"]) for preset in skill["presets"]],
Change.parse_dict(skill["nextChange"]) if "nextChange" in skill.keys() else None,
TemperatureDropDetection.parse_dict(skill["temperatureDropDetection"]),
skill["targetTemp"],
TimeControl.parse_dict(skill["timeControl"]),
skill["holidayActive"],
ThermostatSkillMode[skill["mode"]],
skill["summerActive"],
Unit.parse_dict(skill["usedTempSensor"], None)
)
class TemperatureSkill(Skill):
def __init__(self, offset: float, current_in_celsius: int):
super().__init__(SkillType.SmartHomeTemperatureSensor)
self.offset: float = offset
self.current_in_celsius: int = current_in_celsius
def __repr__(self):
return f"{self.type.name}: " \
f"offset: {self.offset}; currentInCelsius: {self.current_in_celsius}"
def to_json(self):
return {
"type": self.type.name,
"offset": self.offset,
"currentInCelsius": self.current_in_celsius
}
def to_web_data(self, device_id: int):
data = {
"Roomtemp": self.current_in_celsius,
"Offset": self.offset
}
return data
@staticmethod
def parse_dict(skill: Dict):
return TemperatureSkill(skill["offset"], skill["currentInCelsius"])
class BatterySkill(Skill):
def __init__(self, charge_level_in_percent: int):
super().__init__(SkillType.SmartHomeBattery)
self.charge_level_in_percent: int = charge_level_in_percent
def __repr__(self):
return f"{self.type.name}: chargeLevelInPercent: {self.charge_level_in_percent}"
def to_json(self):
return {"type": self.type.name, "chargeLevelInPercent": self.charge_level_in_percent}
def to_web_data(self, device_id: int):
return {}
@staticmethod
def parse_dict(skill: Dict):
return BatterySkill(skill["chargeLevelInPercent"])
class UnitTypes(Enum):
BATTERY = auto()
TEMPERATURE_SENSOR = auto()
THERMOSTAT = auto()
class Unit:
def __init__(self, unit_type: UnitTypes, idx: int, display_name: str,
device: Optional[Device], skills: List[Union[Dict, Skill]],
actor_identification_number: str, interaction_controls: Optional[List] = None):
self.type: UnitTypes = unit_type
self.id: int = idx
self.display_name: str = display_name
self.device: Optional[Device] = device
self.skills: List[Skill] = skills
self.actor_identification_number: str = actor_identification_number
self.interaction_controls: Optional[List] = interaction_controls # ToDo: Type for this
def __repr__(self):
desc = f"type: {self.type.name}; id: {self.id}; displayName: {self.display_name};"
if self.device is not None:
desc += f"device: {self.device.short_description()}; "
desc += f"skills: {self.skills}; "
if self.interaction_controls is not None:
desc += f"interactionControls: {self.interaction_controls} ;"
desc += f"actorIdentificationNumber: {self.actor_identification_number}"
return desc
def to_json(self):
state = {
"type": self.type.name,
"id": self.id,
"displayName": self.display_name,
"skills": [skill.to_json() for skill in self.skills],
"actorIdentificationNumber": self.actor_identification_number
}
if self.device is not None:
state["device"] = self.device.to_short_json()
if self.interaction_controls is not None:
state["interactionControls"] = self.interaction_controls
return state
def to_web_data(self, device_id: int):
data = {}
for skill in self.skills:
data.update(skill.to_web_data(device_id))
return data
@staticmethod
def parse_dict(unit: Dict, device: Optional[Device]):
return Unit(UnitTypes[unit["type"]],
unit["id"],
unit["displayName"],
device,
[Skill.parse_dict(skill) for skill in unit["skills"]],
unit["actorIdentificationNumber"],
unit["interactionControls"] if "interactionControls" in unit.keys() else None)
class DeviceType(Enum):
SmartHomeDevice = auto()
class DeviceCategory(Enum):
THERMOSTAT = auto()
class Device:
def __init__(self, state):
self.type: Optional[DeviceType] = None
self.is_deletable: bool = False
self.id: int = -1
self.master_connection_state = None # ToDo
self.display_name: Optional[str] = None
self.category: Optional[str] = None
self.units = None # ToDo
self.firmware_version: Optional[FirmwareVersion] = None
self.model: Optional[str] = None
self.is_editable: bool = False
self.manufacturer: Optional[Manufacturer] = None
self.push_service: Optional[PushService] = None
self.actor_identification_number: Optional[str] = None
if state is not None:
self.parse_state(state)
def __repr__(self):
# ToDo
pass
def short_description(self):
desc = f"masterConnectionState: {self.master_connection_state}; " \
f"type: {self.type}; model: {self.model}; id: {self.id}; " \
f"manufacturer: {self.manufacturer}; " \
f"actorIdentificationNumber: {self.actor_identification_number}; " \
f"displayName: {self.display_name}"
return desc
def parse_state(self, state):
self.type = DeviceType[state["type"]]
self.is_deletable = state["isDeletable"]
self.id = state["id"]
self.master_connection_state = state["masterConnectionState"]
self.display_name = state["displayName"]
self.category = state["category"]
self.units = [Unit.parse_dict(unit, self) for unit in state["units"]]
self.firmware_version = FirmwareVersion.parse_dict(state["firmwareVersion"])
self.model = state["model"]
self.is_editable = state["isEditable"]
self.manufacturer = Manufacturer.parse_dict(state["manufacturer"])
self.push_service = PushService.parse_dict(state["pushService"])
self.actor_identification_number = state["actorIdentificationNumber"]
def to_web_data(self):
data = {
"device": self.id,
"ule_device_name": self.display_name
}
for unit in self.units:
data.update(unit.to_web_data(self.id))
return data
def to_json(self):
state = {"type": self.type.name,
"isDeletable": self.is_deletable,
"id": self.id,
"masterConnectionState": self.master_connection_state,
"displayName": self.display_name,
"category": self.category,
"units": [unit.to_json() for unit in self.units],
"firmwareVersion": self.firmware_version.to_json(),
"model": self.model,
"isEditable": self.is_editable,
"manufacturer": self.manufacturer.to_json(),
"pushService": self.push_service.to_json(),
"actorIdentificationNumber": self.actor_identification_number}
return state
def set_offset(self, offset: float):
temp_sensor: Optional[Unit] = None
for unit in self.units:
if unit.type == UnitTypes.TEMPERATURE_SENSOR:
temp_sensor = unit
break
temp_skill: Optional[TemperatureSkill] = None
for skill in temp_sensor.skills:
if skill.type == SkillType.SmartHomeTemperatureSensor:
temp_skill = typing.cast(TemperatureSkill, skill)
break
temp_skill.offset = offset
def get_offset(self):
temp_sensor: Optional[Unit] = None
for unit in self.units:
if unit.type == UnitTypes.TEMPERATURE_SENSOR:
temp_sensor = unit
break
temp_skill: Optional[TemperatureSkill] = None
for skill in temp_sensor.skills:
if skill.type == SkillType.SmartHomeTemperatureSensor:
temp_skill = typing.cast(TemperatureSkill, skill)
break
return temp_skill.offset
def get_temperature(self):
temp_sensor: Optional[Unit] = None
for unit in self.units:
if unit.type == UnitTypes.TEMPERATURE_SENSOR:
temp_sensor = unit
break
temp_skill: Optional[TemperatureSkill] = None
for skill in temp_sensor.skills:
if skill.type == SkillType.SmartHomeTemperatureSensor:
temp_skill = typing.cast(TemperatureSkill, skill)
break
return temp_skill.current_in_celsius
def to_short_json(self):
return {
"masterConnectionState": self.master_connection_state,
"type": self.type.name,
"model": self.model,
"id": self.id,
"manufacturer": self.manufacturer.to_json(),
"actorIdentificationNumber": self.actor_identification_number,
"displayName": self.display_name
}
from device import Device
class FritzBox:
def __init__(self, url: str, password: str, update_timeout: int, user: str = None, dry_run: bool = False) -> None:
@ -757,6 +34,7 @@ class FritzBox:
async def hold_connection_alive(self) -> None:
while True:
# Session automatically destroyed after 20m of inactivity
# according to the manual
await asyncio.sleep(19 * 60)
self.check_session()
@ -799,7 +77,7 @@ class FritzBox:
if not self.login():
logging.error("Failed to login to Fritz!Box")
else:
logging.info("Already logged in")
logging.debug("Already logged in")
def login(self) -> bool:
logging.info(f"Login user {self.user} to Fritz!Box")
@ -814,7 +92,7 @@ class FritzBox:
elif self.user is None and elem.tag == "Users":
for user_elem in elem:
if "fritz" in user_elem.text:
user = user_elem.text
self.user = user_elem.text
assert challenge is not None and self.user is not None

11
fritz_temp_sync/homeassistant.py
View File

@ -27,22 +27,24 @@ class HomeAssistantAPI:
async def connect(self):
retries = 5
logging.info("Connect to home assistant...")
while True:
try:
self.ws = await websockets.connect(self.url)
self.sender = asyncio.create_task(self.sending())
await self.auth()
self.receiver = asyncio.create_task(self.receiving())
return
except InvalidStatusCode:
return True
except (InvalidStatusCode, TimeoutError):
if retries > 0:
retries -= 1
await asyncio.sleep(10)
logging.info("Retry home assistant connection...")
await asyncio.sleep(30)
logging.info(f"Retry home assistant connection... ({retries})")
continue
else:
logging.error("Invalid status code while connecting to Home Assistant")
await self.exit_loop()
return False
async def wait_for_close(self):
await self.ws.wait_closed()
@ -85,7 +87,6 @@ class HomeAssistantAPI:
self.sender.cancel()
if self.receiver is not None:
self.receiver.cancel()
asyncio.get_running_loop().stop()
async def auth(self):
msg = json.loads(await self.ws.recv())

2
fritz_temp_sync/run.sh
View File

@ -1,3 +1,3 @@
#!/usr/bin/env bashio
#!/usr/bin/with-contenv bashio
python3 /srv/sync_ha_fb.py /data/options.json

135
fritz_temp_sync/sync_ha_fb.py
View File

@ -3,9 +3,8 @@
import asyncio
import os
import sys
from typing import Dict
from fritzbox import FritzBox, Device
from fritzbox import FritzBox
from homeassistant import HomeAssistantAPI
import logging
import json
@ -18,62 +17,40 @@ async def handle_event(idx: int):
logging.debug(f"Wait for events for {idx}")
while event := await ha.events[idx].get():
entity_id = event["data"]["entity_id"]
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.info(f"received changed state from {entity_id}")
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_temp = round(float(sensor_state["attributes"]["temperature"]) * 2) / 2
if therm_temp != sensor_temp:
logging.info(f"{therm_name}: {therm_temp}")
logging.info(f"{sensor}: {sensor_state['attributes']['temperature']} ({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
"""
fb.login()
logged = False
"""
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"]
try:
entity_id = event["data"]["entity_id"]
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":
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_temp = round(float(sensor_state["attributes"]["temperature"]) * 2) / 2
if therm_temp != sensor_temp:
logging.info(f"{therm_name}: {therm_temp}")
logging.info(f"{entity_id}: {new_state['attributes']['temperature']} ({sensor_temp})")
logging.info(f"{therm_name}: {therm_temp}\n{sensor}: {sensor_state['attributes']['temperature']} ({sensor_temp})")
fb.correct_offset(therm_name, sensor_temp)
"""
current_temp, current_offset, id, name = fb.get_device_data(name=thermostate)
if not logged:
logging.info(
f"Current measurement from {entity_id}: {new_state['attributes']['temperature']} ({rounded})")
logged = True
logging.info(f"Current measurement from {thermostate}: {current_temp}")
new_offset = current_offset + rounded - current_temp
if new_offset != current_offset:
old_offset = current_offset
logging.debug(f"Set offset for {thermostate} from {current_offset} to {new_offset}")
fb.set_offset(current_temp, new_offset, id, name)
current_temp, current_offset, id, name = fb.get_device_data(name=thermostate)
logging.debug(f"Target: {new_offset} ; Set: {current_offset}")
if new_offset == current_offset:
logging.info(f"Adjustet offset from {old_offset} to {new_offset}")
else:
logging.warning(f"Failed to adjust offset from {old_offset} to {new_offset}")
fb.logout()
"""
elif entity_id in sensor_mappings.keys():
sensor_temp = round(float(new_state["attributes"]["temperature"]) * 2) / 2
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})")
fb.correct_offset(therm_name, sensor_temp)
except KeyError:
pass
async def init(ha: HomeAssistantAPI, fb: FritzBox):
await ha.connect()
if not await ha.connect():
return
logging.debug("Subscribe")
state_changed_id = await ha.subscribe_event("state_changed")
logging.debug(state_changed_id)
@ -81,31 +58,37 @@ async def init(ha: HomeAssistantAPI, fb: FritzBox):
fb.login()
await ha.wait_for_close()
logging.info("Websocket closed, shutting down..")
asyncio.get_running_loop().stop()
async def main():
config_path = sys.argv[1]
config = json.load(open(config_path))
level = logging.INFO
if "log_level" in config:
if config["log_level"] == "DEBUG":
level = logging.DEBUG
logging.basicConfig(level=level, format="[%(asctime)s] [%(levelname)s] %(message)s")
logging.debug(config)
logging.basicConfig(level=logging.INFO, format="[%(asctime)s] [%(levelname)s] %(message)s")
config_path = sys.argv[1]
config = json.load(open(config_path))
logging.debug(config)
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"]
global ha
ha = HomeAssistantAPI(supervisor_token, supervisor_url)
loop = asyncio.get_event_loop()
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"
ha = HomeAssistantAPI(os.environ["SUPERVISOR_TOKEN"], supervisor_url)
for mapping in config["mappings"]:
if mapping["sensor"] not in sensor_mappings.keys():
sensor_mappings[mapping["sensor"]] = []
sensor_mappings[mapping["sensor"]].append(mapping["thermostate"])
thermostate_mappings[mapping["thermostate"]] = mapping["sensor"]
for mapping in config["mappings"]:
if mapping["sensor"] not in sensor_mappings.keys():
sensor_mappings[mapping["sensor"]] = []
sensor_mappings[mapping["sensor"]].append(mapping["thermostate"])
thermostate_mappings[mapping["thermostate"]] = mapping["sensor"]
try:
await init(ha, fb)
except KeyboardInterrupt:
pass
loop.create_task(init(ha, fb))
try:
loop.run_forever()
except KeyboardInterrupt:
pass
asyncio.run(main())