ModbusDevice#

Bases: Instrument

Config-driven Modbus client. Semantic access by register alias from a ModbusConfig.

Initialize a ModbusDevice.

Parameters:

config

A ModbusConfig, a dict (validated via Pydantic), or a path to a JSON config.
connection

Overrides config.connection. Accepts a ModbusTransport (pass the same instance to several devices to share one line; unit_id is then required), or a dict (with transport = "tcp" / "rtu"), which builds a private transport. Required if the config has no connection section.
name

Channel-name prefix; falls back to config.device.name.
publishers

Publishers that receive emitted Measurement/Command data.
autostart

When True, open the connection and start background polling. Requires a timing section (with poll_interval) — passing autostart=True without one is an error.
unit_id

Modbus unit/slave address (0-255; 0-247 on serial). May also come from the connection block’s unit_id field; values given in both places must agree. Required alongside a shared transport; defaults to 1 when this device builds its own private transport.
**kwargs

Default tags applied to every emitted Measurement/Command.

Raises:

ValueError – No connection in args or config, an address that is missing, conflicting, or out of range, or autostart=True with no timing section.

Attributes

`background_interval`	Seconds between background daemon iterations (0 = no wait).
`channel_names`	Return an insertion-ordered snapshot of all published channel names.
`unit_id`	set once at construction, via the constructor kwarg or the config.

Methods

`add_background_daemon_function`	Append `method` to the daemon's call list.
`add_publisher`	Register a publisher to receive this instrument's Measurement/Command data.
`close`	Release this device's hold on the transport and stop the daemon; the line closes when the last device leaves.
`define_background_daemon`	Replace all daemon functions with a single `method` (called with the given args).
`get_channel`	Return the most recent `length` samples for `channel_name` from the in-memory buffer.
`get_single_channel_value`	Return the most recent sample for `channel_name` from the in-memory buffer.
`open`	Open the Modbus TCP/RTU connection, registering this device as an owner of the transport.
`publish`	Fan `data` out to every configured publisher; `kwargs` pass through.
`read`	Read the register named `alias` and return the scaled value.
`start`	Start the background daemon thread.
`stop`	Signal the background daemon to stop and join it.
`write`	Write `value` to the register named `alias`.

property background_interval#: Seconds between background daemon iterations (0 = no wait).

property unit_id: int#

set once at construction, via the constructor kwarg or the config.

Type:: Modbus unit/slave address this device is bound to. Read-only

open() → None[source]#: Open the Modbus TCP/RTU connection, registering this device as an owner of the transport.

close() → None[source]#: Release this device’s hold on the transport and stop the daemon; the line closes when the last device leaves.

read(alias: str, **kwargs) → Measurement[source]#: Read the register named alias and return the scaled value.

write(

alias: str,

value: float | int | bool | str,

**kwargs,

) → Command[source]#

Write value to the register named alias.

value is in physical units when a scale is configured. For coils, pass True/False. For registers with a write_value_map, pass the string key to look up the mapped value.

Raises:

TypeError – Value type does not match the register’s data type.
KeyError – String value not found in write_value_map.
ValueError – Read-only register, value violates write_min/write_max, or scaled raw is out of range for the data type.

add_background_daemon_function(

method: Callable,

*args,

**kwargs,

)#: Append method to the daemon’s call list. Use define_background_daemon to replace instead.

add_publisher( publisher: Publisher, )#: Register a publisher to receive this instrument’s Measurement/Command data.

property channel_names: tuple[str, ...]#

Return an insertion-ordered snapshot of all published channel names.

Names are complete after one full daemon iteration: {name}.loop_time publishes every iteration, so call get_channel(“loop_time”, wait_for_new_samples=True) then read channel_names to ensure the complete set has arrived.

Raises:: RuntimeError – No background buffer; start() was not called.

define_background_daemon(

method: Callable,

*args,

**kwargs,

)#: Replace all daemon functions with a single method (called with the given args).

get_channel( channel_name: str, length: int = 1, wait_for_new_samples: bool = False, timeout: float = 10.0, ) → Measurement#

Return the most recent length samples for channel_name from the in-memory buffer.

If the channel does not exist yet, or no sample is available, the code will always block until timeout expires.

Parameters:

channel_name

Name of the channel to retrieve.
length

Number of trailing samples to return.
wait_for_new_samples

Block until at least length new values arrive.
timeout

Seconds to wait when insufficient data exists or wait_for_new_samples=True.

Raises:

RuntimeError – No background buffer; start() was not called.
ChannelNotFoundError – channel had no values and no data appeared before timeout. wait_for_new_samples=True and channel did not appear within timeout.
ChannelValueTimeoutError – wait_for_new_samples=True and values did not arrive within timeout.

get_single_channel_value( channel_name: str, ) → float | str | None#

Return the most recent sample for channel_name from the in-memory buffer.

This will not wait, if data is not available then None is returned.

Parameters:

channel_name

Name of the channel to retrieve.

Raises:

RuntimeError – No background buffer; start() was not called.

publish(

data: Measurement | Command,

**kwargs,

)#: Fan data out to every configured publisher; kwargs pass through.

start()#: Start the background daemon thread. No-op if already running.

stop()#: Signal the background daemon to stop and join it. No-op if not running.