ModbusRTUTransport#

class ModbusRTUTransport(
port: str,
baudrate: int = 9600,
parity: Literal['N', 'E', 'O'] = 'N',
stopbits: Literal[1, 2] = 1,
bytesize: Literal[5, 6, 7, 8] = 8,
timeout: float = 3.0,
framer: Literal['rtu', 'ascii'] = 'rtu',
)[source]#

Bases: ModbusTransport

A Modbus serial line (RTU or ASCII framing). One port, addressed per wire op via unit_id.

Validate and store the serial port’s fields.

Attributes

is_open

Whether the client has been opened and not closed.

Methods

check_unit_id

Validate unit_id against this physical layer's range (0 to _max_unit_id) and return it.

close

Close and teardown; with a holder, remove it and tear down only when the last owner leaves.

lock

Return the reentrant resource lock for atomic multi-step sequences; the holder may write/query/read inside it.

open

Open the connection (idempotent); with a holder, admit it and return True only for the first owner.

read_coils

Read coils by address (FC01).

read_discrete_inputs

Read discrete inputs by address (FC02).

read_holding_registers

Read holding registers by address (FC03).

read_input_registers

Read input registers by address (FC04).

read_typed

Read address as data_type, dispatching by register_type and decoding across registers.

write_coil

Write a single coil by address (FC05).

write_coils

Write multiple coils by address (FC15).

write_holding_register

Write a single holding register by address (FC06).

write_holding_registers

Write multiple holding registers by address (FC16).

write_typed

Encode value as data_type and write it to address, dispatching by register_type.

check_unit_id(unit_id: int) → int#

Validate unit_id against this physical layer’s range (0 to _max_unit_id) and return it.

close(holder: object | None = None) → None#

Close and teardown; with a holder, remove it and tear down only when the last owner leaves.

A holder that needs to talk to the instrument before the connection goes (releasing a remote lock, say) does that in its own close before calling this, while the session is still up.

Raises UnknownHolderError when an unrecognized holder closes while others still own the connection, because that is the case where ignoring it silently strands the real owners and leaves the connection open forever. Once the last owner has left there is nothing to strand, so a repeat close is a no-op and close stays idempotent.

property is_open: bool#

Whether the client has been opened and not closed. Stays True across a dropped socket, which the next op reconnects.

lock() → RLock#

Return the reentrant resource lock for atomic multi-step sequences; the holder may write/query/read inside it.

open(holder: object | None = None) → bool#

Open the connection (idempotent); with a holder, admit it and return True only for the first owner.

read_coils(
address: int,
count: int,
*,
unit_id: int,
) → list[bool]#

Read coils by address (FC01).

read_discrete_inputs(
address: int,
count: int,
*,
unit_id: int,
) → list[bool]#

Read discrete inputs by address (FC02).

read_holding_registers(
address: int,
count: int,
*,
unit_id: int,
) → list[int]#

Read holding registers by address (FC03).

read_input_registers(
address: int,
count: int,
*,
unit_id: int,
) → list[int]#

Read input registers by address (FC04).

read_typed(
register_type: Literal['holding', 'input', 'coil', 'discrete'],
address: int,
data_type: Literal['uint16', 'int16', 'uint32', 'int32', 'uint64', 'int64', 'float32', 'float64', 'bool'],
*,
unit_id: int,
byte_swap: bool = False,
word_swap: bool = False,
long_swap: bool = False,
) → int | float | bool#

Read address as data_type, dispatching by register_type and decoding across registers.

Coils and discrete inputs are single-bit, so data_type must be "bool" and the read returns a bool.

write_coil(
address: int,
value: bool,
*,
unit_id: int,
) → None#

Write a single coil by address (FC05).

write_coils(
address: int,
values: list[bool],
*,
unit_id: int,
) → None#

Write multiple coils by address (FC15).

write_holding_register(
address: int,
value: int,
*,
unit_id: int,
) → None#

Write a single holding register by address (FC06).

write_holding_registers(
address: int,
values: list[int],
*,
unit_id: int,
) → None#

Write multiple holding registers by address (FC16).

write_typed(
register_type: Literal['holding', 'input', 'coil', 'discrete'],
address: int,
value: int | float | bool,
data_type: Literal['uint16', 'int16', 'uint32', 'int32', 'uint64', 'int64', 'float32', 'float64', 'bool'],
*,
unit_id: int,
byte_swap: bool = False,
word_swap: bool = False,
long_swap: bool = False,
) → None#

Encode value as data_type and write it to address, dispatching by register_type.

Coils are single-bit, so data_type must be "bool" and value must be a bool (no numeric coercion).