산업 현장 AI 통합 아키텍처 가이드 2편: 실시간 이상 감지와 예측 유지보수

import paho.mqtt.client as mqtt
import json
from datetime import datetime, timezone

def on_message(client, userdata, msg):
    raw = json.loads(msg.payload)
    event = {
        "specversion": "1.0",
        "type": "sensor.measurement",
        "source": f"factory/line-A/{msg.topic}",
        "time": datetime.now(timezone.utc).isoformat(),
        "data": {
            "value": raw["v"],
            "unit": raw.get("u", "?"),
            "quality": raw.get("q", 192),
        }
    }
    kafka_producer.send("sensor-events", json.dumps(event).encode())

client = mqtt.Client()
client.on_message = on_message
client.connect("broker.factory.local", 1883)
client.subscribe("sensors/#", qos=1)

import sqlite3

class EdgeBuffer:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.execute(
            "CREATE TABLE IF NOT EXISTS events ("
            "id INTEGER PRIMARY KEY AUTOINCREMENT,"
            "payload TEXT NOT NULL,"
            "sent INTEGER DEFAULT 0)"
        )

    def write(self, payload: str):
        self.conn.execute("INSERT INTO events(payload) VALUES(?)", (payload,))
        self.conn.commit()

    def flush(self, producer, batch: int = 1000):
        rows = self.conn.execute(
            "SELECT id, payload FROM events WHERE sent=0 LIMIT ?", (batch,)
        ).fetchall()
        for _, p in rows:
            producer.send("sensor-events", p.encode())
        producer.flush()
        ids = [str(r[0]) for r in rows]
        self.conn.execute(
            f"UPDATE events SET sent=1 WHERE id IN ({','.join(ids)})"
        )
        self.conn.commit()

CREATE TABLE sensor_readings (
    time        TIMESTAMPTZ NOT NULL,
    device_id   TEXT NOT NULL,
    metric      TEXT NOT NULL,
    value       DOUBLE PRECISION
);

SELECT create_hypertable('sensor_readings', 'time',
    chunk_time_interval => INTERVAL '1 hour');

-- 자동 압축 (7일 후)
SELECT add_compression_policy('sensor_readings', INTERVAL '7 days');

-- 1분 집계 연속 뷰
CREATE MATERIALIZED VIEW sensor_1min
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 minute', time) AS bucket,
       device_id, metric,
       AVG(value) avg_val, MAX(value) max_val
FROM sensor_readings
GROUP BY bucket, device_id, metric;

OPC-UA / MQTT / Modbus
    ↓
엣지 게이트웨이 (EdgeBuffer + 프로토콜 변환)
    ↓ Kafka
스트림 처리 (Kafka Streams)
    ↓
TimescaleDB (단기 고해상도) + S3 (장기 콜드)
    ↓
AI 모델 (이상 탐지 / 예측 유지보수 / 품질 검사)