윈도우에서 OPC-UA 서버 구축하기 with python

원래 REST 방식으로 서버를 구축하려다, 실시간성이 필요하다고 해서 OPC-UA 서버를 구축하는 것으로 가닥을 잡았다.

 

MQTT와 OPC-UA 중에 뭘 고를지 좀 고민되긴 하는데, 일단 OPC UA로 구축해보고, 불편한점이 있나 확인해봐야지

 

구축 방식은 여러방법이 있지만, python 상에서 서버를 작동하는 방법으로 구현해보았다.

 

1. 파이썬 설치 

그냥 아무거나 받아봤다. 3.13이 깔리더라. 윈도우 스토어에서..

 

2. opc ua 라이브러리 설치

 

pip install opcua

 

다 끝나고 나면 

 

pip list로 잘 설치되었는지 확인해보자

 

 

3. 서버 코드 작성하기

 

 

from opcua import Server, ua  
import random

server = Server()
server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")

# ✅ 네임스페이스 등록
uri = "http://example.org"
idx = server.register_namespace(uri)  # 클라이언트에서 사용할 ns 값

# ✅ 올바른 방식으로 노드 추가 (NodeId 사용)
temp_node = server.nodes.objects.add_variable(ua.NodeId("Temperature", idx), "Temperature", 25.0)
humid_node = server.nodes.objects.add_variable(ua.NodeId("Humidity", idx), "Humidity", 50.0)
height_node = server.nodes.objects.add_variable(ua.NodeId("Height", idx), "Height", 0.0)


# ✅ 클라이언트에서 값을 읽고 쓸 수 있도록 설정
temp_node.set_writable()
humid_node.set_writable()
height_node.set_writable()

# ✅ 노드 ID 출력 (클라이언트에서 반드시 이 값을 사용해야 함)
print(f"✅ Temperature Node ID: {temp_node.nodeid}")  
print(f"✅ Humidity Node ID: {humid_node.nodeid}")
print(f"✅ Height Node ID: {height_node.nodeid}")

server.start()
print("✅ OPC UA Server is running!")

try:
    while True:
        temp_value = round(random.uniform(20, 30), 2)
        humid_value = round(random.uniform(40, 60), 2)

        temp_node.set_value(temp_value)
        humid_node.set_value(humid_value)

except KeyboardInterrupt:
    print("Stop Server")
    server.stop()

 

4. 서버 코드 작동 테스트

작동 확인 완료

 

5. 클라이언트 코드 작성 (데이터 송신만)

 

파이썬 버전

from opcua import Client
import random

# OPC UA 서버 주소 (서버에서 확인한 IP 사용)
client = Client("opc.tcp://192.168.1.108:4840/freeopcua/server/")  
client.connect()
print("✅ OPC UA 서버 연결 성공!")

try:
    # 🔹 서버에서 출력된 노드 ID와 일치하는지 확인
    height_node = client.get_node("ns=2;s=Height")  

    while True:
        height_value = round(random.uniform(10, 60), 2)
        height_node.set_value(height_value)
except KeyboardInterrupt:
    print("클라이언트 종료")
    client.disconnect()

 

 

6. 클라이언트 작동 확인

 

 

7. 클라이언트 코드 (데이터 수신 )

from opcua import Client

# OPC UA 서버 주소 (서버에서 확인한 IP 사용)
client = Client("opc.tcp://192.168.1.108:4840/freeopcua/server/")  
client.connect()
print("✅ OPC UA 서버 연결 성공!")

try:
    # 🔹 서버에서 출력된 노드 ID와 일치하는지 확인
    temp_node = client.get_node("ns=2;s=Temperature")  # 올바른 ID
    humid_node = client.get_node("ns=2;s=Humidity")  # 올바른 ID
    height_node = client.get_node("ns=2;s=Height")  # 올바른 ID
    while True:
        temp_value = temp_node.get_value()
        humid_value = humid_node.get_value()
        height_value = height_node.get_value()
        print(f"🌡️ 온도: {temp_value}°C | 💧 습도: {humid_value}% | 고도: {height_value}m " )
except KeyboardInterrupt:
    print("클라이언트 종료")
    client.disconnect()

 

8. 작동 확인

 

 

유니티 클라이언트 샘플 만들기는 다음 글에서 계속..