Modbus TCP in Python con asyncio: client e server da zero

Chi lavora con impianti industriali, fotovoltaico, building automation o contatori di energia prima o poi incontra Modbus. È un protocollo del 1979, nato per i PLC Modicon, eppure è ancora lo standard di fatto per far dialogare dispositivi di produttori diversi. Il motivo è che fa poche cose e le fa in modo prevedibile: legge e scrive bit e registri a 16 bit, con un formato dei messaggi che si impara in un pomeriggio.

Python è un linguaggio molto diffuso in questo campo, perché è comodo per scrivere collettori di dati, script di diagnostica e gateway verso database e sistemi di monitoraggio. In questo articolo implementiamo Modbus TCP da zero con la sola libreria standard: un client asincrono basato su asyncio, un server con una memoria dati in RAM, le conversioni tra registri e tipi numerici e il CRC di Modbus RTU. Useremo il modulo struct per tutta la codifica binaria e scriveremo una suite di test con pytest.

Le quattro tabelle del modello dati

Un dispositivo Modbus non espone oggetti o strutture, ma quattro tabelle piatte, ciascuna indirizzabile da 0 a 65535:

Tabella	Elemento	Accesso	Esempi
Coil	bit	lettura e scrittura	relè, comandi di marcia e arresto
Discrete input	bit	sola lettura	finecorsa, allarmi, stati
Holding register	16 bit	lettura e scrittura	setpoint, parametri
Input register	16 bit	sola lettura	temperature, tensioni, contatori

Le schede tecniche dei dispositivi usano spesso una numerazione storica a cinque o sei cifre: 40001 indica il primo holding register, 30001 il primo input register. Questi numeri partono da 1, mentre l'indirizzo che viaggia nel pacchetto parte da 0: il registro 40001 si legge all'indirizzo 0. Se dopo aver collegato un dispositivo nuovo i valori sembrano sfasati di una posizione, il problema è quasi sempre questa differenza.

Com'è fatto un messaggio Modbus TCP

Ogni messaggio Modbus contiene una PDU (Protocol Data Unit): un byte con il codice funzione seguito dai parametri, per un massimo di 253 byte. La PDU è identica su qualunque trasporto. Su linea seriale, in Modbus RTU, viene preceduta dall'indirizzo del dispositivo e seguita da un CRC. Su TCP viene invece preceduta da un'intestazione di sette byte chiamata MBAP:

• transaction identifier, 2 byte, scelto dal client e ricopiato dal server nella risposta;
• protocol identifier, 2 byte, sempre 0;
• length, 2 byte, il numero di byte che seguono: unit identifier più PDU;
• unit identifier, 1 byte, l'indirizzo del dispositivo finale, usato soprattutto dai gateway TCP verso linee seriali.

Tutto è codificato in big-endian. In Python l'intestazione si descrive con una sola stringa di formato di struct: ">HHHB", cioè big-endian, tre interi senza segno a 16 bit e uno a 8 bit. Ecco una richiesta di lettura di dieci holding register dall'indirizzo 0, verso l'unità 1:

>>> import struct
>>> pdu = struct.pack(">BHH", 0x03, 0, 10)
>>> frame = struct.pack(">HHHB", 1, 0, len(pdu) + 1, 1) + pdu
>>> frame.hex(" ")
'00 01 00 00 00 06 01 03 00 00 00 0a'

Se la richiesta non può essere soddisfatta, il server risponde con il codice funzione con il bit più alto acceso, per esempio 0x83 invece di 0x03, seguito da un codice di eccezione: 0x01 funzione non supportata, 0x02 indirizzo non valido, 0x03 valore non valido, 0x04 guasto del dispositivo.

Struttura del progetto

Usiamo il layout con la directory src, che evita di importare per errore il codice dalla directory corrente invece che dal pacchetto installato:

modbus-tcp/
├── pyproject.toml
├── src/modbus/
│   ├── __init__.py
│   ├── pdu.py       codifica e decodifica delle PDU
│   ├── client.py    client asincrono
│   ├── server.py    memoria dati e server
│   ├── codec.py     conversioni tra registri e tipi numerici
│   └── crc.py       CRC16 per Modbus RTU
├── tests/test_modbus.py
└── examples/demo.py

Il modulo delle conversioni si chiama codec.py e non types.py: un modulo con lo stesso nome di uno della libreria standard è una fonte classica di errori di import difficili da diagnosticare.

[project]
name = "modbus-tcp"
version = "0.1.0"
requires-python = ">=3.12"

[project.optional-dependencies]
test = ["pytest>=8"]

[tool.pytest.ini_options]
pythonpath = ["src"]
testpaths = ["tests"]

Codici, limiti ed eccezioni

Nel modulo pdu.py definiamo i codici funzione e di eccezione con IntEnum: si confrontano direttamente con gli interi letti dal socket ma hanno un nome leggibile nei messaggi di errore. I limiti di quantità derivano dalla dimensione massima della PDU: 125 registri in lettura producono una risposta di 252 byte, il massimo che entra nei 253 disponibili.

"""Codifica e decodifica delle PDU Modbus."""

from __future__ import annotations

import struct
from collections.abc import Sequence
from enum import IntEnum


class FunctionCode(IntEnum):
    READ_COILS = 0x01
    READ_DISCRETE_INPUTS = 0x02
    READ_HOLDING_REGISTERS = 0x03
    READ_INPUT_REGISTERS = 0x04
    WRITE_SINGLE_COIL = 0x05
    WRITE_SINGLE_REGISTER = 0x06
    WRITE_MULTIPLE_REGISTERS = 0x10


class ExceptionCode(IntEnum):
    ILLEGAL_FUNCTION = 0x01
    ILLEGAL_DATA_ADDRESS = 0x02
    ILLEGAL_DATA_VALUE = 0x03
    SERVER_DEVICE_FAILURE = 0x04


# Quantita' massime per singola richiesta
MAX_READ_BITS = 2000
MAX_READ_REGISTERS = 125
MAX_WRITE_REGISTERS = 123


class ModbusError(Exception):
    """Classe base per tutti gli errori della libreria."""


class ProtocolError(ModbusError):
    """Risposta malformata o incoerente con la richiesta."""


class ModbusException(ModbusError):
    """Il dispositivo ha risposto con un'eccezione Modbus."""

    def __init__(self, function: int, code: int) -> None:
        self.function = function
        self.code = code
        try:
            name = ExceptionCode(code).name.lower().replace("_", " ")
        except ValueError:
            name = "unknown"
        super().__init__(f"exception 0x{code:02X} ({name}) for function 0x{function:02X}")

La distinzione tra ProtocolError e ModbusException è importante. La prima indica che qualcosa non va nello scambio di dati e che la connessione non è più affidabile. La seconda indica che il dispositivo ha capito la richiesta e l'ha rifiutata in modo ordinato: la connessione resta valida e si può continuare a usarla.

Richieste e risposte

Con struct.pack la costruzione delle richieste si riduce a una riga per funzione. Le scritture singole hanno lo stesso layout delle letture, con il valore al posto della quantità. Per i coil la specifica impone i valori 0xFF00 per acceso e 0x0000 per spento.

def read_request(function: int, address: int, quantity: int) -> bytes:
    return struct.pack(">BHH", function, address, quantity)


def write_single_coil_request(address: int, value: bool) -> bytes:
    return struct.pack(">BHH", FunctionCode.WRITE_SINGLE_COIL, address, 0xFF00 if value else 0)


def write_single_register_request(address: int, value: int) -> bytes:
    return struct.pack(">BHH", FunctionCode.WRITE_SINGLE_REGISTER, address, value)


def write_multiple_registers_request(address: int, values: Sequence[int]) -> bytes:
    count = len(values)
    return struct.pack(
        f">BHHB{count}H", FunctionCode.WRITE_MULTIPLE_REGISTERS, address, count, 2 * count, *values
    )

Coil e discrete input viaggiano impacchettati otto per byte, a partire dal bit meno significativo del primo byte. Le due funzioni seguenti convertono in entrambe le direzioni:

def pack_bits(bits: Sequence[bool]) -> bytes:
    out = bytearray((len(bits) + 7) // 8)
    for i, bit in enumerate(bits):
        if bit:
            out[i // 8] |= 1 << (i % 8)
    return bytes(out)


def unpack_bits(data: bytes, count: int) -> list[bool]:
    return [bool(data[i // 8] & (1 << (i % 8))) for i in range(count)]

Sul lato delle risposte il client verifica tutto ciò che può: codice funzione, byte di conteggio, lunghezza effettiva e, per le scritture, l'eco della richiesta. Per le funzioni 0x05 e 0x06 la risposta coincide con la richiesta; per la 0x10 contiene i primi cinque byte della richiesta, cioè funzione, indirizzo e quantità. In entrambi i casi il confronto con request[:5] è sufficiente.

def check_function(function: int, pdu: bytes) -> None:
    """Verifica il codice funzione e trasforma le eccezioni Modbus in errori Python."""
    if not pdu:
        raise ProtocolError("empty response")
    if pdu[0] == function | 0x80:
        if len(pdu) != 2:
            raise ProtocolError("malformed exception response")
        raise ModbusException(function, pdu[1])
    if pdu[0] != function:
        raise ProtocolError(f"unexpected function 0x{pdu[0]:02X}")


def parse_bits(function: int, pdu: bytes, quantity: int) -> list[bool]:
    check_function(function, pdu)
    expected = (quantity + 7) // 8
    if len(pdu) != 2 + expected or pdu[1] != expected:
        raise ProtocolError("unexpected byte count")
    return unpack_bits(pdu[2:], quantity)


def parse_registers(function: int, pdu: bytes, quantity: int) -> list[int]:
    check_function(function, pdu)
    expected = 2 * quantity
    if len(pdu) != 2 + expected or pdu[1] != expected:
        raise ProtocolError("unexpected byte count")
    return list(struct.unpack(f">{quantity}H", pdu[2:]))


def parse_echo(request: bytes, pdu: bytes) -> None:
    check_function(request[0], pdu)
    if pdu != request[:5]:
        raise ProtocolError("write echo mismatch")

Il client asincrono

Un collettore di dati interroga spesso decine di dispositivi, e con asyncio possiamo gestirli tutti in un solo thread: mentre una richiesta attende la risposta di un inverter lento, il ciclo degli eventi serve gli altri. Ogni istanza del client possiede una coppia StreamReader e StreamWriter e un asyncio.Lock che serializza le transazioni sulla stessa connessione. Molti dispositivi embedded elaborano una richiesta alla volta e si confondono se ne ricevono più di una in coda, quindi è la scelta più sicura.

"""Client Modbus TCP asincrono."""

from __future__ import annotations

import asyncio
import struct
from collections.abc import Sequence
from types import TracebackType
from typing import Self

from .pdu import (
    MAX_READ_BITS,
    MAX_READ_REGISTERS,
    MAX_WRITE_REGISTERS,
    FunctionCode,
    ModbusError,
    ProtocolError,
    parse_bits,
    parse_echo,
    parse_registers,
    read_request,
    write_multiple_registers_request,
    write_single_coil_request,
    write_single_register_request,
)

MBAP = struct.Struct(">HHHB")


class ClientClosedError(ModbusError):
    """Il client e' stato chiuso, oppure e' fallito per un errore di trasporto."""


class AsyncClient:
    def __init__(
        self,
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
        unit_id: int,
        timeout: float,
    ) -> None:
        self._reader = reader
        self._writer = writer
        self._lock = asyncio.Lock()
        self._tid = 0
        self._closed = False
        self.unit_id = unit_id
        self.timeout = timeout

    @classmethod
    async def connect(
        cls, host: str, port: int = 502, unit_id: int = 1, timeout: float = 2.0
    ) -> AsyncClient:
        async with asyncio.timeout(timeout):
            reader, writer = await asyncio.open_connection(host, port)
        return cls(reader, writer, unit_id, timeout)

    async def close(self) -> None:
        if not self._closed:
            self._closed = True
            self._writer.close()
        try:
            await self._writer.wait_closed()
        except (ConnectionError, OSError):
            pass

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        await self.close()

L'import di TracebackType dal modulo types della libreria standard funziona senza ambiguità proprio perché il nostro modulo si chiama codec, mentre Self, introdotto in Python 3.11, fa sì che __aenter__ restituisca il tipo corretto anche in eventuali sottoclassi. Precompilare il formato dell'intestazione con struct.Struct evita di riparsare la stringa di formato a ogni richiesta.

Il metodo _do esegue una transazione completa. Il blocco asyncio.timeout, disponibile da Python 3.11, copre scrittura e lettura con un'unica scadenza. Il metodo readexactly risolve il problema del framing: TCP è uno stream e un frame può arrivare spezzato in più segmenti, ma readexactly restituisce esattamente il numero di byte richiesto oppure solleva IncompleteReadError se la connessione si chiude prima.

    async def _do(self, pdu: bytes) -> bytes:
        async with self._lock:
            if self._closed:
                raise ClientClosedError("client closed")
            self._tid = (self._tid + 1) & 0xFFFF
            tid = self._tid
            frame = MBAP.pack(tid, 0, len(pdu) + 1, self.unit_id) + pdu
            try:
                async with asyncio.timeout(self.timeout):
                    self._writer.write(frame)
                    await self._writer.drain()
                    while True:
                        header = await self._reader.readexactly(MBAP.size)
                        rtid, protocol, length, unit = MBAP.unpack(header)
                        if protocol != 0 or not 2 <= length <= 254:
                            raise ProtocolError("invalid MBAP header")
                        body = await self._reader.readexactly(length - 1)
                        if rtid != tid:
                            # risposta tardiva a una transazione precedente
                            continue
                        if unit != self.unit_id:
                            raise ProtocolError(f"unexpected unit id {unit}")
                        return body
            except BaseException:
                # timeout, errore di rete o annullamento: lo stream non e' piu' affidabile
                self._closed = True
                self._writer.close()
                raise

Il blocco except BaseException è voluto, e il raise finale lo rende innocuo. Oltre agli errori ordinari intercetta anche asyncio.CancelledError, che deriva da BaseException. Se il task che esegue la richiesta viene annullato a metà lettura, nello stream restano byte di una risposta incompleta. La richiesta successiva li leggerebbe come se fossero l'inizio della propria risposta. Chiudere la connessione è l'unica strategia davvero sicura, e lasciamo al chiamante il compito di riconnettersi.

L'API pubblica valida i parametri prima di toccare la rete. In Python questa validazione ha un motivo in più: struct.pack solleverebbe un generico struct.error per un valore fuori intervallo, poco chiaro per chi usa la libreria.

    @staticmethod
    def _check(address: int, quantity: int, maximum: int) -> None:
        if not 0 <= address <= 0xFFFF:
            raise ValueError(f"address {address} out of range 0-65535")
        if not 1 <= quantity <= maximum:
            raise ValueError(f"quantity {quantity} out of range 1-{maximum}")

    async def _read_bits(self, function: int, address: int, quantity: int) -> list[bool]:
        self._check(address, quantity, MAX_READ_BITS)
        response = await self._do(read_request(function, address, quantity))
        return parse_bits(function, response, quantity)

    async def _read_registers(self, function: int, address: int, quantity: int) -> list[int]:
        self._check(address, quantity, MAX_READ_REGISTERS)
        response = await self._do(read_request(function, address, quantity))
        return parse_registers(function, response, quantity)

    async def read_coils(self, address: int, quantity: int) -> list[bool]:
        return await self._read_bits(FunctionCode.READ_COILS, address, quantity)

    async def read_discrete_inputs(self, address: int, quantity: int) -> list[bool]:
        return await self._read_bits(FunctionCode.READ_DISCRETE_INPUTS, address, quantity)

    async def read_holding_registers(self, address: int, quantity: int) -> list[int]:
        return await self._read_registers(FunctionCode.READ_HOLDING_REGISTERS, address, quantity)

    async def read_input_registers(self, address: int, quantity: int) -> list[int]:
        return await self._read_registers(FunctionCode.READ_INPUT_REGISTERS, address, quantity)

    async def write_single_coil(self, address: int, value: bool) -> None:
        self._check(address, 1, 1)
        request = write_single_coil_request(address, value)
        parse_echo(request, await self._do(request))

    async def write_single_register(self, address: int, value: int) -> None:
        self._check(address, 1, 1)
        if not 0 <= value <= 0xFFFF:
            raise ValueError(f"register value {value} out of range 0-65535")
        request = write_single_register_request(address, value)
        parse_echo(request, await self._do(request))

    async def write_multiple_registers(self, address: int, values: Sequence[int]) -> None:
        self._check(address, len(values), MAX_WRITE_REGISTERS)
        if any(not 0 <= v <= 0xFFFF for v in values):
            raise ValueError("register values must be in range 0-65535")
        request = write_multiple_registers_request(address, values)
        parse_echo(request, await self._do(request))

La memoria dati

Il server ha bisogno di un posto in cui tenere le quattro tabelle. Una dataclass con quattro liste è sufficiente, e il metodo handle trasforma una PDU di richiesta in una PDU di risposta senza sapere nulla di socket o intestazioni MBAP.

In un programma asyncio non serve alcun lock: handle è una funzione sincrona, e nessun'altra coroutine può essere eseguita mentre è in corso. Ogni richiesta viene quindi applicata in modo atomico, e lo stesso vale per un aggiornamento fatto da un'altra coroutine che scrive due registri consecutivi senza await in mezzo. Se invece la memoria venisse aggiornata da un thread separato, per esempio quello di un driver hardware, gli aggiornamenti andrebbero passati al ciclo degli eventi con loop.call_soon_threadsafe.

"""Memoria dati e server Modbus TCP."""

from __future__ import annotations

import asyncio
import struct
from dataclasses import dataclass

from .pdu import (
    MAX_READ_BITS,
    MAX_READ_REGISTERS,
    MAX_WRITE_REGISTERS,
    ExceptionCode,
    FunctionCode,
    pack_bits,
)

MBAP = struct.Struct(">HHHB")


def exception(function: int, code: int) -> bytes:
    return bytes([function | 0x80, code])


@dataclass
class DataStore:
    coils: list[bool]
    discrete_inputs: list[bool]
    holding_registers: list[int]
    input_registers: list[int]

    @classmethod
    def create(cls, size: int) -> DataStore:
        return cls([False] * size, [False] * size, [0] * size, [0] * size)

    def handle(self, pdu: bytes) -> bytes:
        """Elabora una PDU di richiesta e restituisce la PDU di risposta."""
        if not pdu:
            return exception(0, ExceptionCode.ILLEGAL_FUNCTION)
        function = pdu[0]

        if function in (FunctionCode.READ_COILS, FunctionCode.READ_DISCRETE_INPUTS):
            if len(pdu) != 5:
                return exception(function, ExceptionCode.ILLEGAL_DATA_VALUE)
            address, quantity = struct.unpack_from(">HH", pdu, 1)
            if not 1 <= quantity <= MAX_READ_BITS:
                return exception(function, ExceptionCode.ILLEGAL_DATA_VALUE)
            bits = self.coils if function == FunctionCode.READ_COILS else self.discrete_inputs
            if address + quantity > len(bits):
                return exception(function, ExceptionCode.ILLEGAL_DATA_ADDRESS)
            data = pack_bits(bits[address : address + quantity])
            return bytes([function, len(data)]) + data

        if function in (FunctionCode.READ_HOLDING_REGISTERS, FunctionCode.READ_INPUT_REGISTERS):
            if len(pdu) != 5:
                return exception(function, ExceptionCode.ILLEGAL_DATA_VALUE)
            address, quantity = struct.unpack_from(">HH", pdu, 1)
            if not 1 <= quantity <= MAX_READ_REGISTERS:
                return exception(function, ExceptionCode.ILLEGAL_DATA_VALUE)
            if function == FunctionCode.READ_HOLDING_REGISTERS:
                registers = self.holding_registers
            else:
                registers = self.input_registers
            if address + quantity > len(registers):
                return exception(function, ExceptionCode.ILLEGAL_DATA_ADDRESS)
            values = registers[address : address + quantity]
            return struct.pack(f">BB{quantity}H", function, 2 * quantity, *values)

        if function == FunctionCode.WRITE_SINGLE_COIL:
            if len(pdu) != 5:
                return exception(function, ExceptionCode.ILLEGAL_DATA_VALUE)
            address, value = struct.unpack_from(">HH", pdu, 1)
            if value not in (0xFF00, 0x0000):
                return exception(function, ExceptionCode.ILLEGAL_DATA_VALUE)
            if address >= len(self.coils):
                return exception(function, ExceptionCode.ILLEGAL_DATA_ADDRESS)
            self.coils[address] = value == 0xFF00
            return bytes(pdu)

        if function == FunctionCode.WRITE_SINGLE_REGISTER:
            if len(pdu) != 5:
                return exception(function, ExceptionCode.ILLEGAL_DATA_VALUE)
            address, value = struct.unpack_from(">HH", pdu, 1)
            if address >= len(self.holding_registers):
                return exception(function, ExceptionCode.ILLEGAL_DATA_ADDRESS)
            self.holding_registers[address] = value
            return bytes(pdu)

        if function == FunctionCode.WRITE_MULTIPLE_REGISTERS:
            if len(pdu) < 6:
                return exception(function, ExceptionCode.ILLEGAL_DATA_VALUE)
            address, quantity, byte_count = struct.unpack_from(">HHB", pdu, 1)
            if (
                not 1 <= quantity <= MAX_WRITE_REGISTERS
                or byte_count != 2 * quantity
                or len(pdu) != 6 + byte_count
            ):
                return exception(function, ExceptionCode.ILLEGAL_DATA_VALUE)
            if address + quantity > len(self.holding_registers):
                return exception(function, ExceptionCode.ILLEGAL_DATA_ADDRESS)
            written = struct.unpack_from(f">{quantity}H", pdu, 6)
            self.holding_registers[address : address + quantity] = written
            return bytes(pdu[:5])

        return exception(function, ExceptionCode.ILLEGAL_FUNCTION)

L'ordine dei controlli segue la specifica: prima il codice funzione (eccezione 0x01), poi la forma della richiesta e le quantità (0x03), infine gli indirizzi (0x02). In Python non dobbiamo preoccuparci di overflow nella somma tra indirizzo e quantità, perché gli interi hanno precisione arbitraria.

Il server TCP

Il server si appoggia ad asyncio.start_server, che chiama una coroutine per ogni nuova connessione. La coroutine legge frame in sequenza, applica il filtro sull'unit identifier e risponde ricopiando transaction identifier e unit identifier della richiesta. Un'intestazione non valida chiude la connessione, perché dopo un errore di framing non c'è modo affidabile di capire dove inizi il frame successivo.

class Server:
    def __init__(
        self, store: DataStore, unit_id: int = 0, idle_timeout: float | None = None
    ) -> None:
        self.store = store
        # 0 accetta qualunque unit identifier
        self.unit_id = unit_id
        # None disabilita la chiusura delle connessioni inattive
        self.idle_timeout = idle_timeout

    async def handle_connection(
        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
    ) -> None:
        try:
            while True:
                async with asyncio.timeout(self.idle_timeout):
                    header = await reader.readexactly(MBAP.size)
                tid, protocol, length, unit = MBAP.unpack(header)
                if protocol != 0 or not 2 <= length <= 254:
                    break
                pdu = await reader.readexactly(length - 1)
                if self.unit_id and unit != self.unit_id:
                    # un dispositivo reale ignora le richieste destinate ad altri
                    continue
                response = self.store.handle(pdu)
                writer.write(MBAP.pack(tid, 0, len(response) + 1, unit) + response)
                await writer.drain()
        except (asyncio.IncompleteReadError, ConnectionError, TimeoutError):
            pass
        finally:
            writer.close()
            try:
                await writer.wait_closed()
            except (ConnectionError, OSError):
                pass

    async def start(self, host: str = "0.0.0.0", port: int = 502) -> asyncio.Server:
        return await asyncio.start_server(self.handle_connection, host, port)

Il timeout di inattività conta più di quanto sembri. Molti dispositivi reali accettano pochissime connessioni contemporanee, a volte una sola. Un client che si riavvia senza chiudere il socket lascia una connessione orfana che occupa quel posto. Lato server, chiudere le connessioni silenziose da troppo tempo evita il problema. Passando None ad asyncio.timeout si disattiva il limite.

Dai registri ai valori

Un registro contiene soltanto 16 bit, e il loro significato dipende dal dispositivo. Le tre codifiche più frequenti sono l'intero con segno a 16 bit, spesso scalato (una temperatura in decimi di grado), l'intero a 32 bit su due registri (un contatore di energia) e il float32 IEEE 754 su due registri. Per i valori a 32 bit la specifica non stabilisce in che ordine vadano i due registri, e i produttori hanno scelto entrambe le strade: parola alta per prima (ABCD) oppure parola bassa per prima (CDAB).

"""Conversioni tra registri Modbus e tipi numerici."""

from __future__ import annotations

import struct
from collections.abc import Sequence
from enum import Enum


class WordOrder(Enum):
    HIGH_WORD_FIRST = "ABCD"
    LOW_WORD_FIRST = "CDAB"


def int16_from_register(value: int) -> int:
    """Interpreta un registro come intero con segno in complemento a due."""
    return value - 0x10000 if value & 0x8000 else value


def _ordered(registers: Sequence[int], order: WordOrder) -> tuple[int, int]:
    if len(registers) != 2:
        raise ValueError("exactly two registers are required")
    high, low = registers
    if order is WordOrder.LOW_WORD_FIRST:
        high, low = low, high
    return high, low


def uint32_from_registers(registers: Sequence[int], order: WordOrder) -> int:
    high, low = _ordered(registers, order)
    return (high << 16) | low


def registers_from_uint32(value: int, order: WordOrder) -> list[int]:
    high, low = (value >> 16) & 0xFFFF, value & 0xFFFF
    return [low, high] if order is WordOrder.LOW_WORD_FIRST else [high, low]


def float32_from_registers(registers: Sequence[int], order: WordOrder) -> float:
    high, low = _ordered(registers, order)
    (value,) = struct.unpack(">f", struct.pack(">HH", high, low))
    return float(value)


def registers_from_float32(value: float, order: WordOrder) -> list[int]:
    raw = struct.unpack(">I", struct.pack(">f", value))[0]
    return registers_from_uint32(raw, order)

Una nota sulla precisione: Python lavora internamente con float a 64 bit, quindi un valore come 230,1 passato attraverso float32 torna indietro come 230,10000610351562. Non è un errore della conversione ma il limite della rappresentazione a 32 bit, ed è bene arrotondare alla precisione dichiarata dal dispositivo prima di mostrare o salvare il valore.

Quando la documentazione non chiarisce l'ordine delle parole, conviene leggere una grandezza nota, come la tensione di rete o la frequenza a 50 Hz, e provare entrambe le interpretazioni: quella sbagliata produce numeri palesemente assurdi.

Il CRC di Modbus RTU

Molti dispositivi parlano solo Modbus RTU su RS-485, e alcuni gateway inoltrano i frame RTU così come sono dentro una connessione TCP. In entrambi i casi serve il CRC-16 di Modbus: polinomio riflesso 0xA001, valore iniziale 0xFFFF, trasmesso con il byte meno significativo per primo.

"""CRC16 di Modbus RTU."""


def crc16(data: bytes) -> int:
    crc = 0xFFFF
    for byte in data:
        crc ^= byte
        for _ in range(8):
            if crc & 1:
                crc = (crc >> 1) ^ 0xA001
            else:
                crc >>= 1
    return crc


def append_crc(frame: bytes) -> bytes:
    """Aggiunge il CRC in coda al frame, byte meno significativo per primo."""
    return frame + crc16(frame).to_bytes(2, "little")

Il metodo int.to_bytes con "little" esprime l'ordine dei byte del CRC in modo esplicito. Il frame RTU della solita lettura di dieci registri diventa 01 03 00 00 00 0A C5 CD. Per volumi elevati il calcolo bit a bit si può sostituire con una tabella precalcolata di 256 valori, ma su una linea seriale il collo di bottiglia è sempre la velocità di trasmissione.

Il file __init__.py espone l'API pubblica del pacchetto:

"""Implementazione minimale di Modbus TCP."""

from .client import AsyncClient, ClientClosedError
from .codec import (
    WordOrder,
    float32_from_registers,
    int16_from_register,
    registers_from_float32,
    registers_from_uint32,
    uint32_from_registers,
)
from .crc import append_crc, crc16
from .pdu import (
    ExceptionCode,
    FunctionCode,
    ModbusError,
    ModbusException,
    ProtocolError,
)
from .server import DataStore, Server

__all__ = [
    "AsyncClient",
    "ClientClosedError",
    "DataStore",
    "ExceptionCode",
    "FunctionCode",
    "ModbusError",
    "ModbusException",
    "ProtocolError",
    "Server",
    "WordOrder",
    "append_crc",
    "crc16",
    "float32_from_registers",
    "int16_from_register",
    "registers_from_float32",
    "registers_from_uint32",
    "uint32_from_registers",
]

I test

Per testare il codice asincrono non serve alcun plugin: ogni test definisce una coroutine e la esegue con asyncio.run. Il context manager running avvia un server su una porta libera scelta dal sistema, indicando la porta 0, apre un client e alla fine chiude tutto nell'ordine giusto. Da Python 3.12 Server.wait_closed attende la chiusura di tutte le connessioni, quindi il client va chiuso prima del server.

import asyncio
import math
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import pytest

from modbus import (
    AsyncClient,
    ClientClosedError,
    DataStore,
    ExceptionCode,
    ModbusException,
    Server,
    WordOrder,
    append_crc,
    float32_from_registers,
    int16_from_register,
    registers_from_float32,
)


@asynccontextmanager
async def running(size: int = 100) -> AsyncIterator[tuple[DataStore, AsyncClient]]:
    store = DataStore.create(size)
    server = await Server(store).start("127.0.0.1", 0)
    port = server.sockets[0].getsockname()[1]
    try:
        async with await AsyncClient.connect("127.0.0.1", port, timeout=1.0) as client:
            yield store, client
    finally:
        server.close()
        await server.wait_closed()


def test_registers_round_trip() -> None:
    async def scenario() -> None:
        async with running() as (_, client):
            await client.write_multiple_registers(10, [1, 2, 0xBEEF])
            await client.write_single_register(13, 42)
            assert await client.read_holding_registers(10, 4) == [1, 2, 0xBEEF, 42]

    asyncio.run(scenario())


def test_coils_and_inputs() -> None:
    async def scenario() -> None:
        async with running() as (store, client):
            store.discrete_inputs[9] = True
            store.input_registers[0] = 0xFF38
            await client.write_single_coil(3, True)
            assert await client.read_coils(0, 10) == [i == 3 for i in range(10)]
            inputs = await client.read_discrete_inputs(0, 10)
            assert inputs[9] and not inputs[8]
            (raw,) = await client.read_input_registers(0, 1)
            assert int16_from_register(raw) == -200

    asyncio.run(scenario())


def test_illegal_address_keeps_connection() -> None:
    async def scenario() -> None:
        async with running() as (_, client):
            with pytest.raises(ModbusException) as info:
                await client.read_holding_registers(95, 10)
            assert info.value.code == ExceptionCode.ILLEGAL_DATA_ADDRESS
            # dopo un'eccezione Modbus la connessione resta valida
            assert await client.read_holding_registers(0, 1) == [0]

    asyncio.run(scenario())


def test_client_side_validation() -> None:
    async def scenario() -> None:
        async with running() as (_, client):
            with pytest.raises(ValueError):
                await client.read_holding_registers(0, 126)
            with pytest.raises(ValueError):
                await client.write_single_register(0, 70000)

    asyncio.run(scenario())


def test_timeout_closes_client() -> None:
    async def silent(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        # legge fino alla chiusura senza mai rispondere
        await reader.read()
        writer.close()

    async def scenario() -> None:
        server = await asyncio.start_server(silent, "127.0.0.1", 0)
        port = server.sockets[0].getsockname()[1]
        client = await AsyncClient.connect("127.0.0.1", port, timeout=0.1)
        with pytest.raises(TimeoutError):
            await client.read_coils(0, 1)
        with pytest.raises(ClientClosedError):
            await client.read_coils(0, 1)
        await client.close()
        server.close()
        await server.wait_closed()

    asyncio.run(scenario())


def test_handle_illegal_function() -> None:
    assert DataStore.create(10).handle(bytes([0x2B, 0x0E])) == bytes([0xAB, 0x01])


def test_float_word_order() -> None:
    for order in WordOrder:
        registers = registers_from_float32(230.5, order)
        assert math.isclose(float32_from_registers(registers, order), 230.5)
    assert registers_from_float32(1.0, WordOrder.HIGH_WORD_FIRST) == [0x3F80, 0x0000]
    assert registers_from_float32(1.0, WordOrder.LOW_WORD_FIRST) == [0x0000, 0x3F80]


def test_crc16() -> None:
    frame = append_crc(bytes([0x01, 0x03, 0x00, 0x00, 0x00, 0x0A]))
    assert frame[-2:] == bytes([0xC5, 0xCD])

Il test sul timeout verifica due cose: che la chiamata fallisca con il TimeoutError predefinito, che da Python 3.11 è lo stesso sollevato da asyncio.timeout, e che il client rifiuti le chiamate successive. Si installano le dipendenze di test e si esegue la suite:

python -m venv .venv
source .venv/bin/activate
pip install -e ".[test]"
pytest -v

Un collettore di esempio

Il programma dimostrativo avvia un server che simula un misuratore di energia, con la tensione come float32 negli input register 0 e 1, e un client che la interroga una volta al secondo. Le tre coroutine convivono nello stesso ciclo degli eventi.

"""Simulatore di misuratore di energia e relativo client di polling."""

import asyncio
import random

from modbus import (
    AsyncClient,
    DataStore,
    Server,
    WordOrder,
    float32_from_registers,
    registers_from_float32,
)


async def simulate(store: DataStore) -> None:
    while True:
        voltage = 228 + random.random() * 4
        # un'unica assegnazione senza await: nessun client legge un valore a meta'
        store.input_registers[0:2] = registers_from_float32(voltage, WordOrder.HIGH_WORD_FIRST)
        await asyncio.sleep(1)


async def main() -> None:
    store = DataStore.create(10)
    server = await Server(store, idle_timeout=60).start("127.0.0.1", 1502)
    simulator = asyncio.create_task(simulate(store))

    async with await AsyncClient.connect("127.0.0.1", 1502) as client:
        for _ in range(5):
            registers = await client.read_input_registers(0, 2)
            voltage = float32_from_registers(registers, WordOrder.HIGH_WORD_FIRST)
            print(f"voltage: {voltage:.2f} V (raw {registers[0]:04X} {registers[1]:04X})")
            await asyncio.sleep(1)

    simulator.cancel()
    server.close()
    await server.wait_closed()


if __name__ == "__main__":
    asyncio.run(main())

La porta 1502 evita di dover eseguire lo script con privilegi di amministratore, necessari su Linux per le porte inferiori a 1024. Per interrogare più dispositivi in parallelo basta creare un client per ciascuno e raccogliere le letture con asyncio.gather o con un asyncio.TaskGroup.

Considerazioni per la produzione

• Riconnessione: il client si chiude al primo errore di trasporto. Il ciclo di polling deve quindi ricrearlo con un backoff esponenziale. Le ModbusException non richiedono la riconnessione, mentre TimeoutError, ProtocolError e gli errori di rete sì.
• Letture raggruppate: ogni transazione costa almeno un round trip, che dietro un gateway seriale può valere decine di millisecondi. Leggere un blocco contiguo di registri con una sola richiesta è molto più efficiente che leggerli uno alla volta.
• Rispetto dei dispositivi: molti apparati hanno microcontrollori modesti, e un polling troppo frequente può rallentarli o bloccarli.
• Sicurezza: Modbus TCP non ha autenticazione né cifratura, e chiunque raggiunga la porta 502 può scrivere registri. I dispositivi vanno isolati in una rete dedicata, con accessi filtrati dal firewall. La variante Modbus/TCP Security usa TLS sulla porta 802, ma pochi dispositivi la supportano.
• Funzioni aggiuntive: la scrittura di più coil (0x0F), la scrittura con maschera (0x16) e la lettura e scrittura combinata (0x17) si aggiungono con lo stesso schema delle altre.

Se servono il trasporto seriale, la modalità ASCII o funzioni meno comuni, la libreria di riferimento in Python è pymodbus, che fornisce client e server sincroni e asincroni per TCP, UDP, RTU e TLS. Conoscere il formato dei byte resta comunque prezioso, perché è quello che permette di leggere una cattura di Wireshark e capire in pochi secondi se il problema è nel dispositivo, nella rete o nel proprio codice.

Conclusione

Abbiamo costruito un'implementazione di Modbus TCP completa per le funzioni più usate, con un client asincrono che gestisce timeout, annullamenti e risposte tardive, un server che rispetta la semantica delle eccezioni, le conversioni per interi con segno, interi a 32 bit e float con entrambi gli ordini di parola, e il CRC di Modbus RTU. Il modulo struct si è rivelato lo strumento ideale per descrivere i formati binari in modo compatto, mentre asyncio ci ha permesso di gestire connessioni concorrenti in un solo thread, senza lock e senza callback.