Feature modbusserver (openWB#1115)

* Modbus-Steuerung

pytest

fix

flake8

pytest

topic

add update reg

fix

fix

improve tester

flake8

fix typo

fix folder structure

start modbus server only in secondary mode

automate modbus test

* Apply suggestions from code review

Co-authored-by: benderl <[email protected]>

* fixes

* fix pytest

---------

Co-authored-by: benderl <[email protected]>

.github/workflows/github-actions-python.yml

@@ -15,6 +15,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install flake8 pytest paho-mqtt requests-mock jq pyjwt==2.6.0 bs4 pkce typing_extensions python-dateutil==2.8.2
+          pip install umodbus
       - name: Flake8 with annotations in packages folder
         uses: TrueBrain/[email protected]
         with:

packages/control/chargepoint/chargepoint.py

@@ -132,6 +132,7 @@ class Get:
     currents: List[float] = field(default_factory=currents_list_factory)
     daily_imported: float = 0
     daily_exported: float = 0
+    evse_current: float = 0
     exported: float = 0
     fault_str: str = "Kein Fehler."
     fault_state: int = 0

packages/helpermodules/command_test.py

@@ -16,7 +16,7 @@
 
 @pytest.fixture
 def subdata_fixture() -> None:
-    SubData(*([Mock()]*16))
+    SubData(*([Mock()]*17))
     SubData.cp_data = {"cp0":  Mock(spec=ChargepointStateUpdate, chargepoint=Mock(
         spec=Chargepoint, chargepoint_module=Mock(spec=ChargepointModulePro)))}
 

packages/helpermodules/hardware_configuration.py

@@ -16,5 +16,13 @@ def get_hardware_configuration_setting(name: str):
         return json.loads(f.read())[name]
 
 
+def get_serial_number() -> str:
+    try:
+        with open("/home/openwb/snnumber", "r") as file:
+            return file.read().replace("\n", "")
+    except FileNotFoundError:
+        return "noSerialNumber"
+
+
 if __name__ == "__main__":
     update_hardware_configuration(json.loads(sys.argv[1]))

packages/helpermodules/logger.py

@@ -62,6 +62,7 @@ def mb_to_bytes(megabytes: int) -> int:
     urllib3_log.addHandler(urllib3_file_handler)
 
     logging.getLogger("pymodbus").setLevel(logging.WARNING)
+    logging.getLogger("uModbus").setLevel(logging.WARNING)
 
 
 log = logging.getLogger(__name__)

packages/helpermodules/modbusserver.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python
+import logging
+from socketserver import TCPServer
+from collections import defaultdict
+import struct
+from umodbus import conf
+from umodbus.server.tcp import RequestHandler, get_server
+from umodbus.utils import log_to_stream
+
+from helpermodules import timecheck
+from helpermodules.hardware_configuration import get_serial_number
+from helpermodules.pub import Pub
+from helpermodules.subdata import SubData
+
+
+log = logging.getLogger(__name__)
+
+try:
+    log_to_stream(level=logging.DEBUG)
+    data_store = defaultdict(int)
+    conf.SIGNED_VALUES = True
+    TCPServer.allow_reuse_address = True
+    app = get_server(TCPServer, ('0.0.0.0', 1502), RequestHandler)
+
+    serial_number = get_serial_number().replace("snnumber=", "")
+except (Exception, OSError):
+    log.exception("Fehler im Modbus-Server")
+
+
+def _form_int32(value, startreg):
+    secondreg = startreg + 1
+    try:
+        binary32 = struct.pack('>l', int(value))
+        high_byte, low_byte = struct.unpack('>hh', binary32)
+        data_store[startreg] = high_byte
+        data_store[secondreg] = low_byte
+    except Exception:
+        log.exception("Fehler beim Füllen der Register")
+        data_store[startreg] = -1
+        data_store[secondreg] = -1
+
+
+def _form_int16(value, startreg):
+    try:
+        value = int(value)
+        if (value > 32767 or value < -32768):
+            raise Exception("Number to big")
+        data_store[startreg] = value
+    except Exception:
+        log.exception("Fehler beim Füllen der Register")
+        data_store[startreg] = -1
+
+
+def _form_str(value: str, startreg):
+    bytes = value.encode("utf-8")
+    length = len(bytes)
+    if length > 20:
+        raise ValueError("String darf max 20 Zeichen enthalten.")
+    register_offset = 0
+    for i in range(0, length, 2):
+        try:
+            if i < length-1:
+                stream_two_bytes = struct.pack(">bb", bytes[i], bytes[i+1])
+                stream_one_word = struct.unpack(">h", stream_two_bytes)[0]
+            else:
+                stream_two_bytes = struct.pack(">bb", bytes[i], 0)
+                stream_one_word = struct.unpack(">h", stream_two_bytes)[0]
+            data_store[startreg+register_offset] = stream_one_word
+        except Exception:
+            data_store[startreg+register_offset] = -1
+        finally:
+            register_offset += 1
+
+
+def _get_pos(number, n):
+    return number // 10**n % 10 - 1
+
+
+try:
+    @app.route(slave_ids=[1], function_codes=[3, 4], addresses=list(range(0, 32000)))
+    def read_data_store(slave_id, function_code, address):
+        """" Return value of address. """
+        if address > 10099:
+            Pub().pub("openWB/set/internal_chargepoint/global_data",
+                      {"heartbeat": timecheck.create_timestamp_unix(), "parent_ip": None})
+            chargepoint = SubData.internal_chargepoint_data[f"cp{_get_pos(address, 2)}"]
+            askedvalue = int(str(address)[-2:])
+            if askedvalue == 00:
+                _form_int32(chargepoint.get.power, address)
+            elif askedvalue == 2:
+                _form_int32(chargepoint.get.imported, address)
+            elif 4 <= askedvalue <= 6:
+                _form_int16(chargepoint.get.voltages[askedvalue-4]*100, address)
+            elif 7 <= askedvalue <= 9:
+                _form_int16(chargepoint.get.currents[askedvalue-7]*100, address)
+            elif askedvalue == 14:
+                _form_int16(chargepoint.get.plug_state, address)
+            elif askedvalue == 15:
+                _form_int16(chargepoint.get.charge_state, address)
+            elif askedvalue == 16:
+                _form_int16(chargepoint.get.evse_current, address)
+            elif 30 <= askedvalue <= 32:
+                _form_int16(chargepoint.get.powers[askedvalue-30], address)
+            elif askedvalue == 41:
+                _form_int32(chargepoint.get.exported, address)
+            elif askedvalue == 43:
+                _form_int16(1, address)
+            elif askedvalue == 50:
+                _form_str(serial_number, address)
+            elif askedvalue == 60:
+                _form_str(chargepoint.get.rfid, address)
+
+        return data_store[address]
+except Exception:
+    log.exception("Fehler im Modbus-Server")
+
+try:
+    @app.route(slave_ids=[1], function_codes=[6, 16], addresses=list(range(0, 32000)))
+    def write_data_store(slave_id, function_code, address, value):
+        """" Set value for address. """
+        if 10170 < address:
+            cp_topic = f"openWB/set/internal_chargepoint/{_get_pos(address, 2)}/data/"
+            askedvalue = int(str(address)[-2:])
+            if askedvalue == 71:
+                Pub().pub(f"{cp_topic}set_current", value/100)
+            elif askedvalue == 80:
+                Pub().pub(f"{cp_topic}phases_to_use", value)
+            elif askedvalue == 81:
+                Pub().pub(f"{cp_topic}trigger_phase_switch", value)
+            elif askedvalue == 98:
+                Pub().pub(f"{cp_topic}cp_interruption_duration", value)
+            elif askedvalue == 99:
+                Pub().pub("openWB/set/command/modbus_server/todo", {"command": "systemUpdate", "data": {}})
+except Exception:
+    log.exception("Fehler im Modbus-Server")
+
+
+def start_modbus_server(event_modbus_server):
+    try:
+        # Wenn start_modbus_server aus SubData aufegrufen wird, wenn das Topic gesetzt wird, führt das zu einem
+        # circular Import.
+        event_modbus_server.wait()
+        event_modbus_server.clear()
+        log.debug("Starte Modbus-Server")
+        app.serve_forever()
+    finally:
+        try:
+            app.shutdown()
+            app.server_close()
+        except Exception:
+            log.exception("Fehler im Modbus-Server")

packages/helpermodules/setdata.py

@@ -581,6 +581,8 @@ def process_chargepoint_get_topics(self, msg):
             self._validate_value(msg, bool)
         elif "/get/fault_state" in msg.topic:
             self._validate_value(msg, int, [(0, 2)])
+        elif "/get/evse_current" in msg.topic:
+            self._validate_value(msg, int, [(0, 0), (6, 32), (600, 3200)])
         elif ("/get/fault_str" in msg.topic or
                 "/get/state_str" in msg.topic or
                 "/get/heartbeat" in msg.topic):
@@ -698,7 +700,8 @@ def process_general_topic(self, msg: mqtt.MQTTMessage):
         try:
             if "openWB/set/general/extern_display_mode" in msg.topic:
                 self._validate_value(msg, str)
-            elif "openWB/set/general/extern" in msg.topic:
+            elif ("openWB/set/general/modbus_control" in msg.topic or
+                  "openWB/set/general/extern" in msg.topic):
                 self._validate_value(msg, bool)
             elif "openWB/set/general/control_interval" in msg.topic:
                 self._validate_value(msg, int, [(10, 10), (20, 20), (60, 60)])

packages/helpermodules/subdata.py

@@ -31,7 +31,7 @@
 from modules.common.abstract_vehicle import CalculatedSocState, GeneralVehicleConfig
 from modules.common.simcount.simcounter_state import SimCounterState
 from modules.internal_chargepoint_handler.internal_chargepoint_handler_config import (
-    GlobalHandlerData, InternalChargepointHandlerData, RfidData)
+    GlobalHandlerData, InternalChargepoint, RfidData)
 from modules.vehicles.manual.config import ManualSoc
 
 log = logging.getLogger(__name__)
@@ -57,9 +57,9 @@ class SubData:
     bat_all_data = bat_all.BatAll()
     bat_data: Dict[str, bat.Bat] = {}
     general_data = general.General()
-    internal_chargepoint_data: Dict[str, Union[InternalChargepointHandlerData, GlobalHandlerData, RfidData]] = {
-        "cp0": InternalChargepointHandlerData(),
-        "cp1": InternalChargepointHandlerData(),
+    internal_chargepoint_data: Dict[str, Union[InternalChargepoint, GlobalHandlerData, RfidData]] = {
+        "cp0": InternalChargepoint(),
+        "cp1": InternalChargepoint(),
         "global_data": GlobalHandlerData(),
         "rfid_data": RfidData()}
     optional_data = optional.Optional()
@@ -82,7 +82,8 @@ def __init__(self,
                  event_stop_internal_chargepoint: threading.Event,
                  event_update_config_completed: threading.Event,
                  event_soc: threading.Event,
-                 event_jobs_running: threading.Event):
+                 event_jobs_running: threading.Event,
+                 event_modbus_server: threading.Event,):
         self.event_ev_template = event_ev_template
         self.event_charge_template = event_charge_template
         self.event_cp_config = event_cp_config
@@ -99,6 +100,7 @@ def __init__(self,
         self.event_update_config_completed = event_update_config_completed
         self.event_soc = event_soc
         self.event_jobs_running = event_jobs_running
+        self.event_modbus_server = event_modbus_server
         self.heartbeat = False
 
     def sub_topics(self):
@@ -573,6 +575,9 @@ def process_general_topic(self, var: general.General, msg: mqtt.MQTTMessage):
                     subprocess.run([
                         str(Path(__file__).resolve().parents[2] / "runs" / "setup_network.sh")
                     ])
+                elif "openWB/general/modbus_control" == msg.topic:
+                    if decode_payload(msg.payload) and self.general_data.data.extern:
+                        self.event_modbus_server.set()
                 else:
                     self.set_json_payload_class(var.data, msg)
         except Exception:
@@ -781,14 +786,17 @@ def process_internal_chargepoint_topic(self, client, var, msg):
         try:
             if re.search("/internal_chargepoint/[0-1]/data/parent_cp", msg.topic) is not None:
                 index = get_index(msg.topic)
-                if decode_payload(msg.payload) != var[f"cp{index}"].parent_cp:
+                if decode_payload(msg.payload) != var[f"cp{index}"].data.parent_cp:
                     log.debug("Neustart des Handlers für den internen Ladepunkt.")
                     self.event_stop_internal_chargepoint.set()
                     self.event_start_internal_chargepoint.set()
                 self.set_json_payload_class(var[f"cp{index}"], msg)
             elif re.search("/internal_chargepoint/[0-1]/", msg.topic) is not None:
                 index = get_index(msg.topic)
-                self.set_json_payload_class(var[f"cp{index}"], msg)
+                if re.search("/internal_chargepoint/[0-1]/data/", msg.topic) is not None:
+                    self.set_json_payload_class(var[f"cp{index}"].data, msg)
+                elif re.search("/internal_chargepoint/[0-1]/get/", msg.topic) is not None:
+                    self.set_json_payload_class(var[f"cp{index}"].get, msg)
             elif "internal_chargepoint/global_data" in msg.topic:
                 self.set_json_payload_class(var["global_data"], msg)
                 if decode_payload(msg.payload)["parent_ip"] != var["global_data"].parent_ip:

packages/helpermodules/update_config.py

@@ -132,6 +132,7 @@ class UpdateConfig:
         "^openWB/general/external_buttons_hw$",
         "^openWB/general/grid_protection_configured$",
         "^openWB/general/grid_protection_active$",
+        "^openWB/general/modbus_control$",
         "^openWB/general/mqtt_bridge$",
         "^openWB/general/grid_protection_timestamp$",
         "^openWB/general/grid_protection_random_stop$",
@@ -401,6 +402,7 @@ class UpdateConfig:
         ("openWB/general/extern_display_mode", "primary"),
         ("openWB/general/external_buttons_hw", False),
         ("openWB/general/grid_protection_configured", True),
+        ("openWB/general/modbus_control", False),
         ("openWB/general/notifications/selected", "none"),
         ("openWB/general/notifications/plug", False),
         ("openWB/general/notifications/start_charging", False),

packages/main.py

@@ -11,7 +11,6 @@
 from threading import Thread
 from helpermodules.measurement_logging.update_daily_yields import update_daily_yields
 from helpermodules.measurement_logging.write_log import save_log
-from helpermodules.pub import Pub
 
 from modules import loadvars
 from modules import configuration
@@ -20,6 +19,8 @@
 from helpermodules import setdata
 from helpermodules import logger
 from helpermodules import command
+from helpermodules.modbusserver import start_modbus_server
+from helpermodules.pub import Pub
 from control import prepare
 from control import data
 from control import process
@@ -188,6 +189,7 @@ def schedule_jobs():
     event_command_completed.set()
     event_subdata_initialized = threading.Event()
     event_update_config_completed = threading.Event()
+    event_modbus_server = threading.Event()
     event_jobs_running = threading.Event()
     event_jobs_running.set()
     prep = prepare.Prepare()
@@ -204,7 +206,7 @@ def schedule_jobs():
                           general_internal_chargepoint_handler.event_stop,
                           event_update_config_completed,
                           event_soc,
-                          event_jobs_running)
+                          event_jobs_running, event_modbus_server)
     comm = command.Command(event_command_completed)
     t_sub = Thread(target=sub.sub_topics, args=(), name="Subdata")
     t_set = Thread(target=set.set_data, args=(), name="Setdata")
@@ -224,6 +226,7 @@ def schedule_jobs():
     t_comm.start()
     t_soc.start()
     t_internal_chargepoint.start()
+    threading.Thread(target=start_modbus_server, args=(event_modbus_server,), name="Modbus Control Server").start()
     # Warten, damit subdata Zeit hat, alle Topics auf dem Broker zu empfangen.
     event_update_config_completed.wait(300)
     Pub().pub("openWB/set/system/boot_done", True)