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
import threading
# 配置 Flask 日志级别
log = logging.getLogger('werkzeug')
log.setLevel(logging.WARNING)


# 添加连接池管理类
class ModbusConnectionPool:
    _instance = None
    _lock = threading.Lock()
    
    @classmethod
    def get_instance(cls):
        """获取单例实例"""
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = ModbusConnectionPool()
        return cls._instance
    
    def __init__(self):
        """初始化连接池"""
        self._connections = {}  # 保存每个目标地址的连接
        self._locks = {}        # 每个连接的锁，避免并发问题
        self._in_use = {}       # 标记连接是否在使用中
        self._last_used = {}    # 记录连接最后使用时间
        
        # 启动清理线程
        self._cleanup_thread = threading.Thread(target=self._cleanup_idle_connections, daemon=True)
        self._cleanup_thread.start()
    
    def get_connection(self, host, port):
        """获取指定地址的连接，如果不存在则创建"""
        key = f"{host}:{port}"
        
        # 如果不存在该地址的锁，创建一个
        if key not in self._locks:
            with self._lock:
                if key not in self._locks:
                    self._locks[key] = threading.Lock()
        
        # 获取该地址的锁
        with self._locks[key]:
            # 如果连接不存在或已断开，创建新连接
            if key not in self._connections or not self._connections[key].is_socket_open():
                try:
                    logging.info(f"Creating new Modbus connection to {host}:{port}")
                    client = ModbusTcpClient(host, port=port, timeout=10)
                    connected = client.connect()
                    if not connected:
                        logging.error(f"Failed to connect to {host}:{port}")
                        return None
                    self._connections[key] = client
                    self._in_use[key] = True
                    self._last_used[key] = time.time()
                    return client
                except Exception as e:
                    logging.error(f"Error creating Modbus connection to {host}:{port}: {e}")
                    return None
            else:
                # 连接存在，标记为使用中并更新时间
                self._in_use[key] = True
                self._last_used[key] = time.time()
                return self._connections[key]
    
    def release_connection(self, host, port):
        """释放连接，将其标记为不再使用"""
        key = f"{host}:{port}"
        with self._locks.get(key, self._lock):
            if key in self._in_use:
                self._in_use[key] = False
                self._last_used[key] = time.time()
    
    def _cleanup_idle_connections(self):
        """清理空闲连接的后台线程"""
        idle_timeout = 300  # 5分钟空闲超时
        while True:
            time.sleep(60)  # 每分钟检查一次
            current_time = time.time()
            keys_to_check = list(self._connections.keys())
            
            for key in keys_to_check:
                try:
                    with self._locks.get(key, self._lock):
                        # 如果连接空闲超过5分钟，关闭并移除
                        if (key in self._in_use and not self._in_use[key] and 
                            current_time - self._last_used.get(key, 0) > idle_timeout):
                            logging.info(f"Closing idle connection: {key}")
                            try:
                                if self._connections[key].is_socket_open():
                                    self._connections[key].close()
                            except:
                                pass  # 忽略关闭连接时的错误
                            
                            # 从所有字典中移除
                            self._connections.pop(key, None)
                            self._in_use.pop(key, None)
                            self._last_used.pop(key, None)
                except Exception as e:
                    logging.error(f"Error during connection cleanup for {key}: {e}")


# 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"))
        self.connection_pool = ModbusConnectionPool.get_instance()
        self.retry_max = 3  # 最大重试次数
        self.retry_delay = 0.5  # 初始重试延迟（秒）
    
    def get_client(self):
        """获取Modbus客户端连接（使用连接池）"""
        # 使用指数退避重试获取连接
        retry_count = 0
        delay = self.retry_delay
        
        while retry_count < self.retry_max:
            client = self.connection_pool.get_connection(self.MODBUS_HOST, self.MODBUS_PORT)
            if client and client.is_socket_open():
                return client
                
            # 连接失败，等待后重试
            logging.warning(f"Connection attempt {retry_count+1} failed, retrying in {delay:.1f}s...")
            time.sleep(delay)
            delay *= 2  # 指数退避
            retry_count += 1
        
        logging.error(f"Failed to connect to Modbus server {self.MODBUS_HOST}:{self.MODBUS_PORT} after {self.retry_max} attempts")
        return None

    def close_client(self, client):
        """释放客户端连接回连接池，不再实际关闭连接"""
        if client:
            self.connection_pool.release_connection(self.MODBUS_HOST, self.MODBUS_PORT)

    # 新增十进制转成二进制
    @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