제조 현장 시스템 지도

┌─────────────────────────────────────────────────────────┐
│                    제조 현장 시스템                       │
│                                                          │
│  ┌──────┐    ┌──────┐    ┌──────┐    ┌──────────┐       │
│  │ PLC  │───▶│ SCADA│───▶│ MES  │───▶│ ERP      │       │
│  │설비  │    │모니터│    │실행  │    │경영      │       │
│  └──────┘    └──────┘    └──────┘    └──────────┘       │
│     │                       │              │             │
│     │  OPC-UA           REST API      REST API          │
│     │                       │              │             │
│     └───────────────────────┴──────────────┘             │
│                             │                            │
│                      ┌──────┴──────┐                     │
│                      │  AI Agent   │                     │
│                      │ (우리가 만들)│                     │
│                      └─────────────┘                     │
└─────────────────────────────────────────────────────────┘

Agent가 데이터에 접근하는 3가지 방법:

---

패턴 1: REST API (가장 일반적)

MES/ERP가 REST API를 제공하는 경우

Agent  ──HTTP──>  MES API Gateway  ──>  MES DB
                      │
                  인증, 속도제한, 로깅

import httpx
from typing import Optional

class MesApiClient:
    """MES REST API 클라이언트"""

    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.client = httpx.Client(timeout=5.0)

    def get_equipment_status(self, equipment_id: str) -> dict:
        """설비 상태 조회"""
        try:
            response = self.client.get(
                f"{self.base_url}/api/v1/equipment/{equipment_id}/status",
                headers=self.headers
            )
            response.raise_for_status()
            return response.json()
        except httpx.TimeoutException:
            raise TimeoutError(f"MES 응답 시간 초과 (설비: {equipment_id})")
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 404:
                raise ValueError(f"설비 '{equipment_id}'를 찾을 수 없습니다")
            raise ConnectionError(f"MES API 오류: {e.response.status_code}")

    def get_alarm_history(self, equipment_id: str, days: int = 30) -> list:
        """알람 이력 조회"""
        response = self.client.get(
            f"{self.base_url}/api/v1/alarms",
            params={"equipment_id": equipment_id, "days": days},
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()["data"]

장점	단점
표준화된 인터페이스	API가 제공되어야 함
보안 (인증/인가)	레거시 MES는 미지원
문서화 용이	실시간성 한계 (polling)

---

패턴 2: 데이터베이스 직접 접근

MES API가 없거나 불충분한 경우

Agent  ──SQL──>  MES Database (읽기 전용)

import asyncpg  # PostgreSQL 비동기 드라이버

class MesDbClient:
    """MES DB 직접 조회 (읽기 전용)"""

    def __init__(self, dsn: str):
        self.dsn = dsn
        self.pool = None

    async def connect(self):
        self.pool = await asyncpg.create_pool(
            self.dsn,
            min_size=2,
            max_size=10,
            command_timeout=5
        )

    async def get_equipment_status(self, equipment_id: str) -> dict:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow("""
                SELECT
                    e.equipment_id,
                    e.equipment_name,
                    e.line_id,
                    es.status,
                    es.utilization,
                    es.temperature,
                    es.vibration,
                    es.updated_at
                FROM equipment e
                JOIN equipment_status es ON e.equipment_id = es.equipment_id
                WHERE e.equipment_id = $1
                  AND es.updated_at = (
                      SELECT MAX(updated_at)
                      FROM equipment_status
                      WHERE equipment_id = $1
                  )
            """, equipment_id)

            if not row:
                return None
            return dict(row)

    async def get_active_alarms(self, equipment_id: str) -> list:
        async with self.pool.acquire() as conn:
            rows = await conn.fetch("""
                SELECT alarm_code, alarm_desc, severity, occurred_at
                FROM alarm_log
                WHERE equipment_id = $1
                  AND resolved_at IS NULL
                ORDER BY occurred_at DESC
            """, equipment_id)
            return [dict(r) for r in rows]

장점	단점
API 없어도 가능	DB 스키마 이해 필요
복잡한 쿼리 가능	보안 위험 (DB 직접 접근)
빠른 조회	MES 성능에 영향 줄 수 있음

주의: 반드시 읽기 전용 계정을 사용하세요! 실수로 DELETE나 UPDATE를 실행하면 생산 라인이 멈출 수 있습니다.

---

패턴 3: OPC-UA (실시간 설비 데이터)

PLC/SCADA에서 실시간 센서 데이터를 직접 읽는 경우

Agent  ──OPC-UA──>  SCADA/PLC  ──>  센서

from opcua import Client

class OpcUaClient:
    """OPC-UA 실시간 설비 데이터 조회"""

    def __init__(self, endpoint: str):
        self.endpoint = endpoint
        self.client = Client(endpoint)

    def connect(self):
        self.client.connect()

    def get_sensor_data(self, node_id: str) -> dict:
        """센서 데이터 실시간 조회"""
        node = self.client.get_node(node_id)
        value = node.get_value()
        return {
            "node_id": node_id,
            "value": value,
            "timestamp": node.get_data_value().SourceTimestamp
        }

    def get_equipment_sensors(self, equipment_id: str) -> dict:
        """설비의 전체 센서 데이터 조회"""
        base_node = f"ns=2;s={equipment_id}"
        return {
            "temperature": self.get_sensor_data(f"{base_node}.Temperature"),
            "vibration": self.get_sensor_data(f"{base_node}.Vibration"),
            "pressure": self.get_sensor_data(f"{base_node}.Pressure"),
            "rpm": self.get_sensor_data(f"{base_node}.SpindleRPM"),
        }

장점	단점
실시간 데이터	설정 복잡
PLC 직접 접근	OPC-UA 서버 필요
산업 표준	보안 설정 까다로움

---

패턴 선택 가이드

상황	권장 패턴
MES에 REST API가 있음	패턴 1 (REST API)
API는 없지만 DB 접근 가능	패턴 2 (DB 직접)
실시간 센서 데이터 필요	패턴 3 (OPC-UA)
여러 시스템 통합	패턴 1 + 2 혼합
신규 스마트팩토리	패턴 1 + 3

현실: 대부분의 제조 현장은 레거시 시스템이라 패턴 2(DB 직접)부터 시작하는 경우가 많다.