jiateng_ws/utils/modbus_utils.py

from flask import Flask, jsonify, request, make_response
from pymodbus.client import ModbusTcpClient
import time
import logging
from .config_loader import ConfigLoader
# 配置 Flask 日志级别
log = logging.getLogger('werkzeug')
log.setLevel(logging.WARNING)


# Modbus TCP 配置

class ModbusUtils:
    
    def __init__(self, host=None, port=None) -> None:
        # 初始化 modbus 配置 
        config = ConfigLoader.get_instance()
        self.MODBUS_HOST = host if host is not None else config.get_value("modbus.host")
        self.MODBUS_PORT = port if port is not None else int(config.get_value("modbus.port"))
    
    def get_client(self):
        # 创建Modbus TCP客户端实例，指定服务器的IP地址和端口号
        # client = ModbusTcpClient('localhost', port=5020)
        client = ModbusTcpClient(self.MODBUS_HOST, port=self.MODBUS_PORT, timeout=10) # 增加超时时间
        logging.info(f"Attempting to connect to Modbus server {self.MODBUS_HOST}:{self.MODBUS_PORT}")
        try:
            is_connected = client.connect()  #确保客户端已连接
            if is_connected:
                logging.info(f"Successfully connected to Modbus server {self.MODBUS_HOST}:{self.MODBUS_PORT}")
            else:
                logging.error(f"Failed to connect to Modbus server {self.MODBUS_HOST}:{self.MODBUS_PORT}. client.connect() returned False.")
        except Exception as e:
            logging.error(f"Exception during connection to Modbus server {self.MODBUS_HOST}:{self.MODBUS_PORT}: {e}")
            # Optionally, re-raise or handle as appropriate
            return None # Or raise an exception
        return client

    def close_client(self, client):
        # 关闭客户端连接
        if client:
            client.close()

    # 新增十进制转成二进制
    @staticmethod
    def decimal_to_binary(decimal):
        """十进制转16位二进制，右对齐"""
        return format(decimal, '016b')

    # 新增二进制转成十进制  
    @staticmethod
    def binary_to_decimal(binary):
        """二进制字符串转十进制"""
        return int(binary, 2)
    
    # 新增十进制转16进制
    @staticmethod
    def decimal_to_hex(decimal):
        return hex(decimal)[2:]


    @staticmethod
    def split_data(data):
        '''
        解析位置编码获取寄存器地址
        :param data: 位置编码，例如 "01-003-02"
        :return: 寄存器地址
        
        规则说明：
        - column: 库区编号，01表示A库（D100-D108），02表示B库（D110-D118）
        - level: 层号(1-9)，对应寄存器最后一位，如第2层对应D101
        - row: 排号(1-16)，对应寄存器的位索引
        '''
        data_list = data.split('-')
        column = data_list[0]
        row = data_list[1]
        level = data_list[2]
        return column, row, level

    def get_hex_str(self, data):
        """
        解析位置编码获取寄存器地址
        :param data: 位置编码，例如 "01-003-02"
        :return: 寄存器地址
        
        规则说明：
        - column: 库区编号，01表示A库（D100-D108），02表示B库（D110-D118）
        - level: 层号(1-9)，对应寄存器最后一位，如第2层对应D101
        - row: 排号(1-16)，对应寄存器的位索引
        
        示例：
        01-003-02 => D101.2 (A库第2层，第3排)
        02-005-06 => D115.4 (B库第6层，第5排)
        """
        column, row, level = self.split_data(data)
        # 计算基础地址：A库从100开始，B库从110开始
        base_address = 100 + (int(column) - 1) * 10
        # 加上层号得到最终地址
        register_address = base_address + (int(level) - 1)
        return register_address


    @staticmethod
    def check_register_value(client, address, expected_value, timeout_seconds=5):
        """
        校验指定寄存器地址中的数据是否达到期望值
        :param client: ModbusTcp客户端
        :param address: 寄存器地址
        :param expected_value: 期望值
        :param timeout_seconds: 超时时间（秒）
        :return: True表示达到期望值，False表示超时
        """
        try:
            start_time = time.time()
            while True:
                result = client.read_holding_registers(address=address, count=1)
                if not result.isError() and result.registers[0] == expected_value:
                    logging.info(f"寄存器D{address}值: {result.registers[0]} 达到期望值{expected_value}")
                    return True
                
                if time.time() - start_time > timeout_seconds:
                    logging.warning(f"寄存器D{address} {timeout_seconds}秒内未达到期望值{expected_value}")
                    return False
                
                time.sleep(0.2)  # 添加短暂延时，避免过于频繁的读取
        except Exception as e:
            logging.error(f"校验寄存器D{address}值时发生错误: {str(e)}")
            return False

    def write_register_until_success(self, client, address, value, expected_value=None, timeout_seconds=5):
        """
        循环写入寄存器直到成功或超时
        :param client: ModbusTcp客户端
        :param address: 寄存器地址
        :param value: 要写入的值
        :param expected_value: 期望值（如果为None，则使用value作为期望值）
        :param timeout_seconds: 超时时间（秒）
        :return: True表示写入成功，False表示超时
        """
        if expected_value is None:
            expected_value = value
            
        start_time = time.time()
        while True:
            # 写入值
            logging.info(f"写入寄存器D{address}值: {value}")
            client.write_registers(address=address, values=[value])
            logging.info(f"写入寄存器D{address}值: {value}成功")
            # 检查是否达到期望值
            if self.check_register_value(client, address, expected_value):
                return True
                
            if time.time() - start_time > timeout_seconds:
                logging.warning(f"寄存器D{address} {timeout_seconds}秒内写入失败")
                return False
                
            time.sleep(0.2)  # 添加短暂延时，避免过于频繁的写入


    @staticmethod
    def handle_error(error_msg, task_number='', is_emergency_stop=False):
        """通用错误处理函数
        :param error_msg: 错误信息
        :param task_number: 任务编号，默认为空
        :param is_emergency_stop: 是否是急停引起的错误
        """
        logging.error(error_msg)
        pass
        # return response, 500

    # 添加便捷方法，用于读取保持寄存器
    def read_holding_register(self, client, address, count=1):
        """
        读取保持寄存器
        :param client: ModbusTcp客户端
        :param address: 寄存器地址
        :param count: 读取的寄存器数量
        :return: 读取结果，失败返回None
        """
        try:
            result = client.read_holding_registers(address=address, count=count)
            if result.isError():
                logging.error(f"读取寄存器D{address}失败: {result}")
                return None
            return result.registers
        except Exception as e:
            logging.error(f"读取寄存器D{address}时发生错误: {str(e)}")
            return None

    # 添加便捷方法，用于写入保持寄存器
    def write_register(self, client, address, value):
        """
        写入保持寄存器
        :param client: ModbusTcp客户端
        :param address: 寄存器地址
        :param value: 要写入的值
        :return: 是否写入成功
        """
        try:
            result = client.write_registers(address=address, values=[value])
            if result.isError():
                logging.error(f"写入寄存器D{address}值{value}失败: {result}")
                return False
            logging.info(f"写入寄存器D{address}值{value}成功")
            return True
        except Exception as e:
            logging.error(f"写入寄存器D{address}值{value}时发生错误: {str(e)}")
            return False