Server (3.x)

Pymodbus offers servers with transport protocols for

  • Serial (RS-485) typically using a dongle

  • TCP

  • TLS

  • UDP

  • possibility to add a custom transport protocol

communication in 2 versions:

  • synchronous server,

  • asynchronous server using asyncio.

Remark All servers are implemented with asyncio, and the synchronous servers are just an interface layer allowing synchronous applications to use the server as if it was synchronous.

Warning The current serial server implementation offer only limited support for running the server on a shared rs485 line (multipoint).

Server classes.

class pymodbus.server.ModbusBaseServer(params: CommParams, context: ModbusServerContext | SimDevice | list[SimDevice], ignore_missing_devices: bool, broadcast_enable: bool, identity: ModbusDeviceIdentification | None, framer: FramerType, trace_packet: Callable[[bool, bytes], bytes] | None, trace_pdu: Callable[[bool, ModbusPDU], ModbusPDU] | None, trace_connect: Callable[[bool], None] | None, custom_pdu: list[type[ModbusPDU]] | None)

Bases: ModbusProtocol

Common functionality for all server classes.

active_server: ModbusBaseServer | None = None
async async_getValues(device_id: int, func_code: int, address: int, count: int = 1) → list[int] | list[bool]

Get count values from datastore.

Parameters:

  • device_id – the device being addressed

  • func_code – The function we are working with

  • address – The starting address

  • count – The number of values to retrieve

Returns:

The requested values from a:a+c

async async_setValues(device_id: int, func_code: int, address: int, values: list[int] | list[bool]) → None

Set the datastore with the supplied values.

Parameters:

  • device_id – the device being addressed

  • func_code – The function we are working with

  • address – The starting address

  • values – The new values to be set

callback_connected() → None

Call when connection is successful.

callback_data(data: bytes, addr: tuple | None = None) → int

Handle received data.

callback_disconnected(exc: Exception | None) → None

Call when connection is lost.

callback_new_connection()

Handle incoming connect.

async serve_forever(*, background: bool = False)

Start endless loop.

async shutdown()

Close server.

class pymodbus.server.ModbusSerialServer(context: ModbusServerContext | SimDevice | list[SimDevice], *, framer: FramerType = FramerType.RTU, ignore_missing_devices: bool = False, identity: ModbusDeviceIdentification | None = None, broadcast_enable: bool = False, trace_packet: Callable[[bool, bytes], bytes] | None = None, trace_pdu: Callable[[bool, ModbusPDU], ModbusPDU] | None = None, trace_connect: Callable[[bool], None] | None = None, custom_pdu: list[type[ModbusPDU]] | None = None, **kwargs)

Bases: ModbusBaseServer

A modbus threaded serial socket server.

Tip

Remember to call serve_forever to start server.

class pymodbus.server.ModbusSimulatorServer(modbus_server: str = 'server', modbus_device: str = 'device', http_host: str = '0.0.0.0', http_port: int = 8080, log_file: str = 'server.log', json_file: str = 'setup.json', custom_actions_module: str | None = None)

Bases: object

ModbusSimulatorServer.

Parameters:

  • modbus_server – Server name in json file (default: “server”)

  • modbus_device – Device name in json file (default: “client”)

  • http_host – TCP host for HTTP (default: “localhost”)

  • http_port – TCP port for HTTP (default: 8080)

  • json_file – setup file (default: “setup.json”)

  • custom_actions_module – python module with custom actions (default: none)

if either http_port or http_host is none, HTTP will not be started. This class starts a http server, that serves a couple of endpoints:

  • “<addr>/” static files

  • “<addr>/api/log” log handling, HTML with GET, REST-API with post

  • “<addr>/api/registers” register handling, HTML with GET, REST-API with post

  • “<addr>/api/calls” call (function code / message) handling, HTML with GET, REST-API with post

  • “<addr>/api/server” server handling, HTML with GET, REST-API with post

Example:

from pymodbus import ModbusSimulatorServer

async def run():
    simulator = ModbusSimulatorServer(
        modbus_server="my server",
        modbus_device="my device",
        http_host="localhost",
        http_port=8080)
    await simulator.run_forever(only_start=True)
    ...
    await simulator.stop()
action_add(params, range_start, range_stop)

Build list of registers matching filter.

action_clear(_params, _range_start, _range_stop)

Clear register filter.

action_monitor(params, range_start, range_stop)

Start monitoring calls.

action_reset(_params, _range_start, _range_stop)

Reset call simulation.

action_set(params, _range_start, _range_stop)

Set register value.

action_simulate(params, _range_start, _range_stop)

Simulate responses.

action_stop(_params, _range_start, _range_stop)

Stop call monitoring.

build_html_calls(params: dict, html: str) → str

Build html calls page.

build_html_log(_params, html)

Build html log page.

build_html_registers(params, html)

Build html registers page.

build_html_server(_params, html)

Build html server page.

build_json_calls(params: dict) → dict

Build json calls response.

build_json_log(params)

Build json log page.

build_json_registers(params)

Build json registers response.

build_json_server(params)

Build html server page.

async handle_html(request: Request)

Handle html.

async handle_html_static(request: Request)

Handle static html.

async handle_json(request: Request)

Handle api registers.

helper_handle_submit(params, submit_actions)

Build html register submit.

async run_forever(only_start=False)

Start modbus and http servers.

async start_modbus_server(app)

Start Modbus server as asyncio task.

async stop()

Stop modbus and http servers.

async stop_modbus_server(app)

Stop modbus server.

class pymodbus.server.ModbusTcpServer(context: ModbusServerContext | SimDevice | list[SimDevice], *, framer=FramerType.SOCKET, identity: ModbusDeviceIdentification | None = None, address: tuple[str, int] = ('', 502), ignore_missing_devices: bool = False, broadcast_enable: bool = False, trace_packet: Callable[[bool, bytes], bytes] | None = None, trace_pdu: Callable[[bool, ModbusPDU], ModbusPDU] | None = None, trace_connect: Callable[[bool], None] | None = None, custom_pdu: list[type[ModbusPDU]] | None = None)

Bases: ModbusBaseServer

A modbus threaded tcp socket server.

Tip

Remember to call serve_forever to start server.

class pymodbus.server.ModbusTlsServer(context: ModbusServerContext | SimDevice | list[SimDevice], *, framer=FramerType.TLS, identity: ModbusDeviceIdentification | None = None, address: tuple[str, int] = ('', 502), sslctx=None, certfile=None, keyfile=None, password=None, ignore_missing_devices=False, broadcast_enable=False, trace_packet: Callable[[bool, bytes], bytes] | None = None, trace_pdu: Callable[[bool, ModbusPDU], ModbusPDU] | None = None, trace_connect: Callable[[bool], None] | None = None, custom_pdu: list[type[ModbusPDU]] | None = None)

Bases: ModbusTcpServer

A modbus threaded tls socket server.

Tip

Remember to call serve_forever to start server.

class pymodbus.server.ModbusUdpServer(context: ModbusServerContext | SimDevice | list[SimDevice], *, framer=FramerType.SOCKET, identity: ModbusDeviceIdentification | None = None, address: tuple[str, int] = ('', 502), ignore_missing_devices: bool = False, broadcast_enable: bool = False, trace_packet: Callable[[bool, bytes], bytes] | None = None, trace_pdu: Callable[[bool, ModbusPDU], ModbusPDU] | None = None, trace_connect: Callable[[bool], None] | None = None, custom_pdu: list[type[ModbusPDU]] | None = None)

Bases: ModbusBaseServer

A modbus threaded udp socket server.

Tip

Remember to call serve_forever to start server.

async pymodbus.server.ServerAsyncStop() → None

Terminate server.

pymodbus.server.ServerStop() → None

Terminate server.

async pymodbus.server.StartAsyncSerialServer(context: ModbusServerContext | SimDevice | list[SimDevice], **kwargs) → None

Start and run a serial modbus server.

Parameters:

  • context – Datastore object

  • kwargs – for parameter explanation see ModbusSerialServer

Tip

Only handles a single server !

Use ModbusSerialServer to allow multiple servers in one app.

async pymodbus.server.StartAsyncTcpServer(context: ModbusServerContext | SimDevice | list[SimDevice], **kwargs) → None

Start and run a tcp modbus server.

Parameters:

  • context – Datastore object

  • kwargs – for parameter explanation see ModbusTcpServer

Tip

Only handles a single server !

Use ModbusTcpServer to allow multiple servers in one app.

async pymodbus.server.StartAsyncTlsServer(context: ModbusServerContext | SimDevice | list[SimDevice], **kwargs) → None

Start and run a tls modbus server.

Parameters:

  • context – Datastore object

  • kwargs – for parameter explanation see ModbusTlsServer

Tip

Only handles a single server !

Use ModbusTlsServer to allow multiple servers in one app.

async pymodbus.server.StartAsyncUdpServer(context: ModbusServerContext | SimDevice | list[SimDevice], **kwargs) → None

Start and run a udp modbus server.

Parameters:

  • context – Datastore object

  • kwargs – for parameter explanation see ModbusUdpServer

Tip

Only handles a single server !

Use ModbusUdpServer to allow multiple servers in one app.

pymodbus.server.StartSerialServer(context: ModbusServerContext | SimDevice | list[SimDevice], **kwargs) → None

Start and run a modbus serial server.

Parameters:

  • context – Datastore object

  • kwargs – for parameter explanation see ModbusSerialServer

Tip

Only handles a single server !

Use ModbusSerialServer to allow multiple servers in one app.

pymodbus.server.StartTcpServer(context: ModbusServerContext | SimDevice | list[SimDevice], **kwargs) → None

Start and run a modbus TCP server.

Parameters:

  • context – Datastore object

  • kwargs – for parameter explanation see ModbusTcpServer

Tip

Only handles a single server !

Use ModbusTcpServer to allow multiple servers in one app.

pymodbus.server.StartTlsServer(context: ModbusServerContext | SimDevice | list[SimDevice], **kwargs) → None

Start and run a modbus TLS server.

Parameters:

  • context – Datastore object

  • kwargs – for parameter explanation see ModbusTlsServer

Tip

Only handles a single server !

Use ModbusTlsServer to allow multiple servers in one app.

pymodbus.server.StartUdpServer(context: ModbusServerContext | SimDevice | list[SimDevice], **kwargs) → None

Start and run a modbus UDP server.

Parameters:

  • context – Datastore object

  • kwargs – for parameter explanation see ModbusUdpServer

Tip

Only handles a single server !

Use ModbusUdpServer to allow multiple servers in one app.

pymodbus.server.get_simulator_commandline(cmdline=None)

Get command line arguments.

Datastore for server

Note

The legacy datastores will be replaced in v4.0.0 by SimData/SimDevice (already available) to provide a more flexible data definition.

Datastore definitions