ModbusTransport#

class ModbusTransport[source]#

Bases: TransportBase, ABC

Abstract Modbus line: owns the session, locking, and every wire op, addressed per call via unit_id. Annotate against it; construct a concrete subclass.

Initialize holder tracking, the lock, and the (not yet connected) client slot.

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`.

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

check_unit_id(unit_id: int) → int[source]#: Validate unit_id against this physical layer’s range (0 to _max_unit_id) and return it.

read_holding_registers( address: int, count: int, *, unit_id: int, ) → list[int][source]#: Read holding registers by address (FC03).

read_input_registers( address: int, count: int, *, unit_id: int, ) → list[int][source]#: Read input registers by address (FC04).

write_holding_register( address: int, value: int, *, unit_id: int, ) → None[source]#: Write a single holding register by address (FC06).

write_holding_registers( address: int, values: list[int], *, unit_id: int, ) → None[source]#: Write multiple holding registers by address (FC16).

read_coils( address: int, count: int, *, unit_id: int, ) → list[bool][source]#: Read coils by address (FC01).

write_coil( address: int, value: bool, *, unit_id: int, ) → None[source]#: Write a single coil by address (FC05).

write_coils( address: int, values: list[bool], *, unit_id: int, ) → None[source]#: Write multiple coils by address (FC15).

read_discrete_inputs( address: int, count: int, *, unit_id: int, ) → list[bool][source]#: Read discrete inputs by address (FC02).

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[source]#

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_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[source]#

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).

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.

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.