Finalize Climate thermostat

custom_components/luxtronik/climate.py

@@ -190,7 +190,7 @@ async def async_set_hvac_mode(self, hvac_mode: str) -> None:
                               self.__get_luxmode(hvac_mode, self.preset_mode), debounce=False, update_immediately_after_write=True)
 
     @property
-    def preset_mode(self) -> str | None:
+    def preset_mode(self) -> str:  # | None:
         """Return current preset mode."""
         luxmode = self._luxtronik.get_value(self._heater_sensor)
         if luxmode == LUX_MODE_OFF and self._control_mode_home_assistant and self.hvac_action == CURRENT_HVAC_IDLE:

custom_components/luxtronik/helpers/debounce.py

@@ -0,0 +1,47 @@
+import time
+from threading import Timer
+
+def debounce(wait):
+    """ Decorator that will postpone a functions
+        execution until after wait seconds
+        have elapsed since the last time it was invoked. """
+    def decorator(fn):
+        def debounced(*args, **kwargs):
+            def call_it():
+                fn(*args, **kwargs)
+            try:
+                debounced.t.cancel()
+            except(AttributeError):
+                pass
+            debounced.t = Timer(wait, call_it)
+            debounced.t.start()
+        return debounced
+    return decorator
+
+
+# def debounce(wait):
+#     """ Decorator that will postpone a functions
+#         execution until after wait seconds
+#         have elapsed since the last time it was invoked. """
+
+#     def decorator(fn):
+#         def debounced(*args, **kwargs):
+#             def call_it():
+#                 debounced._timer = None
+#                 debounced._last_call = time.time()
+#                 return fn(*args, **kwargs)
+
+#             time_since_last_call = time.time() - debounced._last_call
+#             if time_since_last_call >= wait:
+#                 return call_it()
+
+#             if debounced._timer is None:
+#                 debounced._timer = Timer(wait - time_since_last_call, call_it)
+#                 debounced._timer.start()
+
+#         debounced._timer = None
+#         debounced._last_call = 0
+
+#         return debounced
+
+#     return decorator

custom_components/luxtronik/helpers/lux_helper.py

@@ -0,0 +1,56 @@
+import socket
+
+from ..const import LOGGER
+
+
+def discover():
+    """Broadcast discovery for luxtronik heatpumps."""
+
+    for p in (4444, 47808):
+        LOGGER.debug(f"Send discovery packets to port {p}")
+        server = socket.socket(
+            socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
+        server.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+        server.bind(("", p))
+        server.settimeout(2)
+
+        # send AIT magic brodcast packet
+        data = "2000;111;1;\x00"
+        server.sendto(data.encode(), ("<broadcast>", p))
+        LOGGER.debug(f"Sending broadcast request \"{data.encode()}\"")
+
+        while True:
+            try:
+                res, con = server.recvfrom(1024)
+                res = res.decode("ascii", errors="ignore")
+                # if we receive what we just sent, continue
+                if res == data:
+                    continue
+                ip = con[0]
+                # if the response starts with the magic nonsense
+                if res.startswith("2500;111;"):
+                    res = res.split(";")
+                    LOGGER.debug(f"Received answer from {ip} \"{res}\"")
+                    try:
+                        port = int(res[2])
+                    except ValueError:
+                        LOGGER.debug(
+                            "Response did not contain a valid port number, an old Luxtronic software version might be the reason.")
+                        port = None
+                    return (ip, port)
+                # if not, continue
+                else:
+                    LOGGER.debug(
+                        f"Received answer, but with wrong magic bytes, from {ip} skip this one")
+                    continue
+            # if the timout triggers, go on an use the other broadcast port
+            except socket.timeout:
+                break
+
+
+def get_manufacturer_by_model(model: str) -> str:
+    if model is None:
+        return None
+    if model.startswith('LD'):
+        return 'Novelan'
+    return None

custom_components/luxtronik/services.yaml

@@ -0,0 +1,9 @@
+write:
+  description: Write a parameter on the luxtronik heatpump.
+  fields:
+    parameter: 
+      description: ID of the value to write.
+      example: "ID_Ba_Bw_akt"
+    value: 
+      description: Value to write. 
+      example: "Automatic"