OPC UA ↔ AAS 연동 실시간 데이터 게이트웨이

PLC의 OPC UA 데이터를 AAS Property로 실시간 동기화하는 게이트웨이 아키텍처 | AAS Advanced J04

OPC UA → AAS 연동 아키텍처

PLC

Siemens S7-1500
Allen-Bradley ControlLogix
XGI-CPUE (LS ELECTRIC)

IEC 61131-3

→OPC UA
TCP 4840

OPC UA Server

Node tree 노출
ns=2;s=AR-S513.AssemblyForce
Subscription 지원

IEC 62541

→UA Client
SDK

OPC UA Client
(Gateway)

Python asyncua
Node Subscribe
값 변경 콜백

asyncio

→HTTP PUT
REST API

AAS Server

DemoServer / BaSyx
Submodel Property
업데이트

Part 2 API

→REST API
JSON

REST Client

대시보드
Postman
외부 시스템

HTTP/JSON

OPC UA NodeId ↔ AAS semanticId 매핑 예시

OPC UA NodeId	데이터 유형	AAS 경로	semanticId (IRDI)	단위
ns=2;s=AR-S513.AssemblyForce	Float	AR01/OperationalData/AssemblyForce	0173-1#02-AAI648#001	A
ns=2;s=AR-S513.AssemblyVoltage	Float	AR01/OperationalData/AssemblyVoltage	0173-1#02-AAI649#001	V
ns=2;s=AR-S513.ArcTime	Float	AR01/OperationalData/ArcTime	0173-1#02-AAI651#001	s
ns=2;s=AR-S513.MotorTemp	Float	AR01/OperationalData/MotorTemperature	0173-1#02-AAI652#001	°C
ns=2;s=AR-S513.CycleCount	Int32	AR01/OperationalData/CycleCount	0173-1#02-AAI655#001	count
ns=2;s=AR-S513.Status	String	AR01/OperationalData/Status	0173-1#02-AAI660#001	enum

OPC UA → AAS 게이트웨이 Python 예시

import asyncio
from asyncua import Client, ua
import aiohttp OPCUA_URL = "opc.tcp://plc-142:4840/freeopcua/server/"
AAS_URL = "http://localhost:8081"
AAS_ID = "dXJuOmZhY3Rvcnk..." # base64(AAS id)
SM_ID = "dXJuOmZhY3Rvcnk..." # base64(Submodel id) MAPPING = { "ns=2;s=AR-S513.AssemblyForce": "AssemblyForce", "ns=2;s=AR-S513.AssemblyVoltage": "AssemblyVoltage", "ns=2;s=AR-S513.MotorTemp": "MotorTemperature",
} async def sync_to_aas(prop_name, value): url = (f"{AAS_URL}/shells/{AAS_ID}" f"/submodels/{SM_ID}" f"/submodel-elements/{prop_name}/$value") async with aiohttp.ClientSession as s: await s.patch(url, json=str(value)) async def main: async with Client(OPCUA_URL) as client: handles = [] sub = await client.create_subscription( 500, lambda node, val, data: asyncio.create_task( sync_to_aas(MAPPING[node.nodeid], val) ) ) for nodeid in MAPPING: node = client.get_node(nodeid) h = await sub.subscribe_data_change(node) handles.append(h) await asyncio.sleep(float("inf")) # 영구 실행 asyncio.run(main)

OPC UA Subscription (Push)

값 변경 시 즉시 콜백 — 실시간성 우수
불필요한 네트워크 트래픽 최소화
MonitoredItem 개수 제한 있음 (서버별)
연결 끊김 시 재구독 로직 필요
적합: 고빈도 변화 데이터 (전류, 온도)

AAS 폴링 (Pull)

일정 주기로 GET 요청 — 구현 단순
변화 없어도 항상 트래픽 발생
폴링 주기가 실시간성 상한 결정
부하 분산 가능 (임의 시간차 요청)
적합: 저빈도 상태값 (사이클 카운트)

BaSyx OPC UA Connector

BaSyx 공식 컴포넌트 — 별도 코딩 불필요
YAML 매핑 파일로 NodeId→Property 설정
Docker 컨테이너로 배포 가능
OPC UA → AAS 자동 동기화 (주기/이벤트)

# BaSyx OPC UA Connector 설정 (YAML)
aas: id: "urn:factory142:aas:AR01"
opcua: serverUrl: "opc.tcp://plc-142:4840/"
mapping: - nodeId: "ns=2;s=AR-S513.AssemblyForce" submodelId: "urn:aas-edu:sm:OperationalData" idShortPath: "AssemblyForce" updateRate: 500ms - nodeId: "ns=2;s=AR-S513.AssemblyVoltage" submodelId: "urn:aas-edu:sm:OperationalData" idShortPath: "AssemblyVoltage" updateRate: 500ms

DemoServer에 OPC UA 지원 추가하는 방법

방법 1: Python 게이트웨이 (권장)

asyncua 라이브러리로 OPC UA 구독 → DemoServer REST API PUT 방식. 의존성 최소, 유연한 변환 로직 구현 가능.

방법 2: BaSyx OPC UA Connector

DemoServer 대신 BaSyx로 교체 후 공식 Connector 사용. 안정성 높지만 설정 복잡도 증가.

방법 3: Node-RED 연동

node-red-contrib-opcua → node-red-contrib-aas 노드 체인. 비개발자도 GUI로 플로우 구성 가능.

AAS J04 — OPC UA Integration | IEC 62541 (OPC UA) | asyncua Python SDK | BaSyx OPC UA Connector | 조립 셀 S513 AR01 실시간 데이터 연동