diff --git a/packages/helpermodules/broker.py b/packages/helpermodules/broker.py index c9b6940c6f..1a39b7f49e 100644 --- a/packages/helpermodules/broker.py +++ b/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") diff --git a/packages/helpermodules/command.py b/packages/helpermodules/command.py index 2b429f135e..37ba17f090 100644 --- a/packages/helpermodules/command.py +++ b/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) diff --git a/packages/helpermodules/create_debug.py b/packages/helpermodules/create_debug.py index cb032927a5..6f7f47cb61 100644 --- a/packages/helpermodules/create_debug.py +++ b/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): diff --git a/packages/helpermodules/data_migration/data_migration.py b/packages/helpermodules/data_migration/data_migration.py index ec197a1fb7..40da4697bc 100644 --- a/packages/helpermodules/data_migration/data_migration.py +++ b/packages/helpermodules/data_migration/data_migration.py @@ -15,7 +15,6 @@ 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 @@ -23,7 +22,6 @@ 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}) diff --git a/packages/helpermodules/measurement_logging/write_log.py b/packages/helpermodules/measurement_logging/write_log.py index fd8d51aa4b..3d44f9a941 100644 --- a/packages/helpermodules/measurement_logging/write_log.py +++ b/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): diff --git a/packages/helpermodules/setdata.py b/packages/helpermodules/setdata.py index 7973936181..3b02297c77 100644 --- a/packages/helpermodules/setdata.py +++ b/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) diff --git a/packages/helpermodules/subdata.py b/packages/helpermodules/subdata.py index 4c59046940..1e9de3ec3e 100644 --- a/packages/helpermodules/subdata.py +++ b/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([ diff --git a/packages/helpermodules/update_config.py b/packages/helpermodules/update_config.py index f85dd8c8f0..318413c71f 100644 --- a/packages/helpermodules/update_config.py +++ b/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) diff --git a/packages/modules/chargepoints/external_openwb/chargepoint_module.py b/packages/modules/chargepoints/external_openwb/chargepoint_module.py index c0b162e4c8..57fd3f0caa 100644 --- a/packages/modules/chargepoints/external_openwb/chargepoint_module.py +++ b/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): diff --git a/packages/modules/chargepoints/mqtt/chargepoint_module.py b/packages/modules/chargepoints/mqtt/chargepoint_module.py index 92526a020a..8cad0269df 100644 --- a/packages/modules/chargepoints/mqtt/chargepoint_module.py +++ b/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): diff --git a/packages/modules/devices/generic/mqtt/device.py b/packages/modules/devices/generic/mqtt/device.py index af862f342f..f71f3da376 100644 --- a/packages/modules/devices/generic/mqtt/device.py +++ b/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}/#") diff --git a/packages/modules/internal_chargepoint_handler/chargepoint_module.py b/packages/modules/internal_chargepoint_handler/chargepoint_module.py index 5dea675284..3296ab1ed9 100644 --- a/packages/modules/internal_chargepoint_handler/chargepoint_module.py +++ b/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): diff --git a/packages/modules/io_devices/add_on/api.py b/packages/modules/io_devices/add_on/api.py index a94371e115..8f4802593d 100644 --- a/packages/modules/io_devices/add_on/api.py +++ b/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) diff --git a/packages/modules/io_devices/eebus/api.py b/packages/modules/io_devices/eebus/api.py index 12b8a68b8d..28d2867c3b 100644 --- a/packages/modules/io_devices/eebus/api.py +++ b/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): diff --git a/packages/modules/smarthome/mqtt/off.py b/packages/modules/smarthome/mqtt/off.py index c10c28f263..0c50c67232 100644 --- a/packages/modules/smarthome/mqtt/off.py +++ b/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() diff --git a/packages/modules/smarthome/mqtt/on.py b/packages/modules/smarthome/mqtt/on.py index 10dcc0b0c5..8606c4602a 100644 --- a/packages/modules/smarthome/mqtt/on.py +++ b/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() diff --git a/packages/modules/smarthome/mqtt/watt.py b/packages/modules/smarthome/mqtt/watt.py index e11635424b..059e634e13 100644 --- a/packages/modules/smarthome/mqtt/watt.py +++ b/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() diff --git a/packages/modules/vehicles/mqtt/soc.py b/packages/modules/vehicles/mqtt/soc.py index fbfd56bb96..b70b240dd9 100644 --- a/packages/modules/vehicles/mqtt/soc.py +++ b/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): diff --git a/packages/smarthome/smartcommon.py b/packages/smarthome/smartcommon.py index 4b95036ee8..e79c3a3d87 100644 --- a/packages/smarthome/smartcommon.py +++ b/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() diff --git a/requirements.txt b/requirements.txt index 36e59cb480..5472f33381 100644 --- a/requirements.txt +++ b/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 diff --git a/runs/remoteSupport/remoteSupport.py b/runs/remoteSupport/remoteSupport.py index f20ab349b6..68e287752a 100755 --- a/runs/remoteSupport/remoteSupport.py +++ b/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) diff --git a/simpleAPI/simpleAPI_mqtt.py b/simpleAPI/simpleAPI_mqtt.py index 580c09dfc5..0bbccfb94a 100755 --- a/simpleAPI/simpleAPI_mqtt.py +++ b/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")