palo/nixos-config
palo/nixos-config
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fiddeling with heater and fyrtur

Browse source
This commit is contained in:
Ingolf Wagner 2021-03-18 13:15:51 +01:00
parent 4003c41c3d
commit d1e5ff35ae
No known key found for this signature in database
GPG key ID: 76BF5F1928B9618B
8 changed files with 86 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
configs/pepe/home-assistant/light-control.nix
View file

@ -24,6 +24,14 @@
"zigbee2mqtt/door_sensor_4"
"zigbee2mqtt/door_sensor_5"
];
disabled_switches = [
"zigbee2mqtt/led_1"
"zigbee2mqtt/led_2"
"zigbee2mqtt/light_2"
"zigbee2mqtt/light_4"
"zigbee2mqtt/light_5"
"zigbee2mqtt/light_7"
];
}
{
name = "down";

1
configs/pepe/home-assistant/zigbee2mqtt/heater.nix
View file

@ -18,6 +18,7 @@ in {
{ id, ... }: {
name = id;
value = {
legacy = false;
retain = false;
friendly_name = name;
transition = 1;

7
mqtt/fyrtur.py
View file

@ -6,10 +6,6 @@ import paho.mqtt.client as mqtt
import json
import threading
# initial scene
from heater.modules.heater import Heater
from heater.modules.watcher import Watcher
scene = "up-dark"
@ -60,7 +56,7 @@ class FyrturWatcher:
elif position == Position.DOWN:
fyrtur.wanted_position = fyrtur.bottom
elif position == Position.HALF:
fyrtur.wanted_position = round((fyrtur.top - fyrtur.bottom) / 2)
fyrtur.wanted_position = round((fyrtur.top - fyrtur.bottom) / 2 + fyrtur.bottom)
def publish(self, client):
for fyrtur in self.fyrturs.values():
@ -69,7 +65,6 @@ class FyrturWatcher:
time.sleep(2)
watcher = FyrturWatcher({
"office1": Fyrtur(topic="zigbee2mqtt/fyrtur1", set_topic="zigbee2mqtt/fyrtur1/set", top=100, bottom=16),
"office2": Fyrtur(topic="zigbee2mqtt/fyrtur4", set_topic="zigbee2mqtt/fyrtur4/set", top=100, bottom=22),

82
mqtt/heater.py
View file

@ -1,12 +1,77 @@
import time
import paho.mqtt.client as mqtt
import json
import paho.mqtt.client as mqtt
import threading
import time
from typing import Dict
class Heater:
def __init__(self, topic, set_topic):
self.not_initialized_yet = True
self.wanted_temperature = 15
self.actual_temperature = 15
self.set_topic = set_topic
self.topic = topic
def payload(self):
payload = {
"system_mode": "auto",
"current_heating_setpoint": self.wanted_temperature,
"eurotronic_host_flags": {"window_open": True}
}
return json.dumps(payload)
def needs_publish(self):
if self.not_initialized_yet:
return True
else:
return self.wanted_temperature != self.actual_temperature
def update_actual_heating_point(self, payload):
heating_setpoint = int(payload["current_heating_setpoint"])
if self.not_initialized_yet:
self.not_initialized_yet = False
self.wanted_temperature = heating_setpoint
print("%s: update wanted temperature %d" % (self.topic, self.actual_temperature))
self.actual_temperature = heating_setpoint
print("%s: update actual temperature %d" % (self.topic, self.actual_temperature))
def topic_and_payload_for_query(self):
payload = {
"current_heating_setpoint": ""
}
return ("%s/get" % self.topic), json.dumps(payload)
class Watcher:
def __init__(self, heater: Dict[str, Heater]):
self.heater = heater
def publish(self, client):
for heater in self.heater.values():
if heater.needs_publish():
client.publish(heater.set_topic, heater.payload())
time.sleep(2)
def update(self, name, temperature):
heater: Heater = self.heater.get(name)
heater.wanted_temperature = temperature
def get_topics(self):
return [heater.topic for heater in self.heater.values()]
def update_actual_heating_point_for_topic(self, topic, payload):
for heater in self.heater.values():
if heater.topic == topic:
heater.update_actual_heating_point(payload)
return
def pull_values(self, client):
for heater in self.heater.values():
topic, payload = heater.topic_and_payload_for_query()
client.publish(topic, payload)
# initial scene
from heater.modules.heater import Heater
from heater.modules.watcher import Watcher
scene = "default"
@ -57,8 +122,8 @@ def update_scene(client):
watcher.update("office2", 10)
watcher.update("bedroom", 13)
elif scene in ["default", "up-bright", "up-dark", "half", "down"]:
watcher.update("office1", 23)
watcher.update("office2", 23)
watcher.update("office1", 26)
watcher.update("office2", 26)
watcher.update("bedroom", 18)
else:
watcher.update("office1", 10)
@ -80,6 +145,7 @@ if __name__ == "__main__":
mqttClient.on_message = on_message
mqttClient.username_pw_set("homeassistant", password="password")
mqttClient.connect("pepe.private", 1883, 60)
# Blocking call that processes network traffic, dispatches callbacks and
# handles reconnecting.
# Other loop*() functions are available that give a threaded interface and a

0
mqtt/heater/__init__.py
View file

0
mqtt/heater/modules/__init__.py
View file

39
mqtt/heater/modules/heater.py
View file

@ -1,39 +0,0 @@
import json
class Heater:
def __init__(self, topic, set_topic):
self.not_initialized_yet = True
self.wanted_temperature = 15
self.actual_temperature = 15
self.set_topic = set_topic
self.topic = topic
def payload(self):
payload = {
"system_mode": "auto",
"current_heating_setpoint": str(self.wanted_temperature),
"eurotronic_host_flags": {"window_open": True}
}
return json.dumps(payload)
def needs_publish(self):
if self.not_initialized_yet:
return True
else:
return self.wanted_temperature != self.actual_temperature
def update_actual_heating_point(self, payload):
heating_setpoint = int(payload["current_heating_setpoint"])
if self.not_initialized_yet:
self.not_initialized_yet = False
self.wanted_temperature = heating_setpoint
print("%s: update wanted temperature %d" % (self.topic, self.actual_temperature))
self.actual_temperature = heating_setpoint
print("%s: update actual temperature %d" % (self.topic, self.actual_temperature))
def topic_and_payload_for_query(self):
payload = {
"current_heating_setpoint": ""
}
return ("%s/get" % self.topic), json.dumps(payload)

34
mqtt/heater/modules/watcher.py
View file

@ -1,34 +0,0 @@
import time
from typing import List, Dict
from heater.modules.heater import Heater
class Watcher:
def __init__(self, heater: Dict[str, Heater]):
self.heater = heater
def publish(self, client):
for heater in self.heater.values():
if heater.needs_publish():
client.publish(heater.set_topic, heater.payload())
time.sleep(2)
def update(self, name, temperature):
heater: Heater = self.heater.get(name)
heater.wanted_temperature = temperature
def get_topics(self):
return [heater.topic for heater in self.heater.values()]
def update_actual_heating_point_for_topic(self, topic, payload):
for heater in self.heater.values():
if heater.topic == topic:
heater.update_actual_heating_point(payload)
return
def pull_values(self, client):
for heater in self.heater.values():
topic, payload = heater.topic_and_payload_for_query()
client.publish(topic, payload)