Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/main.yml
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ jobs:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
fail-fast: false
steps:
- name: Check out
Expand Down
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ Run the simulator without Docker with `python -m modsim --comm tcp --port 5020`.
The default TCP simulator uses one sequential datastore shared by all Unit IDs.
The CLI currently supports only `--store sequential`; `--slaves` and `--context`
are not implemented and are rejected. TCP and UDP use the socket framer, TLS uses
the TLS framer, and serial accepts RTU, ASCII, or binary framing. Serial requires a suitable local device and is not covered by the local protocol regression suite. TLS uses `examples/certificates/pymodbus.crt` and `pymodbus.key` relative to the working directory. A local test covers a self-signed server certificate and a verified client connection; the server does not authenticate client certificates.
the TLS framer, and serial accepts RTU or ASCII framing. Serial requires a suitable local device and is not covered by the local protocol regression suite. TLS uses `examples/certificates/pymodbus.crt` and `pymodbus.key` relative to the working directory. A local test covers a self-signed server certificate and a verified client connection; the server does not authenticate client certificates.

UDP mode (`--comm udp`) listens on the loopback address and has local read and
shutdown coverage. TCP and UDP report a startup error if the requested port is
Expand Down
2 changes: 1 addition & 1 deletion docs/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ addresses return a Modbus exception.
TCP and UDP use socket framing; UDP listens on loopback. TLS uses TLS framing
and expects `examples/certificates/pymodbus.crt` and `pymodbus.key` relative
to the working directory. The TLS server does not authenticate clients. Serial
supports RTU, ASCII and binary framing; physical serial devices have not been
supports RTU and ASCII framing; physical serial devices have not been
validated in the local regression suite.

Only the sequential datastore is implemented. The CLI rejects non-default
Expand Down
131 changes: 131 additions & 0 deletions modsim/compat.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
"""Compatibility layer for PyModbus API changes.

This module provides backward-compatible wrappers for APIs that changed
between PyModbus 3.6.9 and 3.16.1.
"""

import struct
from typing import List


class Endian:
"""Compatibility class for endianness constants removed in PyModbus 3.16.1."""

BIG = "big"
LITTLE = "little"


class BinaryPayloadBuilder:
"""Compatibility wrapper for PyModbus BinaryPayloadBuilder.

This class was removed in PyModbus 3.9.0+ but is needed for backward
compatibility with existing datastore initialization code.
"""

def __init__(self, byteorder: str = "big", wordorder: str = "big") -> None:
"""Initialize the builder.

Args:
byteorder: Either "big" or "little" for byte order
wordorder: Either "big" or "little" for word order (register order)
"""
self.byteorder = byteorder
self.wordorder = wordorder
self._bits: List[bool] = []
self._registers: List[int] = []

def add_bits(self, bits: List[bool]) -> "BinaryPayloadBuilder":
"""Add a list of bits to the payload.

Args:
bits: List of boolean values

Returns:
Self for chaining
"""
self._bits.extend(bits)
return self

def add_16bit_uint(self, value: int) -> "BinaryPayloadBuilder":
"""Add a 16-bit unsigned integer."""
self._registers.append(value & 0xFFFF)
return self

def add_16bit_int(self, value: int) -> "BinaryPayloadBuilder":
"""Add a 16-bit signed integer."""
# Convert negative to unsigned representation (two's complement)
if value < 0:
value = (1 << 16) + value
self._registers.append(value & 0xFFFF)
return self

def add_32bit_uint(self, value: int) -> "BinaryPayloadBuilder":
"""Add a 32-bit unsigned integer (uses 2 registers)."""
if self.wordorder == "big":
self._registers.append((value >> 16) & 0xFFFF)
self._registers.append(value & 0xFFFF)
else:
self._registers.append(value & 0xFFFF)
self._registers.append((value >> 16) & 0xFFFF)
return self

def add_32bit_int(self, value: int) -> "BinaryPayloadBuilder":
"""Add a 32-bit signed integer (uses 2 registers)."""
# Convert to unsigned if negative
if value < 0:
value = (1 << 32) + value
return self.add_32bit_uint(value)

def add_32bit_float(self, value: float) -> "BinaryPayloadBuilder":
"""Add a 32-bit float (uses 2 registers)."""
fmt = ">f" # big-endian float
packed = struct.pack(fmt, value)
value_int = struct.unpack(">I", packed)[0]
return self.add_32bit_uint(value_int)

def add_64bit_uint(self, value: int) -> "BinaryPayloadBuilder":
"""Add a 64-bit unsigned integer (uses 4 registers)."""
if self.wordorder == "big":
self._registers.append((value >> 48) & 0xFFFF)
self._registers.append((value >> 32) & 0xFFFF)
self._registers.append((value >> 16) & 0xFFFF)
self._registers.append(value & 0xFFFF)
else:
self._registers.append(value & 0xFFFF)
self._registers.append((value >> 16) & 0xFFFF)
self._registers.append((value >> 32) & 0xFFFF)
self._registers.append((value >> 48) & 0xFFFF)
return self

def add_64bit_int(self, value: int) -> "BinaryPayloadBuilder":
"""Add a 64-bit signed integer (uses 4 registers)."""
# Convert to unsigned if negative
if value < 0:
value = (1 << 64) + value
return self.add_64bit_uint(value)

def add_64bit_float(self, value: float) -> "BinaryPayloadBuilder":
"""Add a 64-bit float (uses 4 registers)."""
packed = struct.pack(">d", value)
value_int = struct.unpack(">Q", packed)[0]
return self.add_64bit_uint(value_int)

def add_string(self, value: str) -> "BinaryPayloadBuilder":
"""Add a string (encoded as UTF-8)."""
encoded = value.encode("utf-8")
# Pad to even number of bytes
if len(encoded) % 2:
encoded += b"\x00"
# Convert bytes to registers (2 bytes per register)
for i in range(0, len(encoded), 2):
reg_value = (encoded[i] << 8) | encoded[i + 1]
self._registers.append(reg_value)
return self

def to_coils(self) -> List[bool]:
"""Return the bits as a list of booleans."""
return self._bits

def to_registers(self) -> List[int]:
"""Return the registers as a list of integers."""
return self._registers
51 changes: 32 additions & 19 deletions modsim/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,20 +2,19 @@
"""Modbus simulator server and process lifecycle."""

import os
import ssl
import asyncio
import signal
import logging
from contextlib import suppress
from typing import Any

from pymodbus.constants import Endian
from pymodbus.datastore import (
ModbusDeviceContext,
ModbusSequentialDataBlock,
ModbusServerContext,
ModbusSlaveContext,
)
from pymodbus.device import ModbusDeviceIdentification
from pymodbus.payload import BinaryPayloadBuilder
from pymodbus.pdu.device import ModbusDeviceIdentification
from pymodbus.server import (
StartAsyncSerialServer,
ModbusTcpServer,
Expand All @@ -25,6 +24,7 @@
)

from . import __version__
from .compat import BinaryPayloadBuilder, Endian
from .utils import get_commandline, get_utc_time, get_logger

_logger = logging.getLogger(__name__)
Expand All @@ -44,7 +44,7 @@ def setup_server(args: Any) -> Any:
for _ in range(3):
# The bits within each byte are big-endian
# The coils status (from the first position) = [1,0,1,0,0,0,0,1]
co_builder.add_bits([True, False, False, False, False, True, False, True])
co_builder.add_bits([True, False, True, False, False, False, False, True])

hr_builder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
# address=0, bytes=8
Expand Down Expand Up @@ -79,14 +79,16 @@ def setup_server(args: Any) -> Any:
# use that payload in the data store
co_values = co_builder.to_coils()
hr_values = hr_builder.to_registers()
store = ModbusSlaveContext(
co=ModbusSequentialDataBlock(0, co_values),
di=ModbusSequentialDataBlock(10000, co_values),
ir=ModbusSequentialDataBlock(30000, hr_values),
hr=ModbusSequentialDataBlock(40000, hr_values),
zero_mode=True,
# Note: In PyModbus 3.16.1+, ModbusSequentialDataBlock uses address - 1 internally
# to convert to SimData. To maintain address ranges, we use address + 1 here.
# This maps Modbus addresses the same way as zero_mode=True did in older versions.
device_context = ModbusDeviceContext(
co=ModbusSequentialDataBlock(1, co_values),
di=ModbusSequentialDataBlock(10001, co_values),
ir=ModbusSequentialDataBlock(30001, hr_values),
hr=ModbusSequentialDataBlock(40001, hr_values),
)
args.context = ModbusServerContext(slaves=store, single=True)
args.context = ModbusServerContext(devices=device_context, single=True)

# initialize the server information
# If you don't set this or any fields, they are defaulted to empty strings.
Expand All @@ -108,7 +110,7 @@ async def run_async_server(args: Any) -> Any:
_logger.info(txt)
server: Any
if args.comm == "tcp":
address = ("", args.port) if args.port else None
address: tuple[str, int] = ("", args.port)
server = ModbusTcpServer(
context=args.context,
identity=args.identity,
Expand All @@ -123,8 +125,9 @@ async def run_async_server(args: Any) -> Any:
await server.serving
finally:
await server.shutdown()
return server
elif args.comm == "udp":
address = ("127.0.0.1", args.port) if args.port else None
address = ("127.0.0.1", args.port)
server = ModbusUdpServer(
context=args.context,
identity=args.identity,
Expand All @@ -139,29 +142,38 @@ async def run_async_server(args: Any) -> Any:
await server.serving
finally:
await server.shutdown()
return server
elif args.comm == "serial":
server = await StartAsyncSerialServer(
await StartAsyncSerialServer(
context=args.context,
identity=args.identity,
port=args.port,
framer=args.framer,
)
return None
elif args.comm == "tls":
address = ("", args.port) if args.port else None
address = ("", args.port)
cwd = os.getcwd().split("/")[-1]
if cwd == "examples":
path = "."
elif cwd == "test":
path = "../examples"
else:
path = "examples"

# Create SSL context from cert and key files
sslctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
sslctx.load_cert_chain(
certfile=f"{path}/certificates/pymodbus.crt",
keyfile=f"{path}/certificates/pymodbus.key",
)

server = ModbusTlsServer(
context=args.context,
identity=args.identity,
address=address,
framer=args.framer,
certfile=f"{path}/certificates/pymodbus.crt",
keyfile=f"{path}/certificates/pymodbus.key",
sslctx=sslctx,
)
try:
if not await server.listen():
Expand All @@ -171,7 +183,8 @@ async def run_async_server(args: Any) -> Any:
await server.serving
finally:
await server.shutdown()
return server
return server
raise ValueError(f"Unknown communication type: {args.comm}")


def app() -> None:
Expand Down
21 changes: 8 additions & 13 deletions modsim/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,8 @@
import logging.config
from datetime import datetime, timezone
from typing import Any, Dict
from pymodbus.transaction import (
ModbusAsciiFramer,
ModbusBinaryFramer,
ModbusRtuFramer,
ModbusSocketFramer,
ModbusTlsFramer,
from pymodbus.framer import (
FramerType,
)

from . import __version__
Expand All @@ -33,7 +29,7 @@ def get_parser() -> argparse.ArgumentParser:
)
parser.add_argument(
"--framer",
choices=["ascii", "binary", "rtu", "socket", "tls"],
choices=["ascii", "rtu", "socket", "tls"],

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated both README.md and docs/index.md to drop the binary framing option from the serial transport description.

help="set framer, default depends on --comm",
type=str,
)
Expand Down Expand Up @@ -86,7 +82,7 @@ def get_commandline() -> argparse.Namespace:
allowed_framers = {
"tcp": {"socket"},
"udp": {"socket"},
"serial": {"rtu", "ascii", "binary"},
"serial": {"rtu", "ascii"},
"tls": {"tls"},
}
if args.framer and args.framer not in allowed_framers[args.comm]:
Expand All @@ -100,11 +96,10 @@ def get_commandline() -> argparse.Namespace:
"tls": ["tls", 5020],
}
framers: Dict[str, Any] = {
"ascii": ModbusAsciiFramer,
"binary": ModbusBinaryFramer,
"rtu": ModbusRtuFramer,
"socket": ModbusSocketFramer,
"tls": ModbusTlsFramer,
"ascii": FramerType.ASCII,
"rtu": FramerType.RTU,
"socket": FramerType.SOCKET,
"tls": FramerType.TLS,
}
args.framer = framers[args.framer or comm_defaults[args.comm][0]]
args.port = args.port or comm_defaults[args.comm][1]
Expand Down
Loading
Loading