moddef.pymodbus module

pymodbus adapter: implements the moddef [Transport] protocol over a pymodbus 3.x async client (spec §32.3). Install with pip install moddef[pymodbus].

• Constructors: [PymodbusTransport.tcp], [PymodbusTransport.serial] (RTU), or wrap any connected pymodbus async client directly.

• Reads are chunked to honor [Options.max_read_words] (Modbus caps a read at 125 registers; devices like the SDM630/EM24 need less — the same knob as the TS/Rust adapters).

• Every request runs under [Options.timeout] (asyncio.timeout).

• Modbus exception responses raise [ModbusExceptionError] carrying the device’s exception code.

One in-flight request per transport: Modbus is a serial request/response channel; concurrent access is a caller decision (e.g. an asyncio.Lock).

exception moddef.pymodbus.ModbusExceptionError(exception_code: int)

Bases: ModDefError

The device answered a Modbus exception (illegal address, busy, …).

exception moddef.pymodbus.ModbusTimeoutError

Bases: ModDefError

[Options.timeout] elapsed before the device answered.

class moddef.pymodbus.Options(unit_id: int = 1, timeout: float | None = 5.0, max_read_words: int = 125)

Bases: object

Connection options shared by all constructors.

max_read_words: int = 125

timeout: float | None = 5.0

unit_id: int = 1

class moddef.pymodbus.PymodbusTransport(client, options: Options | None = None)

Bases: object

moddef Transport over a pymodbus async client.

close() → None

async read_coils(offset: int, count: int) → list[bool]

async read_discrete(offset: int, count: int) → list[bool]

async read_holding(offset: int, count: int) → list[int]

async read_input(offset: int, count: int) → list[int]

async classmethod serial(port: str, baudrate: int = 9600, options: Options | None = None, **serial_kwargs) → PymodbusTransport

Modbus RTU over a serial port (parity/stopbits via serial_kwargs).

async classmethod tcp(host: str, port: int = 502, options: Options | None = None) → PymodbusTransport

async write_coil(offset: int, on: bool) → None

async write_holding(offset: int, words: Sequence[int]) → None