API reference

Auto-generated from the source. The top-level pymod package re-exports every public symbol below — from pymod import Holding, AsyncClient, ... just works.

Clients

class pymod.AsyncClient(transport, *, unit_id=1, timeout_s=0.5, retry=None)[source]

Bases: object

Async Modbus client. Construct via the tcp, rtu, or rtu_over_tcp factories.

Parameters:
classmethod tcp(host, port=502, *, unit_id=1, timeout_s=0.5, retry=None, pipeline=True, connect_timeout_s=3.0)[source]

Modbus TCP client. One auto-reconnecting connection per (host, port). Pipelining can be disabled per-PLC.

Parameters:
Return type:

Self

classmethod rtu(port, baudrate=9600, *, bytesize=8, parity='N', stopbits=1, unit_id=1, timeout_s=0.5, retry=None, inter_frame_delay_s=None)[source]

Modbus RTU client over a serial port. Internal lock serializes the bus across callers.

Parameters:
Return type:

Self

classmethod rtu_over_tcp(host, port, *, unit_id=1, timeout_s=0.5, retry=None, connect_timeout_s=3.0)[source]

RTU framing over a TCP socket — for serial-to-Ethernet gateways.

Parameters:
Return type:

Self

async connect()[source]
Return type:

None

async close()[source]
Return type:

None

property is_connected: bool
async execute(pdu, *, unit_id=None, timeout_s=None, retry=None)[source]

Send a request PDU, return the response PDU.

The retry policy is applied to errors listed in policy.retry_on (default: ModbusTimeoutError, ModbusConnectionError). Other errors — notably ModbusExceptionResponse subclasses — are NOT retried, on the assumption that retrying an illegal address yields the same illegal address.

Per-call overrides default to the values supplied at client construction time.

Parameters:
Return type:

bytes

async read(items, *, unit_id=None, timeout_s=None, retry=None)[source]

Issue a heterogeneous batch read.

Adjacent same-area ranges are coalesced into the minimum number of Modbus reads; per-FC PDU limits are honored by automatic splitting. Returns a list parallel to items — one item failing does not abort the batch.

Parameters:
Return type:

list[ReadResult]

async write(items, *, unit_id=None, timeout_s=None, retry=None)[source]

Issue a batch write. FC05/06 vs FC15/16 selected per item by value count. Per-item results parallel items.

Parameters:
Return type:

list[WriteResult]

class pymod.Client(async_client, loop)[source]

Bases: object

Synchronous facade over AsyncClient.

Owns a private daemon-thread event loop. Each sync call schedules the underlying coroutine on that loop and blocks the caller until it completes. The event loop is stopped on close().

Parameters:
classmethod tcp(host, port=502, *, unit_id=1, timeout_s=0.5, retry=None, pipeline=True, connect_timeout_s=3.0)[source]
Parameters:
Return type:

Self

classmethod rtu(port, baudrate=9600, *, bytesize=8, parity='N', stopbits=1, unit_id=1, timeout_s=0.5, retry=None, inter_frame_delay_s=None)[source]
Parameters:
Return type:

Self

classmethod rtu_over_tcp(host, port, *, unit_id=1, timeout_s=0.5, retry=None, connect_timeout_s=3.0)[source]
Parameters:
Return type:

Self

connect()[source]
Return type:

None

close()[source]
Return type:

None

property is_connected: bool
execute(pdu, *, unit_id=None, timeout_s=None, retry=None)[source]
Parameters:
Return type:

bytes

read(items, *, unit_id=None, timeout_s=None, retry=None)[source]
Parameters:
Return type:

list[ReadResult]

write(items, *, unit_id=None, timeout_s=None, retry=None)[source]
Parameters:
Return type:

list[WriteResult]

Read items

class pymod.Holding(start, count, dtype='uint16', word_order='big', byte_order='big', bit_index=None, bit_indices=None, bit_numbering='lsb_first')[source]

Bases: object

Read from holding registers (FC03), with typed decoding.

Parameters:

  • start (int)

  • count (int)

  • dtype (Literal['int16', 'uint16', 'int32', 'uint32', 'float32', 'int64', 'uint64', 'float64', 'bit', 'bits'])

  • word_order (Literal['big', 'little'])

  • byte_order (Literal['big', 'little'])

  • bit_index (int | None)

  • bit_indices (Sequence[int] | None)

  • bit_numbering (Literal['lsb_first', 'msb_first'])

start: int
count: int
dtype: Literal['int16', 'uint16', 'int32', 'uint32', 'float32', 'int64', 'uint64', 'float64', 'bit', 'bits']
word_order: Literal['big', 'little']
byte_order: Literal['big', 'little']
bit_index: int | None
bit_indices: Sequence[int] | None
bit_numbering: Literal['lsb_first', 'msb_first']
class pymod.Input(start, count, dtype='uint16', word_order='big', byte_order='big', bit_index=None, bit_indices=None, bit_numbering='lsb_first')[source]

Bases: object

Read from input registers (FC04), with typed decoding.

Parameters:

  • start (int)

  • count (int)

  • dtype (Literal['int16', 'uint16', 'int32', 'uint32', 'float32', 'int64', 'uint64', 'float64', 'bit', 'bits'])

  • word_order (Literal['big', 'little'])

  • byte_order (Literal['big', 'little'])

  • bit_index (int | None)

  • bit_indices (Sequence[int] | None)

  • bit_numbering (Literal['lsb_first', 'msb_first'])

start: int
count: int
dtype: Literal['int16', 'uint16', 'int32', 'uint32', 'float32', 'int64', 'uint64', 'float64', 'bit', 'bits']
word_order: Literal['big', 'little']
byte_order: Literal['big', 'little']
bit_index: int | None
bit_indices: Sequence[int] | None
bit_numbering: Literal['lsb_first', 'msb_first']
class pymod.Coil(start, count)[source]

Bases: object

Read coils (FC01).

Parameters:
start: int
count: int
class pymod.Discrete(start, count)[source]

Bases: object

Read discrete inputs (FC02).

Parameters:
start: int
count: int
class pymod.ReadResult(item, values=<factory>, ok=True, error=None)[source]

Bases: object

Per-item read result. Parallel to the input list passed to read().

Parameters:
item: Holding | Input | Coil | Discrete
values: Sequence[int | float | bool]
ok: bool
error: ModbusError | None

Write items

class pymod.WriteHolding(start, values, dtype='uint16', word_order='big', byte_order='big')[source]

Bases: object

Write holding registers. Planner picks FC06 (single) vs FC16 (multiple).

Parameters:
start: int
values: Sequence[int | float]
dtype: Literal['int16', 'uint16', 'int32', 'uint32', 'float32']
word_order: Literal['big', 'little']
byte_order: Literal['big', 'little']
class pymod.WriteCoils(start, values)[source]

Bases: object

Write coils. Planner picks FC05 (single) vs FC15 (multiple).

Parameters:
start: int
values: Sequence[bool]
class pymod.WriteResult(item, ok=True, error=None)[source]

Bases: object

Per-item write result. Parallel to the input list passed to write().

Parameters:
item: WriteHolding | WriteCoils
ok: bool
error: ModbusError | None

Server

class pymod.Server(*, host='0.0.0.0', port=502, on_read, on_write=None, unit_id=None, max_connections=32)[source]

Bases: object

Modbus TCP server with callback-based data access.

Parameters:

  • host (str)

  • port (int)

  • on_read (ReadCallback)

  • on_write (WriteCallback | None)

  • unit_id (int | None)

  • max_connections (int)

async start()[source]
Return type:

None

async stop()[source]
Return type:

None

property is_running: bool
property active_connections: int
property port: int

Bound port. Useful when port=0 was used (ephemeral port).

pymod.Area = <enum 'Area'>[source]

Modbus address space. Determines which function code is used.

Retry policy

class pymod.RetryPolicy(max_attempts=2, backoff_initial_s=0.05, backoff_factor=2.0, backoff_cap_s=1.0, retry_on=(<class 'pymod.errors.ModbusTimeoutError'>, <class 'pymod.errors.ModbusConnectionError'>))[source]

Bases: object

Per-call retry policy.

Defaults match the user-stated requirement: 1 retry (= 2 total attempts), transport errors only. Modbus exception responses are NOT retried by default — the same illegal address will fail forever.

Parameters:
max_attempts: int
backoff_initial_s: float
backoff_factor: float
backoff_cap_s: float
retry_on: tuple[type[ModbusError], ...]
delay_for(attempt)[source]

Backoff delay before the n-th retry (attempt is 1-indexed; first retry is attempt=1).

Parameters:

attempt (int)

Return type:

float

Exceptions

exception pymod.ModbusError[source]

Bases: Exception

Base class for everything pymod raises.

exception pymod.ModbusTransportError[source]

Bases: ModbusError

The request did not produce a usable reply on the wire.

exception pymod.ModbusTimeoutError[source]

Bases: ModbusTransportError

No response within the configured timeout.

exception pymod.ModbusConnectionError[source]

Bases: ModbusTransportError

TCP connection refused/dropped, or serial port unavailable.

exception pymod.ModbusProtocolError[source]

Bases: ModbusError

A response was received but is malformed or out-of-spec.

exception pymod.ModbusExceptionResponse(message='', code=None)[source]

Bases: ModbusError

Slave returned a valid Modbus exception response. Code in code.

Parameters:
Return type:

None

code: int = 0
exception pymod.IllegalFunction(message='', code=None)[source]

Bases: ModbusExceptionResponse

Parameters:
Return type:

None

code: int = 1
exception pymod.IllegalDataAddress(message='', code=None)[source]

Bases: ModbusExceptionResponse

Parameters:
Return type:

None

code: int = 2
exception pymod.IllegalDataValue(message='', code=None)[source]

Bases: ModbusExceptionResponse

Parameters:
Return type:

None

code: int = 3
exception pymod.SlaveDeviceFailure(message='', code=None)[source]

Bases: ModbusExceptionResponse

Parameters:
Return type:

None

code: int = 4
exception pymod.SlaveDeviceBusy(message='', code=None)[source]

Bases: ModbusExceptionResponse

Parameters:
Return type:

None

code: int = 6
exception pymod.MemoryParityError(message='', code=None)[source]

Bases: ModbusExceptionResponse

Parameters:
Return type:

None

code: int = 8
exception pymod.GatewayPathUnavailable(message='', code=None)[source]

Bases: ModbusExceptionResponse

Parameters:
Return type:

None

code: int = 10
exception pymod.GatewayTargetFailedToRespond(message='', code=None)[source]

Bases: ModbusExceptionResponse

Parameters:
Return type:

None

code: int = 11

Codec helpers

Lower-level encode/decode functions, useful for custom function-code codecs and integration tests.

Encode/decode int16, uint16, int32, uint32, float32 across registers.

Pure functions. Inputs are 16-bit register values (0..0xFFFF); outputs are typed Python values. word_order and byte_order together select the ABCD / CDAB / BADC / DCBA layout — see codec/_common.py.

pymod.codec.values.regs_per_item(dtype)[source]

How many 16-bit registers are needed to represent one dtype item.

Parameters:

dtype (Literal['int16', 'uint16', 'int32', 'uint32', 'float32', 'int64', 'uint64', 'float64'])

Return type:

int

pymod.codec.values.decode_registers(registers, dtype, *, word_order='big', byte_order='big')[source]

Decode a register block into a list of typed values.

The block size must be a multiple of regs_per_item(dtype).

Parameters:

  • registers (Sequence[int])

  • dtype (Literal['int16', 'uint16', 'int32', 'uint32', 'float32', 'int64', 'uint64', 'float64'])

  • word_order (Literal['big', 'little'])

  • byte_order (Literal['big', 'little'])

Return type:

list[int | float]

pymod.codec.values.encode_registers(values, dtype, *, word_order='big', byte_order='big')[source]

Encode a list of typed values into a register block.

Length of returned register list = len(values) * regs_per_item(dtype).

Parameters:

  • values (Sequence[int | float])

  • dtype (Literal['int16', 'uint16', 'int32', 'uint32', 'float32', 'int64', 'uint64', 'float64'])

  • word_order (Literal['big', 'little'])

  • byte_order (Literal['big', 'little'])

Return type:

list[int]

Extract bits from a register block.

Algorithm (per ADR 04):

  1. Each register is two bytes, MSB-first per the Modbus spec.

  2. Apply byte_order per-register, then word_order across registers, producing one big-endian byte string of count * 16 bits — equivalently one large unsigned integer.

  3. Index into that integer per bit_numbering. The default is lsb_first: bit 0 is the LSB, equivalent to (value >> bit_index) & 1. msb_first flips so bit 0 is the most-significant bit of the assembled integer (vendors that label bits from the top).

pymod.codec.bits.extract_bit(registers, bit_index, *, word_order='big', byte_order='big', bit_numbering='lsb_first')[source]

Return a single bit from a register block.

Parameters:
Return type:

bool

pymod.codec.bits.extract_bits(registers, bit_indices, *, word_order='big', byte_order='big', bit_numbering='lsb_first')[source]

Return a list of bits at the given indices from a register block.

Returned list is parallel to bit_indices.

Parameters:
Return type:

list[bool]

Protocol helpers

The pure-functions PDU layer. Most users never need these directly — they’re exposed for advanced cases (custom FCs, building a non-standard slave, etc.).

Per-function-code PDU codecs (pure, no I/O).

Each standard function code has an encode_* and decode_* pair operating on bytes. Exception responses (FC | 0x80) are detected via detect_exception and surfaced as the appropriate ModbusExceptionResponse subclass.

Custom function codes are supported via register_codec().

Function codes covered:

FC01 read coils FC02 read discrete inputs FC03 read holding registers FC04 read input registers FC05 write single coil FC06 write single register FC15 write multiple coils FC16 write multiple registers

pymod.protocol.pdu.detect_exception(pdu, expected_fc)[source]

If pdu is an exception response for expected_fc, return the corresponding ModbusExceptionResponse instance. If it is a normal response for expected_fc, return None. Anything else is a protocol error.

Parameters:
Return type:

ModbusExceptionResponse | None

pymod.protocol.pdu.encode_read_coils(start, count)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.encode_read_discrete_inputs(start, count)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.decode_read_coils(pdu, count)[source]
Parameters:
Return type:

list[bool]

pymod.protocol.pdu.decode_read_discrete_inputs(pdu, count)[source]
Parameters:
Return type:

list[bool]

pymod.protocol.pdu.encode_read_holding_registers(start, count)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.encode_read_input_registers(start, count)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.decode_read_holding_registers(pdu, count)[source]
Parameters:
Return type:

list[int]

pymod.protocol.pdu.decode_read_input_registers(pdu, count)[source]
Parameters:
Return type:

list[int]

pymod.protocol.pdu.encode_write_single_coil(address, value)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.decode_write_single_coil(pdu)[source]

Returns (address, value). Slaves echo the request as the response.

Parameters:

pdu (bytes)

Return type:

tuple[int, bool]

pymod.protocol.pdu.encode_write_single_register(address, value)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.decode_write_single_register(pdu)[source]

Returns (address, value).

Parameters:

pdu (bytes)

Return type:

tuple[int, int]

pymod.protocol.pdu.encode_write_multiple_coils(start, values)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.decode_write_multiple_coils(pdu)[source]

Returns (start, count).

Parameters:

pdu (bytes)

Return type:

tuple[int, int]

pymod.protocol.pdu.encode_write_multiple_registers(start, values)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.decode_write_multiple_registers(pdu)[source]

Returns (start, count).

Parameters:

pdu (bytes)

Return type:

tuple[int, int]

class pymod.protocol.pdu.CustomCodec(code, encode, decode)[source]

Bases: object

Codec for a custom or vendor-specific function code.

encode produces the request PDU bytes (function code byte first). decode is called with the response PDU bytes (function code byte first) and must surface exceptions itself, typically by calling detect_exception(pdu, code).

Parameters:
code: int
encode: Callable[[...], bytes]
decode: Callable[[bytes], object]
pymod.protocol.pdu.register_codec(codec)[source]

Register a codec for a non-standard function code.

Codes 0x01..0x10 are reserved for the standard FCs in this library. Codes with the exception bit (0x80) set are reserved for exception responses.

Parameters:

codec (CustomCodec)

Return type:

None

pymod.protocol.pdu.get_codec(code)[source]
Parameters:

code (int)

Return type:

CustomCodec | None

pymod.protocol.pdu.unregister_codec(code)[source]

Remove a previously registered codec. Mainly for tests.

Parameters:

code (int)

Return type:

None

pymod.protocol.pdu.decode_request_read_coils(pdu)[source]

Returns (start, count) for a FC01 request PDU.

Parameters:

pdu (bytes)

Return type:

tuple[int, int]

pymod.protocol.pdu.decode_request_read_discrete_inputs(pdu)[source]
Parameters:

pdu (bytes)

Return type:

tuple[int, int]

pymod.protocol.pdu.decode_request_read_holding_registers(pdu)[source]
Parameters:

pdu (bytes)

Return type:

tuple[int, int]

pymod.protocol.pdu.decode_request_read_input_registers(pdu)[source]
Parameters:

pdu (bytes)

Return type:

tuple[int, int]

pymod.protocol.pdu.decode_request_write_single_coil(pdu)[source]

Returns (address, value) for a FC05 request PDU.

Parameters:

pdu (bytes)

Return type:

tuple[int, bool]

pymod.protocol.pdu.decode_request_write_single_register(pdu)[source]

Returns (address, value) for a FC06 request PDU.

Parameters:

pdu (bytes)

Return type:

tuple[int, int]

pymod.protocol.pdu.decode_request_write_multiple_coils(pdu)[source]

Returns (start, values) for a FC15 request PDU.

Parameters:

pdu (bytes)

Return type:

tuple[int, list[bool]]

pymod.protocol.pdu.decode_request_write_multiple_registers(pdu)[source]

Returns (start, values) for a FC16 request PDU.

Parameters:

pdu (bytes)

Return type:

tuple[int, list[int]]

pymod.protocol.pdu.encode_response_read_coils(values)[source]
Parameters:

values (Sequence[bool])

Return type:

bytes

pymod.protocol.pdu.encode_response_read_discrete_inputs(values)[source]
Parameters:

values (Sequence[bool])

Return type:

bytes

pymod.protocol.pdu.encode_response_read_holding_registers(values)[source]
Parameters:

values (Sequence[int])

Return type:

bytes

pymod.protocol.pdu.encode_response_read_input_registers(values)[source]
Parameters:

values (Sequence[int])

Return type:

bytes

pymod.protocol.pdu.encode_response_write_single_coil(address, value)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.encode_response_write_single_register(address, value)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.encode_response_write_multiple_coils(start, count)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.encode_response_write_multiple_registers(start, count)[source]
Parameters:
Return type:

bytes

pymod.protocol.pdu.encode_exception_response(fc, exception_code)[source]

Encode a Modbus exception response (FC | 0x80, code).

Parameters:

  • fc (int)

  • exception_code (int)

Return type:

bytes

Logging helpers

Convenience helpers for pymod’s logging.

The library is silent by default — both pymod and pymod.wire start with NullHandler attached, so library users pay nothing unless they opt in.

This module provides a one-line configure() for users who want quick visibility, plus a JsonFormatter for structured output. Power users who already have a logging setup can simply attach their own handlers to the pymod and pymod.wire loggers and skip this module entirely.

Examples

import pymod.logging
pymod.logging.configure(level="INFO")                         # human
pymod.logging.configure(level="DEBUG", json=True)             # JSON
pymod.logging.configure(level="INFO", wire_trace=True)        # + hex dumps
class pymod.logging.JsonFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)[source]

Bases: Formatter

One log record per line, JSON-encoded. Suitable for log aggregators.

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{’ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

format(record)[source]

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

Parameters:

record (LogRecord)

Return type:

str

pymod.logging.configure(*, level='INFO', json=False, wire_trace=False, stream=None)[source]

Attach a handler to the pymod logger.

Parameters:

  • level (log level for the main pymod logger.)

  • json (emit structured JSON instead of human-readable lines.)

  • wire_trace (also enable the pymod.wire logger at DEBUG (hex dumps) – of every PDU sent and received).

  • stream (destination; defaults to sys.stderr.)

Return type:

None