From c232891a3da9890c6e5143913ea257de644f370e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Petr=20Kub=C3=A1nek?= Date: Tue, 25 Aug 2026 00:08:26 +0200 Subject: [PATCH] Reduces complexity of adding simulator datatypes. - Uses enum features to replace ad-hoc helper objects. - Introduce defaults for simulator config - .json parsing doesn't fail if some values or sections are missing (e.g. not all project need int64). --- pymodbus/datastore/simulator.py | 207 +++++++++------------ pymodbus/server/simulator/http_server.py | 8 +- test/datastore/test_simulator_datastore.py | 6 + 3 files changed, 93 insertions(+), 128 deletions(-) diff --git a/pymodbus/datastore/simulator.py b/pymodbus/datastore/simulator.py index 13785612b..28d5bddc1 100644 --- a/pymodbus/datastore/simulator.py +++ b/pymodbus/datastore/simulator.py @@ -3,11 +3,11 @@ from __future__ import annotations import dataclasses +import enum import random import struct from collections.abc import Callable from datetime import datetime -from enum import IntEnum from typing import Any from ..constants import ExcCodes @@ -17,17 +17,21 @@ WORD_SIZE = 16 -class CellType(IntEnum): +class CellType(enum.IntEnum): """Define single cell types.""" - INVALID = 0 - BITS = 1 - UINT16 = 2 - UINT32 = 3 - FLOAT32 = 4 - FLOAT64 = 5 - STRING = 6 - NEXT = 7 + INVALID = enum.auto() + BITS = enum.auto() + INT16 = enum.auto() + UINT16 = enum.auto() + INT32 = enum.auto() + UINT32 = enum.auto() + INT64 = enum.auto() + UINT64 = enum.auto() + FLOAT32 = enum.auto() + FLOAT64 = enum.auto() + STRING = enum.auto() + NEXT = enum.auto() @classmethod def register_count(cls, celltype: CellType) -> int: @@ -37,7 +41,7 @@ def register_count(cls, celltype: CellType) -> int: f"Invalid call to register_count with type {CellType(celltype).name} ({celltype})." ) - if celltype in [cls.BITS, cls.UINT16]: + if celltype in [cls.BITS, cls.INT16, cls.UINT16]: return 1 if cls.is_64(celltype): return 4 @@ -51,7 +55,7 @@ def is_int(cls, celltype: CellType) -> bool: @classmethod def is_64(cls, celltype: CellType) -> bool: """Return True if the given cell type represents 64=bit (8 bytes, 4 registers) value.""" - return celltype in [cls.FLOAT64] + return celltype in [cls.INT64, cls.UINT64, cls.FLOAT64] @dataclasses.dataclass(repr=False, eq=False) @@ -110,8 +114,12 @@ class Label: # pylint: disable=too-many-instance-attributes type: str = "type" type_bits = "bits" type_exception: str = "type exception" + type_int16: str = "int16" type_uint16: str = "uint16" + type_int32: str = "int32" type_uint32: str = "uint32" + type_int64: str = "int64" + type_uint64: str = "uint64" type_float32: str = "float32" type_float64: str = "float64" type_string: str = "string" @@ -120,11 +128,24 @@ class Label: # pylint: disable=too-many-instance-attributes write: str = "write" @classmethod - def try_get(cls, key, config_part): + def try_get(cls, key: str, config_part: dict[str, Any]) -> Any: """Check if entry is present in config.""" if key not in config_part: - txt = f"ERROR Configuration invalid, missing {key} in {config_part}" - raise RuntimeError(txt) + raise RuntimeError( + f"ERROR Configuration invalid, missing {key} in {config_part}" + ) + return config_part[key] + + @classmethod + def try_get_default( + cls, + key: str, + config_part: dict[str, Any], + default_value: Any, + ) -> Any: + """Check if entry is present in config.""" + if key not in config_part: + return default_value return config_part[key] @@ -136,6 +157,7 @@ class Setup: def __init__(self, runtime: Any) -> None: """Initialize.""" + super().__init__() self.runtime = runtime self.config: Any = {} self.config_types: dict[str, dict[str, Any]] = { @@ -144,120 +166,77 @@ def __init__(self, runtime: Any) -> None: Label.next: None, Label.value: 0, Label.action: None, - Label.method: self.handle_type_bits, + }, + Label.type_int16: { + Label.type: CellType.UINT16, + Label.next: None, + Label.value: 0, + Label.action: None, }, Label.type_uint16: { Label.type: CellType.UINT16, Label.next: None, Label.value: 0, Label.action: None, - Label.method: self.handle_type_uint16, }, Label.type_uint32: { Label.type: CellType.UINT32, Label.next: CellType.NEXT, Label.value: 0, Label.action: None, - Label.method: self.handle_type_uint32, }, Label.type_float32: { Label.type: CellType.FLOAT32, Label.next: CellType.NEXT, Label.value: 0, Label.action: None, - Label.method: self.handle_type_float32, }, Label.type_float64: { Label.type: CellType.FLOAT64, Label.next: CellType.NEXT, Label.value: 0, Label.action: None, - Label.method: self.handle_type_float64, }, Label.type_string: { Label.type: CellType.STRING, Label.next: CellType.NEXT, Label.value: 0, Label.action: None, - Label.method: self.handle_type_string, }, } - def handle_type_bits(self, start, stop, value, action, action_parameters): - """Handle type bits.""" - for reg in self.runtime.registers[start:stop]: - if reg.type != CellType.INVALID: - raise RuntimeError(f'ERROR "{Label.type_bits}" {reg} used') - reg.value = value - reg.type = CellType.BITS - reg.action = action - reg.action_parameters = action_parameters - - def handle_type_uint16(self, start, stop, value, action, action_parameters): - """Handle type uint16.""" - for reg in self.runtime.registers[start:stop]: - if reg.type != CellType.INVALID: - raise RuntimeError(f'ERROR "{Label.type_uint16}" {reg} used') - reg.value = value - reg.type = CellType.UINT16 - reg.action = action - reg.action_parameters = action_parameters - - def handle_type_uint32(self, start, stop, value, action, action_parameters): - """Handle type uint32.""" - regs_value = ModbusSimulatorContext.build_registers_from_value( - value, CellType.UINT32 - ) - for i in range(start, stop, 2): - regs = self.runtime.registers[i : i + 2] - if regs[0].type != CellType.INVALID or regs[1].type != CellType.INVALID: - raise RuntimeError(f'ERROR "{Label.type_uint32}" {i},{i + 1} used') - regs[0].value = regs_value[0] - regs[0].type = CellType.UINT32 - regs[0].action = action - regs[0].action_parameters = action_parameters - regs[1].value = regs_value[1] - regs[1].type = CellType.NEXT - - def handle_type_float32(self, start, stop, value, action, action_parameters): - """Handle type float32.""" - regs_value = ModbusSimulatorContext.build_registers_from_value( - value, CellType.FLOAT32 - ) - for i in range(start, stop, 2): - regs = self.runtime.registers[i : i + 2] - if regs[0].type != CellType.INVALID or regs[1].type != CellType.INVALID: - raise RuntimeError(f'ERROR "{Label.type_float32}" {i},{i + 1} used') - regs[0].value = regs_value[0] - regs[0].type = CellType.FLOAT32 - regs[0].action = action - regs[0].action_parameters = action_parameters - regs[1].value = regs_value[1] - regs[1].type = CellType.NEXT - - def handle_type_float64(self, start, stop, value, action, action_parameters): - """Handle type float64.""" - regs_value = ModbusSimulatorContext.build_registers_from_value( - value, CellType.FLOAT64 - ) - for i in range(start, stop, 4): - regs = self.runtime.registers[i : i + 4] - if ( - regs[0].type != CellType.INVALID - or regs[1].type != CellType.INVALID - or regs[2].type != CellType.INVALID - or regs[3].type != CellType.INVALID - ): - raise RuntimeError( - f'ERROR "{Label.type_float64}" {i},{i + 1},{i + 2},{i + 3} used' - ) - regs[0].value = regs_value[0] - regs[0].type = CellType.FLOAT64 - regs[0].action = action - regs[0].action_parameters = action_parameters - for i in range(1, 4): - regs[i].value = regs_value[i] - regs[i].type = CellType.NEXT + def handle_type( + self, + celltype: CellType, + start: int, + stop: int, + value: int | float | str, + action, + action_parameters, + ) -> None: + """Handle type int16.""" + if celltype == CellType.STRING: + self.handle_type_string(start, stop, value, action, action_parameters) + return + + reg_count = CellType.register_count(celltype) + regs_value = ModbusSimulatorContext.build_registers_from_value(value, celltype) + for i in range(start, stop, reg_count): + regs = self.runtime.registers[i : i + reg_count] + + for value_index, reg in enumerate(regs): + if reg.type != CellType.INVALID: + raise RuntimeError( + f'ERROR "{CellType(celltype).name}" {i + value_index} used' + ) + + reg.value = regs_value[value_index] + if not value_index: + reg.type = celltype + reg.action = action + reg.action_parameters = action_parameters + else: + reg.type = CellType.NEXT def handle_type_string(self, start, stop, value, action, action_parameters): """Handle type string.""" @@ -309,9 +288,9 @@ def handle_setup_section(self): defaults_value = Label.try_get(Label.value, defaults) defaults_action = Label.try_get(Label.action, defaults) for key, entry in self.config_types.items(): - entry[Label.value] = Label.try_get(key, defaults_value) + entry[Label.value] = Label.try_get_default(key, defaults_value, None) if ( - action := Label.try_get(key, defaults_action) + action := Label.try_get_default(key, defaults_action, None) ) not in self.runtime.action_name_to_id: raise RuntimeError(f"ERROR illegal action {key} in {defaults_action}") entry[Label.action] = action @@ -351,7 +330,8 @@ def handle_write_allowed(self): def handle_types(self): """Handle the different types.""" for section, type_entry in self.config_types.items(): - layout = Label.try_get(section, self.config) + if (layout := Label.try_get_default(section, self.config, None)) is None: + continue for entry in layout: if not isinstance(entry, dict): entry = {Label.addr: entry} @@ -361,7 +341,8 @@ def handle_types(self): start = regs[0] if (stop := regs[1]) >= self.runtime.register_count: raise RuntimeError(f'Error "{section}" {start}, {stop} illegal') - type_entry[Label.method]( + getattr(self, "handle_type")( + CellType[section.upper()], start, stop + 1, entry.get(Label.value, type_entry[Label.value]), @@ -411,26 +392,6 @@ def setup(self, config, custom_actions) -> None: self.runtime.action_methods.append(method) self.runtime.action_name_to_id.update({None: 0}) - self.runtime.registerType_name_to_id = { - Label.type_bits: CellType.BITS, - Label.type_uint16: CellType.UINT16, - Label.type_uint32: CellType.UINT32, - Label.type_float32: CellType.FLOAT32, - Label.type_string: CellType.STRING, - Label.next: CellType.NEXT, - Label.invalid: CellType.INVALID, - } - self.runtime.registerType_id_to_name = [ - "invalid", # 0 - "bits", # 1 - "uint16", # 2 - "uint32", # 3 - "float32", # 4 - "float64", # 5 - "string", # 6 - "next", # 7 - ] - self.config = config self.handle_setup_section() self.handle_invalid_address() @@ -547,8 +508,6 @@ def __init__( self.action_name_to_id: dict[str, int] = {} self.action_id_to_name: list[str] = [] self.action_methods: list[Callable] = [] - self.registerType_name_to_id: dict[str, int] = {} - self.registerType_id_to_name: list[str] = [] if config: Setup(self).setup(config, custom_actions) Log.warning( @@ -565,7 +524,7 @@ def get_text_register(self, register): """Get raw register.""" reg = self.registers[register] text_cell = TextCell() - text_cell.type = self.registerType_id_to_name[reg.type] + text_cell.type = CellType(reg.type).name.lower() text_cell.access = str(reg.access) text_cell.count_read = str(reg.count_read) text_cell.count_write = str(reg.count_write) @@ -833,7 +792,7 @@ def action_uptime(cls, registers, inx, **_parameters): @classmethod def build_registers_from_value( - cls, value: int | float, celltype: CellType + cls, value: int | float | str, celltype: CellType ) -> list[int]: """Build registers from int32, float32 or float64.""" reg_count = CellType.register_count(celltype) diff --git a/pymodbus/server/simulator/http_server.py b/pymodbus/server/simulator/http_server.py index 36a4029ed..8157f99d4 100644 --- a/pymodbus/server/simulator/http_server.py +++ b/pymodbus/server/simulator/http_server.py @@ -15,7 +15,7 @@ from aiohttp import web from ...datastore import ModbusServerContext, ModbusSimulatorContext -from ...datastore.simulator import Label +from ...datastore.simulator import CellType, Label from ...logging import Log from ...pdu import DecodePDU from ...pdu.device import ModbusDeviceIdentification @@ -310,8 +310,8 @@ def build_html_registers(self, params, html): # pragma: no cover else: foot = "Nothing selected" register_types = "".join( - f"" - for name, reg_id in self.datastore_context.registerType_name_to_id.items() + f"" + for reg_id in CellType ) register_actions = "".join( f"" @@ -487,7 +487,7 @@ def build_json_registers(self, params): register_rows.append(row) # Generate register types and actions (assume these are predefined mappings) - register_types = dict(self.datastore_context.registerType_name_to_id) + register_types = {celltype.name: celltype.value for celltype in CellType} register_actions = dict(self.datastore_context.action_name_to_id) # Build the JSON response diff --git a/test/datastore/test_simulator_datastore.py b/test/datastore/test_simulator_datastore.py index 4b6deaab2..707066b7d 100644 --- a/test/datastore/test_simulator_datastore.py +++ b/test/datastore/test_simulator_datastore.py @@ -585,6 +585,12 @@ async def test_simulator_action_random(self, celltype, minval, maxval, device): ) assert minval <= new_value <= maxval + def test_simulator_label_try_get(self): + """Test Label.try_get method.""" + assert Label.try_get("valid", {"valid": 42}) == 42 + with pytest.raises(RuntimeError): + Label.try_get("invalid", {"valid": 10}) + def test_simulator_loop_validate(self, simulator): """Test simulator set values.""" assert not simulator.loop_validate(59, 60, False)