제조 현장 시스템 지도

┌─────────────────────────────────────────────────────────┐
│                    제조 현장 시스템                       │
│                                                          │
│  ┌──────┐    ┌──────┐    ┌──────┐    ┌──────────┐       │
│  │ 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"]

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

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

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

먼저: async def와 await 아래 코드에 처음 보는 문법이 나옵니다. async def로 선언한 함수는 바로 실행되지 않고, await를 만나면 "응답을 기다리는 동안 다른 일을 하라"고 제어를 넘깁니다. DB나 네트워크처럼 기다리는 시간이 대부분인 작업을 여러 건 겹쳐 처리할 때 씁니다. async with pool.acquire() as conn은 연결 하나를 빌려 쓰고 자동으로 돌려준다는 뜻입니다.

이 주차의 실습은 async 없이 진행합니다. 아래 코드는 실제 MES DB 연동이 어떤 모양인지 보여주는 참고용이고, 지금은 "기다림이 긴 호출은 이렇게 겹쳐 처리한다" 정도로 읽으면 됩니다. 동기 방식으로 쓰려면 asyncpg 대신 psycopg를 쓰고 async/await를 빼면 됩니다.

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]

주의: 반드시 읽기 전용 계정을 사용하세요! 실수로 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:
        """설비의 전체 센서 데이터 조회

        node_id 형식 "ns=2;s=CNC-001.Temperature" 읽는 법:
          ns=2  네임스페이스 번호. 서버가 태그 이름을 묶어 둔 구역 번호다
          s=... 문자열 식별자. 이 서버가 그 구역에서 쓰는 태그 이름
        번호와 이름은 설비·서버마다 다르므로, 현장 OPC-UA 서버의 주소 공간을
        브라우징해 실제 값을 확인한 뒤 코드에 넣는다. 아래는 예시 형식이다.
        """
        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"),
        }

패턴 선택 가이드

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