OPC-UA con Python: Tutorial Práctico para Automatización Industrial

Keywords secundarias: opc-ua python ejemplo, opcua asyncio tutorial, python plc comunicación, opcua servidor cliente python, opc ua industria 4.0 Fecha: 2026-09-01 Autor: Carlos — automatizatodo.com Categoría: Automatización Industrial / Python

---

Si trabajas en automatización industrial y todavía estás intercambiando datos entre PLCs y software de supervisión con protocolos propietarios, drivers de pago o archivos CSV manuales, tienes un problema que OPC-UA con Python resuelve en una tarde. Este tutorial de OPC-UA con Python para automatización industrial te lleva desde cero hasta tener un servidor y un cliente funcionales, con código que puedes copiar y adaptar a tu planta.

OPC-UA (Open Platform Communications – Unified Architecture) es el estándar abierto que la industria lleva años adoptando para comunicar máquinas, PLCs, SCADA y sistemas de gestión. Python, con la librería asyncua, te da acceso directo a ese protocolo sin necesidad de licencias ni software de terceros. La combinación es potente: lees datos de un Siemens S7-1500, un Allen-Bradley o cualquier PLC con servidor OPC-UA integrado usando unas pocas líneas de código.

Por qué OPC-UA es el estándar que necesitas conocer

Antes de meternos en código, conviene entender por qué OPC-UA se ha convertido en el idioma universal de la industria 4.0.

Los protocolos clásicos de comunicación industrial — Modbus, Profibus, DeviceNet — fueron diseñados para un mundo donde cada fabricante tenía su ecosistema cerrado. Funcionan, pero con limitaciones serias cuando quieres integrar datos de planta con sistemas IT (bases de datos, dashboards, IA, ERP).

OPC-UA resuelve esto con varias ventajas clave:

• Independiente de plataforma: Funciona en Windows, Linux, dispositivos embebidos y en la nube. No estás atado a un sistema operativo.
• Seguridad integrada: Autenticación, cifrado y firma de mensajes de serie. No es un parche añadido después, como ocurre con Modbus TCP.
• Modelo de información rico: No solo transporta valores numéricos — estructura los datos con tipos, relaciones y metadatos. Un nodo OPC-UA te dice qué es, qué unidades tiene y cuándo se actualizó.
• Escalable: Desde un sensor hasta una planta entera con miles de variables, sin cambiar de protocolo.
• Soportado por todos los grandes: Siemens, Rockwell, Beckhoff, B&R, Schneider, Phoenix Contact… todos sus PLCs modernos traen servidor OPC-UA integrado.

En la práctica, si tu PLC tiene firmware de los últimos 5-6 años, probablemente ya tiene un servidor OPC-UA que puedes activar desde la configuración del proyecto. Y si no lo tiene, existen gateways OPC-UA para prácticamente cualquier protocolo legacy.

Qué necesitas antes de empezar

Requisitos mínimos para seguir este tutorial:

• Python 3.10 o superior. La librería asyncua requiere esta versión como mínimo por el uso extensivo de asyncio.
• pip o uv para instalar paquetes.
• Un editor de código. VS Code, PyCharm o el que prefieras.
• Conocimientos básicos de Python. Variables, funciones, bucles. Si sabes qué es async/await, mejor, pero lo explicamos sobre la marcha.
• Opcional: un PLC con servidor OPC-UA. Si no tienes uno a mano, no pasa nada — vamos a crear nuestro propio servidor OPC-UA en Python para practicar.

Instalación de asyncua

Abre tu terminal y ejecuta:

# Crear un entorno virtual (recomendado)
python -m venv opcua-env
source opcua-env/bin/activate  # Linux/Mac
# opcua-env\Scripts\activate   # Windows

# Instalar la librería
pip install asyncua

asyncua es el sucesor moderno de python-opcua. Está basada en asyncio, soporta Python 3.10+ y tiene más del 95% de cobertura de tests. Es la librería que debes usar en 2026 — la versión antigua (python-opcua) ya no recibe actualizaciones.

Para verificar que todo está bien:

import asyncua
print(asyncua.__version__)

Si imprime un número de versión sin errores, estás listo.

Crear un servidor OPC-UA en Python paso a paso

Vamos a construir un servidor OPC-UA que simule los datos de una línea de producción: temperatura de un horno, velocidad de un motor y estado de una válvula. Esto es exactamente lo que harías si quisieras exponer datos de un proceso industrial a otros sistemas.

Código del servidor

import asyncio
import random
from asyncua import Server, ua

async def main():
    # 1. Crear el servidor
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://0.0.0.0:4840/automatizatodo/server/")
    server.set_server_name("Servidor OPC-UA - Línea de Producción")

    # 2. Configurar el namespace
    uri = "https://automatizatodo.com/opcua/tutorial"
    idx = await server.register_namespace(uri)

    # 3. Crear la estructura de nodos
    # Objeto principal: Línea de Producción
    linea = await server.nodes.objects.add_object(idx, "LineaProduccion")

    # Variables del proceso
    temperatura = await linea.add_variable(
        idx, "TemperaturaHorno", 25.0,
        varianttype=ua.VariantType.Double
    )
    velocidad_motor = await linea.add_variable(
        idx, "VelocidadMotor", 0.0,
        varianttype=ua.VariantType.Double
    )
    estado_valvula = await linea.add_variable(
        idx, "EstadoValvula", False,
        varianttype=ua.VariantType.Boolean
    )
    contador_piezas = await linea.add_variable(
        idx, "ContadorPiezas", 0,
        varianttype=ua.VariantType.Int32
    )

    # Hacer las variables escribibles (para que un cliente pueda modificarlas)
    await temperatura.set_writable()
    await velocidad_motor.set_writable()
    await estado_valvula.set_writable()

    print(f"Servidor OPC-UA iniciado en opc.tcp://0.0.0.0:4840")
    print(f"Namespace: {uri} (índice {idx})")

    # 4. Simular datos cambiantes
    async with server:
        piezas = 0
        while True:
            # Simular temperatura del horno (fluctúa entre 180-220°C)
            temp = 200 + random.uniform(-20, 20)
            await temperatura.write_value(temp)

            # Simular velocidad del motor (RPM)
            vel = 1450 + random.uniform(-50, 50)
            await velocidad_motor.write_value(vel)

            # La válvula se abre si la temperatura supera 210°C
            valvula_abierta = temp > 210
            await estado_valvula.write_value(valvula_abierta)

            # Contador de piezas incrementa
            piezas += 1
            await contador_piezas.write_value(piezas)

            print(f"T={temp:.1f}°C | Motor={vel:.0f}RPM | "
                  f"Válvula={'ABIERTA' if valvula_abierta else 'CERRADA'} | "
                  f"Piezas={piezas}")

            await asyncio.sleep(1)  # Actualizar cada segundo

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

Qué hace cada parte

1. Inicialización: Crea el servidor y lo configura para escuchar en el puerto 4840 (el puerto estándar OPC-UA).
2. Namespace: Registra un espacio de nombres propio. Esto evita conflictos si el servidor expone nodos de distintas fuentes.
3. Nodos: Crea un objeto LineaProduccion con cuatro variables: temperatura, velocidad, estado de válvula y contador de piezas. Cada variable tiene su tipo de dato definido explícitamente.
4. Simulación: En un bucle infinito, actualiza los valores cada segundo con datos que simulan un proceso real. La válvula reacciona a la temperatura — exactamente como lo haría una lógica de control simple.

Guarda este archivo como servidor_opcua.py y ejecútalo:

python servidor_opcua.py

Deberías ver los datos actualizándose en la terminal cada segundo. Tu servidor OPC-UA ya está funcionando.

Crear un cliente OPC-UA para leer datos del PLC

Ahora viene la parte que más interesa en automatización industrial: conectarte a un servidor OPC-UA (ya sea el que acabamos de crear o el de un PLC real) y leer datos.

Código del cliente básico

import asyncio
from asyncua import Client

async def main():
    url = "opc.tcp://localhost:4840/automatizatodo/server/"

    async with Client(url=url) as client:
        print(f"Conectado a {url}")

        # Obtener el namespace index
        uri = "https://automatizatodo.com/opcua/tutorial"
        idx = await client.get_namespace_index(uri)

        # Acceder a los nodos por su ruta (browse path)
        temperatura = await client.nodes.objects.get_child(
            [f"{idx}:LineaProduccion", f"{idx}:TemperaturaHorno"]
        )
        velocidad = await client.nodes.objects.get_child(
            [f"{idx}:LineaProduccion", f"{idx}:VelocidadMotor"]
        )
        estado_valvula = await client.nodes.objects.get_child(
            [f"{idx}:LineaProduccion", f"{idx}:EstadoValvula"]
        )
        contador = await client.nodes.objects.get_child(
            [f"{idx}:LineaProduccion", f"{idx}:ContadorPiezas"]
        )

        # Leer valores en bucle
        for i in range(10):
            temp = await temperatura.read_value()
            vel = await velocidad.read_value()
            valv = await estado_valvula.read_value()
            piezas = await contador.read_value()

            print(f"[Lectura {i+1}] Temp={temp:.1f}°C | "
                  f"Motor={vel:.0f}RPM | "
                  f"Válvula={'ABIERTA' if valv else 'CERRADA'} | "
                  f"Piezas={piezas}")

            await asyncio.sleep(2)

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

Ejecuta el cliente en otra terminal mientras el servidor sigue corriendo:

python cliente_opcua.py

Verás cómo lee los valores que el servidor está generando en tiempo real. Si en vez de nuestro servidor quisieras conectarte a un PLC Siemens, solo cambiarías la URL y la ruta de los nodos — el protocolo es el mismo.

Suscripciones: recibir datos en tiempo real sin hacer polling

Leer valores en bucle funciona, pero no es eficiente. En un entorno industrial real con cientos de variables, lo que quieres son suscripciones: el servidor te avisa cuando un valor cambia, sin que tengas que preguntar constantemente.

Código con suscripciones

import asyncio
from asyncua import Client

class ManejadorDatos:
    """Recibe notificaciones cuando cambian los valores suscritos."""

    def datachange_notification(self, node, val, data):
        nombre = node.nodeid.Identifier
        print(f"  ⚡ Cambio detectado → {nombre}: {val}")

async def main():
    url = "opc.tcp://localhost:4840/automatizatodo/server/"

    async with Client(url=url) as client:
        print(f"Conectado a {url}")
        print("Esperando cambios en los datos...\n")

        uri = "https://automatizatodo.com/opcua/tutorial"
        idx = await client.get_namespace_index(uri)

        # Obtener nodos
        temperatura = await client.nodes.objects.get_child(
            [f"{idx}:LineaProduccion", f"{idx}:TemperaturaHorno"]
        )
        estado_valvula = await client.nodes.objects.get_child(
            [f"{idx}:LineaProduccion", f"{idx}:EstadoValvula"]
        )

        # Crear suscripción con intervalo de 500ms
        manejador = ManejadorDatos()
        suscripcion = await client.create_subscription(500, manejador)

        # Suscribirse a los nodos que nos interesan
        await suscripcion.subscribe_data_change([temperatura, estado_valvula])

        # Mantener la conexión activa
        await asyncio.sleep(30)  # Escuchar durante 30 segundos

        # Limpiar
        await suscripcion.delete()

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

La diferencia es fundamental: con polling lees cada X segundos te haya cambiado el valor o no. Con suscripciones, solo recibes datos cuando hay un cambio real. En una planta con 2.000 variables, esto reduce el tráfico de red y la carga del servidor de forma drástica.

El parámetro 500 en create_subscription es el intervalo mínimo de publicación en milisegundos. Para datos críticos de control puedes bajarlo a 100ms. Para monitorización general, 1.000-5.000ms es suficiente.

Escribir valores: enviar comandos al PLC desde Python

OPC-UA no solo sirve para leer. También puedes escribir valores en los nodos — por ejemplo, cambiar un setpoint de temperatura o activar una válvula desde un script Python.

import asyncio
from asyncua import Client

async def main():
    url = "opc.tcp://localhost:4840/automatizatodo/server/"

    async with Client(url=url) as client:
        uri = "https://automatizatodo.com/opcua/tutorial"
        idx = await client.get_namespace_index(uri)

        velocidad = await client.nodes.objects.get_child(
            [f"{idx}:LineaProduccion", f"{idx}:VelocidadMotor"]
        )

        # Leer valor actual
        valor_actual = await velocidad.read_value()
        print(f"Velocidad actual: {valor_actual:.0f} RPM")

        # Escribir nuevo setpoint
        nuevo_setpoint = 1200.0
        await velocidad.write_value(nuevo_setpoint)
        print(f"Nuevo setpoint enviado: {nuevo_setpoint:.0f} RPM")

        # Verificar
        valor_nuevo = await velocidad.read_value()
        print(f"Velocidad después de escritura: {valor_nuevo:.0f} RPM")

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

Precaución importante: En un entorno real, escribir valores en un PLC desde un script externo requiere medidas de seguridad serias. Nunca expongas un servidor OPC-UA sin autenticación a la red. La escritura debe pasar por validaciones en el propio PLC (límites, permisos, interlocks). Python envía el valor, pero la lógica de seguridad siempre debe estar en el controlador.

Conectar con un PLC real: Siemens S7-1500 como ejemplo

Todo lo anterior funciona igual con un PLC real. Lo que cambia es la configuración del lado del PLC y la URL de conexión.

Configurar el servidor OPC-UA en TIA Portal

Para un S7-1500 con TIA Portal:

1. Activar el servidor OPC-UA: En las propiedades del PLC, ve a OPC UA > Server > Activar servidor OPC UA.
2. Configurar el puerto: Por defecto es 4840. Puedes cambiarlo si hay conflictos.
3. Seleccionar variables a exponer: En TIA Portal v17+, puedes marcar variables individuales del DB como accesibles vía OPC-UA. En versiones anteriores se expone todo el DB con acceso externo habilitado.
4. Seguridad: Configura al menos autenticación por usuario/contraseña. Para producción, usa certificados X.509.
5. Compilar y descargar al PLC.

Conectar desde Python

import asyncio
from asyncua import Client

async def main():
    # URL del PLC (ajusta la IP a tu red)
    url = "opc.tcp://192.168.0.10:4840"

    async with Client(url=url) as client:
        # Si el PLC requiere autenticación:
        # client.set_user("usuario_opcua")
        # client.set_password("tu_contraseña_segura")

        # Explorar el espacio de direcciones
        root = client.nodes.root
        objetos = await root.get_children()

        print("Nodos raíz del PLC:")
        for obj in objetos:
            nombre = await obj.read_browse_name()
            print(f"  - {nombre}")

        # Leer una variable específica por NodeId
        # El NodeId depende de cómo hayas configurado el PLC
        # Ejemplo: ns=3;s="DB_Proceso"."Temperatura"
        nodo = client.get_node('ns=3;s="DB_Proceso"."Temperatura"')
        valor = await nodo.read_value()
        print(f"\nTemperatura del proceso: {valor}°C")

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

El formato del NodeId varía según el fabricante. En Siemens suele ser ns=3;s="NombreDB"."NombreVariable". En Beckhoff TwinCAT es diferente. Puedes usar herramientas como UaExpert (gratuito) para explorar el espacio de direcciones del PLC y encontrar los NodeIds correctos.

Caso práctico: registrar datos de producción en PostgreSQL

Vamos a juntar todo en un ejemplo del mundo real. Un script que se conecta al servidor OPC-UA, se suscribe a las variables del proceso y guarda cada lectura en una base de datos PostgreSQL. Esto es la base de cualquier sistema de historización industrial.

import asyncio
import asyncpg
from asyncua import Client
from datetime import datetime

DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "database": "produccion",
    "user": "admin",
    "password": "tu_contraseña"
}

class RegistradorProduccion:
    def __init__(self, pool):
        self.pool = pool

    def datachange_notification(self, node, val, data):
        nombre = str(node.nodeid.Identifier)
        timestamp = datetime.now()
        asyncio.create_task(
            self._guardar(nombre, val, timestamp)
        )

    async def _guardar(self, nombre, valor, timestamp):
        async with self.pool.acquire() as conn:
            await conn.execute(
                """INSERT INTO datos_proceso
                   (variable, valor, timestamp)
                   VALUES ($1, $2, $3)""",
                nombre, float(valor), timestamp
            )

async def main():
    # Conectar a la base de datos
    pool = await asyncpg.create_pool(**DB_CONFIG)

    # Crear tabla si no existe
    async with pool.acquire() as conn:
        await conn.execute("""
            CREATE TABLE IF NOT EXISTS datos_proceso (
                id SERIAL PRIMARY KEY,
                variable VARCHAR(100),
                valor DOUBLE PRECISION,
                timestamp TIMESTAMPTZ DEFAULT NOW()
            )
        """)

    url = "opc.tcp://localhost:4840/automatizatodo/server/"

    async with Client(url=url) as client:
        uri = "https://automatizatodo.com/opcua/tutorial"
        idx = await client.get_namespace_index(uri)

        # Suscribirse a todas las variables
        nodos = []
        for var in ["TemperaturaHorno", "VelocidadMotor",
                     "ContadorPiezas"]:
            nodo = await client.nodes.objects.get_child(
                [f"{idx}:LineaProduccion", f"{idx}:{var}"]
            )
            nodos.append(nodo)

        registrador = RegistradorProduccion(pool)
        sub = await client.create_subscription(1000, registrador)
        await sub.subscribe_data_change(nodos)

        print("Registrando datos de producción en PostgreSQL...")
        print("Presiona Ctrl+C para detener.\n")

        try:
            while True:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            print("\nDetenido. Cerrando conexiones...")
            await sub.delete()

    await pool.close()

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

Este patrón — suscripción OPC-UA + registro en base de datos — es exactamente lo que hacen los historians industriales comerciales como Wonderware, FactoryTalk Historian o AVEVA. La diferencia es que tú lo construyes con herramientas open-source, sin licencias de miles de euros al año.

Si ya tienes un dashboard con Grafana conectado a PostgreSQL, puedes visualizar estos datos en tiempo real con paneles que se actualizan cada segundo.

Seguridad en OPC-UA: lo que no puedes ignorar

Exponer datos industriales a la red sin seguridad es una invitación al desastre. OPC-UA tiene mecanismos de seguridad robustos — úsalos.

Autenticación por usuario y contraseña

async with Client(url=url) as client:
    client.set_user("operador")
    client.set_password("contraseña_segura_2026")
    # La conexión se autentica automáticamente

Autenticación por certificado X.509

Para producción, los certificados son la opción recomendada:

from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256

async with Client(url=url) as client:
    await client.set_security(
        SecurityPolicyBasic256Sha256,
        certificate_path="cliente_cert.pem",
        private_key_path="cliente_key.pem",
        server_certificate_path="servidor_cert.pem"
    )

Buenas prácticas de seguridad OPC-UA

• Nunca expongas el puerto 4840 a Internet. Usa VPN o redes segmentadas.
• Segmenta la red OT. El servidor OPC-UA del PLC debe estar en una VLAN dedicada, accesible solo desde la DMZ industrial.
• Usa cifrado. SecurityPolicyBasic256Sha256 como mínimo. Evita None en producción.
• Rota certificados. Establece una política de renovación (anual como mínimo).
• Monitoriza conexiones. Registra quién se conecta, cuándo y qué lee. Si usas una estrategia de ciberseguridad OT, OPC-UA encaja perfectamente en tu arquitectura.

Errores comunes y cómo solucionarlos

Después de implementar OPC-UA en varias plantas, estos son los problemas que más se repiten:

Error: BadTimeout o desconexiones frecuentes El timeout de sesión por defecto es demasiado corto. Ajústalo:

client.session_timeout = 60000  # 60 segundos

Error: BadNodeIdUnknown al acceder a variables del PLC El NodeId que estás usando no existe. Usa get_children() para explorar la estructura real del servidor antes de intentar acceder a un nodo específico.

Error: BadSecurityPolicyRejected El servidor requiere un nivel de seguridad que tu cliente no está ofreciendo. Configura la política de seguridad que coincida con la del servidor.

Rendimiento lento con muchas variables Lee múltiples nodos en una sola llamada en vez de uno por uno:

valores = await client.read_values([nodo1, nodo2, nodo3])

El PLC no aparece en la red Verifica que el firewall del PC no bloquea el puerto 4840. En Windows, añade una excepción. En Linux:

sudo ufw allow 4840/tcp

OPC-UA vs Modbus vs MQTT: cuándo usar cada uno

Una pregunta frecuente es cuándo elegir OPC-UA frente a otros protocolos. La respuesta corta:

Criterio	OPC-UA	Modbus TCP	MQTT
Modelo de datos	Rico (tipos, estructura, metadatos)	Plano (registros numéricos)	Flexible (payload libre)
Seguridad	Integrada (cifrado + autenticación)	Ninguna nativa	TLS opcional
Descubrimiento	Sí (browse del address space)	No	No
Complejidad	Media-alta	Baja	Baja
Uso típico	Comunicación PLC↔MES/SCADA	Dispositivos simples, sensores	IoT, telemetría, cloud
Latencia	Baja (ms)	Muy baja (ms)	Variable (depende del broker)

Usa OPC-UA cuando: necesitas integrar PLCs modernos con sistemas IT, quieres seguridad de serie, o trabajas con estándares de industria 4.0.

Usa Modbus cuando: los dispositivos son simples, el presupuesto es limitado, o ya tienes infraestructura Modbus funcionando. Tenemos un tutorial de Modbus TCP con Python y n8n si es tu caso.

Usa MQTT cuando: el escenario es IoT con muchos dispositivos ligeros enviando datos a la nube, o necesitas un patrón pub/sub desacoplado. Puedes ver ejemplos prácticos de MQTT industrial en otro de nuestros artículos.

Próximos pasos y recursos

Con lo que has aprendido en este tutorial ya puedes:

• Montar un servidor OPC-UA para simular o exponer datos de proceso.
• Conectarte a PLCs reales y leer variables en tiempo real.
• Usar suscripciones para recibir cambios sin polling.
• Registrar datos en bases de datos para historización y análisis.
• Aplicar seguridad básica con autenticación y cifrado.

Para ir más allá:

• Documentación oficial de asyncua: opcua-asyncio en GitHub — referencia completa con ejemplos.
• Especificación OPC-UA: OPC Foundation — estándares y documentación oficial del protocolo.
• UaExpert: Unified Automation — cliente OPC-UA gratuito para explorar servidores, imprescindible para debugging.

---

¿Necesitas integrar OPC-UA con tus sistemas de planta? En automatizatodo.com ayudamos a empresas industriales a conectar sus PLCs y SCADA con bases de datos, dashboards y herramientas de IA — sin depender de software propietario caro. Contacta con nosotros y te ayudamos a montar tu arquitectura OPC-UA.