nixos-config/mqtt/heater.py

import json
import paho.mqtt.client as mqtt
import threading
import time
from typing import Dict


class Heater:
    def __init__(self, topic):
        self.not_initialized_yet = True
        self.wanted_temperature = 14
        self.actual_temperature = 14
        self.topic = topic

    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": "",
            "occupied_heating_setpoint": "",
            "unoccupied_heating_setpoint": "",
            "local_temperature": "",
            # "pi_heating_demand": "",
            # "system_mode": "",
        }
        return ("%s/get" % self.topic), json.dumps(payload)

    def topic_and_payload_for_set(self):
        payload = {
            "system_mode": "auto",
            # "current_heating_setpoint": str(self.wanted_temperature),
            "occupied_heating_setpoint": str(self.wanted_temperature),
            "unoccupied_heating_setpoint": str(self.wanted_temperature),
            "eurotronic_host_flags": {"window_open": True},
        }
        return ("%s/set" % 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():
                topic, payload = heater.topic_and_payload_for_set()
                client.publish(topic, 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)


scene = "default"

watcher = Watcher(
    {
        "office1": Heater(topic="zigbee2mqtt/office_heater_1"),
        "office2": Heater(topic="zigbee2mqtt/office_heater_2"),
        "bedroom": Heater(topic="zigbee2mqtt/bedroom_heater_1"),
        "storage": Heater(topic="zigbee2mqtt/storage_heater_1"),
    }
)


# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, _userdata, _flags, rc):
    print("Connected with result code " + str(rc))

    threading.Thread(target=loop_thread, args=(client,), daemon=True).start()

    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("control/lights/set")
    for topic in watcher.get_topics():
        client.subscribe(topic)
    watcher.pull_values(client)


# The callback for when a PUBLISH message is received from the server.
def on_message(client, _userdata, msg):
    global scene
    (topic, payload) = parse_message(msg)
    if topic == "control/lights/set":
        print("set scene %s -> %s" % (scene, payload["scene"]))
        scene = payload["scene"]
        update_scene(client)
    else:
        print("got %s" % topic)
        watcher.update_actual_heating_point_for_topic(topic, payload)


def parse_message(msg):
    m_decode = str(msg.payload.decode("utf-8", "ignore"))
    payload = json.loads(m_decode)  # decode json data
    return msg.topic, payload


def update_scene(client):
    if scene in ["night", "outside"]:
        watcher.update("office1", 14)
        watcher.update("office2", 14)
        watcher.update("bedroom", 14)
        watcher.update("storage", 14)
    elif scene in ["default", "up-bright", "up-dark", "half", "down"]:
        watcher.update("office1", 25)
        watcher.update("office2", 25)
        watcher.update("bedroom", 18)
        watcher.update("storage", 18)
    else:
        watcher.update("office1", 14)
        watcher.update("office2", 14)
        watcher.update("bedroom", 14)
        watcher.update("storage", 14)

    watcher.publish(client)


def loop_thread(client):
    while True:
        watcher.publish(client)
        watcher.pull_values(client)
        time.sleep(120)


if __name__ == "__main__":
    mqttClient = mqtt.Client()
    mqttClient.on_connect = on_connect
    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
    # manual interface.
    mqttClient.loop_forever()