Ir al contenido

Cómo usar PyModbus con un Raspberry Pi PLC: ejemplos asyncio

Modbus TCP y RTU desde Python en un Raspberry PLC: instalación, servidor y cliente asyncio, probado con pymodbus 3.16
25 de septiembre de 2023 por
Cómo usar PyModbus con un Raspberry Pi PLC: ejemplos asyncio
Boot & Work Corp. S.L, Ricard Franch Argullol

Guía completa: Modbus TCP y RTU en los Raspberry PLC, con Python y Node-RED

PyModbus es la librería de Python para Modbus TCP y Modbus RTU. En un Raspberry PLC se instala en un entorno virtual de Python con pip install "pymodbus[serial]", y a partir de ahí el PLC puede ser cliente Modbus (maestro) que lee contadores y variadores, o servidor Modbus (esclavo) que lee un SCADA. Usa el cliente síncrono para scripts sencillos, y el cliente asyncio cuando un programa tiene que consultar varios equipos o servir datos a la vez.

Esta guía muestra la parte asíncrona de pymodbus con dos programas probados: un servidor Modbus TCP que publica la temperatura de la CPU del PLC, y un cliente asyncio que consulta dos PLC a la vez. Los dos se han ejecutado con pymodbus 3.16.1 y Python 3.13, por Modbus TCP y por Modbus RTU sobre un enlace serie virtual. Para el cableado, los nombres de los puertos RS-485 y los ejemplos síncronos, consulta la guía completa de Modbus para el Raspberry PLC.

¿Qué es pymodbus?

pymodbus es una pila Modbus de código abierto escrita en Python, con licencia BSD-3-Clause. Tiene clientes y servidores para Modbus TCP, Modbus RTU y ASCII en líneas serie, Modbus sobre TLS y sobre UDP, cada uno con una API síncrona y una API asyncio. En un Raspberry PLC funciona sobre Raspberry Pi OS, como cualquier programa de Python.

Función en el Raspberry PLCClase de pymodbusUso típico
Cliente Modbus TCP (maestro)ModbusTcpClient, AsyncModbusTcpClientLeer contadores de energía, variadores y E/S remotas por Ethernet
Cliente Modbus RTU (maestro)ModbusSerialClient, AsyncModbusSerialClientConsultar esclavos en la línea RS-485
Servidor Modbus TCP (esclavo)StartTcpServer, StartAsyncTcpServerDar valores del PLC a un SCADA, un HMI u otro PLC
Servidor Modbus RTU (esclavo)StartSerialServer, StartAsyncSerialServerResponder a un maestro en la línea RS-485

Instalar pymodbus en el Raspberry PLC

Raspberry Pi OS basado en Debian 12 (Bookworm) y posteriores bloquea pip install fuera de un entorno virtual, así que crea uno. pymodbus 3.9 y posteriores necesitan Python 3.10 o posterior.

python3 -m venv ~/modbus-venv
source ~/modbus-venv/bin/activate
pip install "pymodbus[serial]==3.16.1"
python -c "import pymodbus; print(pymodbus.__version__)"

La opción [serial] instala pyserial, que Modbus RTU necesita. Fija la versión en tu proyecto: pymodbus cambia su API a menudo, y la próxima versión mayor (v4) elimina las clases antiguas del datastore.

pymodbus síncrono o asíncrono

Las dos API envían las mismas tramas Modbus. La diferencia es cómo espera tu programa las respuestas.

SíncronoAsíncrono (asyncio)
Cómo espera una peticiónLa llamada se bloquea hasta la respuesta o el timeoutLa llamada se espera con await, y otras tareas se ejecutan mientras tanto
Varios equiposUno tras otro: un equipo caído retrasa a todos los demás con su timeoutUna tarea por equipo: un equipo caído solo retrasa su propia tarea
Cliente y servidor en un programaNecesita hilos y bloqueosTareas en el mismo bucle de eventos
CódigoMás corto, más fácil de depurarasync y await en cada función que habla Modbus
Adecuado paraUn script que lee un equipo, pruebas, tareas cronPasarelas, registradores de datos, un PLC que es cliente y servidor a la vez

Un cliente de pymodbus no es seguro entre hilos: no compartas un cliente entre hilos. Con asyncio, una tarea por equipo y un cliente por tarea evitan el problema.

Servidor Modbus TCP en el Raspberry PLC con valores en vivo

Este servidor responde como equipo 1 en el puerto TCP 5020. Tiene cuatro tablas separadas de 10 valores: coils, entradas discretas, holding registers e input registers. Antes de cada lectura de registros, la función refresh escribe la temperatura de la CPU del PLC en el holding register 0 y los segundos desde el arranque en el registro 1. De la misma forma puedes publicar las entradas del PLC o valores de tu propio programa.

#!/usr/bin/env python3
"""Modbus TCP server (slave) on a Raspberry PLC with pymodbus 3.16.

Holding register 0: CPU temperature in tenths of a degree (452 = 45.2 C)
Holding register 1: seconds since the server started
Coils 0 to 9: free coils that a client can write and read
"""
import asyncio
import time

from pymodbus.server import StartAsyncTcpServer
from pymodbus.simulator import DataType, SimData, SimDevice

PORT = 5020          # 502 needs root, see the note in the text
DEVICE_ID = 1        # unit identifier that this server answers to
TEMP_FILE = "/sys/class/thermal/thermal_zone0/temp"
START = time.monotonic()


def cpu_temp_tenths():
    with open(TEMP_FILE) as f:
        return int(f.read()) // 100  # millidegrees to tenths of a degree


async def refresh(function_code, start_address, address, count,
                  current_registers, set_values):
    """Called by pymodbus on every request, before it answers."""
    if function_code in (3, 4):  # read holding or input registers
        current_registers[0] = cpu_temp_tenths()
        current_registers[1] = int(time.monotonic() - START) & 0xFFFF
    return None  # no exception: answer normally


device = SimDevice(
    DEVICE_ID,
    # Four separate tables: coils, discrete inputs, holding, input registers
    simdata=(
        [SimData(0, count=10, values=False, datatype=DataType.BITS)],
        [SimData(0, count=10, values=False, datatype=DataType.BITS)],
        [SimData(0, count=10, values=0, datatype=DataType.REGISTERS)],
        [SimData(0, count=10, values=0, datatype=DataType.REGISTERS)],
    ),
    action=refresh,
)


async def main():
    # 0.0.0.0 listens on all interfaces: eth0, eth1 and localhost
    await StartAsyncTcpServer(device, address=("0.0.0.0", PORT))


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

En Linux, solo root puede abrir puertos TCP por debajo de 1024, así que un usuario normal que arranca un servidor en el puerto 502 recibe un PermissionError. La prueba usa el 5020. En producción, ejecuta el servicio con los permisos que necesita o redirige el puerto 502 al puerto de tu servidor. Este servidor usa SimDevice y SimData, el datastore de pymodbus 3.12 y posteriores. El antiguo ModbusSlaveContext ya no existe.

Cliente Modbus asyncio que consulta dos equipos a la vez

El cliente ejecuta una tarea por equipo con asyncio.gather. Cada 2 segundos, cada tarea invierte la coil 0 (FC05), la vuelve a leer (FC01) y lee los dos holding registers (FC03). Si un equipo no responde, su tarea espera 5 segundos y se vuelve a conectar, y la otra tarea sigue.

#!/usr/bin/env python3
"""Asynchronous Modbus TCP client (master) with pymodbus 3.16.

Polls two devices at the same time. Each device has its own task, so a
slow or missing device does not delay the other one.
"""
import asyncio

from pymodbus.client import AsyncModbusTcpClient
from pymodbus.exceptions import ModbusException

DEVICES = [
    # name, IP address, TCP port, unit identifier
    ("plc-a", "127.0.0.1", 5020, 1),
    ("plc-b", "127.0.0.1", 5021, 1),
]
POLL_S = 2.0


async def poll(name, host, port, device_id):
    client = AsyncModbusTcpClient(host, port=port, timeout=1, retries=1)
    coil = False
    while True:
        if not client.connected and not await client.connect():
            print(f"{name}: not reachable, new try in 5 s")
            await asyncio.sleep(5)
            continue
        try:
            coil = not coil
            await client.write_coil(0, coil, device_id=device_id)            # FC05
            rc = await client.read_coils(0, count=1, device_id=device_id)    # FC01
            rr = await client.read_holding_registers(0, count=2, device_id=device_id)  # FC03
            if rc.isError() or rr.isError():
                print(f"{name}: exception response {rc} {rr}")
            else:
                temp = rr.registers[0] / 10
                print(f"{name}: coil 0 = {rc.bits[0]}, CPU {temp:.1f} C, "
                      f"uptime {rr.registers[1]} s")
        except ModbusException as exc:
            print(f"{name}: no valid answer ({exc})")
            client.close()  # reconnect on the next cycle
        await asyncio.sleep(POLL_S)


async def main():
    await asyncio.gather(*(poll(*d) for d in DEVICES))


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

Salida con los dos servidores en marcha, y después con el segundo servidor parado:

plc-a: coil 0 = True, CPU 45.2 C, uptime 1 s
plc-b: coil 0 = True, CPU 45.2 C, uptime 1 s
plc-a: coil 0 = False, CPU 45.2 C, uptime 3 s
plc-b: coil 0 = False, CPU 45.2 C, uptime 3 s
...
plc-b: not reachable, new try in 5 s
plc-a: coil 0 = True, CPU 45.2 C, uptime 9 s
plc-a: coil 0 = False, CPU 45.2 C, uptime 11 s

Para probarlo en un solo PLC, arranca el servidor dos veces, con PORT = 5020 y PORT = 5021, y ejecuta el cliente en un tercer terminal. En una red real, escribe la dirección IP de cada PLC o equipo en DEVICES y usa el puerto 502.

Modbus RTU en el puerto RS-485 con asyncio

Para Modbus RTU, cambia solo el constructor del cliente o del servidor. El puerto RS-485 es de /dev/ttySC0 a /dev/ttySC3, según la versión del Raspberry PLC (lístalos con ls /dev/ttySC*).

import asyncio

from pymodbus.client import AsyncModbusSerialClient
from pymodbus.framer import FramerType


async def main():
    client = AsyncModbusSerialClient(
        "/dev/ttySC0",            # RS-485 port, check the name for your PLC version
        framer=FramerType.RTU,
        baudrate=9600, bytesize=8, parity="N", stopbits=1,
        timeout=1,
    )
    if not await client.connect():
        print("RS-485 port not available")
        return
    rr = await client.read_holding_registers(0, count=2, device_id=1)
    if rr.isError():
        print(f"no valid answer: {rr}")
    else:
        print(f"CPU {rr.registers[0] / 10:.1f} C, uptime {rr.registers[1]} s")
    client.close()


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

En el lado del servidor, StartAsyncSerialServer(device, port="/dev/ttySC0", framer=FramerType.RTU, baudrate=9600) sustituye a StartAsyncTcpServer. Una línea RS-485 es semidúplex y tiene un solo maestro: asyncio no hace la línea más rápida. Consulta los esclavos de una línea uno tras otro en una sola tarea, y usa tareas en paralelo para líneas distintas o para equipos Modbus TCP.

Los ejemplos antiguos de pymodbus fallan con pymodbus 3.16

La primera versión de este post usaba pymodbus 3.5. Con pymodbus 3.16.1 ese código se para en la primera línea que ha cambiado:

Código antiguoError con 3.16.1Código actual
from pymodbus.transaction import ModbusSocketFramerImportErrorfrom pymodbus.framer import FramerType, y después framer=FramerType.SOCKET o FramerType.RTU
ModbusSlaveContext(di=..., co=..., hr=..., ir=...)ImportErrorSimDevice(1, simdata=(...)) con SimData
client.write_coil(0, a, 0x01)TypeError: admite 3 argumentos posicionalesclient.write_coil(0, a, device_id=1)
client.read_coils(0, 1, slave=1)TypeError: argumento 'slave' inesperadoclient.read_coils(0, count=1, device_id=1)

La lista completa de cambios por versión está en la guía de Modbus para el Raspberry PLC y en la documentación de pymodbus.

Mira Modbus RTU y TCP en una red real. El simulador de red Modbus muestra un PLC que consulta cuatro esclavos RTU y sirve los valores por Modbus TCP, con tramas reales, timeouts y errores de orden de palabras. En el caso de la pasarela de datos a bordo, un Raspberry PLC lee por Modbus los datos de propulsión de un barco y los envía a tierra por MQTT.

Abrir el simulador ModbusLeer el caso de la pasarela de datos

Preguntas frecuentes sobre pymodbus en el Raspberry PLC

¿Qué versión de pymodbus debo usar?

Los ejemplos se han probado con pymodbus 3.16.1, la última versión en octubre de 2026. Fija la versión con pip install "pymodbus[serial]==3.16.1", para que una actualización no rompa tu programa.

¿Por qué falla pip install pymodbus en el Raspberry PLC?

Raspberry Pi OS basado en Bookworm o posterior muestra "externally-managed-environment" cuando usas pip fuera de un entorno virtual. Crea uno con python3 -m venv e instala pymodbus ahí.

¿Puede el Raspberry PLC ser cliente y servidor Modbus a la vez?

Sí. Con asyncio, ejecuta las tareas del servidor y del cliente en el mismo bucle de eventos, o ejecuta dos programas. En un puerto serie, solo un programa puede usar el puerto.

¿Por qué falla un servidor pymodbus en el puerto 502?

Los puertos por debajo de 1024 necesitan permisos de root en Linux. Usa un puerto como el 5020 para las pruebas, y da al servicio los permisos que necesita o redirige el puerto 502 en producción.

¿Es más rápido asyncio en una línea RS-485?

No. Una línea RS-485 lleva una petición cada vez. Asyncio ayuda cuando un programa habla con varias líneas o con varios equipos Modbus TCP.

Buscar en nuestro blog

Cómo usar PyModbus con un Raspberry Pi PLC: ejemplos asyncio
Boot & Work Corp. S.L, Ricard Franch Argullol 25 de septiembre de 2023
Compartir esta publicación
Etiquetas

¿Estás buscando tu Controlador Lógico Programable ideal?

Echa un vistazo a esta comparativa de producto de varios controladores industriales basados en Arduino.

Comparamos entradas, salidas, comunicaciones y otras especificaciones con las de los equipos de otras marcas destacadas.

Comparativa de PLCs