API Reference#

Modbus Application Protocol

Abstract: Modbus is a simple, vendor-neutral request/reply protocol that is widely used for communication with PLCs and other field devices.

Added in version 0.3.0.

High-level Modbus TCP/UDP connection wrapper built on top of pymodbus.

Added in version 0.3.0.

class icspacket.proto.modbus.connection.AddressRange(start: int, count: int)[source]#

A contiguous range of Modbus data addresses confirmed present on a device, as returned by Modbus_Connection.scan_table().

Variables:

start (int) – Zero-based starting address of the range.
count (int) – Number of consecutive addresses in the range.

property end: int#: Address immediately following the last address in this range (exclusive).

exception icspacket.proto.modbus.connection.ModbusProtocolError(*args: object, response: ModbusPDU | None = None)[source]#

Raised when a Modbus request fails at the application/protocol level.

This covers both explicit Modbus exception responses (e.g. illegal data address/value, slave device failure) and cases where no response was received at all.

Variables:: response (ModbusPDU | None) – The raw response PDU that caused this error, if any.

class icspacket.proto.modbus.connection.Modbus_Connection(unit_id: int = 1, timeout: float = 10.0, transport: Literal['tcp', 'udp'] = 'tcp')[source]#

Synchronous Modbus TCP/UDP connection built on pymodbus.

This class provides a thin, icspacket-style wrapper around pymodbus.client.ModbusTcpClient and pymodbus.client.ModbusUdpClient, exposing the most common read/write services (coils, discrete inputs, holding/input registers).

All application-level failures (Modbus exception responses, missing responses) are normalized into ModbusProtocolError.

Example:

conn = Modbus_Connection(unit_id=1)
conn.connect(("192.168.1.50", 502))
values = conn.read_holding_registers(0, count=4)
conn.write_register(0, 1234)
conn.close()

# Modbus/UDP
udp_conn = Modbus_Connection(unit_id=1, transport="udp")
udp_conn.connect(("192.168.1.50", 502))

Parameters:

unit_id (int) – Default Modbus unit/slave identifier used for requests that do not explicitly override it.
timeout (float) – Socket-level timeout (in seconds) for the underlying connection.
transport (ModbusTransport) – Transport protocol to use, either "tcp" (default) or "udp".

property client: ModbusBaseSyncClient#: The underlying pymodbus sync client instance (TCP or UDP).

connect(address: tuple[str, int]) → None[source]#

Connect to a Modbus TCP or UDP server.

Parameters:: address (tuple[str, int]) – Target (host, port) tuple (default Modbus port is 502).
Raises:: ConnectionError – If the connection cannot be established.

close() → None[source]#: Close the underlying Modbus TCP/UDP connection.

read_coils(address: int, count: int = 1, unit_id: int | None = None) → list[bool][source]#

Read one or more coils (0x addresses, function code 1).

Parameters:

address (int) – Zero-based starting coil address.
count (int) – Number of coils to read.
unit_id (int | None) – Override the default unit/slave identifier.

Returns:

List of coil states.

Return type:

list[bool]

Raises:

ModbusProtocolError – On an exception response or no response.

read_discrete_inputs(address: int, count: int = 1, unit_id: int | None = None) → list[bool][source]#: Read one or more discrete inputs (1x addresses, function code 2).

read_holding_registers(address: int, count: int = 1, unit_id: int | None = None) → list[int][source]#: Read one or more holding registers (4x addresses, function code 3).

read_input_registers(address: int, count: int = 1, unit_id: int | None = None) → list[int][source]#: Read one or more input registers (3x addresses, function code 4).

write_coil(address: int, value: bool, unit_id: int | None = None) → None[source]#: Write a single coil (function code 5).

write_coils(address: int, values: list[bool], unit_id: int | None = None) → None[source]#: Write multiple coils (function code 15).

Note

A copy of values is passed to the underlying pymodbus client, since it pads the bit list in place to a full byte boundary as a side effect of encoding the request - without the copy, the caller’s list would be silently mutated.

write_register(address: int, value: int, unit_id: int | None = None) → None[source]#: Write a single holding register (function code 6).

write_registers(address: int, values: list[int], unit_id: int | None = None) → None[source]#: Write multiple holding registers (function code 16).

read_device_information(unit_id: int | None = None) → dict[int, bytes][source]#

Read basic device identification objects (function code 43/14).

Returns:: Mapping of object id to raw identification value (e.g. vendor name, product code, revision).
Return type:: dict[int, bytes]

get_unit_ids(start: int = 1, end: int = 248) → list[int][source]#

Discover which unit/slave identifiers respond on this connection.

Modbus has no service to enumerate connected slaves, so this probes every candidate id in [start, end) with a minimal read_coils(0, count=1) request. Any well-formed Modbus reply including exception responses such as illegal data address or slave device failure is treated as evidence that a device with that unit id is present and answering, since it means the request was received and processed. Only a timeout (no reply at all) is treated as absence.

Parameters:

start (int) – First unit id to probe, inclusive.
end (int) – Last unit id to probe, exclusive. Defaults to 248 (one past the highest valid Modbus unit id, 247).

Returns:

Sorted list of responsive unit ids.

Return type:

list[int]

Raises:

ValueError – If start/end are out of bounds.

get_table(table: Literal['coils', 'discrete', 'holding', 'input'], start: int = 0, end: int = 65536, unit_id: int | None = None, block_size: int | None = None) → list[AddressRange][source]#

Discover which addresses of a data table are implemented.

Warning

Because exact discovery requires probing until every boundary is resolved, the number of requests is bounded by end - start in the worst case (e.g. a table that is entirely absent across the whole scanned range). Keep the scanned range modest (the default end is deliberately not the full address space) unless you are prepared for a slow, thorough scan.

Example:

conn = Modbus_Connection(unit_id=1)
conn.connect(("192.168.1.50", 502))
for r in conn.scan_table("holding", end=1000):
    print(f"holding registers: {r}")

Parameters:

table (ModbusTable) – Table to scan: "coils", "discrete", "holding" or "input".
start (int) – Zero-based starting address, inclusive.
end (int) – Ending address, exclusive. Defaults to 0x10000 (the full 16-bit Modbus address space) - narrow this for a faster scan.
unit_id (int | None) – Override the default unit/slave identifier.
block_size (int | None) – Maximum number of addresses to probe per request. Defaults to (and is capped at) the table’s protocol maximum; lower it if the target rejects maximum-size requests even within its valid range.

Returns:

Coalesced list of address ranges that responded successfully, in ascending order.

Return type:

list[AddressRange]

Raises:

ValueError – If table is unknown, the address bounds are invalid, or block_size is less than 1.
ModbusProtocolError – If a probe fails for a reason other than an illegal-address exception (e.g. a device failure response), since that does not indicate an absent address.