Python + OPC-UA: Leer Datos de PLC Siemens S7-1500
Tutorial para leer datos de un PLC Siemens S7-1500 con Python y OPC-UA. Configuración del servidor, código funcional y buenas prácticas.
Estás en una planta con 20 PLCs Siemens S7-1500 controlando líneas de envasado. El SCADA muestra los datos en tiempo real, pero tú necesitas algo más: alimentar un dashboard de producción con Grafana, guardar históricos en PostgreSQL, o entrenar un modelo de mantenimiento predictivo con datos reales. El SCADA no te lo pone fácil. Python sí.
OPC-UA es el estándar de comunicación industrial que Siemens ha integrado de serie en el S7-1500. No necesitas hardware extra, no necesitas licencias adicionales, no necesitas un servidor OPC intermedio. Activas el servidor en TIA Portal, escribes unas líneas de Python y empiezas a leer variables del PLC como si fueran campos de un JSON.
Vamos a hacerlo paso a paso, desde la configuración en TIA Portal hasta un script de producción con gestión de errores y reconexión automática.
Qué es OPC-UA y por qué el S7-1500 lo trae de serie
OPC-UA (Open Platform Communications - Unified Architecture) es el sucesor de OPC clásico (el que dependía de COM/DCOM de Windows y solo funcionaba en la misma red). OPC-UA es multiplataforma, seguro y funciona sobre TCP/IP estándar.
Lo que lo hace relevante para nosotros:
- El S7-1500 tiene un servidor OPC-UA integrado. No necesitas un PC intermedio corriendo KEPServerEX o similar.
- Funciona sobre TCP, normalmente en el puerto 4840. Cualquier máquina que tenga acceso a la red del PLC puede conectarse.
- Soporte de seguridad: autenticación por usuario/contraseña, cifrado con certificados X.509, y control de acceso granular por nodo.
- Suscripciones: en vez de hacer polling (preguntar cada X ms), puedes suscribirte a cambios y recibir notificaciones solo cuando un valor cambia.
La alternativa más común para comunicarse con un S7-1500 es el protocolo S7 (el que usan librerías como python-snap7). Funciona, pero tiene limitaciones: no tiene estándar de seguridad, Siemens no lo documenta oficialmente y puede romperse entre versiones de firmware. OPC-UA es la vía oficial y recomendada por Siemens para integración con sistemas externos.
Si vienes de trabajar con Modbus TCP, OPC-UA te parecerá mucho más cómodo: en vez de direcciones de registro (40001, 40002…), trabajas con nodos con nombre legible ("Temperature_Reactor_1", "Pressure_Tank_A").
Configurar el servidor OPC-UA en TIA Portal
Esto es lo primero. Sin el servidor activo en el PLC, no hay nada que leer.
Activar el servidor
- Abre tu proyecto en TIA Portal V16 o superior (V17/V18 recomendado).
- En la vista de dispositivos, haz doble clic en tu CPU S7-1500.
- Ve a Propiedades → General → OPC UA → Servidor.
- Marca “Activar servidor OPC UA”.
- El puerto por defecto es 4840. Déjalo a menos que tengas conflicto con otro servicio.
Configurar la seguridad
TIA Portal te da tres opciones de seguridad:
- Sin seguridad: cualquiera que tenga acceso a la red puede conectarse. Solo para desarrollo o redes completamente aisladas.
- Sign: los mensajes van firmados pero no cifrados. Protege contra manipulación pero no contra sniffing.
- SignAndEncrypt: firma + cifrado. Para producción siempre debería ser esta.
Para este tutorial usaremos primero conexión sin seguridad (para verificar que todo funciona) y después la aseguraremos.
En la sección de seguridad:
- Marca “Permitir conexiones sin certificado” para empezar (lo desactivaremos después).
- Si usas usuario/contraseña, crea un usuario en “Gestión de usuarios Runtime” con el rol adecuado.
Exponer variables al servidor OPC-UA
Por defecto, el servidor OPC-UA del S7-1500 no expone ninguna variable. Tienes que marcar explícitamente qué datos quieres publicar. Esto es una medida de seguridad, y es buena idea: no querrás que un script Python pueda ver (o peor, escribir) todas las variables del programa.
Para exponer un bloque de datos (DB):
- Abre el DB que quieres publicar.
- En las propiedades del DB, ve a “Atributos”.
- Marca “Acceso al servidor OPC UA: Accessible from OPC UA”.
- Opcionalmente, marca variables individuales como de solo lectura.
Para exponer variables de un bloque de función o del programa principal, haz lo mismo a nivel de variable individual.
Importante: desactiva la opción “Acceso optimizado al bloque” en los DBs que quieras leer por OPC-UA. Si está activado, las direcciones de offset no son fijas y OPC-UA no puede mapearlas correctamente. Ve a Propiedades del DB → Atributos → desmarca “Acceso optimizado al bloque”.
Compilar y descargar
Compila el proyecto y descárgalo al PLC. El servidor OPC-UA se activará automáticamente al arrancar. Puedes verificar que está corriendo desde el panel web del PLC (si lo tiene habilitado) o simplemente intentando conectar desde Python.
Preparar el entorno Python
Instalar opcua-asyncio
La librería estándar para OPC-UA en Python es asyncua (antes conocida como opcua-asyncio). Es asíncrona, bien mantenida y compatible con Python 3.8+.
# Crear entorno virtual (siempre)
python3 -m venv .venv
source .venv/bin/activate
# Instalar la librería
pip install asyncua
Si prefieres una API síncrona (más sencilla para scripts simples), también existe opcua (FreeOpcUa), pero está menos mantenida. Mi recomendación: usa asyncua desde el principio, incluso si no necesitas async. Es la que tiene futuro.
Verificar la conexión
Antes de escribir código, verifica que puedes alcanzar el puerto 4840 del PLC:
# Comprobar que el puerto responde
nc -zv 192.168.1.10 4840
# Si usas Wireshark, filtra por tcp.port == 4840
# para ver el handshake OPC-UA
Si el puerto no responde, revisa:
- ¿El servidor OPC-UA está activado y descargado al PLC?
- ¿Hay un firewall bloqueando el puerto?
- ¿Estás en la misma VLAN que el PLC?
Lectura básica de variables
Conectar y explorar el espacio de nombres
Lo primero que quieres hacer al conectarte a un servidor OPC-UA nuevo es explorar qué nodos están disponibles. El espacio de nombres (address space) es como un árbol de carpetas con las variables dentro.
import asyncio
from asyncua import Client
async def explorar_servidor():
url = "opc.tcp://192.168.1.10:4840"
async with Client(url=url) as client:
# El nodo raíz del servidor
root = client.nodes.root
# Obtener el nodo Objects (donde están los datos)
objects = client.nodes.objects
# Listar los hijos del nodo Objects
children = await objects.get_children()
for child in children:
name = await child.read_browse_name()
print(f"Nodo: {name.Name} | NodeId: {child.nodeid}")
# Un nivel más de profundidad
sub_children = await child.get_children()
for sub in sub_children:
sub_name = await sub.read_browse_name()
print(f" └─ {sub_name.Name} | {sub.nodeid}")
asyncio.run(explorar_servidor())
La salida te mostrará algo como:
Nodo: DeviceSet | NodeId: ns=0;i=85
Nodo: Server | NodeId: ns=0;i=2253
Nodo: "DB_Produccion" | NodeId: ns=3;s="DB_Produccion"
└─ Temperatura_Reactor | NodeId: ns=3;s="DB_Produccion"."Temperatura_Reactor"
└─ Presion_Tanque_A | NodeId: ns=3;s="DB_Produccion"."Presion_Tanque_A"
└─ Estado_Linea | NodeId: ns=3;s="DB_Produccion"."Estado_Linea"
└─ Contador_Piezas | NodeId: ns=3;s="DB_Produccion"."Contador_Piezas"
Fíjate en el formato del NodeId para Siemens: ns=3;s="NombreDB"."NombreVariable". Este es el identificador que usarás para leer y escribir.
Leer variables individuales
import asyncio
from asyncua import Client
async def leer_variables():
url = "opc.tcp://192.168.1.10:4840"
async with Client(url=url) as client:
# Acceder a un nodo por su NodeId
temp_node = client.get_node('ns=3;s="DB_Produccion"."Temperatura_Reactor"')
presion_node = client.get_node('ns=3;s="DB_Produccion"."Presion_Tanque_A"')
estado_node = client.get_node('ns=3;s="DB_Produccion"."Estado_Linea"')
contador_node = client.get_node('ns=3;s="DB_Produccion"."Contador_Piezas"')
# Leer valores
temperatura = await temp_node.read_value()
presion = await presion_node.read_value()
estado = await estado_node.read_value()
contador = await contador_node.read_value()
print(f"Temperatura reactor: {temperatura}°C")
print(f"Presión tanque A: {presion} bar")
print(f"Estado línea: {'Marcha' if estado else 'Parada'}")
print(f"Piezas producidas: {contador}")
# También puedes leer metadatos
data_type = await temp_node.read_data_type_as_variant_type()
print(f"Tipo de dato temperatura: {data_type}")
asyncio.run(leer_variables())
Leer múltiples variables de golpe
Si necesitas leer 50 variables, hacer 50 llamadas individuales es ineficiente. OPC-UA soporta lectura por lotes:
async def leer_lote():
url = "opc.tcp://192.168.1.10:4840"
async with Client(url=url) as client:
# Definir todos los nodos a leer
nodos = [
client.get_node('ns=3;s="DB_Produccion"."Temperatura_Reactor"'),
client.get_node('ns=3;s="DB_Produccion"."Presion_Tanque_A"'),
client.get_node('ns=3;s="DB_Produccion"."Estado_Linea"'),
client.get_node('ns=3;s="DB_Produccion"."Contador_Piezas"'),
client.get_node('ns=3;s="DB_Produccion"."Velocidad_Motor_1"'),
client.get_node('ns=3;s="DB_Produccion"."Caudal_Bomba_A"'),
]
# Lectura por lotes: una sola petición al servidor
valores = await client.read_values(nodos)
for nodo, valor in zip(nodos, valores):
name = await nodo.read_browse_name()
print(f"{name.Name}: {valor}")
asyncio.run(leer_lote())
La lectura por lotes es significativamente más rápida. En una red típica de planta (100 Mbps), leer 50 variables individuales tarda ~500ms. Por lotes, ~10ms.
Suscripciones: recibir datos en tiempo real
Polling (leer cada X segundos) funciona, pero tiene un problema: o lees demasiado rápido y sobrecargas el PLC, o lees demasiado lento y te pierdes cambios. Las suscripciones OPC-UA resuelven esto: te avisas solo cuando un valor cambia.
import asyncio
from asyncua import Client
class ManejadorCambios:
"""Handler que se ejecuta cuando cambia un valor."""
def datachange_notification(self, node, val, data):
"""Callback automático cuando un valor suscrito cambia."""
print(f"Cambio detectado - Nodo: {node} | Nuevo valor: {val}")
print(f" Timestamp servidor: {data.monitored_item.Value.ServerTimestamp}")
print(f" Status: {data.monitored_item.Value.StatusCode}")
async def suscribir_variables():
url = "opc.tcp://192.168.1.10:4840"
async with Client(url=url) as client:
handler = ManejadorCambios()
# Crear suscripción con intervalo de 500ms
subscription = await client.create_subscription(
period=500, # ms entre comprobaciones
handler=handler
)
# Suscribirse a nodos específicos
nodos = [
client.get_node('ns=3;s="DB_Produccion"."Temperatura_Reactor"'),
client.get_node('ns=3;s="DB_Produccion"."Presion_Tanque_A"'),
client.get_node('ns=3;s="DB_Produccion"."Estado_Linea"'),
]
handles = []
for nodo in nodos:
handle = await subscription.subscribe_data_change(nodo)
handles.append(handle)
print("Suscripción activa. Esperando cambios...")
print("(Ctrl+C para salir)")
# Mantener el script corriendo
try:
while True:
await asyncio.sleep(1)
except KeyboardInterrupt:
print("Cerrando suscripción...")
for handle in handles:
await subscription.unsubscribe(handle)
await subscription.delete()
asyncio.run(suscribir_variables())
El parámetro period=500 no significa que el PLC envíe datos cada 500ms. Significa que el servidor comprueba cambios cada 500ms y solo envía los que han cambiado. Si la temperatura no varía, no se transmite nada. Esto es mucho más eficiente que polling.
Deadband: filtrar ruido
Las señales analógicas (temperatura, presión, caudal) fluctúan constantemente por ruido de la señal. Sin filtro, recibirías cientos de cambios por segundo de un valor que básicamente no ha cambiado.
OPC-UA soporta deadband (banda muerta): solo notifica si el cambio supera un umbral.
from asyncua import ua
# Crear item monitorizado con deadband absoluto de 0.5
monitoring_params = ua.MonitoringParameters()
monitoring_params.SamplingInterval = 500
monitoring_params.QueueSize = 10
# Deadband: ignorar cambios menores a 0.5 unidades
filter_obj = ua.DataChangeFilter()
filter_obj.Trigger = ua.DataChangeTrigger.StatusValue
filter_obj.DeadbandType = ua.DeadbandType.Absolute
filter_obj.DeadbandValue = 0.5 # Solo notificar si cambia más de 0.5°C
monitoring_params.Filter = ua.ExtensionObject(Body=filter_obj)
handle = await subscription.subscribe_data_change(
temp_node,
monitoring_params=monitoring_params
)
Con un deadband de 0.5 en una temperatura que oscila entre 72.1 y 72.4°C, no recibirás ninguna notificación. Si sube a 72.8°C, ahí sí. Esto reduce dramáticamente el tráfico de red y la carga del script.
Script de producción con reconexión automática
Los scripts de desarrollo están bien para probar, pero en producción necesitas algo robusto: reconexión automática cuando la red falla o el PLC se reinicia, logging, y persistencia de datos.
import asyncio
import logging
from datetime import datetime
from asyncua import Client
from asyncua.ua.uaerrors import UaError
# Configurar logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler('opcua_reader.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# Configuración
PLC_URL = "opc.tcp://192.168.1.10:4840"
RECONNECT_DELAY = 5 # Segundos entre intentos de reconexión
MAX_RECONNECT_ATTEMPTS = 0 # 0 = infinito
VARIABLES = {
'temperatura': 'ns=3;s="DB_Produccion"."Temperatura_Reactor"',
'presion': 'ns=3;s="DB_Produccion"."Presion_Tanque_A"',
'estado': 'ns=3;s="DB_Produccion"."Estado_Linea"',
'contador': 'ns=3;s="DB_Produccion"."Contador_Piezas"',
}
class PLCReader:
"""Lector OPC-UA con reconexión automática."""
def __init__(self, url: str, variables: dict):
self.url = url
self.variables = variables
self.client = None
self.connected = False
self.reconnect_count = 0
async def connect(self):
"""Conectar al servidor OPC-UA."""
self.client = Client(url=self.url)
self.client.session_timeout = 30000 # 30 segundos
try:
await self.client.connect()
self.connected = True
self.reconnect_count = 0
logger.info(f"Conectado a {self.url}")
except Exception as e:
self.connected = False
raise ConnectionError(f"No se pudo conectar: {e}")
async def disconnect(self):
"""Desconectar limpiamente."""
if self.client and self.connected:
try:
await self.client.disconnect()
except Exception:
pass
self.connected = False
logger.info("Desconectado del servidor OPC-UA")
async def read_all(self) -> dict:
"""Leer todas las variables configuradas."""
if not self.connected:
raise ConnectionError("No hay conexión activa")
nodos = {
name: self.client.get_node(node_id)
for name, node_id in self.variables.items()
}
resultado = {}
try:
node_list = list(nodos.values())
valores = await self.client.read_values(node_list)
for (name, _), valor in zip(nodos.items(), valores):
resultado[name] = valor
resultado['timestamp'] = datetime.now().isoformat()
return resultado
except UaError as e:
self.connected = False
raise ConnectionError(f"Error de lectura: {e}")
async def run_loop(self, interval: float = 1.0, callback=None):
"""Bucle principal con reconexión automática."""
while True:
try:
if not self.connected:
logger.info("Intentando reconexión...")
await self.connect()
datos = await self.read_all()
if callback:
await callback(datos)
else:
logger.info(f"Datos: {datos}")
await asyncio.sleep(interval)
except ConnectionError as e:
self.reconnect_count += 1
logger.warning(
f"Conexión perdida (intento {self.reconnect_count}): {e}"
)
await self.disconnect()
if (MAX_RECONNECT_ATTEMPTS > 0
and self.reconnect_count >= MAX_RECONNECT_ATTEMPTS):
logger.error("Máximo de reintentos alcanzado. Saliendo.")
break
delay = min(RECONNECT_DELAY * self.reconnect_count, 60)
logger.info(f"Reintentando en {delay}s...")
await asyncio.sleep(delay)
except KeyboardInterrupt:
logger.info("Interrupción manual. Cerrando...")
break
except Exception as e:
logger.error(f"Error inesperado: {e}", exc_info=True)
await asyncio.sleep(RECONNECT_DELAY)
async def guardar_en_csv(datos: dict):
"""Callback: guardar datos en CSV."""
import csv
import os
archivo = f"datos_plc_{datetime.now().strftime('%Y-%m-%d')}.csv"
existe = os.path.exists(archivo)
with open(archivo, 'a', newline='') as f:
writer = csv.DictWriter(f, fieldnames=datos.keys())
if not existe:
writer.writeheader()
writer.writerow(datos)
async def main():
reader = PLCReader(PLC_URL, VARIABLES)
try:
await reader.run_loop(
interval=2.0,
callback=guardar_en_csv
)
finally:
await reader.disconnect()
if __name__ == "__main__":
asyncio.run(main())
Este script hace varias cosas bien:
- Reconexión con backoff exponencial: si la conexión falla, espera cada vez más (5s, 10s, 15s… hasta 60s máximo) para no saturar la red.
- Logging a fichero y consola: puedes revisar qué pasó en la noche sin estar delante.
- Callback flexible: el
guardar_en_csves un ejemplo. Puedes sustituirlo por una función que escriba en PostgreSQL, envíe a un broker MQTT, o llame a una API REST. - Separación de configuración: las variables y la URL están arriba, fáciles de modificar sin tocar la lógica.
Conectar con base de datos
Leer datos está bien, pero el valor real aparece cuando los persistes. Si ya tienes una base de datos de producción, puedes alimentarla directamente desde este script.
import asyncpg
async def guardar_en_postgres(datos: dict):
"""Guardar lectura en PostgreSQL."""
conn = await asyncpg.connect(
host='192.168.1.50',
database='produccion',
user='plc_writer',
password='tu_password'
)
await conn.execute('''
INSERT INTO lecturas_plc
(timestamp, temperatura, presion, estado, contador)
VALUES ($1, $2, $3, $4, $5)
''',
datetime.now(),
datos['temperatura'],
datos['presion'],
datos['estado'],
datos['contador']
)
await conn.close()
Para un sistema más robusto, usa un pool de conexiones en vez de abrir y cerrar por cada lectura. La guía de conectar un PLC a una base de datos SQL cubre esto en detalle.
Si además quieres visualizar los datos en tiempo real, puedes conectar Grafana a esa misma base de datos y tener un dashboard de producción funcional en menos de una hora.
Seguridad: certificados y autenticación
El script que hemos montado funciona sin seguridad. Para desarrollo está bien, pero en producción necesitas:
Autenticación por usuario y contraseña
async def conectar_con_auth():
client = Client(url="opc.tcp://192.168.1.10:4840")
client.set_user("operador_python")
client.set_password("contraseña_segura")
await client.connect()
# ... leer datos
await client.disconnect()
El usuario y contraseña se configuran en TIA Portal, en la sección de gestión de usuarios del servidor OPC-UA.
Certificados X.509
Para la máxima seguridad, usa certificados:
from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256
async def conectar_con_certificado():
client = Client(url="opc.tcp://192.168.1.10:4840")
await client.set_security(
SecurityPolicyBasic256Sha256,
certificate="./certs/client_cert.pem",
private_key="./certs/client_key.pem",
server_certificate="./certs/server_cert.pem"
)
await client.connect()
# ... leer datos
await client.disconnect()
Generar los certificados:
# Generar clave privada y certificado autofirmado
openssl req -x509 -newkey rsa:2048 -keyout client_key.pem \
-out client_cert.pem -days 365 -nodes \
-subj "/CN=PythonOPCUA/O=MiEmpresa"
Después, importa client_cert.pem en TIA Portal como certificado de confianza del servidor OPC-UA. Esto es un proceso manual en TIA Portal, pero solo lo haces una vez.
Errores comunes y cómo solucionarlos
”BadNodeIdUnknown”
El NodeId que estás usando no existe en el servidor. Causas habituales:
- El DB no tiene habilitado el acceso OPC-UA.
- El nombre de la variable tiene un typo (cuidado con mayúsculas/minúsculas).
- Tienes “Acceso optimizado al bloque” activado (desactívalo).
Solución: usa el script de exploración para listar los nodos disponibles y copia el NodeId exacto.
”BadSessionClosed” o desconexiones frecuentes
El timeout de sesión del servidor es más corto que tu intervalo de lectura. Si lees cada 30 segundos pero el timeout es 10 segundos, la sesión muere entre lecturas.
Solución: ajusta client.session_timeout a un valor mayor que tu intervalo de lectura × 3.
”BadTooManySessions”
El S7-1500 tiene un límite de sesiones OPC-UA simultáneas (normalmente 8-15 según el modelo de CPU). Si tienes múltiples clientes conectados, puedes alcanzar el límite.
Solución: cierra siempre las conexiones cuando termines. Usa async with Client(url=url) as client: para que el cierre sea automático. Y revisa que no tengas scripts zombis corriendo.
Rendimiento: lecturas lentas
Si las lecturas tardan más de lo esperado:
- Usa lectura por lotes en vez de individual.
- Reduce el número de variables suscritas si usas suscripciones.
- Comprueba la carga de la CPU del PLC (un programa OB1 muy pesado deja menos recursos para OPC-UA).
- Verifica la red: un cable Ethernet suelto o un switch saturado pueden añadir latencia.
Integrar con n8n para automatización
Si ya tienes n8n conectado con tu PLC vía OPC-UA, puedes usar Python como puente para escenarios que n8n no cubre nativamente. Por ejemplo:
- Python lee datos del PLC cada segundo y los guarda en PostgreSQL.
- n8n consulta PostgreSQL periódicamente y genera alertas, informes o acciones.
- Si detecta una anomalía, n8n dispara un workflow que envía un SMS, crea un ticket o notifica por Telegram.
Otra opción: tu script Python expone los datos via API REST (con FastAPI o Flask) y n8n los consume con un nodo HTTP Request. Así desacoplas la adquisición de datos de la lógica de negocio.
Para una introducción más general a OPC-UA con Python que cubra otros PLCs y simuladores, la guía de OPC-UA con Python complementa bien este tutorial específico para S7-1500.
Lo esencial
Leer datos de un S7-1500 con Python y OPC-UA no requiere hardware adicional, ni licencias, ni software intermedio. Activas el servidor en TIA Portal, instalas asyncua, y empiezas a trabajar.
Los pasos clave:
- Activar OPC-UA en TIA Portal y exponer las variables que necesitas.
- Desactivar “Acceso optimizado al bloque” en los DBs.
- Usar
asyncuacon lectura por lotes para rendimiento. - Implementar reconexión automática para scripts de larga duración.
- Activar seguridad (mínimo usuario/contraseña) antes de ir a producción.
El script de producción que hemos montado es un punto de partida sólido. Modifica el callback para adaptarlo a tu infraestructura: CSV para pruebas rápidas, PostgreSQL para producción, MQTT para integración con otros sistemas.
Una vez que tienes datos del PLC en Python, las posibilidades se multiplican: análisis de calidad, mantenimiento predictivo, dashboards en tiempo real, integración con ERPs. El PLC deja de ser una caja negra y se convierte en una fuente de datos accesible para toda la organización.
Sigue leyendo
Dashboard Industrial con Python y Flask (2026)
Construye un dashboard industrial con Python y Flask. Visualiza datos de producción en tiempo real conectando sensores, PLCs y bases de datos.
Leer artículoOPC-UA con Python: Tutorial de Automatización
Conecta PLCs y SCADA con Python usando OPC-UA. Tutorial con código real, servidor y cliente funcionales para automatización industrial.
Leer artículo