Modbus Interface

class pydoover.docker.ModbusInterface(app_key: str, modbus_uri: str = '127.0.0.1:50054', service_name: str = 'doover.ModbusInterface', timeout: int = 7, config: Schema = None)[source]

ModbusInterface is a gRPC interface for interacting with modbus devices.

It allows for opening and closing modbus buses, reading and writing registers, and subscribing to register updates. It is designed to be used with the modbus_iface gRPC service.

config

Configuration schema for the modbus interface, containing modbus bus definitions. This is loaded from application config automatically and should be specified in your app_config.py file.

Type:: Schema

add_read_register_subscription(bus_id: str = 'default', modbus_id: int = 1, start_address: int = 0, num_registers: int = 1, register_type: int = 4, poll_secs: int = 3, callback: Callable[[list[int]], None] | Coroutine[[list[int]], None] = None, bus=None)[source]

Add a subscription to read registers from a modbus bus.

This method creates a subcscription that will periodically read registers from the specified modbus device and invoke the provided callback when a read request succeeds.

The provided callback can be a regular function or a coroutine.

Examples

>>> def my_callback(values: list[int]):
...     print("Received new register values:", values)
>>> self.modbus_iface.add_read_register_subscription(
...     bus_id="my_bus",
...     modbus_id=1,
...     start_address=0,
...     num_registers=10,
...     callback=my_callback,
... )

Parameters:

bus_id (str, optional) – The bus ID to read registers from (default is “default”)
modbus_id (int, optional) – The modbus ID of the device to read registers from (default is 1)
start_address (int, optional) – The starting address of the registers to read (default is 0)
num_registers (int, optional) – The number of registers to read (default is 1)
register_type (int, optional) – The type of registers to read (default is 4, which is typically holding registers)
poll_secs (int, optional) – The polling interval in seconds for the subscription (default is 3 seconds)
callback (Callback) – The callback function to invoke when a read request succeeds. This accepts a list of integers representing the register values. If only one register is read, this will be a single integer. This callback can be a regular function or a coroutine.

close_bus(bus_id: str = 'default') → bool[source]: Close a modbus bus.

Deprecated since version Buses: are pooled by the modbus interface and no longer need explicit closing; see read_registers().

fetch_bus_status(bus_id: str = 'default') → bool[source]

Get the status of a modbus bus.

Parameters:: bus_id (str, optional) – The bus ID to fetch an OK status for
Returns:: True if the bus is open, False otherwise.
Return type:: bool

open_bus(bus_type='serial', name='default', serial_port='/dev/ttyS0', serial_baud=9600, serial_method='rtu', serial_bits=8, serial_parity='N', serial_stop=1, serial_timeout=0.3, tcp_uri='127.0.0.1:5000', tcp_timeout=2) → bool[source]: Open a modbus bus.

Deprecated since version Buses: now open on demand: pass connection settings to read_registers() / write_registers() (via bus or the application config) instead of pre-opening.

read_registers(bus_id: str = 'default', modbus_id: int = 1, start_address: int = 0, num_registers: int = 1, register_type: int = 4, configure_bus: bool = True, bus=None, retries: int | None = None) → int | list[int] | None[source]

Read registers from a modbus bus.

Examples

>>> self.modbus_iface.read_registers(bus_id="default", modbus_id=1, start_address=0, num_registers=10)

Parameters:

bus_id (str, optional) – Deprecated and ignored — kept for backwards compatibility. The bus is identified by its configured connection settings, not an id.
modbus_id (int, optional) – The modbus ID of the device to read registers from (default is 1)
start_address (int, optional) – The starting address of the registers to read (default is 0)
num_registers (int, optional) – The number of registers to read (default is 1)
register_type (int, optional) – The type of registers to read (default is 4, which is typically holding registers)
configure_bus (bool, optional) – Deprecated and ignored — the bus opens on demand from the request settings.
bus (ModbusConfig, optional) – The bus to read from. If omitted, the bus configured in the application config is used; pass this to select one when several are configured.
retries (int, optional) – How many times the interface retries on failure. 0 fails fast (no retry) — useful when a failure is expected/normal. Left unset, the interface applies its default.

Returns:

The values read from the registers. If only one register is read, returns an int. If multiple registers are read, returns a list of ints. If the response failed, returns None.

Return type:

int | list[int] | None

stub: alias of modbusIfaceStub

test_comms(message: str = 'Comms Check Message') → str | None[source]

Test connection by sending a basic echo response to modbus interface container.

Parameters:: message (str) – Message to send to modbus interface to have echo’d as a response
Returns:: The response from modbus interface.
Return type:: str

write_registers(bus_id: str = 'default', modbus_id: int = 1, start_address: int = 0, values: list[int] = None, register_type: int = 4, configure_bus: bool = True, bus=None, retries: int | None = None) → bool[source]

Write values to registers on a modbus bus.

Examples

>>> self.modbus_iface.write_registers(
...     bus_id="my_bus",
...     modbus_id=1,
...     start_address=0,
...     values=[100, 200, 300],
...     register_type=4,
...     configure_bus=True,
... )

Parameters:

bus_id (str) – Deprecated and ignored — kept for backwards compatibility.
modbus_id (int) – The modbus ID of the device to write registers to (default is 1)
start_address (int) – The starting address of the registers to write (default is 0)
values (list[int]) – Register values to write
register_type (int) – The type of registers to write (default is 4, which is typically holding registers)
configure_bus (bool) – Deprecated and ignored — the bus opens on demand from the request settings.
bus (ModbusConfig, optional) – The bus to write to. If omitted, the bus configured in the application config is used; pass this to select one when several are configured.
retries (int, optional) – How many times the interface retries on failure. 0 fails fast. Left unset, the interface applies its default.

Returns:

True if the write operation was successful, False otherwise.

Return type:

bool

class pydoover.docker.modbus.ModbusConfig(display_name: str = 'Modbus Config')[source]

class pydoover.docker.modbus.ManyModbusConfig(display_name: str = 'Modbus Config')[source]