OPC UA cho kỹ sư Vision: node, đọc ghi, subscription với asyncua

Ở bài "Modbus TCP bằng Python với pymodbus", mọi dữ liệu là thanh ghi 16 bit đánh số: "HR 12–13 là đường kính, float32, ABCD" chỉ tồn tại trong một file Excel mà hai bên phải giữ khớp nhau. OPC UA (Open Platform Communications Unified Architecture, chuẩn IEC 62541) giải quyết đúng chỗ yếu đó: dữ liệu là các biến có tên, có kiểu, có cấu trúc, client có thể duyệt xem server có gì, đăng ký nhận thông báo khi giá trị đổi thay vì hỏi vòng, và kênh truyền có thể ký, mã hóa. Ngày nay nhiều PLC đời mới có sẵn OPC UA server (ví dụ S7-1500, Omron NJ/NX, Beckhoff TwinCAT; một số dòng cần bật tùy chọn, mua license runtime hoặc thêm module), SCADA và MES gần như đều nói được OPC UA. Bài này dạy bạn các khái niệm cốt lõi rồi chạy một server và client hoàn chỉnh bằng thư viện asyncua ngay trên máy. Bài tiếp theo, "TCP socket: gửi kết quả và tọa độ cho robot", quay về kênh đơn giản nhất cho những thiết bị không có OPC UA.

1. Các khái niệm cốt lõi

Hai chi tiết thực tế: (1) chỉ số namespace có thể thay đổi giữa các lần khởi động hoặc giữa các phiên bản firmware, còn URI của namespace thì không; client chuyên nghiệp tra chỉ số từ URI (get_namespace_index) thay vì viết cứng ns=3. (2) Kiểu định danh do server quyết định: PLC Siemens dùng NodeId chuỗi kiểu ns=3;s="DB_Vision"."Result" (có dấu nháy kép), nhiều server khác dùng số. Đừng đoán: duyệt cây bằng một công cụ client OPC UA (có nhiều công cụ miễn phí) hoặc bằng code ở mục 3 để lấy NodeId chính xác.

2. Cài đặt asyncua

asyncua là thư viện OPC UA mã nguồn mở cho Python (dự án FreeOpcUa/opcua-asyncio), có cả client lẫn server, viết theo kiểu asyncio (mọi lệnh mạng là await). Bài này kiểm tra với asyncua 2.1.0 trên Python 3.13.

pip install asyncua
python -c "import importlib.metadata as m; print(m.version('asyncua'))"

Thư viện cũ tên opcua (đồng bộ, không async) đã ngừng phát triển; đừng trộn hai thư viện. Nếu cần gọi từ code đồng bộ, asyncua có lớp bọc asyncua.sync, nhưng trong bài ta dùng thẳng asyncio. Thư viện ghi khá nhiều cảnh báo qua logging; các ví dụ hạ mức log xuống ERROR cho gọn.

3. Server mô phỏng, duyệt cây, đọc và ghi

Ngoài đời, OPC UA server thường nằm trong PLC. Để học, ta tự dựng một server trên 127.0.0.1 với một object Vision và năm biến, rồi cho client kết nối vào. Hàm tao_server được dùng lại ở các ví dụ sau.

import asyncio
import logging
from asyncua import Server, Client, ua

logging.getLogger("asyncua").setLevel(logging.ERROR)        # bớt cảnh báo của thư viện
NS_URI = "urn:hocfree:tram-vision"


async def tao_server(port):
    """OPC UA server mô phỏng 'PLC': object Vision với 5 biến, không bảo mật, chỉ nghe trên 127.0.0.1."""
    server = Server()
    await server.init()
    server.set_endpoint(f"opc.tcp://127.0.0.1:{port}/vision/")
    server.set_security_policy([ua.SecurityPolicyType.NoSecurity])
    ns = await server.register_namespace(NS_URI)
    tram = await server.nodes.objects.add_object(ua.NodeId("Vision", ns), f"{ns}:Vision")
    bien = {}
    for ten, gia_tri, kieu in [("Trigger", False, ua.VariantType.Boolean),     # PLC ghi
                               ("CycleNo", 0, ua.VariantType.UInt32),          # PLC ghi
                               ("Result", 0, ua.VariantType.Int16),            # vision ghi: 1 OK, 2 NG
                               ("DiameterMM", 0.0, ua.VariantType.Double),     # vision ghi
                               ("CycleEcho", 0, ua.VariantType.UInt32)]:       # vision ghi SAU CÙNG
        nut = await tram.add_variable(ua.NodeId(f"Vision.{ten}", ns), f"{ns}:{ten}", ua.Variant(gia_tri, kieu))
        await nut.set_writable()                                 # cho phép client ghi
        bien[ten] = nut
    return server, ns, bien


async def demo_doc_ghi():
    server, ns, bien = await tao_server(4841)
    async with server:                                            # start() ... stop()
        async with Client(url="opc.tcp://127.0.0.1:4841/vision/", timeout=2) as client:
            idx = await client.get_namespace_index(NS_URI)        # tra ns từ URI, không viết cứng
            print("Namespace", NS_URI, "→ ns =", idx)

            print("Duyệt Objects:")
            for nut in await client.nodes.objects.get_children():
                print("  ", nut.nodeid.to_string(), (await nut.read_browse_name()).to_string())

            tram = await client.nodes.objects.get_child(f"{idx}:Vision")
            print("Biến trong Vision:")
            for nut in await tram.get_children():
                ten = (await nut.read_browse_name()).Name
                kieu = await nut.read_data_type_as_variant_type()
                print(f"   {nut.nodeid.to_string():<24} {ten:<11} {kieu.name:<8} = {await nut.read_value()}")

            trigger = client.get_node(f"ns={idx};s=Vision.Trigger")
            result = client.get_node(f"ns={idx};s=Vision.Result")
            print("Đọc Trigger:", await trigger.read_value())

            await result.write_value(ua.Variant(1, ua.VariantType.Int16))      # đúng kiểu Int16
            print("Ghi Result = 1 (Int16) → server thấy:", await bien["Result"].read_value())
            for gia_tri in (2, 2.0):                                            # int Python = Int64, float = Double
                try:
                    await result.write_value(gia_tri)
                except ua.UaStatusCodeError as e:
                    print(f"Ghi Result = {gia_tri!r} → bị từ chối: {type(e).__name__}")


asyncio.run(demo_doc_ghi())

Namespace urn:hocfree:tram-vision → ns = 2
Duyệt Objects:
   i=31915 0:Locations
   i=2253 0:Server
   i=23470 0:Aliases
   ns=2;s=Vision 2:Vision
Biến trong Vision:
   ns=2;s=Vision.Trigger    Trigger     Boolean  = False
   ns=2;s=Vision.CycleNo    CycleNo     UInt32   = 0
   ns=2;s=Vision.Result     Result      Int16    = 0
   ns=2;s=Vision.DiameterMM DiameterMM  Double   = 0.0
   ns=2;s=Vision.CycleEcho  CycleEcho   UInt32   = 0
Đọc Trigger: False
Ghi Result = 1 (Int16) → server thấy: 1
Ghi Result = 2 → bị từ chối: BadTypeMismatch
Ghi Result = 2.0 → bị từ chối: BadTypeMismatch

Đọc kết quả từng phần:

• Namespace của ta được cấp chỉ số 2 (0 là của chuẩn, 1 là của chính server). Dưới Objects ngoài object Vision của ta còn có các node chuẩn như Server (i=2253), nơi chứa trạng thái, thời gian, khả năng của server; danh sách node chuẩn có thể khác đôi chút giữa các phiên bản thư viện.
• Mỗi biến có NodeId dạng chuỗi, BrowseName, kiểu dữ liệu và giá trị: client tự khám phá được tất cả mà không cần file Excel nào.
• Kiểu dữ liệu phải khớp chính xác. Số nguyên Python được gửi mặc định là Int64, số thực là Double; biến Int16 từ chối cả hai bằng mã lỗi BadTypeMismatch. Luôn bọc giá trị trong ua.Variant(giá_trị, ua.VariantType.Int16), hoặc đọc kiểu của biến một lần lúc khởi động (read_data_type_as_variant_type) rồi dùng lại. Đây là lỗi số một khi ghi vào biến của PLC: INT của Siemens là Int16, DINT là Int32, REAL là Float (32 bit), LREAL là Double.

4. Subscription: nhận thông báo thay vì hỏi vòng

Với Modbus, client phải hỏi "Trigger đã bật chưa?" mỗi vài chục ms. OPC UA có subscription: client tạo một subscription, gắn vào đó các monitored item (biến cần theo dõi); server tự lấy mẫu biến và gửi thông báo khi giá trị đổi. Hai thông số quyết định độ trễ:

Hệ quả quan trọng: độ trễ từ lúc PLC bật Trigger đến lúc vision biết có thể tới cỡ sampling + publishing interval, cộng thời gian PLC cập nhật biến vào server OPC UA (nhiều PLC chỉ làm việc đó trong thời gian rảnh của chu kỳ quét). Tức là hàng chục ms và không tất định. Một xung Trigger ngắn hơn sampling interval có thể bị bỏ sót hoàn toàn. Vì vậy trong trạm cần chụp đúng vị trí, trigger chụp vẫn đi bằng dây I/O vào camera; OPC UA chở dữ liệu ngữ cảnh (số chu kỳ, công thức) và kết quả.

Ví dụ dưới là bài thực hành của bài học: server đóng vai PLC, chạy ba chu kỳ: đặt CycleNo, bật Trigger, chờ đến khi CycleEcho bằng số chu kỳ thì đọc kết quả, tắt Trigger. Phía vision là client: đăng ký nhận thay đổi của Trigger, khi thấy sườn lên thì đọc CycleNo, "chụp và đo" đường kính trên một ảnh tổng hợp, ghi DiameterMM, Result và cuối cùng CycleEcho. Cách dùng CycleEcho làm tín hiệu "đã xong" giống hệt quy tắc ở bài I/O số và bài Modbus: dữ liệu ghi trước, tín hiệu hoàn thành ghi sau cùng.

import asyncio
import cv2
import numpy as np
from asyncua import Client, ua

MM_PER_PX = 0.05
DUNG_SAI = (9.90, 10.10)                     # đường kính hợp lệ, mm


def chup_va_do(ban_kinh_px):
    """'Chụp' ảnh tổng hợp một trục tròn và đo đường kính từ diện tích (số pixel trắng)."""
    anh = np.zeros((260, 260), np.uint8)
    cv2.circle(anh, (130, 130), ban_kinh_px, 255, -1)
    dien_tich = np.count_nonzero(anh)
    return 2 * np.sqrt(dien_tich / np.pi) * MM_PER_PX


class BatThayDoi:
    """Handler của subscription: asyncua gọi datachange_notification mỗi khi biến đổi giá trị."""

    def __init__(self):
        self.hang_doi = asyncio.Queue()

    def datachange_notification(self, node, val, data):
        self.hang_doi.put_nowait(val)        # chỉ đẩy vào hàng đợi, xử lý ở vòng lặp chính


async def vision_client(url, so_chu_ky, ban_kinh_theo_chu_ky):
    async with Client(url=url, timeout=2) as client:
        idx = await client.get_namespace_index(NS_URI)
        nut = {ten: client.get_node(f"ns={idx};s=Vision.{ten}")
               for ten in ("Trigger", "CycleNo", "Result", "DiameterMM", "CycleEcho")}
        bat = BatThayDoi()
        sub = await client.create_subscription(20, bat)                 # publishing interval 20 ms
        await sub.subscribe_data_change(nut["Trigger"], sampling_interval=10)
        print("vision: thông báo đầu tiên (giá trị hiện tại) Trigger =", await bat.hang_doi.get())
        truoc, da_xu_ly = False, 0
        while da_xu_ly < so_chu_ky:
            trigger = await asyncio.wait_for(bat.hang_doi.get(), timeout=3)
            if trigger and not truoc:                                    # sườn lên
                chu_ky = await nut["CycleNo"].read_value()
                d = chup_va_do(ban_kinh_theo_chu_ky[chu_ky])
                ok = DUNG_SAI[0] <= d <= DUNG_SAI[1]
                await nut["DiameterMM"].write_value(ua.Variant(round(d, 3), ua.VariantType.Double))
                await nut["Result"].write_value(ua.Variant(1 if ok else 2, ua.VariantType.Int16))
                await nut["CycleEcho"].write_value(ua.Variant(chu_ky, ua.VariantType.UInt32))   # SAU CÙNG
                print(f"vision: chu kỳ {chu_ky}: D = {d:.3f} mm → {'OK' if ok else 'NG'}")
                da_xu_ly += 1
            truoc = trigger
        await sub.delete()


async def plc_gia(bien, so_chu_ky):
    """Logic 'PLC' chạy phía server: ghi thẳng vào node của server."""
    await asyncio.sleep(0.3)                                             # chờ vision kết nối
    for chu_ky in range(1, so_chu_ky + 1):
        await bien["CycleNo"].write_value(ua.Variant(chu_ky, ua.VariantType.UInt32))
        await bien["Trigger"].write_value(True)
        for _ in range(200):                                             # chờ tối đa 2 s
            if await bien["CycleEcho"].read_value() == chu_ky:
                break
            await asyncio.sleep(0.01)
        else:
            print(f"PLC: chu kỳ {chu_ky} quá thời gian → xử lý như NG, báo lỗi")
        kq = await bien["Result"].read_value()
        d = await bien["DiameterMM"].read_value()
        print(f"PLC: chu kỳ {chu_ky}: Result = {kq} ({'OK' if kq == 1 else 'NG'}), D = {d} mm")
        await bien["Trigger"].write_value(False)
        await asyncio.sleep(0.1)


async def demo_subscription():
    server, ns, bien = await tao_server(4842)
    async with server:
        await asyncio.wait_for(asyncio.gather(
            vision_client("opc.tcp://127.0.0.1:4842/vision/", 3, {1: 100, 2: 104, 3: 101}),
            plc_gia(bien, 3)), timeout=15)
    print("Đã dừng server.")


asyncio.run(demo_subscription())

vision: thông báo đầu tiên (giá trị hiện tại) Trigger = False
vision: chu kỳ 1: D = 10.000 mm → OK
PLC: chu kỳ 1: Result = 1 (OK), D = 10.0 mm
vision: chu kỳ 2: D = 10.395 mm → NG
PLC: chu kỳ 2: Result = 2 (NG), D = 10.395 mm
vision: chu kỳ 3: D = 10.095 mm → OK
PLC: chu kỳ 3: Result = 1 (OK), D = 10.095 mm
Đã dừng server.

Bốn điểm cần để ý:

• Thông báo đầu tiên luôn là giá trị hiện tại ngay khi đăng ký (ở đây Trigger = False). Code bắt sườn phải tính đến nó, nếu không lúc vision khởi động giữa chừng khi Trigger đang bật, nó sẽ chụp nhầm một sản phẩm cũ hoặc bỏ sót.
• Handler chỉ đẩy vào hàng đợi. datachange_notification được thư viện gọi từ task xử lý thông báo; làm việc nặng (chụp, xử lý ảnh) ngay trong đó sẽ làm chậm mọi thông báo khác. Đẩy vào asyncio.Queue rồi xử lý ở vòng lặp chính là mẫu chuẩn. (asyncua 2.x còn có chế độ lặp async for trên subscription khi không truyền handler.)
• Thứ tự ghi: DiameterMM và Result trước, CycleEcho sau cùng; PLC chỉ đọc kết quả khi CycleEcho khớp số chu kỳ nó đã gửi. Mỗi lệnh ghi riêng lẻ là một yêu cầu riêng, nên nếu đảo thứ tự, PLC có thể thấy CycleEcho mới đi kèm Result cũ. (Có thể gộp nhiều giá trị vào một lệnh write_values, nhưng chuẩn không bảo đảm server áp dụng chúng một cách nguyên tử, nên vẫn giữ quy ước "echo sau cùng".)
• Mọi thao tác đều có thời hạn: timeout=2 cho từng yêu cầu của client, wait_for khi chờ thông báo, vòng chờ 2 s phía PLC, và timeout=15 cho toàn bộ demo để script luôn tự kết thúc.

Trong dự án thật, phía vision chỉ cần phần vision_client: thay URL bằng endpoint của PLC, thay tên biến bằng NodeId lấy được khi duyệt cây, và bật Client(url, auto_reconnect=True) hoặc tự viết vòng kết nối lại (asyncua 2.x có tùy chọn tự kết nối lại và tạo lại subscription).

5. Bảo mật: ba chế độ và chứng chỉ

OPC UA được thiết kế có bảo mật từ đầu, ở hai tầng:

Ví dụ trong bài dùng NoSecurity vì chạy trên 127.0.0.1. Với PLC thật, cấu hình phổ biến là Basic256Sha256 + SignAndEncrypt + tài khoản người dùng. Trong asyncua, phía client đặt bằng một dòng trước khi kết nối, ví dụ await client.set_security_string("Basic256Sha256,SignAndEncrypt,cert.der,key.pem") cùng client.set_user(...), client.set_password(...); chứng chỉ có thể tạo bằng OpenSSL hoặc công cụ đi kèm thư viện. Khi kết nối bị từ chối với lỗi về chứng chỉ, hãy kiểm tra theo thứ tự: chứng chỉ client đã được tin cậy trên server chưa, đồng hồ hai máy có lệch nhiều không (chứng chỉ có thời hạn hiệu lực), application URI trong chứng chỉ có khớp không.

6. OPC UA hay Modbus TCP?

Lời khuyên thực tế: trạm đơn lẻ với PLC đời cũ hoặc giá rẻ, ít dữ liệu → Modbus TCP. PLC có OPC UA server sẵn, cần nhiều biến có cấu trúc, dữ liệu còn đi lên SCADA/MES, hoặc nhà máy có yêu cầu bảo mật → OPC UA. Cả hai trường hợp: trigger chụp cần chính xác thời gian vẫn đi bằng dây I/O.

7. Lỗi thường gặp

8. Bài tập

import asyncio
from asyncua import Client, ua


async def in_cay(node, sau=0, toi_da=3):
    ten = (await node.read_browse_name()).to_string()
    lop = await node.read_node_class()
    dong = f"{'  ' * sau}{ten} [{lop.name}]"
    if lop == ua.NodeClass.Variable:
        dong += f" {(await node.read_data_type_as_variant_type()).name}"
    print(dong)
    if sau < toi_da:
        for con in await node.get_children():
            await in_cay(con, sau + 1, toi_da)


async def bai_1():
    server, ns, bien = await tao_server(4843)
    async with server:
        async with Client(url="opc.tcp://127.0.0.1:4843/vision/", timeout=2) as client:
            idx = await client.get_namespace_index(NS_URI)
            await in_cay(client.get_node(f"ns={idx};s=Vision"))


asyncio.run(bai_1())

2:Vision [Object]
  2:Trigger [Variable] Boolean
  2:CycleNo [Variable] UInt32
  2:Result [Variable] Int16
  2:DiameterMM [Variable] Double
  2:CycleEcho [Variable] UInt32

Tóm tắt

• OPC UA: server giữ address space gồm các node; client duyệt, đọc, ghi, đăng ký. NodeId = namespace + định danh; tra chỉ số namespace từ URI, lấy NodeId bằng cách duyệt cây.
• Ghi phải đúng kiểu: ua.Variant(v, ua.VariantType.Int16/Float/Double…); INT ↔ Int16, DINT ↔ Int32, REAL ↔ Float, LREAL ↔ Double. Mỗi giá trị có StatusCode.
• Subscription thay cho hỏi vòng; sampling và publishing interval quyết định độ trễ (hàng chục ms, không tất định); thông báo đầu tiên là giá trị hiện tại; handler chỉ đẩy vào hàng đợi.
• Quy ước bắt tay vẫn như cũ: dữ liệu ghi trước, CycleEcho sau cùng, mọi chờ đợi có timeout.
• Bảo mật: None / Sign / SignAndEncrypt, chứng chỉ ứng dụng và trust list, tài khoản người dùng; không để None trên mạng sản xuất.
• Modbus TCP: đơn giản, có ở mọi PLC. OPC UA: dữ liệu có cấu trúc, bảo mật, hợp tích hợp SCADA/MES. Trigger chụp chính xác thời gian vẫn đi bằng dây I/O.

Bài tiếp theo: TCP socket: gửi kết quả và tọa độ cho robot, kênh tự định nghĩa mà robot của gần như mọi hãng đều nói được.