Modbus TCP bằng Python với pymodbus cho trạm vision

I/O số chỉ gửi được vài bit. Khi PLC cần biết đường kính đo được là bao nhiêu, mã lỗi nào, hay trạm đang chạy công thức (recipe) nào, bạn cần một giao thức mạng. Modbus TCP là lựa chọn đơn giản và phổ biến nhất: gần như PLC nào cũng hỗ trợ, đặc tả mở, và Python có thư viện pymodbus rất đầy đủ. Bài này dùng lại đúng trình tự bắt tay ở bài "I/O số và handshake với PLC", nhưng thay dây điện bằng các ô nhớ Modbus; bài "Các chuẩn giao tiếp: Modbus TCP, EtherNet/IP, PROFINET" đã cho bạn bức tranh tổng quát, ở đây là code chạy được. Mọi ví dụ đều chạy trên máy bạn với một PLC giả khởi động ngay trong script.

1. Modbus TCP trong năm phút

• Client / server (tài liệu cũ gọi là master / slave). Client gửi yêu cầu, server trả lời. Server không bao giờ tự gửi gì. Trong trạm vision, PLC thường là server, PC vision là client và phải hỏi vòng (polling) liên tục: "có trigger chưa?".
• Chạy trên TCP, cổng mặc định 502. Trên Linux, mở cổng dưới 1024 cần quyền root, nên các ví dụ dùng cổng 5020.
• Mỗi yêu cầu mang một Unit ID (pymodbus gọi là device_id). Với PLC nối thẳng Ethernet, giá trị này thường bị bỏ qua hoặc phải là 1 hay 255 tùy hãng; nó chỉ thực sự quan trọng khi đi qua gateway Modbus TCP → RTU.
• Không có kiểu dữ liệu: mọi thứ là bit hoặc thanh ghi 16 bit. Số thực, số 32 bit, chuỗi… là do hai bên tự thống nhất cách ghép.

2. Mô hình dữ liệu và cách đánh địa chỉ

Một lần đọc được tối đa 125 thanh ghi hoặc 2000 bit; một lần ghi nhiều tối đa 123 thanh ghi. Mỗi PLC ánh xạ các vùng này vào bộ nhớ riêng của nó (thanh ghi D, khối DB, vùng %MW…): hãy xin kỹ sư PLC bảng ánh xạ cụ thể.

Bẫy kinh điển: 0 hay 1? Trên dây, địa chỉ luôn bắt đầu từ 0. Ký hiệu "40001" là cách viết của tài liệu: chữ số đầu chỉ vùng, phần còn lại đếm từ 1. Vì vậy:

Lệch một địa chỉ là lỗi rất khó thấy: bạn vẫn đọc/ghi thành công, chỉ là vào ô bên cạnh. Cách kiểm tra khi chạy thử: nhờ kỹ sư PLC ghi một giá trị "lạ" (ví dụ 1234) vào đúng ô, bạn đọc ra phải thấy 1234.

3. Cài đặt và lưu ý phiên bản

pip install pymodbus
python -c "import pymodbus; print(pymodbus.__version__)"

Bài viết dùng pymodbus 3.15.0. Thư viện này đổi API khá thường xuyên, nên nếu bạn đọc code trên mạng hãy để ý:

• Tham số chọn thiết bị hiện là device_id=. Các bản 3.x cũ dùng slave=, bản rất cũ dùng unit=. Gặp TypeError: ... unexpected keyword argument 'slave' là dấu hiệu code viết cho bản cũ.
• Phía server, bản 3.15 dùng SimData / SimDevice để khai báo bộ nhớ. Các lớp cũ ModbusSequentialDataBlock, ModbusDeviceContext (trước nữa là ModbusSlaveContext) vẫn chạy nhưng in cảnh báo "deprecated" và sẽ bị bỏ ở bản 4.
• Các tham số sau địa chỉ (count, device_id) phải truyền theo tên: read_coils(0, count=4, device_id=1).

4. PLC giả và các lệnh đọc/ghi cơ bản

Lớp PlcGia chạy một Modbus TCP server trong thread riêng (pymodbus server dùng asyncio), với bộ nhớ gồm 16 coil, 16 discrete input, 32 holding register, 16 input register, đều bắt đầu từ địa chỉ 0. Hàm doc_coil, ghi_hr… đóng vai "chương trình PLC" tự đọc/ghi bộ nhớ của nó; phía vision chỉ dùng ModbusTcpClient như với PLC thật.

import asyncio
import threading
import time

from pymodbus.client import ModbusTcpClient
from pymodbus.server import ModbusTcpServer
from pymodbus.simulator import DataType, SimData, SimDevice


class PlcGia:
    """PLC giả: một Modbus TCP server chạy trong thread riêng, có hàm đọc/ghi bộ nhớ của chính nó."""

    def __init__(self, port=5020):
        self.port = port

    def bat(self):
        self._da_nghe = threading.Event()
        self._thread = threading.Thread(target=lambda: asyncio.run(self._main()), daemon=True)
        self._thread.start()
        if not self._da_nghe.wait(timeout=3):
            raise RuntimeError("PLC giả không khởi động được")

    async def _main(self):
        self.loop = asyncio.get_running_loop()
        bo_nho = SimDevice(id=1, simdata=(
            [SimData(0, count=16, values=False, datatype=DataType.BITS)],   # coils 0..15
            [SimData(0, count=16, values=False, datatype=DataType.BITS)],   # discrete inputs 0..15
            [SimData(0, count=32, values=0, datatype=DataType.REGISTERS)],  # holding registers 0..31
            [SimData(0, count=16, values=0, datatype=DataType.REGISTERS)],  # input registers 0..15
        ))
        self.server = ModbusTcpServer(bo_nho, address=("127.0.0.1", self.port))
        await self.server.serve_forever(background=True)          # mở cổng rồi trả về ngay
        self._da_nghe.set()
        await self.server.serving                                 # chờ tới khi shutdown()

    def tat(self):
        asyncio.run_coroutine_threadsafe(self.server.shutdown(), self.loop).result(timeout=3)
        self._thread.join(timeout=3)

    # "Chương trình PLC" đọc/ghi bộ nhớ của chính nó (mã hàm 1 = coil, 3 = holding register)
    def _goi(self, coro):
        return asyncio.run_coroutine_threadsafe(coro, self.loop).result(timeout=2)

    def doc_coil(self, dia_chi):
        return self._goi(self.server.async_getValues(1, 1, dia_chi, 1))[0]

    def ghi_coil(self, dia_chi, gia_tri):
        self._goi(self.server.async_setValues(1, 5, dia_chi, [bool(gia_tri)]))

    def doc_hr(self, dia_chi, so_luong=1):
        return self._goi(self.server.async_getValues(1, 3, dia_chi, so_luong))

    def ghi_hr(self, dia_chi, gia_tri):
        self._goi(self.server.async_setValues(1, 16, dia_chi, list(gia_tri)))


plc = PlcGia(port=5020)
plc.bat()
plc.ghi_hr(0, [7, 1])                 # "chương trình PLC": mã sản phẩm 7, chu kỳ số 1

client = ModbusTcpClient("127.0.0.1", port=5020, timeout=1)
print("Kết nối:", client.connect())

# Đọc 2 holding register từ địa chỉ 0 (= 40001, 40002)
rr = client.read_holding_registers(0, count=2, device_id=1)
print("HR[0..1] =", rr.registers)

# Ghi coil 1 (READY) và đọc lại 4 coil đầu
client.write_coil(1, True, device_id=1)
rr = client.read_coils(0, count=4, device_id=1)
print("Coil[0..3] =", rr.bits[:4], "(nhận về", len(rr.bits), "bit)")

# Ghi một và nhiều holding register (mã hàm 06 và 16)
client.write_register(10, 1, device_id=1)
client.write_registers(11, [42, 500], device_id=1)
print("PLC thấy HR[10..12] =", plc.doc_hr(10, 3))

# Đọc địa chỉ không tồn tại: KHÔNG có exception, mà là một phản hồi lỗi
rr = client.read_holding_registers(100, count=1, device_id=1)
print("Đọc HR 100:", rr.isError(), rr)

Kết nối: True
HR[0..1] = [7, 1]
Coil[0..3] = [False, True, False, False] (nhận về 8 bit)
PLC thấy HR[10..12] = [1, 42, 500]
Đọc HR 100: True ExceptionResponse(dev_id=1, function_code=131, exception_code=2)

Ba điều cần nhớ từ kết quả trên:

• read_coils(0, count=4) trả về 8 bit: các bit được đóng gói theo byte, phần dư là 0. Luôn cắt bits[:count].
• Đọc sai địa chỉ không ném exception: pymodbus trả về ExceptionResponse. Mã hàm 131 = 0x80 + 3 (lỗi của lệnh đọc holding register), exception_code=2 là ILLEGAL DATA ADDRESS. Phải kiểm tra rr.isError() sau mỗi lệnh. Ngược lại, mất kết nối thì có exception (ConnectionException).
• pymodbus tự ghi log lỗi kết nối ra stderr (ví dụ "Connection refused"). Đó là log của thư viện, không phải lỗi của code; trong dự án bạn sẽ chuyển chúng vào file log.

5. Số thực 32 bit qua hai thanh ghi

Một số float32 theo chuẩn IEEE 754 chiếm 4 byte, gọi là A B C D (A chứa bit dấu). Modbus chỉ quy định trong một thanh ghi byte cao đi trước; còn thanh ghi nào chứa nửa nào của số thực thì chuẩn không nói. Vì vậy ngoài thực tế có bốn kiểu:

import struct

gia_tri = 123.456
b = struct.pack(">f", gia_tri)                     # 4 byte IEEE 754, big-endian: A B C D
print("Byte:", b.hex(" ").upper())

def thanh_ghi(bon_byte):
    """Ghép 4 byte thành 2 thanh ghi 16 bit (mỗi thanh ghi big-endian như chuẩn Modbus)."""
    return [int.from_bytes(bon_byte[0:2], "big"), int.from_bytes(bon_byte[2:4], "big")]

A, B, C, D = b
thu_tu = {
    "ABCD (big-endian)":     bytes([A, B, C, D]),
    "CDAB (đảo word)":       bytes([C, D, A, B]),
    "BADC (đảo byte)":       bytes([B, A, D, C]),
    "DCBA (little-endian)":  bytes([D, C, B, A]),
}
for ten, bb in thu_tu.items():
    regs = thanh_ghi(bb)
    hieu_sai = struct.unpack(">f", bb)[0]           # nếu bên nhận cứ đọc như ABCD
    print(f"{ten:<22} thanh ghi = [0x{regs[0]:04X}, 0x{regs[1]:04X}]   đọc như ABCD = {hieu_sai:.6g}")

Byte: 42 F6 E9 79
ABCD (big-endian)      thanh ghi = [0x42F6, 0xE979]   đọc như ABCD = 123.456
CDAB (đảo word)        thanh ghi = [0xE979, 0x42F6]   đọc như ABCD = -1.88337e+25
BADC (đảo byte)        thanh ghi = [0xF642, 0x79E9]   đọc như ABCD = -9.86112e+32
DCBA (little-endian)   thanh ghi = [0x79E9, 0xF642]   đọc như ABCD = 1.5185e+35

Hãy nhìn cột cuối: đọc sai thứ tự không báo lỗi gì, chỉ ra một con số vô lý (hoặc tệ hơn, một con số trông hợp lý nếu giá trị nhỏ). Hai kiểu hay gặp nhất là ABCD (big-endian, nhiều tài liệu gọi là "float, big endian") và CDAB ("word swap", thường gặp ở PLC lưu dữ liệu little-endian theo word). pymodbus hỗ trợ trực tiếp hai kiểu này qua word_order; kiểu đảo byte (BADC, DCBA) phải tự xử lý bằng struct như trên.

F32 = client.DATATYPE.FLOAT32

regs_big = client.convert_to_registers(123.456, F32)                          # ABCD
regs_swap = client.convert_to_registers(123.456, F32, word_order="little")   # CDAB
print("big   :", [hex(r) for r in regs_big])
print("little:", [hex(r) for r in regs_swap])

# Gửi số đo lên HR 12-13 trong MỘT lệnh ghi (mã hàm 16)
client.write_registers(12, regs_big, device_id=1)
tren_plc = plc.doc_hr(12, 2)
print("PLC nhận:", tren_plc)
print("Giải mã đúng thứ tự :", round(client.convert_from_registers(tren_plc, F32), 4))
print("Giải mã sai thứ tự  :", client.convert_from_registers(tren_plc, F32, word_order="little"))

# Kiểm tra chéo bằng struct: không phụ thuộc thư viện
raw = b"".join(r.to_bytes(2, "big") for r in tren_plc)
print("struct >f           :", round(struct.unpack(">f", raw)[0], 4))

client.close()
plc.tat()

big   : ['0x42f6', '0xe979']
little: ['0xe979', '0x42f6']
PLC nhận: [17142, 59769]
Giải mã đúng thứ tự : 123.456
Giải mã sai thứ tự  : -1.8833671448184637e+25
struct >f           : 123.456

Cách thống nhất thứ tự byte khi chạy thử: vision gửi một "giá trị mồi" dễ nhận ra như 123.456 hoặc 1.0 (0x3F800000) vào ô số thực, kỹ sư PLC xem trên màn hình giám sát. Nếu thấy 123.456 là xong; nếu thấy số khổng lồ, thử đổi word_order. Một cách khác được nhiều đội chọn để tránh hẳn vấn đề: gửi số nguyên đã nhân hệ số, ví dụ đường kính 10.004 mm gửi thành 10004 (đơn vị µm) trong một thanh ghi INT16 hoặc hai thanh ghi INT32. PLC nào cũng hiểu số nguyên và dễ hiển thị trên HMI.

6. Bảng địa chỉ cho trạm vision

Bảng dưới đây là "hợp đồng" giữa bạn và kỹ sư PLC, dùng cho phần còn lại của bài và cho dự án cuối chặng. Nguyên tắc vàng: mỗi ô nhớ chỉ có một bên ghi.

HR 11 là chi tiết đáng giá: PLC so HR 11 với số chu kỳ nó đã gửi. Nếu khác, kết quả thuộc về sản phẩm khác, xử lý như NG. Đây là bản mạng của quy tắc "bỏ kết quả muộn" ở bài trước.

# Bảng địa chỉ thống nhất với kỹ sư PLC (địa chỉ giao thức, bắt đầu từ 0)
TRIGGER, READY, BUSY, DONE = 0, 1, 2, 3        # coil 00001..00004
HR_MA_SP, HR_CHU_KY = 0, 1                     # 40001, 40002  (PLC ghi)
HR_KET_QUA = 10                                # 40011: 1 = OK, 2 = NG (vision ghi)
HR_CHU_KY_TRA = 11                             # 40012: vision trả lại số chu kỳ đã xử lý
HR_DO = 12                                     # 40013-40014: đường kính (float32, ABCD)
TB = 1                                         # device_id (Unit ID)

for ten, dc in [("TRIGGER", TRIGGER), ("DONE", DONE)]:
    print(f"coil {ten:<8} địa chỉ {dc} = {dc + 1:05d}")
for ten, dc in [("HR_KET_QUA", HR_KET_QUA), ("HR_DO", HR_DO)]:
    print(f"HR   {ten:<10} địa chỉ {dc} = {40001 + dc}")

coil TRIGGER  địa chỉ 0 = 00001
coil DONE     địa chỉ 3 = 00004
HR   HR_KET_QUA địa chỉ 10 = 40011
HR   HR_DO      địa chỉ 12 = 40013

7. Vòng lặp chờ trigger

Vòng lặp phía vision làm đúng trình tự bắt tay 4 pha, chỉ khác là "đọc dây" thành read_coils và "bật dây" thành write_coils:

1. Đọc coil TRIGGER mỗi 10 ms. Chỉ bắt đầu chu kỳ khi thấy sườn lên.
2. BUSY = 1, đọc mã sản phẩm và số chu kỳ, chụp, xử lý.
3. Ghi kết quả + số chu kỳ + số thực trong một lệnh write_registers; chờ server xác nhận rồi mới bật DONE.
4. Chờ PLC tắt TRIGGER (có timeout), rồi DONE = 0, READY = 1.
5. Mất kết nối: đóng socket, chờ 0,5 s, kết nối lại; sau khi kết nối lại ghi lại trạng thái READY vì PLC có thể đã khởi động lại và mất bộ nhớ.

Để thử phần kết nối lại, chương trình PLC giả sẽ "mất điện" 1 giây sau sản phẩm thứ 2 (tắt server rồi bật lại với bộ nhớ trắng). Code dùng tiếp lớp PlcGia và bảng địa chỉ ở trên.

import threading
import time

import cv2
import numpy as np
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException

MM_MOI_PX = 0.05


def chup_anh_gia(chu_ky):
    """Ảnh giả: một vòng đệm, bán kính thay đổi theo chu kỳ (chu kỳ 3 bị to)."""
    ban_kinh = {3: 104}.get(chu_ky, 100)
    anh = np.full((300, 300), 30, np.uint8)
    cv2.circle(anh, (150, 150), ban_kinh, 220, -1)
    return anh


def do_duong_kinh(anh):
    _, nhi_phan = cv2.threshold(anh, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
    cs, _ = cv2.findContours(nhi_phan, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE)
    (_, _), r = cv2.minEnclosingCircle(max(cs, key=cv2.contourArea))
    return 2 * r * MM_MOI_PX


def kiem(rr):
    """pymodbus trả phản hồi lỗi chứ không ném exception: tự đổi thành exception."""
    if rr.isError():
        raise ModbusException(f"PLC trả lỗi: {rr}")
    return rr


class TramVision:
    def __init__(self, host, port, timeout_trigger_tat=2.0, chu_ky_doc=0.01):
        self.client = ModbusTcpClient(host, port=port, timeout=0.5, retries=1)
        self.timeout_trigger_tat = timeout_trigger_tat
        self.chu_ky_doc = chu_ky_doc
        self.da_xu_ly = 0

    def doc_bit(self, dia_chi):
        return kiem(self.client.read_coils(dia_chi, count=1, device_id=TB)).bits[0]

    def trang_thai(self, ready, busy, done):
        kiem(self.client.write_coils(READY, [ready, busy, done], device_id=TB))  # coil 1..3 một lệnh

    def mot_chu_ky(self):
        self.trang_thai(ready=False, busy=True, done=False)
        ma_sp, chu_ky = kiem(self.client.read_holding_registers(HR_MA_SP, count=2, device_id=TB)).registers
        d = do_duong_kinh(chup_anh_gia(chu_ky))
        ok = 9.90 <= d <= 10.10
        regs = [1 if ok else 2, chu_ky] + self.client.convert_to_registers(d, self.client.DATATYPE.FLOAT32)
        kiem(self.client.write_registers(HR_KET_QUA, regs, device_id=TB))   # 1) kết quả TRƯỚC
        self.trang_thai(ready=False, busy=False, done=True)                  # 2) rồi mới DONE
        han = time.monotonic() + self.timeout_trigger_tat
        trigger = True
        while trigger:                                                       # 3) chờ PLC tắt TRIGGER
            if time.monotonic() > han:
                print("  ! PLC không tắt TRIGGER, tự xóa DONE")
                break
            time.sleep(self.chu_ky_doc)
            trigger = self.doc_bit(TRIGGER)
        self.trang_thai(ready=True, busy=False, done=False)                  # 4) sẵn sàng lại
        self.da_xu_ly += 1
        print(f"Chu kỳ {chu_ky}: SP mã {ma_sp}, d = {d:.3f} mm → {'OK' if ok else 'NG'}")
        return trigger                    # mức TRIGGER thấy lần cuối (thường là False)

    def chay(self, so_chu_ky, thoi_han_s=20):
        trigger_truoc = False
        het_han = time.monotonic() + thoi_han_s
        while self.da_xu_ly < so_chu_ky and time.monotonic() < het_han:
            try:
                if not self.client.connected:
                    if not self.client.connect():
                        time.sleep(0.5)                  # PLC chưa lên: thử lại sau
                        continue
                    print("Đã kết nối PLC, ghi lại READY")
                    self.trang_thai(ready=True, busy=False, done=False)
                    trigger_truoc = False
                trigger = self.doc_bit(TRIGGER)
                if trigger and not trigger_truoc:        # chỉ bắt SƯỜN LÊN
                    trigger = self.mot_chu_ky()          # cập nhật mức đã thấy SAU chu kỳ
                trigger_truoc = trigger
                time.sleep(self.chu_ky_doc)
            except (ModbusException, OSError) as e:      # ConnectionException, socket bị reset...
                print(f"Mất kết nối PLC ({type(e).__name__}), sẽ kết nối lại")
                self.client.close()
                time.sleep(0.5)
        self.client.close()


def chuong_trinh_plc(plc, so_sp, nhat_ky):
    """Logic PLC giả: bật trigger, chờ DONE, đọc kết quả. Sau SP2 thì 'khởi động lại'."""
    def cho(dieu_kien, timeout):
        han = time.monotonic() + timeout
        while not dieu_kien():
            if time.monotonic() > han:
                return False
            time.sleep(0.005)
        return True

    for k in range(1, so_sp + 1):
        if not cho(lambda: plc.doc_coil(READY), 5):
            nhat_ky.append(f"SP{k}: vision không READY")
            continue
        plc.ghi_hr(HR_MA_SP, [7, k])
        plc.ghi_coil(TRIGGER, 1)
        if cho(lambda: plc.doc_coil(DONE), 2.0):
            kq, k_tra, *f = plc.doc_hr(HR_KET_QUA, 4)
            d = ModbusTcpClient.convert_from_registers(f, ModbusTcpClient.DATATYPE.FLOAT32)
            khop = "khớp" if k_tra == k else "LỆCH"
            nhat_ky.append(f"SP{k}: {'OK' if kq == 1 else 'NG'}, d={d:.3f}, số chu kỳ {khop}")
        else:
            nhat_ky.append(f"SP{k}: TIMEOUT → loại")
        plc.ghi_coil(TRIGGER, 0)
        cho(lambda: not plc.doc_coil(DONE), 2.0)
        if k == 2:                                   # mô phỏng PLC mất điện 1 giây
            plc.tat()
            time.sleep(1.0)
            plc.bat()                                # bộ nhớ khởi tạo lại về 0


plc = PlcGia(port=5020)
plc.bat()
nhat_ky_plc = []
t_plc = threading.Thread(target=chuong_trinh_plc, args=(plc, 4, nhat_ky_plc), daemon=True)
t_plc.start()

tram = TramVision("127.0.0.1", 5020)
tram.chay(so_chu_ky=4)
t_plc.join(timeout=5)
plc.tat()
print("Nhật ký PLC:", *nhat_ky_plc, sep="\n  ")

Đã kết nối PLC, ghi lại READY
Chu kỳ 1: SP mã 7, d = 10.000 mm → OK
Chu kỳ 2: SP mã 7, d = 10.000 mm → OK
Mất kết nối PLC (ConnectionException), sẽ kết nối lại
Đã kết nối PLC, ghi lại READY
Chu kỳ 3: SP mã 7, d = 10.400 mm → NG
Chu kỳ 4: SP mã 7, d = 10.000 mm → OK
Nhật ký PLC:
  SP1: OK, d=10.000, số chu kỳ khớp
  SP2: OK, d=10.000, số chu kỳ khớp
  SP3: NG, d=10.400, số chu kỳ khớp
  SP4: OK, d=10.000, số chu kỳ khớp

Chu kỳ 3 có đường kính 10,4 mm nằm ngoài dung sai 10,00 ± 0,10 mm nên NG; nhật ký phía PLC xác nhận nó nhận đúng số thực và số chu kỳ khớp. Khi PLC "mất điện", lệnh đọc kế tiếp ném ConnectionException, vòng lặp thử lại đến khi PLC lên, ghi lại READY và chạy tiếp như chưa có gì xảy ra. Trên stderr bạn sẽ thấy vài dòng log "Connection refused" của pymodbus trong lúc PLC tắt: đó là bình thường.

Một vài con số về thời gian

Mỗi lệnh Modbus TCP trên mạng LAN công nghiệp thường mất cỡ vài ms (phụ thuộc chu kỳ xử lý truyền thông của PLC). Hỏi vòng mỗi 10 ms tức là khoảng 100 yêu cầu mỗi giây: đa số PLC xử lý thoải mái, nhưng đừng hạ xuống 1 ms "cho nhanh" vì PLC có thể phải dành thời gian cho truyền thông thay vì chạy chương trình. Độ trễ phát hiện trigger trung bình bằng nửa chu kỳ hỏi vòng cộng thời gian một lệnh. Nếu ứng dụng cần phản ứng dưới 1 ms (chụp vật đang chạy nhanh), dùng trigger phần cứng vào camera như bài trước, còn Modbus chỉ dùng để trả kết quả.

8. Chạy với PLC thật

• Thay "127.0.0.1", 5020 bằng IP của PLC và cổng 502. Đặt IP tĩnh cho PC vision cùng dải mạng, kiểm tra bằng ping trước.
• Thử bằng một công cụ "Modbus poll / Modbus master" có giao diện trước khi chạy code: nếu công cụ đọc được đúng ô, code của bạn cũng sẽ đọc được.
• Bật Modbus server trên PLC (nhiều PLC phải khai báo khối chức năng hoặc tùy chọn trong phần mềm cấu hình) và hỏi kỹ sư PLC giới hạn số kết nối đồng thời.
• Wireshark với bộ lọc mbtcp cho bạn xem từng gói: mã hàm, địa chỉ, giá trị. Đây là bằng chứng khách quan nhất khi hai bên tranh luận "ai gửi sai".

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

10. Bài tập

from pymodbus.client import ModbusTcpClient

# Tài liệu: "Đường kính: 40101–40102, REAL, word swap" → địa chỉ giao thức 100, word_order="little"
regs_doc_duoc = [0xE979, 0x42F6]                  # giả sử read_holding_registers(100, count=2) trả về
d = ModbusTcpClient.convert_from_registers(regs_doc_duoc, ModbusTcpClient.DATATYPE.FLOAT32,
                                           word_order="little")
print(round(d, 3))

123.456

import struct

regs = [0x79E9, 0xF642]
raw = b"".join(r.to_bytes(2, "big") for r in regs)          # đúng thứ tự trên dây
cach_doc = {
    "ABCD": raw,
    "CDAB": raw[2:4] + raw[0:2],
    "BADC": bytes([raw[1], raw[0], raw[3], raw[2]]),
    "DCBA": raw[::-1],
}
for ten, bb in cach_doc.items():
    print(ten, f"{struct.unpack('>f', bb)[0]:.6g}")

ABCD 1.5185e+35
CDAB -9.86112e+32
BADC -1.88337e+25
DCBA 123.456

Tóm tắt

• Modbus TCP: PC vision là client hỏi vòng, PLC là server; cổng 502; dữ liệu chỉ là bit và thanh ghi 16 bit.
• Bốn vùng: coil (đọc/ghi bit), discrete input (chỉ đọc bit), input register (chỉ đọc), holding register (đọc/ghi). 40001 trong tài liệu = địa chỉ 0 trên dây.
• pymodbus 3.15: device_id= (không còn slave=), server khai báo bằng SimData/SimDevice; luôn kiểm tra isError(), bắt cả ModbusException và OSError khi mất kết nối (socket bị reset giữa chừng ném ConnectionResetError, không phải lỗi của pymodbus).
• float32 = 2 thanh ghi; thống nhất ABCD/CDAB/BADC/DCBA bằng giá trị mồi, kiểm tra bằng struct, ghi trong một lệnh.
• Vòng lặp: bắt sườn lên TRIGGER, ghi kết quả trước rồi DONE, chờ TRIGGER tắt có timeout, trả lại số chu kỳ, kết nối lại và khởi tạo lại tín hiệu sau sự cố.

Bài tiếp theo: OPC UA cho kỹ sư Vision, chuẩn giao tiếp có kiểu dữ liệu và tên biến rõ ràng, so sánh trực tiếp với Modbus TCP bạn vừa học.