modbus-server-nd/modbus_context.py

"""Readonly Modbus server context."""

import logging

from pymodbus.constants import ExcCodes
from pymodbus.datastore import ModbusServerContext

from register_store import RegisterStore

logger = logging.getLogger(__name__)


class ReadonlyHoldingRegisterContext(ModbusServerContext):
    def __init__(self, store: RegisterStore):
        self.simdevices = []
        self.store = store
        self._illegal_address_log_count = 0

    def device_ids(self) -> list[int]:
        return sorted(self.store._registers.keys())

    async def async_getValues(self, device_id: int, func_code: int, address: int, count: int = 1):
        if func_code != 3:
            return ExcCodes.ILLEGAL_FUNCTION
        result = self.store.read_holding_registers(device_id, address, count)
        if result == ExcCodes.ILLEGAL_ADDRESS and self._illegal_address_log_count < 20:
            self._illegal_address_log_count += 1
            logger.warning(
                "Modbus读取非法地址，slave_id=%s，address=%s，count=%s，寄存器概况=%s",
                device_id,
                address,
                count,
                self.store.describe(),
            )
        return result

    async def async_setValues(self, device_id: int, func_code: int, address: int, values: list[int]):
        return ExcCodes.ILLEGAL_FUNCTION