Ingesta de Datos en Tiempo Real con Python, SQL e Ignition SCADA

Matriz de Decisión: Patrones de Ingesta y Consulta en SCADA

Al diseñar flujos de telemetría entre sistemas de planta y plataformas SCADA empresariales, la elección del mecanismo de lectura/escritura impacta directamente en la latencia, concurrencia y saturación del Gateway.

Antipatrones Críticos en Arquitecturas OT/IT

Implementación Práctica del Pipeline

1. Infraestructura con Docker Compose

Definición del entorno con base de datos PostgreSQL e Ignition Gateway exponiendo el puerto web 8088 y el puerto de base de datos 5432:

services:
  scada_db:
    image: postgres:15
    container_name: scada_postgres_db
    environment:
      POSTGRES_USER: scada_user
      POSTGRES_PASSWORD: scada_password
      POSTGRES_DB: scada_database
    ports:
      - "5432:5432"

  ignition:
    image: inductiveautomation/ignition:latest
    container_name: ignition_scada_demo
    ports:
      - "8088:8088"
    environment:
      IGNITION_EDITION: standard
      ACCEPT_IGNITION_EULA: "Y"
    depends_on:
      - scada_db

2. Productor de Telemetría en Tiempo Real (Python)

Script que simula mediciones periódicas (temperatura y presión) de motores industriales y las inserta con marca de tiempo precisa en PostgreSQL:

import psycopg2
import time
import random
from datetime import datetime

# Configuración de conexión relacional
conn = psycopg2.connect(
    dbname="scada_database",
    user="scada_user",
    password="scada_password",
    host="localhost",
    port="5432"
)
cursor = conn.cursor()

# Creación de tabla para ingesta de sensores
cursor.execute("""
CREATE TABLE IF NOT EXISTS sensor_data (
    id SERIAL PRIMARY KEY,
    machine_name VARCHAR(50),
    temperature FLOAT,
    pressure FLOAT,
    timestamp TIMESTAMP
);
""")
conn.commit()

# Simulación de transmisión en streaming continuo
machine_id = "Motor_A"
print(f"Iniciando streaming de telemetría para {machine_id}...")

try:
    while True:
        temp = round(random.uniform(50.0, 60.0), 2)
        press = round(random.uniform(2.0, 3.5), 2)
        current_time = datetime.now()

        cursor.execute(
            "INSERT INTO sensor_data (machine_name, temperature, pressure, timestamp) VALUES (%s, %s, %s, %s)",
            (machine_id, temp, press, current_time)
        )
        conn.commit()
        print(f"[{current_time}] Ingestado -> Temp: {temp} °C, Presión: {press} bar")
        time.sleep(2)
except KeyboardInterrupt:
    cursor.close()
    conn.close()
    print("Transmisión finalizada de forma segura.")

3. Lógica Analítica en Ignition Scripting Library

Script modular en la librería de Ignition para consumir la Named Query parametrizada, computar métricas y generar logs estructurados en el Gateway:

def analyzeMachine(machine_name):
    # Ejecución de Named Query parametrizada
    params = {"machineParam": machine_name}
    dataset = system.db.runNamedQuery("GetMachineData", params)
    
    if dataset.getRowCount() == 0:
        return 0.0, 0.0
        
    total_temp = 0.0
    max_pressure = 0.0
    
    for row in range(dataset.getRowCount()):
        temp = dataset.getValueAt(row, "temperature")
        pressure = dataset.getValueAt(row, "pressure")
        total_temp += temp
        if pressure > max_pressure:
            max_pressure = pressure
            
    avg_temp = total_temp / dataset.getRowCount()
    
    # Registro en el visor de diagnósticos del Gateway
    logger = system.util.getLogger("analytics")
    logger.info("Análisis {} -> Temp Media: {:.2f} °C, Presión Max: {:.2f} bar".format(
        machine_name, avg_temp, max_pressure
    ))
    
    return avg_temp, max_pressure

Framework de Implementación Paso a Paso

1. Levantar el stack: Ejecuta docker compose up -d e instala las dependencias de cliente (psycopg2, faker).
2. Configurar Database Connection en Ignition: Accede a http://localhost:8088, ve a Config > Database > Connections y crea la conexión JDBC apuntando a PostgreSQL.
3. Abrir Designer Launcher: Lanza el proyecto e implementa la Named Query (GetMachineData) con el parámetro machineParam de tipo String.
4. Diseñar la vista en Perspective: Inserta un componente Table, añade un Binding a la Named Query con sondeo periódico (Polling de 2 segundos) y retorno en formato Dataset.
5. Ejecutar el simulador Python: Inicia el script stream.py y verifica la reactividad instantánea tanto en la vista web como en los logs de diagnóstico de Ignition.