Upgrade paho mqtt

packages/helpermodules/broker.py

@@ -25,7 +25,10 @@ def __init__(self,
                  port: int = 1886) -> None:
         try:
             self.name = f"openWB-{name}-{get_name_suffix()}"
-            self.client = mqtt.Client(self.name)
+            self.client = mqtt.Client(
+                callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+                client_id=self.name,
+            )
             self.client.on_connect = on_connect
             self.client.on_message = on_message
             self.client.connect(host, port)
@@ -48,7 +51,10 @@ def disconnect(self) -> None:
 class InternalBrokerPublisher:
     def __init__(self) -> None:
         try:
-            self.client = mqtt.Client(f"openWB-python-bulk-publisher-{get_name_suffix()}")
+            self.client = mqtt.Client(
+                callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+                client_id=f"openWB-python-bulk-publisher-{get_name_suffix()}",
+            )
             self.client.connect("localhost", 1886)
         except Exception:
             log.exception("Fehler beim Verbindungsaufbau zum Bulk-Publisher")

packages/helpermodules/command.py

@@ -139,7 +139,7 @@ def sub_commands(self):
         except Exception:
             log.exception("Fehler im Command-Modul")
 
-    def on_connect(self, client: mqtt.Client, userdata, flags: dict, rc: int):
+    def on_connect(self, client: mqtt.Client, userdata, flags, reason_code, properties):
         """ connect to broker and subscribe to set topics
         """
         try:
@@ -1234,7 +1234,7 @@ def get_cloud_id(self):
             log.exception("Fehler im Command-Modul")
             return []
 
-    def on_connect(self, client, userdata, flags, rc):
+    def on_connect(self, client, userdata, flags, reason_code, properties):
         """ connect to broker and subscribe to set topics
         """
         client.subscribe(f'openWB/{self.topic_str}#', 2)

packages/helpermodules/create_debug.py

@@ -485,7 +485,7 @@ def get_broker(self):
         BrokerClient("processBrokerBranch", self.__on_connect_broker, self.__get_content).start_finite_loop()
         return self.content
 
-    def __on_connect_broker(self, client, userdata, flags, rc):
+    def __on_connect_broker(self, client, userdata, flags, reason_code, properties):
         client.subscribe("openWB/#", 2)
 
     def __get_content(self, client, userdata, msg):
@@ -496,7 +496,7 @@ def get_broker_essentials(self):
                      self.__get_content).start_finite_loop()
         return self.content
 
-    def __on_connect_broker_essentials(self, client, userdata, flags, rc):
+    def __on_connect_broker_essentials(self, client, userdata, flags, reason_code, properties):
         client.subscribe("openWB/system/ip_address", 2)
         client.subscribe("openWB/system/current_commit", 2)
         client.subscribe("openWB/system/boot_done", 2)
@@ -511,7 +511,7 @@ def __on_connect_broker_essentials(self, client, userdata, flags, rc):
         client.subscribe("openWB/bat/#", 2)
         client.subscribe("openWB/optional/ep/flexible_tariff/provider", 2)
 
-    def __on_connect_bridges(self, client, userdata, flags, rc):
+    def __on_connect_bridges(self, client, userdata, flags, reason_code, properties):
         client.subscribe("openWB/system/mqtt/#", 2)
 
     def get_cloud(self):

packages/helpermodules/data_migration/data_migration.py

@@ -15,15 +15,13 @@
 import pathlib
 import shutil
 import tarfile
-from paho.mqtt.client import Client as MqttClient, MQTTMessage
 from threading import Thread
 from typing import Callable, Dict, List, Optional, Union
 
 from control import data
 from control.ev import ev
 from dataclass_utils import dataclass_from_dict
 import dataclass_utils
-from helpermodules.broker import BrokerClient
 from helpermodules.data_migration.id_mapping import MapId
 from helpermodules.hardware_configuration import update_hardware_configuration
 from helpermodules.measurement_logging.process_log import get_totals
@@ -639,21 +637,3 @@ def merge_list_of_records(self, key):
             reduce(self._merge_records_by(key), records)
             for _, records in groupby(sorted(lst, key=key_prop), key_prop)
         ]
-
-
-class BrokerCphargepoints:
-    def get_configured_cp_ids(self) -> List:
-        self.all_received_topics = {}
-        BrokerClient("update-config", self.on_connect, self.on_message).start_finite_loop()
-        cp_ids = []
-        for topic, payload in self.all_received_topics.items():
-            cp_ids.append(get_index(topic))
-        return cp_ids
-
-    def on_connect(self, client: MqttClient, userdata, flags: dict, rc: int):
-        """ connect to broker and subscribe to set topics
-        """
-        client.subscribe("openWB/chargepoint/+/config", 2)
-
-    def on_message(self, client: MqttClient, userdata, msg: MQTTMessage):
-        self.all_received_topics.update({msg.topic: msg.payload})

packages/helpermodules/measurement_logging/write_log.py

@@ -126,7 +126,7 @@ def __init__(self) -> None:
         except Exception:
             log.exception("Fehler im Werte-Logging-Modul für SmartHome")
 
-    def on_connect(self, client: MqttClient, userdata, flags: dict, rc: int):
+    def on_connect(self, client: MqttClient, userdata, flags, reason_code, properties):
         client.subscribe("openWB/LegacySmartHome/#", 2)
 
     def on_message(self, client: MqttClient, userdata, msg: MQTTMessage):

packages/helpermodules/setdata.py

@@ -45,7 +45,7 @@ def set_data(self):
     def disconnect(self) -> None:
         self.internal_broker_client.disconnect()
 
-    def on_connect(self, client: mqtt.Client, userdata, flags: dict, rc: int):
+    def on_connect(self, client: mqtt.Client, userdata, flags, reason_code, properties):
         """ connect to broker and subscribe to set topics
         """
         client.subscribe("openWB/set/#", 2)

packages/helpermodules/subdata.py

@@ -121,7 +121,7 @@ def sub_topics(self):
     def disconnect(self) -> None:
         self.internal_broker_client.disconnect()
 
-    def on_connect(self, client: mqtt.Client, userdata, flags: dict, rc: int):
+    def on_connect(self, client: mqtt.Client, userdata, flags, reason_code, properties):
         """ subscribe topics
         """
         client.subscribe([

packages/helpermodules/update_config.py

@@ -736,7 +736,7 @@ def update(self):
         finally:
             self.__update_topic("openWB/system/update_config_completed", True)
 
-    def on_connect(self, client: MqttClient, userdata, flags: dict, rc: int):
+    def on_connect(self, client: MqttClient, userdata, flags, reason_code, properties):
         """ connect to broker and subscribe to set topics
         """
         client.subscribe("openWB/#", 2)

packages/modules/chargepoints/external_openwb/chargepoint_module.py

@@ -59,7 +59,7 @@ def parse_received_topics(value: str):
                 pub.pub_single(f"openWB/set/isss/parentCPlp{self.config.configuration.duo_num + 1}",
                                str(num), hostname=ip_address)
 
-                def on_connect(client, userdata, flags, rc):
+                def on_connect(client, userdata, flags, reason_code, properties):
                     client.subscribe(f"openWB/internal_chargepoint/{self.config.configuration.duo_num}/get/#")
 
                 def on_message(client, userdata, message):

packages/modules/chargepoints/mqtt/chargepoint_module.py

@@ -22,7 +22,7 @@ def __init__(self, config: Mqtt) -> None:
         self.store = get_chargepoint_value_store(self.config.id)
         self.fault_state = FaultState(ComponentInfo(self.config.id, "Ladepunkt", "chargepoint"))
 
-        def on_connect(client, userdata, flags, rc):
+        def on_connect(client, userdata, flags, reason_code, properties):
             client.subscribe(f"openWB/mqtt/chargepoint/{self.config.id}/#")
 
         def on_message(client, userdata, message):
@@ -49,7 +49,7 @@ def get_values(self) -> None:
         def parse_received_topics(value: str):
             return received_topics.get(f"{topic_prefix}{value}", get_default(ChargepointState, value))
         with SingleComponentUpdateContext(self.fault_state):
-            def on_connect(client, userdata, flags, rc):
+            def on_connect(client, userdata, flags, reason_code, properties):
                 client.subscribe(f"openWB/mqtt/chargepoint/{self.config.id}/get/#")
 
             def on_message(client, userdata, message):

packages/modules/devices/generic/mqtt/device.py

@@ -25,7 +25,7 @@ def create_inverter_component(component_config: MqttInverterSetup):
         return inverter.MqttInverter(component_config, device_id=device_config.id)
 
     def update_components(components: Iterable[Union[bat.MqttBat, counter.MqttCounter, inverter.MqttInverter]]):
-        def on_connect(client, userdata, flags, rc):
+        def on_connect(client, userdata, flags, reason_code, properties):
             for component in components:
                 client.subscribe(f"openWB/mqtt/{type_to_topic_mapping(component.component_config.type)}/"
                                  f"{component.component_config.id}/#")

packages/modules/internal_chargepoint_handler/chargepoint_module.py

@@ -61,7 +61,7 @@ def __init__(self, local_charge_point_num: int,
             self.perform_phase_switch(1)
             self.old_phases_in_use = 1
         else:
-            def on_connect(client, userdata, flags, rc):
+            def on_connect(client, userdata, flags, reason_code, properties):
                 client.subscribe(f"openWB/internal_chargepoint/{self.local_charge_point_num}/get/phases_in_use")
 
             def on_message(client, userdata, message):

packages/modules/io_devices/add_on/api.py

@@ -42,7 +42,7 @@ def get(self, host: str) -> IoState:
                      1886 if host == "localhost" else 1883).start_finite_loop()
         return self.io_state
 
-    def on_connect(self, client, userdata, flags, rc):
+    def on_connect(self, client, userdata, flags, reason_code, properties):
         """ connect to broker and subscribe to set topics
         """
         client.subscribe('openWB/internal_io/states/#', 2)

packages/modules/io_devices/eebus/api.py

@@ -150,7 +150,7 @@ def initializer():
         Path(f"{Path(__file__).resolve().parents[4]}/ramdisk/eebus_hems_client.log").touch(exist_ok=True)
         run_eebus()
 
-        def on_connect(client, userdata, flags, rc):
+        def on_connect(client, userdata, flags, reason_code, properties):
             client.subscribe(f"openWB/eebus/{config.id}/#")
 
         def on_message(client, userdata, message):

packages/modules/smarthome/mqtt/off.py

@@ -5,7 +5,7 @@
 numberOfSupportedDevices = 9  # limit number of smart home devices
 
 
-def on_connect(client, userdata, flags, rc) -> None:
+def on_connect(client, userdata, flags, reason_code, properties) -> None:
     client.subscribe("openWB/set/LegacySmartHome/Devices/#", 2)
 
 
@@ -16,7 +16,10 @@ def on_message(client, userdata, msg) -> None:
 devicenumber = str(sys.argv[1])
 ipadr = str(sys.argv[2])
 uberschuss = int(sys.argv[3])
-client = mqtt.Client("openWB-mqttsmarthomecust")
+client = mqtt.Client(
+    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+    client_id="openWB-mqttsmarthomecust",
+)
 client.on_connect = on_connect
 client.on_message = on_message
 startTime = time.time()

packages/modules/smarthome/mqtt/on.py

@@ -5,7 +5,7 @@
 numberOfSupportedDevices = 9  # limit number of smart home devices
 
 
-def on_connect(client, userdata, flags, rc) -> None:
+def on_connect(client, userdata, flags, reason_code, properties) -> None:
     client.subscribe("openWB/set/LegacySmartHome/Devices/#", 2)
 
 
@@ -16,7 +16,10 @@ def on_message(client, userdata, msg) -> None:
 devicenumber = str(sys.argv[1])
 ipadr = str(sys.argv[2])
 uberschuss = int(sys.argv[3])
-client = mqtt.Client("openWB-mqttsmarthomecust")
+client = mqtt.Client(
+    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+    client_id="openWB-mqttsmarthomecust",
+)
 client.on_connect = on_connect
 client.on_message = on_message
 startTime = time.time()

packages/modules/smarthome/mqtt/watt.py

@@ -8,7 +8,7 @@
 numberOfSupportedDevices = 9  # limit number of smart home devices
 
 
-def on_connect(client, userdata, flags, rc) -> None:
+def on_connect(client, userdata, flags, reason_code, properties) -> None:
     client.subscribe("openWB/LegacySmartHome/Devices/"+str(devicenumber) + "/#", 2)
 
 
@@ -48,7 +48,10 @@ def on_message(client, userdata, msg) -> None:
 devicenumber = int(sys.argv[1])
 ipadr = str(sys.argv[2])
 uberschuss = int(sys.argv[3])
-client = mqtt.Client("openWB-mqttsmarthomecust" + str(devicenumber))
+client = mqtt.Client(
+    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+    client_id="openWB-mqttsmarthomecust" + str(devicenumber),
+)
 client.on_connect = on_connect
 client.on_message = on_message
 startTime = time.time()

packages/modules/vehicles/mqtt/soc.py

@@ -14,7 +14,7 @@
 
 def create_vehicle(vehicle_config: MqttSocSetup, vehicle: int):
     def updater(vehicle_update_data: VehicleUpdateData) -> CarState:
-        def on_connect(client, userdata, flags, rc):
+        def on_connect(client, userdata, flags, reason_code, properties):
             client.subscribe(f"openWB/mqtt/vehicle/{vehicle}/get/#")
 
         def on_message(client, userdata, message):

packages/smarthome/smartcommon.py

@@ -43,7 +43,7 @@
 firststart = True
 
 
-def on_connect(client: mqtt.Client, userdata: Any, flags: Dict, rc: int) -> None:
+def on_connect(client: mqtt.Client, userdata: Any, flags, reason_code, properties) -> None:
     #  mqttcg = 'openWB/config/get/SmartHome/'
     #  client.subscribe("openWB/config/get/SmartHome/#", 2)
     client.subscribe(mqttcg + '#', 2)
@@ -199,7 +199,10 @@ def pub(client: mqtt.Client, key: str, value: str) -> None:
 
 
 def sendmq(mqtt_input: Dict[str, str]) -> None:
-    client = mqtt.Client("openWB-SmartHome-bulkpublisher-" + str(os.getpid()))
+    client = mqtt.Client(
+        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+        client_id="openWB-SmartHome-bulkpublisher-" + str(os.getpid()),
+    )
     client.connect("localhost", mqttport)
     for key, value in mqtt_input.items():
         valueold = mqtt_cache.get(key, 'not in cache')
@@ -223,7 +226,10 @@ def conditions(speichersoc: int) -> None:
 
 
 def update_devices() -> None:
-    client = mqtt.Client("openWB-SmartHome-bulkpublisher-" + str(os.getpid()))
+    client = mqtt.Client(
+        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+        client_id="openWB-SmartHome-bulkpublisher-" + str(os.getpid()),
+    )
     client.connect("localhost", mqttport)
     # statische daten einschaltgruppe
     Sbase.ausdevices = 0
@@ -336,7 +342,10 @@ def readmq() -> None:
         with open(bp+'/ramdisk/smartparam.sh', 'w') as f:
             print('%s' % ('#!/bin/bash'), file=f)
     parammqtt = []
-    client = mqtt.Client("openWB-mqttsmarthome")
+    client = mqtt.Client(
+        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+        client_id="openWB-mqttsmarthome",
+    )
     client.on_connect = on_connect
     client.on_message = on_message
     startTime = time.time()

requirements.txt

@@ -1,7 +1,7 @@
 # don't update python distribution packages in here
 typing-extensions==4.13.2
 jq==1.1.3
-paho_mqtt==1.6.1
+paho_mqtt==2.1.0
 pymodbus==2.5.3
 pytest==8.4.2
 requests_mock==1.9.3

runs/remoteSupport/remoteSupport.py

@@ -101,7 +101,7 @@ def stop_tunnel(tunnel: Optional[Popen], tunnel_name: str) -> None:
         log.error(f"tunnel {tunnel_name} is not running.")
 
 
-def on_connect(client: mqtt.Client, userdata, flags: dict, rc: int):
+def on_connect(client: mqtt.Client, userdata, flags, reason_code, properties):
     """connect to broker and subscribe to set topics"""
     log.info("Connected")
     client.subscribe([
@@ -222,7 +222,10 @@ def on_message(client: mqtt.Client, userdata, msg: mqtt.MQTTMessage):
 signal(SIGTERM, handle_terminate)  # Handle SIGTERM from systemctl for graceful shutdown
 signal(SIGINT, handle_terminate)  # Handle SIGINT from keyboard (Strg+C) for graceful shutdown
 lt_executable = get_lt_executable()
-client = mqtt.Client(f"openWB-remote-{get_serial()}-{datetime.today().timestamp()}")
+client = mqtt.Client(
+    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+    client_id=f"openWB-remote-{get_serial()}-{datetime.today().timestamp()}",
+)
 client.on_connect = on_connect
 client.on_message = on_message
 client.will_set(STATE_TOPIC, json.dumps("offline"), qos=2, retain=True)

simpleAPI/simpleAPI_mqtt.py

@@ -47,7 +47,7 @@ def __init__(self, config_file: str):
         self.charge_template_cache: Dict[str, Dict[str, Any]] = {}
 
         # MQTT client setup
-        self.client = mqtt.Client()
+        self.client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
         self.client.on_connect = self._on_connect
         self.client.on_message = self._on_message
         self.client.on_disconnect = self._on_disconnect
@@ -88,9 +88,9 @@ def _load_config_file(self, config_file: str):
             log.error(f"Failed to load config file {config_file}: {e}")
             sys.exit(1)
 
-    def _on_connect(self, client, userdata, flags, rc):
+    def _on_connect(self, client, userdata, flags, reason_code, properties):
         """Callback for successful MQTT connection."""
-        if rc == 0:
+        if reason_code == 0:
             log.info(f"Connected to MQTT broker at {self.host}:{self.port}")
 
             # Publish API revision
@@ -108,12 +108,12 @@ def _on_connect(self, client, userdata, flags, rc):
             log.info(
                 "Subscribed to openWB component topics (bat, pv, chargepoint, counter) and simpleAPI set topics")
         else:
-            log.error(f"Failed to connect to MQTT broker. Return code: {rc}")
+            log.error(f"Failed to connect to MQTT broker. Reason code: {reason_code}")
 
-    def _on_disconnect(self, client, userdata, rc):
+    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
         """Callback for MQTT disconnection."""
-        if rc != 0:
-            log.warning(f"Unexpected MQTT disconnection (code: {rc}). Attempting to reconnect...")
+        if reason_code != 0:
+            log.warning(f"Unexpected MQTT disconnection (code: {reason_code}). Attempting to reconnect...")
             self._reconnect()
         else:
             log.info("Clean MQTT disconnection")