From 9b1707e0b5bd3c9e69c25309b66b4278cdf5e4a0 Mon Sep 17 00:00:00 2001 From: Lutz Bender Date: Fri, 6 Mar 2026 09:20:55 +0100 Subject: [PATCH 1/3] cleanup pub_single --- packages/helpermodules/command.py | 6 +++--- packages/helpermodules/setdata.py | 8 ++++---- packages/modules/common/fault_state.py | 15 ++++++++------- .../internal_chargepoint_handler.py | 11 +++++------ .../modules/internal_chargepoint_handler/rfid.py | 4 ++-- packages/modules/io_devices/add_on/api.py | 7 ++++--- packages/modules/io_devices/eebus/api.py | 6 +++--- 7 files changed, 29 insertions(+), 28 deletions(-) diff --git a/packages/helpermodules/command.py b/packages/helpermodules/command.py index 80c612dbc5..f29450b11c 100644 --- a/packages/helpermodules/command.py +++ b/packages/helpermodules/command.py @@ -1238,15 +1238,15 @@ def __on_message_rm(self, client, userdata, msg): elif re.search("openWB/chargepoint/template/[0-9]+$", msg.topic) is not None: for cp in SubData.cp_data.values(): if cp.chargepoint.data.config.template == int(msg.topic.split("/")[-1]): - pub_single(f'openWB/set/chargepoint/{cp.chargepoint.num}/config/template', 0) + Pub().pub(f'openWB/set/chargepoint/{cp.chargepoint.num}/config/template', 0) elif re.search("openWB/vehicle/template/charge_template/[0-9]+$", msg.topic) is not None: for vehicle in SubData.ev_data.values(): if vehicle.data.charge_template == int(msg.topic.split("/")[-1]): - pub_single(f'openWB/set/vehicle/{vehicle.num}/charge_template', 0) + Pub().pub(f'openWB/set/vehicle/{vehicle.num}/charge_template', 0) elif re.search("openWB/vehicle/template/ev_template/[0-9]+$", msg.topic) is not None: for vehicle in SubData.ev_data.values(): if vehicle.data.ev_template == int(msg.topic.split("/")[-1]): - pub_single(f'openWB/set/vehicle/{vehicle.num}/ev_template', 0) + Pub().pub(f'openWB/set/vehicle/{vehicle.num}/ev_template', 0) except Exception: log.exception("Fehler in ProcessBrokerBranch") diff --git a/packages/helpermodules/setdata.py b/packages/helpermodules/setdata.py index 5d6e0152fe..01f9d4a540 100644 --- a/packages/helpermodules/setdata.py +++ b/packages/helpermodules/setdata.py @@ -13,7 +13,7 @@ from control import data from helpermodules import hardware_configuration, subdata from helpermodules.broker import BrokerClient -from helpermodules.pub import Pub, pub_single +from helpermodules.pub import Pub from helpermodules.utils.topic_parser import decode_payload, get_index, get_index_position from helpermodules.update_config import UpdateConfig import dataclass_utils @@ -1183,9 +1183,9 @@ def process_legacy_smart_home_topic(self, msg): if ("openWB/set/LegacySmartHome/config" in msg.topic or "openWB/set/LegacySmartHome/Devices" in msg.topic): index = get_index(msg.topic) if "openWB/set/LegacySmartHome/config" in msg.topic: - pub_single(msg.topic.replace('openWB/set/', 'openWB/', 1), msg.payload.decode("utf-8"), - retain=True, no_json=True, port=1886) - pub_single(msg.topic, "", no_json=True, port=1886) + Pub().pub(msg.topic.replace('openWB/set/', 'openWB/', 1), msg.payload.decode("utf-8"), + retain=True, no_json=True) + Pub().pub(msg.topic, "", no_json=True) with open(self._get_ramdisk_path()/"rereadsmarthomedevices", 'w') as f: f.write(str(1)) if f"openWB/set/LegacySmartHome/config/set/Devices/{index}/mode" in msg.topic: diff --git a/packages/modules/common/fault_state.py b/packages/modules/common/fault_state.py index ec944c1385..854787f104 100644 --- a/packages/modules/common/fault_state.py +++ b/packages/modules/common/fault_state.py @@ -2,7 +2,8 @@ import traceback from typing import Optional, Callable, TypeVar -from helpermodules import exceptions, pub +from helpermodules import exceptions +from helpermodules.pub import Pub from helpermodules.constants import NO_ERROR from modules.common import component_type from modules.common.component_setup import ComponentSetup @@ -52,13 +53,13 @@ def store_error(self) -> None: topic_prefix = f"openWB/set/{topic}" else: topic_prefix = f"openWB/set/{topic}/{self.component_info.id}" - pub.Pub().pub(f"{topic_prefix}/get/fault_str", self.fault_str) - pub.Pub().pub(f"{topic_prefix}/get/fault_state", self.fault_state.value) + Pub().pub(f"{topic_prefix}/get/fault_str", self.fault_str) + Pub().pub(f"{topic_prefix}/get/fault_state", self.fault_state.value) if self.component_info.type == "internal_chargepoint": - pub.pub_single(f"openWB/set/chargepoint/{self.component_info.hierarchy_id}/get/fault_str", - self.fault_str) - pub.pub_single(f"openWB/set/chargepoint/{self.component_info.hierarchy_id}/get/fault_state", - self.fault_state.value) + Pub().pub(f"openWB/set/chargepoint/{self.component_info.hierarchy_id}/get/fault_str", + self.fault_str) + Pub().pub(f"openWB/set/chargepoint/{self.component_info.hierarchy_id}/get/fault_state", + self.fault_state.value) except Exception: log.exception("Fehler im Modul fault_state") diff --git a/packages/modules/internal_chargepoint_handler/internal_chargepoint_handler.py b/packages/modules/internal_chargepoint_handler/internal_chargepoint_handler.py index 4d1be1d371..30fcf6b12b 100644 --- a/packages/modules/internal_chargepoint_handler/internal_chargepoint_handler.py +++ b/packages/modules/internal_chargepoint_handler/internal_chargepoint_handler.py @@ -5,10 +5,9 @@ import time from typing import Optional from helpermodules import timecheck -from helpermodules import pub from helpermodules.logger import clear_in_memory_log_handler -from helpermodules.pub import Pub, pub_single +from helpermodules.pub import Pub from helpermodules.subdata import SubData from modules.chargepoints.internal_openwb.config import InternalChargepointMode from modules.common.component_context import SingleComponentUpdateContext @@ -55,11 +54,11 @@ def update_state(self, data: InternalChargepoint, heartbeat_expired: bool) -> No " noch aktiv. Es muss erst gewartet werden, bis die CP-Unterbrechung abgeschlossen ist.") return self.cp_module.set_current(set_current) - pub_single(f"openWB/set/chargepoint/{self.hierarchy_id}/set/current", payload=set_current) + Pub().pub(f"openWB/set/chargepoint/{self.hierarchy_id}/set/current", payload=set_current) if data.trigger_phase_switch: log.debug("Switch Phases from "+str(self.old_phases_to_use) + " to " + str(data.phases_to_use)) self.__thread_phase_switch(data.phases_to_use) - pub.pub_single( + Pub().pub( f"openWB/set/internal_chargepoint/{self.cp_module.local_charge_point_num}/data/trigger_phase_switch", False) @@ -209,8 +208,8 @@ def __init__(self, self.update_state = UpdateState(self.module, hierarchy_id) self.old_plug_state = False if global_data.parent_ip != "localhost": - pub_single(f"openWB/set/chargepoint/{hierarchy_id}/get/state_str", - payload="Statusmeldungen bitte auf der Primary-openWB einsehen.") + Pub().pub(f"openWB/set/chargepoint/{hierarchy_id}/get/state_str", + payload="Statusmeldungen bitte auf der Primary-openWB einsehen.") def update(self, global_data: GlobalHandlerData, data: InternalChargepointData, rfid_data: RfidData) -> bool: def __thread_active(thread: Optional[Thread]) -> bool: diff --git a/packages/modules/internal_chargepoint_handler/rfid.py b/packages/modules/internal_chargepoint_handler/rfid.py index 25f303c55a..46878c99e7 100644 --- a/packages/modules/internal_chargepoint_handler/rfid.py +++ b/packages/modules/internal_chargepoint_handler/rfid.py @@ -2,7 +2,7 @@ import logging from evdev import InputDevice, ecodes, list_devices, categorize -from helpermodules import pub +from helpermodules.pub import Pub log = logging.getLogger(__name__) @@ -128,7 +128,7 @@ async def _read_events(self, device: InputDevice): if data.scancode in (ecodes.KEY_ENTER, ecodes.KEY_KPENTER): if len(key_string) > 0: log.debug(f"RFID-String: {key_string}") - pub.pub_single("openWB/set/internal_chargepoint/last_tag", key_string) + Pub().pub("openWB/set/internal_chargepoint/last_tag", key_string) key_string = "" else: log.debug(f"new key: {data.scancode} - {ecodes.KEY[data.scancode]}") diff --git a/packages/modules/io_devices/add_on/api.py b/packages/modules/io_devices/add_on/api.py index 80552614d1..a94371e115 100644 --- a/packages/modules/io_devices/add_on/api.py +++ b/packages/modules/io_devices/add_on/api.py @@ -2,7 +2,7 @@ import logging from typing import Dict, Optional, Tuple -from helpermodules import pub +from helpermodules.pub import Pub, pub_single from helpermodules.broker import BrokerClient from helpermodules.utils.topic_parser import decode_payload from modules.common.abstract_device import DeviceDescriptor @@ -22,9 +22,10 @@ def read() -> Tuple[bool, bool]: def write(analog_output: Optional[Dict[str, int]], digital_output: Optional[Dict[str, bool]]): if config.configuration.host is None: raise ValueError("No host configured") - pub.pub_single("openWB/set/internal_io/states/set/digital_output", digital_output, + if config.configuration.host != "localhost": + pub_single("openWB/set/internal_io/states/set/digital_output", digital_output, hostname=config.configuration.host) - pub.pub_single("openWB/set/internal_io/states/set/digital_output", digital_output) + Pub().pub("openWB/set/internal_io/states/set/digital_output", digital_output) return ConfigurableIo(config=config, component_reader=read, component_writer=write) diff --git a/packages/modules/io_devices/eebus/api.py b/packages/modules/io_devices/eebus/api.py index 2b560f7fa3..9ad639ec95 100644 --- a/packages/modules/io_devices/eebus/api.py +++ b/packages/modules/io_devices/eebus/api.py @@ -8,7 +8,7 @@ from dataclass_utils._dataclass_asdict import asdict from helpermodules import timecheck from helpermodules.broker import BrokerClient -from helpermodules.pub import pub_single +from helpermodules.pub import Pub from helpermodules.utils.run_command import run_command from helpermodules.utils._thread_handler import is_thread_alive, thread_handler from helpermodules.utils.topic_parser import decode_payload @@ -61,7 +61,7 @@ def run(): if e.returncode == 2: msg = ("Zertifikat oder Key ungültig. Wenn das Zertifikat abgelaufen ist, bitte in " "den Einstellungen ein neues Zertifikat generieren und den SKI beim VNB " - "akutalisieren.") + "aktualisieren.") control_command_log.error(msg) thread_exception = ValueError(msg) else: @@ -178,4 +178,4 @@ def create_pub_cert_ski(id: int): config.configuration.cert_info = cert_info with open(f"{cert_path}/ski-{id}", "w") as ski_file: ski_file.write(cert_info.client_ski) - pub_single(f"openWB/set/system/io/{config.id}/config", asdict(config)) + Pub().pub(f"openWB/set/system/io/{config.id}/config", asdict(config)) From 480a00ecd0609356c8b4918aa2285bcc99f13034 Mon Sep 17 00:00:00 2001 From: Lutz Bender Date: Mon, 9 Mar 2026 08:31:26 +0100 Subject: [PATCH 2/3] cleanup pub_single in chargepoint_module.py --- .../external_openwb/chargepoint_module.py | 43 ++++++++----------- 1 file changed, 17 insertions(+), 26 deletions(-) diff --git a/packages/modules/chargepoints/external_openwb/chargepoint_module.py b/packages/modules/chargepoints/external_openwb/chargepoint_module.py index ffda83d504..5814113e22 100644 --- a/packages/modules/chargepoints/external_openwb/chargepoint_module.py +++ b/packages/modules/chargepoints/external_openwb/chargepoint_module.py @@ -33,16 +33,10 @@ def set_current(self, current: float) -> None: current = 0 with SingleComponentUpdateContext(self.fault_state, update_always=False): with self.client_error_context: - if self.config.configuration.duo_num == 0: - pub.pub_single("openWB/set/internal_chargepoint/0/data/set_current", current, - hostname=self.config.configuration.ip_address) - pub.pub_single("openWB/set/isss/Current", current, - hostname=self.config.configuration.ip_address) - else: - pub.pub_single("openWB/set/internal_chargepoint/1/data/set_current", current, - hostname=self.config.configuration.ip_address) - pub.pub_single("openWB/set/isss/Lp2Current", current, - hostname=self.config.configuration.ip_address) + pub.pub_single(f"openWB/set/internal_chargepoint/{self.config.configuration.duo_num}/data/set_current", + current, hostname=self.config.configuration.ip_address) + pub.pub_single(f"openWB/set/isss/{'' if self.config.configuration.duo_num == 0 else 'Lp2'}Current", + current, hostname=self.config.configuration.ip_address) def get_values(self) -> None: def parse_received_topics(value: str): @@ -59,14 +53,11 @@ def parse_received_topics(value: str): {"heartbeat": timecheck.create_timestamp(), "parent_ip": my_ip_address}, hostname=ip_address) pub.pub_single("openWB/set/isss/heartbeat", 0, hostname=ip_address) - pub.pub_single("openWB/set/isss/parentWB", my_ip_address, - hostname=ip_address, no_json=True) - if (self.config.configuration.duo_num == 1): - pub.pub_single("openWB/set/internal_chargepoint/1/data/parent_cp", str(num), hostname=ip_address) - pub.pub_single("openWB/set/isss/parentCPlp2", str(num), hostname=ip_address) - else: - pub.pub_single("openWB/set/internal_chargepoint/0/data/parent_cp", str(num), hostname=ip_address) - pub.pub_single("openWB/set/isss/parentCPlp1", str(num), hostname=ip_address) + pub.pub_single("openWB/set/isss/parentWB", my_ip_address, hostname=ip_address, no_json=True) + pub.pub_single(f"openWB/set/internal_chargepoint/{self.config.configuration.duo_num}/data/parent_cp", + str(num), hostname=ip_address) + 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): client.subscribe(f"openWB/internal_chargepoint/{self.config.configuration.duo_num}/get/#") @@ -152,14 +143,14 @@ def interrupt_cp(self, duration: int) -> None: with SingleComponentUpdateContext(self.fault_state, update_always=False): with self.client_error_context: ip_address = self.config.configuration.ip_address - if (self.config.configuration.duo_num == 1): - pub.pub_single("openWB/set/internal_chargepoint/1/data/cp_interruption_duration", - duration, hostname=ip_address) - pub.pub_single("openWB/set/isss/Cpulp2", duration, hostname=ip_address) - else: - pub.pub_single("openWB/set/internal_chargepoint/0/data/cp_interruption_duration", - duration, hostname=ip_address) - pub.pub_single("openWB/set/isss/Cpulp1", duration, hostname=ip_address) + pub.pub_single( + "openWB/set/internal_chargepoint/" + f"{self.config.configuration.duo_num}/data/cp_interruption_duration", + duration, + hostname=ip_address + ) + pub.pub_single(f"openWB/set/isss/Cpulp{self.config.configuration.duo_num + 1}", + duration, hostname=ip_address) time.sleep(duration) def clear_rfid(self) -> None: From 6b95fd05f8b9ccd92a1354e90499204395e19a2f Mon Sep 17 00:00:00 2001 From: Lutz Bender Date: Mon, 9 Mar 2026 08:31:54 +0100 Subject: [PATCH 3/3] catch localhost in pub_single --- packages/helpermodules/pub.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/packages/helpermodules/pub.py b/packages/helpermodules/pub.py index 1f5f15c2b2..aca8403bdd 100644 --- a/packages/helpermodules/pub.py +++ b/packages/helpermodules/pub.py @@ -32,19 +32,23 @@ def __getattr__(self, name): def pub_single(topic, payload, hostname="localhost", port=1883, no_json=False, retain=True): - """ published eine einzelne Nachricht an einen Host, der nicht der localhost ist. + """ Sendet eine einzelne Nachricht an einen Host. Parameter --------- topic : str - Topic, an das gepusht werden soll + Topic, an das gesendet werden soll payload : int, str, list, float - Payload, der gepusht werden soll. Nicht als json, da ISSS kein json-Payload verwendet. + Payload, der gesendet werden soll. Nicht als json, da ISSS kein json-Payload verwendet. hostname: str IP des Hosts no_json: bool Kompatibilität mit ISSS, die ramdisk verwenden. """ + if hostname == "localhost": + Pub().pub(topic, payload, qos=0, retain=retain) + return + if no_json: publish.single(topic, payload, hostname=hostname, port=port, retain=retain) else: