Пример: Modbus TCP на ODNFC | UnitX - электронные устройства

Здесь собраны две независимые реализации Modbus TCP. Вариант на Lua хранит в регистрах время и UID последней метки. Упрощённый вариант на MicroPython хранит только UID; карта регистров у него отличается.

Шаблон простого Modbus TCP-сервера. Считыватель сохраняет последнюю поднесённую метку: время в Unix-timestamp в регистрах 0–1 и UID в регистрах 2–5. Требуется прошивка 1.6.0 или новее.

local PORT = 502
local UNIT_ID = 1

local wdt_on = Settings.get("sys_wdt") == "on"
local snd = indication.Sound.new()
local leds = indication.Leds.new()

-- обновление holding-регистров: timestamp (4 байта) + UID (до 8 байт)
local function on_uid(uid)
    local uid_bytes = cutils.unhexlify(uid, true)
    if #uid_bytes > 8 then
        eprint("MFRC522: UID does not fit registers 2..5")
        return
    end

    local ts_bytes = os.timestamp("s", "be", true)
    -- очищаем хвост предыдущего UID, затем записываем timestamp и новый UID
    modbus.setregs(0, 0, 0, 0, 0, 0, 0)
    modbus.setregs8(0, ts_bytes .. uid_bytes)
    leds:ok()
    snd:ok()
end

modbus.tcp.init(PORT, UNIT_ID)
leds:start()
snd:start()

local function iteration()
    local callback_ok, err = mfrc522.scan(0, on_uid)
    if callback_ok == false then
        eprint("MFRC522: " .. tostring(err))
        leds:err()
        snd:err()
    end
end

while true do
    if wdt_on then cpu.watchdog.reset() end

    local ok, err = pcall(iteration)
    if not ok then
        local message = tostring(err)
        if message:find("interrupted!", 1, true) then
            error(err, 0)
        end
        eprint("Main loop: " .. message)
        thread.sleepms(250)
    else
        thread.sleepms(50)
    end
end

Клиент на pymodbus читает 6 регистров с адреса 0.

from pymodbus.client import ModbusTcpClient

HOST = "192.168.8.227"
PORT = 502

try:
    # контекстный менеджер сам закроет соединение
    with ModbusTcpClient(HOST, port=PORT) as client:
        if not client.connect():
            raise Exception("Не удалось подключиться к устройству")

        start_address = 0
        count = 6
        response = client.read_holding_registers(address=start_address, count=count)

        if response is not None and not response.isError():
            print("Считанные регистры:", response.registers)
        else:
            print("Ошибка при чтении:", response if response else "Нет ответа от устройства")

except Exception as e:
    print(f"Произошла ошибка: {e}")

Упрощённый Modbus TCP-сервер на MicroPython самостоятельно разбирает заголовок MBAP и поддерживает чтение и запись регистров. UID занимает регистры 0–3; семибайтный UID дополняется нулём до четырёх регистров. При смене карты прежние значения очищаются.

import uasyncio as asyncio
import struct
from rfid_utils import RFID, BUZZ
from machine import WDT
from config import settings

# Коды функций Modbus
READ_HOLDING_REGISTERS = 0x03
WRITE_SINGLE_REGISTER = 0x06
WRITE_MULTIPLE_REGISTERS = 0x10

# Коды ошибок
ILLEGAL_FUNCTION = 0x01
ILLEGAL_DATA_ADDRESS = 0x02
ILLEGAL_DATA_VALUE = 0x03
SERVER_DEVICE_FAILURE = 0x04

MODBUS_REGISTERS = [0, 0, 0, 0, 0, 0, 0, 0]

def uid_to_registers(uid_str, registers):
    # UID должен быть 4 или 7 байт (8 или 14 hex-символов)
    if len(uid_str) not in [8, 14]:
        raise ValueError("UID должен быть 4 или 7 байт (8 или 14 символов в hex)")

    uid_bytes = bytearray()
    for i in range(0, len(uid_str), 2):
        if i + 1 < len(uid_str):
            byte_str = uid_str[i:i+2]
            try:
                byte_val = int(byte_str, 16)
                uid_bytes.append(byte_val)
            except ValueError:
                raise ValueError("UID должен содержать только шестнадцатеричные символы")

    # очищаем UID предыдущей карты
    for i in range(len(registers)):
        registers[i] = 0

    # 7-байтный UID дополняем нулём до чётного числа байтов
    if len(uid_bytes) == 7:
        uid_bytes.append(0)

    # каждый регистр — 2 байта, big-endian
    for i in range(0, len(uid_bytes), 2):
        reg_index = i // 2
        if reg_index < len(registers):
            registers[reg_index] = (uid_bytes[i] << 8) | uid_bytes[i + 1]

    return registers

class ModbusHandler:
    def __init__(self, holding_registers):
        self.holding_registers = holding_registers

    def process_request(self, function_code, data, unit_id):
        """Обработка Modbus-запроса и генерация ответа"""
        try:
            if function_code == READ_HOLDING_REGISTERS:
                return self.handle_read_holding_registers(data)
            elif function_code == WRITE_SINGLE_REGISTER:
                return self.handle_write_single_register(data)
            elif function_code == WRITE_MULTIPLE_REGISTERS:
                return self.handle_write_multiple_registers(data)
            else:
                return struct.pack('BB', function_code | 0x80, ILLEGAL_FUNCTION)
        except (ValueError, TypeError):
            return struct.pack('BB', function_code | 0x80, ILLEGAL_DATA_VALUE)
        except Exception:
            return struct.pack('BB', function_code | 0x80, SERVER_DEVICE_FAILURE)

    def handle_read_holding_registers(self, data):
        """Функция 03 — чтение регистров хранения"""
        if len(data) != 4:
            return struct.pack('BB', READ_HOLDING_REGISTERS | 0x80, ILLEGAL_DATA_VALUE)
        starting_address, quantity = struct.unpack('>HH', data)
        if quantity < 1 or quantity > 125:
            return struct.pack('BB', READ_HOLDING_REGISTERS | 0x80, ILLEGAL_DATA_VALUE)
        if starting_address + quantity > len(self.holding_registers):
            return struct.pack('BB', READ_HOLDING_REGISTERS | 0x80, ILLEGAL_DATA_ADDRESS)
        response = bytearray()
        response.append(READ_HOLDING_REGISTERS)
        response.append(quantity * 2)
        for i in range(quantity):
            register_value = self.holding_registers[starting_address + i]
            response.extend(struct.pack('>H', register_value))
        return response

    def handle_write_single_register(self, data):
        """Функция 06 — запись в один регистр"""
        if len(data) != 4:
            return struct.pack('BB', WRITE_SINGLE_REGISTER | 0x80, ILLEGAL_DATA_VALUE)
        address, value = struct.unpack('>HH', data)
        if address >= len(self.holding_registers):
            return struct.pack('BB', WRITE_SINGLE_REGISTER | 0x80, ILLEGAL_DATA_ADDRESS)
        self.holding_registers[address] = value
        return struct.pack('>BHH', WRITE_SINGLE_REGISTER, address, value)

    def handle_write_multiple_registers(self, data):
        """Функция 16 — запись в несколько регистров"""
        if len(data) < 5:
            return struct.pack('BB', WRITE_MULTIPLE_REGISTERS | 0x80, ILLEGAL_DATA_VALUE)
        starting_address, quantity, byte_count = struct.unpack('>HHB', data[:5])
        if starting_address + quantity > len(self.holding_registers):
            return struct.pack('BB', WRITE_MULTIPLE_REGISTERS | 0x80, ILLEGAL_DATA_ADDRESS)
        if quantity < 1 or quantity > 123 or byte_count != quantity * 2:
            return struct.pack('BB', WRITE_MULTIPLE_REGISTERS | 0x80, ILLEGAL_DATA_VALUE)
        if len(data) != 5 + byte_count:
            return struct.pack('BB', WRITE_MULTIPLE_REGISTERS | 0x80, ILLEGAL_DATA_VALUE)
        for i in range(quantity):
            value = struct.unpack('>H', data[5 + i * 2:7 + i * 2])[0]
            self.holding_registers[starting_address + i] = value
        return struct.pack('>BHH', WRITE_MULTIPLE_REGISTERS, starting_address, quantity)


async def handle(reader, writer):
    modbus_handler = ModbusHandler(MODBUS_REGISTERS)
    print('Новое подключение Modbus')

    while True:
        try:
            # заголовок Modbus TCP (7 байт)
            mbap_header = await read_exactly(reader, 7)
            if not mbap_header:
                print('Клиент разорвал соединение')
                break

            (transaction_id, protocol_id, length, unit_id) = struct.unpack('>HHHB', mbap_header)

            if protocol_id != 0:
                print(f"Неверный ID протокола: {protocol_id}")
                break

            if length < 2 or length > 254:
                print(f"Неверная длина Modbus PDU: {length}")
                break

            data = await read_exactly(reader, length - 1)
            if not data:
                break

            function_code = data[0]
            print(f"Запрос Modbus, функция: {function_code}, адрес: {unit_id}")

            response = modbus_handler.process_request(function_code, data[1:], unit_id)

            response_length = len(response) + 1  # +1 для unit_id
            mbap_response = struct.pack('>HHHB', transaction_id, protocol_id, response_length, unit_id)
            await writer.awrite(mbap_response + response)

        except Exception as e:
            print('Ошибка сервера {} ({})'.format(e, type(e)))
            break

        await asyncio.sleep_ms(50)

    await writer.aclose()
    print('Соединение закрыто')

async def read_exactly(reader, size):
    data = bytearray()
    while len(data) < size:
        chunk = await reader.read(size - len(data))
        if not chunk:
            return None
        data.extend(chunk)
    return data


async def start(host='0.0.0.0', port=502):  # стандартный порт Modbus TCP — 502
    loop = asyncio.get_event_loop()
    loop.create_task(asyncio.start_server(handle, host, port))
    print(f"Modbus TCP сервер запущен на {host}:{port}")


async def run():
    rfid = RFID()
    BUZZ.beeps()
    format_str = settings.get('rfid-format')
    wdt_en = settings.get('sys-i-wdt', False)
    wdt = None

    await start()

    if wdt_en:
        wdt = WDT(timeout=10000)

    while 1:
        if wdt_en:
            wdt.feed()

        uid = rfid.read(fmt=format_str)
        if uid:
            uid_to_registers(uid, MODBUS_REGISTERS)

        await asyncio.sleep_ms(250)

Дополнительные материалы

ODNFC-RS485 Modbus — Modbus RTU по RS485.
Программирование на Lua.

История обновлений ODNFC на Lua

Если установлена версия ниже 1.7.0, сначала обновите её до 1.7.0, а затем до более новой версии. Перед обновлением отключите сторожевой таймер и сотрите пользовательскую программу.

1.8.2

Улучшена работа устройства и исправлены ошибки.
Для устройств, работающих с метками 125 кГц, добавлено обнаружение типовых клонов EM-Marine.

1.8.1

Улучшения:

Новый веб-интерфейс.
Улучшена работа RFID-модулей MFRC и EM-Marine.
Улучшена работа Modbus.
Оптимизирована работа устройства.

1.7.0

Улучшения:

улучшена работа MFRC и EM-Marine;
в раздел «Редактор» добавлен простой файловый менеджер;
улучшена работа сетевых модулей;
оптимизирована работа устройства.

Исправлена ошибка в HTTP-заголовке Content-Length, из-за которой интерфейс не работал в некоторых операционных системах.

1.6.2

Улучшена работа MFRC и EM-Marine.
Улучшена библиотека Modbus.
Оптимизирована работа устройства.
В ODNFC-LAN-LUA добавлена регистрация обработчика успешного чтения.

1.6.1

Добавлена запись в зашифрованные карты MIFARE Classic.
Упрощена инициализация RFID: используются значения по умолчанию для каждой платы, поэтому изменён вызов основного класса и больше не требуются многие Lua-обёртки из rfid.lua.
Добавлена функция mfrc522.scan() — упрощённая версия базового класса для типовых операций.
Исправлены библиотеки EM-Marine и HID.
Оптимизирована библиотека Modbus.
Исправлена работа TCP-сервера.

1.6.0

Улучшения:

обновлён редактор кода: устранена проблема с переводами строк в новых версиях Firefox;
в веб-сервер добавлена поддержка HTTPS и WebSocket;
добавлены новые классы для работы с RFID-метками;
добавлена работа с NDEF и счётчиками Ultralight;
добавлены встроенные библиотеки cbor, cobs, mlib, Heatshrink и microtar;
расширены библиотеки osc, net, cutils, utils, cpu и emmarine;
LuaSocket больше не входит в прошивку;
добавлена загрузка сжатых Heatshrink-библиотек Lua;
добавлен WebSocket REPL для отладки через веб-IDE или расширение VS Code;
улучшена консоль: Unicode, автодополнение и история;
расширена поддержка Modbus;
Paho MQTT заменён собственной реализацией uMqtt;
ускорена работа устройства и веб-интерфейса;
выполнены другие небольшие исправления и оптимизации.

Исправлены ошибки, из-за которых после мягкой перезагрузки Wi-Fi запускался через раз, а RFID-модуль мог не инициализироваться.

1.5.0

Улучшения:

новый интерфейс и редактор кода;
добавлена библиотека tinycobs;
добавлены функции net.tcp.*;
добавлены функции eprint, hprint и wprint для вывода в окно «Инфо»;
трассировка ошибки загрузки теперь показывается в окне «Инфо»;
расширены возможности Telnet, SSH отключён;
в режиме восстановления Wi-Fi работает без пароля;
добавлена поддержка BLE-маяков на устройствах с BLE;
NTP запускается с задержкой 15 секунд, чтобы сеть успела установиться.

Исправлены режим статического Ethernet-адреса и установка яркости NeoPixel.

1.4.0

Улучшен встроенный веб-сервер.
Выполнены небольшие изменения для повышения стабильности.
Исправлена утечка памяти при HTTP-запросах из Windows.

Известная проблема этой версии: режим статического Ethernet-адреса в настройках не работает. Для обхода проблемы в начало пользовательской программы добавляли:

net.en.setup(
  net.packip(Settings.get("net_ip")),
  net.packip(Settings.get("net_mask")),
  net.packip(Settings.get("net_gw")),
  net.packip(Settings.get("net_dns"))
)
net.en.start()

1.3.2

В os.df() добавлен дополнительный параметр.
Выполнены небольшие изменения для повышения стабильности.
Исправлена утечка памяти при запросе метрики в интерфейсе.

Известная проблема этой версии: HTTP-запросы из Windows приводят к утечке памяти. После изменения настроек рекомендовалось закрыть браузер, перезагрузить устройство и не выполнять периодические HTTP-запросы.

1.3.1

Серийный номер синхронизирован с DataMatrix на корпусе модуля, если он есть.

1.3

Библиотека Telegram.lua добавлена в прошивку.
Библиотека LuaSocket теперь распространяется отдельно.