Skip to content

modbus-connection

A small, backend-neutral Modbus connection abstraction — and a typed device-modelling framework on top of it.

modbus-connection is two things in one package:

  1. A connection abstraction — the ModbusConnection and ModbusUnit Protocols — that lets you type against Modbus without committing to a backend. Two interchangeable backends, pymodbus and tmodbus, implement it.
  2. An optional device-modelling framework (modbus_connection.model) that maps a device’s registers and coils to typed Python attributes and reads the whole device — or one sub-system — in as few Modbus calls as possible.

It imports no Modbus library at the top level and nothing from Home Assistant, so you can build a device library once and let the consumer choose the backend.

Model a device once, then connect, update, read, and write it. The optional modbus_connection.model framework maps a device’s registers and coils to typed attributes and reads the whole device in as few Modbus calls as possible.

import asyncio
from modbus_connection.tmodbus import connect_tcp
from modbus_connection.model import Component, gauge, uint32, coil
class Meter(Component):
voltage = gauge(0, 0.1, unit="V") # scaled 16-bit register
current = gauge(1, 0.1, unit="A")
energy = uint32(2, unit="Wh") # 32-bit over two registers
relay = coil(0, writable=True)
async def main() -> None:
conn = await connect_tcp("192.168.1.50", port=502)
try:
meter = Meter(conn.for_unit(1))
await meter.async_update() # one pooled block read
print(meter.voltage, meter.current, meter.energy, meter.relay)
await meter.write("relay", True) # write a writable field
finally:
await conn.close()
asyncio.run(main())

ModbusConnection

The shared, already-connected link to a Modbus network. One physical link addresses many units (1–247). It is transient and owner-held: a backend connect function returns a live instance — there is no connect(). Only the owner ever holds it, and only the owner tears it down with close().

ModbusUnit

A stateless per-unit handle you get from connection.for_unit(unit_id). It has no lifecycle methods — a consumer cannot connect or close the link it rides on. Every method raises on failure; it never returns None.

Requests on one connection are serialized by the backend, so concurrent unit calls can’t interleave. Sharing a single connection across many consumers is strictly better than each opening a competing socket — and that is exactly what this abstraction makes possible while keeping the backend swappable.

Read more about connections and units →