modbus_client.py

import asyncio
import logging
from pyModbusTCP.client import ModbusClient
from typing import Dict, Any

logging.basicConfig(level=logging.INFO)
_logger = logging.getLogger(__name__)

REGISTER_SPECS = {
    "DiscreteInputs": {
        "alias": ["discrete-input:", "1", "1x"],
        "description": "Boolean input value, usually representing a binary input to the PLC",
        "bit_size": 1,
        "permissions": "Read Only",
        "start_address": 10001,
        "read_func": "read_discrete_inputs",
        "write_func": None
    },
    "Coils": {
        "alias": ["coil:", "0", "0x"],
        "description": "Boolean value, usually representing a binary output from the PLC",
        "bit_size": 1,
        "permissions": "Read/Write",
        "start_address": 1,
        "read_func": "read_coils",
        "write_func": "write_single_coil"
    },
    "InputRegisters": {
        "alias": ["input-register:", "3", "3x"],
        "description": "Short input value, usually representing an analog input to the PLC",
        "bit_size": 16,
        "permissions": "Read Only",
        "start_address": 30001,
        "read_func": "read_input_registers",
        "write_func": None
    },
    "HoldingRegisters": {
        "alias": ["holding-register:", "4", "4x"],
        "description": "Short value, usually representing an analog output from the PLC",
        "bit_size": 16,
        "permissions": "Read/Write",
        "start_address": 40001,
        "read_func": "read_holding_registers",
        "write_func": "write_single_register"
    }
}

class ModbusScanner:
    def __init__(self, host: str, port: int = 502, unit_id: int = 1):
        """
        Initializes the Modbus client with the specified host, port, and unit ID.
        """
        self.client = ModbusClient(
            host=host,
            port=port,
            unit_id=unit_id,
            auto_open=True,
            timeout=5
        )
        self.data_dict = {
            "Server": {
                "NamespaceIndex": 0,
                "NodeId": "Server",
                "QualifiedName": "Server",
                "children": {}
            }
        }

    async def scan_registers(self) -> Dict[str, Any]:
        """
        Asynchronously scans all register types and returns a hierarchical dictionary
        of active registers with their addresses and values.
        """
        tasks = []
        for reg_type in REGISTER_SPECS.keys():
            tasks.append(self._scan_register_type(reg_type))
        await asyncio.gather(*tasks)
        return self.data_dict

    async def _scan_register_type(self, reg_type: str):
        """
        Asynchronously scans a specific register type and updates the data_dict.
        """
        specs = REGISTER_SPECS[reg_type]
        aliases = specs["alias"]
        description = specs["description"]
        bit_size = specs["bit_size"]
        permissions = specs["permissions"]
        start_addr = specs["start_address"]
        read_func_name = specs["read_func"]

        read_func = getattr(self.client, read_func_name, None)
        if not callable(read_func):
            _logger.error(f"Read function {read_func_name} not implemented for {reg_type}.")
            return

        _logger.info(f"Scanning {reg_type} starting at address {start_addr}")

        if reg_type not in self.data_dict["Server"]["children"]:
            self.data_dict["Server"]["children"][reg_type] = {
                "NamespaceIndex": 0,
                "NodeId": reg_type,
                "QualifiedName": reg_type,
                "children": {}
            }

        end_addr = self._get_end_address(reg_type)

        batch_size = 100
        for addr in range(start_addr, end_addr, batch_size):
            try:
                count = min(batch_size, end_addr - addr)
                _logger.debug(f"Reading {reg_type} from {addr} to {addr + count -1}")
                response = await self.read_registers_async(read_func, addr, count)
                if response is not None:
                    for i, value in enumerate(response):
                        address = addr + i
                        self.add_to_data_dict(
                            reg_type, address, value, aliases, bit_size, description, permissions
                        )
            except Exception as e:
                _logger.error(f"Error reading {reg_type} at address {addr}: {e}")
                continue

    async def read_registers_async(self, read_func, addr: int, count: int):
        """
        Asynchronously reads registers using the provided read function.
        """
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, read_func, addr, count)

    def add_to_data_dict(self, reg_type, address, value, aliases, bit_size, description, permissions):
        """
        Adds the active node to the hierarchical data dictionary.
        """
        alias_prefix = aliases[0]
        node_id = f"{alias_prefix}{address}[{bit_size}]"
        register_name = f"{reg_type}_{address}"
        node_info = {
            "NamespaceIndex": 0,
            "NodeId": node_id,
            "QualifiedName": register_name,
            "children": {},
            "Value": value,
            "Description": description,
            "Permissions": permissions
        }

        self.data_dict["Server"]["children"][reg_type]["children"][register_name] = node_info
        _logger.debug(f"Active node added: {register_name} -> {node_info}")

    def _get_end_address(self, reg_type: str) -> int:
        """
        Determines the end address based on the register type.
        """
        if reg_type == "Coils":
            return 1000
        elif reg_type == "DiscreteInputs":
            return 11000
        elif reg_type == "InputRegisters":
            return 31000
        elif reg_type == "HoldingRegisters":
            return 41000
        else:
            return REGISTER_SPECS[reg_type]["start_address"] + 1000

    def close(self):
        """
        Closes the Modbus client connection.
        """
        self.client.close()

if __name__ == "__main__":
    import json

    HOST = "10.1.10.93"
    PORT = 502
    UNIT_ID = 1

    async def main():
        scanner = ModbusScanner(host=HOST, port=PORT, unit_id=UNIT_ID)
        try:
            data_dict = await scanner.scan_registers()
            if data_dict:
                with open('data_dict_modbus.json', 'w') as f:
                    json.dump(data_dict, f, indent=4)
                print("Active nodes found and saved to 'data_dict_modbus.json'.")
            else:
                print("No active nodes found.")
        finally:
            scanner.close()

    asyncio.run(main())