python—socket服务器接收以及数据存储2

import yaml
import os
import asyncio

class ConfigLoader:
    def __init__(self, config_dir="config"):
        self.config_dir = config_dir
        self.server_config = {}
        self.data_config = {}
        self.lock = asyncio.Lock()
    
    async def load_configs(self):
        """异步加载所有配置文件"""
        async with self.lock:
            self.server_config = await self._load_config("server_config.yml")
            self.data_config = await self._load_config("data_config.yml")
    
    async def _load_config(self, filename):
        """异步加载单个配置文件"""
        config_path = os.path.join(self.config_dir, filename)
        if not os.path.exists(config_path):
            raise FileNotFoundError(f"配置文件 {config_path} 不存在")
        
        loop = asyncio.get_running_loop()
        # 在单独线程中执行文件I/O
        config = await loop.run_in_executor(
            None, 
            self._load_yaml, 
            config_path
        )
        return config
    
    def _load_yaml(self, path):
        """同步加载YAML文件"""
        with open(path, 'r') as f:
            return yaml.safe_load(f)
    
    def get_server_config(self):
        """获取服务器配置"""
        return self.server_config.get('server', {})
    
    def get_mysql_config(self):
        """获取MySQL配置"""
        return self.server_config.get('mysql', {})
    
    def get_cache_config(self):
        """获取缓存配置"""
        return self.server_config.get('cache', {})
    
    def get_webhooks(self):
        """获取Webhook配置"""
        return self.server_config.get('webhooks', {})
    
    def get_query_tasks(self):
        """获取查询任务配置"""
        tasks = self.data_config.get('query_tasks', {})
        return tasks
    
    def get_data_points(self):
        """获取数据点配置"""
        return self.data_config.get('data_points', {})
    
    def get_web_port(self):
        """获取Web端口"""
        return self.server_config.get('server', {}).get('web_port', 8080)

import asyncio
import socket
import struct
import logging
from datetime import datetime
from collections import deque
from .config_loader import ConfigLoader
from .db_handler import AsyncDBServer
from .cache_manager import DataCache
from .event_manager import EventManager

class AsyncDTUServer:
    def __init__(self, config_loader, event_manager):
        self.config_loader = config_loader
        self.event_manager = event_manager
        
        # 加载配置
        self.server_config = config_loader.get_server_config()
        self.query_tasks = config_loader.get_query_tasks()
        self.data_points = config_loader.get_data_points()
        
        # 服务器配置
        self.host = self.server_config.get('host', '0.0.0.0')
        self.port = self.server_config.get('main_port', 22334)
        self.heartbeat_timeout = self.server_config.get('heartbeat_timeout', 60)
        self.max_missed = self.server_config.get('max_missed', 3)
        self.max_connections = self.server_config.get('max_connections', 1000)
        self.connection_pool_size = self.server_config.get('connection_pool_size', 100)
        
        # 服务器状态
        self.server = None
        self.connections = {}
        self.connection_pool = deque(maxlen=self.connection_pool_size)
        self.running = False
        
        # 初始化数据库连接
        mysql_config = config_loader.get_mysql_config()
        self.db = AsyncDBServer(mysql_config) if mysql_config else None
        
        # 初始化缓存
        cache_config = config_loader.get_cache_config()
        self.cache = DataCache(
            max_size=cache_config.get('max_size', 10000),
            flush_interval=cache_config.get('flush_interval', 10),
            db_handler=self.db,
            event_manager=event_manager
        )
        
        # 初始化连接池
        self._init_connection_pool()
    
    def _init_connection_pool(self):
        """初始化连接池"""
        for _ in range(self.connection_pool_size):
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.connection_pool.append(sock)
    
    async def start(self):
        """启动服务器"""
        self.running = True
        
        # 创建异步服务器
        self.server = await asyncio.start_server(
            self._handle_client,
            self.host,
            self.port,
            reuse_port=True,
            backlog=self.max_connections
        )
        
        addr = self.server.sockets[0].getsockname()
        logging.info(f"异步DTU服务器启动，监听 {addr[0]}:{addr[1]}")
        logging.info(f"最大连接数: {self.max_connections}, 连接池大小: {self.connection_pool_size}")
        
        # 启动后台任务
        asyncio.create_task(self._monitor_heartbeats())
        asyncio.create_task(self._execute_query_tasks())
        asyncio.create_task(self.cache.start())
        
        # 启动服务器
        async with self.server:
            await self.server.serve_forever()
    
    async def stop(self):
        """停止服务器"""
        self.running = False
        
        # 关闭所有客户端连接
        for device_id, conn in list(self.connections.items()):
            await self._close_connection(device_id)
        
        # 停止缓存
        await self.cache.stop()
        
        # 关闭服务器
        if self.server:
            self.server.close()
            await self.server.wait_closed()
        
        logging.info("DTU服务器已停止")
    
    async def _handle_client(self, reader, writer):
        """处理客户端连接"""
        addr = writer.get_extra_info('peername')
        device_id = None
        
        try:
            # 接收注册包 (假设前4字节是设备ID)
            reg_data = await reader.read(4)
            if len(reg_data) < 4:
                logging.warning(f"无效注册包: {addr[0]}:{addr[1]}")
                writer.close()
                await writer.wait_closed()
                return
            
            # 解析设备ID (4字节整数)
            device_id = struct.unpack('>I', reg_data)[0]
            logging.info(f"设备注册: ID={device_id} ({addr[0]}:{addr[1]})")
            
            # 记录注册包
            await self.cache.add_raw_packet(device_id, 'register', reg_data)
            
            # 添加到连接池
            self.connections[device_id] = {
                'reader': reader,
                'writer': writer,
                'last_heartbeat': asyncio.get_running_loop().time(),
                'missed': 0,
                'address': addr,
                'task_status': self._init_task_status()
            }
            
            # 发送注册确认
            writer.write(b'\x01')
            await writer.drain()
            
            # 触发设备上线事件
            await self.event_manager.trigger_event('device_online', {
                'device_id': device_id,
                'ip': addr[0],
                'port': addr[1]
            })
            
            # 主循环处理数据
            while self.running:
                try:
                    # 接收包头 (2字节: 类型 + 长度)
                    header = await reader.read(2)
                    if len(header) < 2:
                        break
                        
                    pkt_type, pkt_len = struct.unpack('>BB', header)
                    
                    # 接收包体
                    data = b''
                    while len(data) < pkt_len:
                        chunk = await reader.read(pkt_len - len(data))
                        if not chunk:
                            break
                        data += chunk
                    
                    if len(data) != pkt_len:
                        break
                        
                    # 处理数据包
                    if pkt_type == 0x02:  # 心跳包
                        await self._process_heartbeat(device_id)
                    elif pkt_type == 0x03:  # 数据包
                        await self._process_data(device_id, data)
                    elif pkt_type == 0x04:  # Modbus响应
                        await self._process_modbus_response(device_id, data)
                    else:
                        logging.warning(f"未知包类型: 0x{pkt_type:02X} 来自设备 {device_id}")
                        
                except (ConnectionResetError, asyncio.IncompleteReadError):
                    logging.warning(f"连接重置: 设备 {device_id}")
                    break
                except Exception as e:
                    logging.error(f"处理设备 {device_id} 时出错: {str(e)}")
                    break
        
        except Exception as e:
            logging.error(f"处理连接时出错: {str(e)}")
        finally:
            if device_id:
                # 触发设备离线事件
                await self.event_manager.trigger_event('device_offline', {
                    'device_id': device_id,
                    'ip': addr[0],
                    'port': addr[1]
                })
                
                await self._close_connection(device_id)
            logging.info(f"连接关闭: 设备 {device_id if device_id else '未知'}")
    
    def _init_task_status(self):
        """初始化任务状态"""
        task_status = {}
        for task_id in self.query_tasks:
            task_status[task_id] = {
                'last_query': 0,
                'last_response': 0,
                'response_count': 0,
                'error_count': 0
            }
        return task_status
    
    async def _process_heartbeat(self, device_id):
        """处理心跳包"""
        if device_id not in self.connections:
            return
        
        conn = self.connections[device_id]
        conn['last_heartbeat'] = asyncio.get_running_loop().time()
        conn['missed'] = 0
        logging.debug(f"收到心跳: 设备 {device_id}")
        
        # 记录心跳包
        await self.cache.add_raw_packet(device_id, 'heartbeat', b'')
        
        try:
            # 发送心跳响应
            conn['writer'].write(b'\x02')
            await conn['writer'].drain()
        except:
            await self._close_connection(device_id)
            return
        
        # 触发心跳相关的查询任务
        await self._trigger_heartbeat_queries(device_id)
    
    async def _trigger_heartbeat_queries(self, device_id):
        """触发心跳相关的查询任务"""
        if device_id not in self.connections:
            return
        
        # 查找所有心跳触发的查询任务
        for task_id, task_info in self.query_tasks.items():
            trigger_type = task_info.get('trigger', 'timer')
            if trigger_type == "heartbeat":
                await self._send_modbus_query(device_id, task_id)
    
    async def _process_data(self, device_id, data):
        """处理普通数据包"""
        logging.info(f"来自设备 {device_id} 的数据: {data.hex()}")
        
        # 记录原始数据包
        await self.cache.add_raw_packet(device_id, 'data', data)
        
        # 发送确认
        try:
            conn = self.connections[device_id]
            conn['writer'].write(b'\x03')
            await conn['writer'].drain()
        except:
            await self._close_connection(device_id)
    
    async def _process_modbus_response(self, device_id, data):
        """处理Modbus响应"""
        if device_id not in self.connections:
            return
        
        try:
            # 记录Modbus响应包
            await self.cache.add_raw_packet(device_id, 'modbus', data)
            
            # 解析Modbus响应
            if len(data) < 5:
                logging.warning(f"设备 {device_id} 的Modbus响应过短: {data.hex()}")
                return
            
            # 提取Modbus字段
            slave_id = data[0]
            function_code = data[1]
            
            # 检查错误响应
            if function_code & 0x80:  # MODBUS_ERROR_FLAG
                error_code = data[2]
                logging.error(f"设备 {device_id} Modbus错误响应: 功能码 0x{function_code:02X}, 错误码 {error_code}")
                return
            
            # 处理正常响应
            if function_code == 0x03:  # MODBUS_READ_HOLDING
                # 读取保持寄存器响应
                byte_count = data[2]
                reg_data = data[3:3+byte_count]
                
                # 更新任务状态
                conn = self.connections[device_id]
                for task_id, task_info in self.query_tasks.items():
                    start_addr = task_info['start_addr']
                    length = task_info['length']
                    
                    # 检查是否包含数据点
                    for addr, dp_info in self.data_points.items():
                        if start_addr <= addr < start_addr + length:
                            await self._parse_data_point(device_id, addr, reg_data, addr - start_addr)
                
                logging.info(f"设备 {device_id} Modbus数据: {reg_data.hex()}")
                
            elif function_code == 0x10:  # MODBUS_WRITE_MULTIPLE
                # 写多个寄存器响应
                start_addr = struct.unpack('>H', data[2:4])[0]
                reg_count = struct.unpack('>H', data[4:6])[0]
                logging.info(f"设备 {device_id} 写入成功: 地址 {start_addr}, 数量 {reg_count}")
                
            else:
                logging.warning(f"设备 {device_id} 未知Modbus功能码: 0x{function_code:02X}")
        
        except Exception as e:
            logging.error(f"解析设备 {device_id} Modbus响应出错: {str(e)}")
    
    async def _parse_data_point(self, device_id, address, reg_data, offset):
        """解析数据点并保存到缓存"""
        if address not in self.data_points:
            return
        
        dp_config = self.data_points[address]
        name = dp_config['name']
        dtype = dp_config['type']
        length = dp_config['length']
        scale = dp_config.get('scale')
        
        try:
            # 计算数据在响应中的位置 (每个寄存器2字节)
            start_index = offset * 2
            end_index = start_index + length
            
            if end_index > len(reg_data):
                logging.warning(f"设备 {device_id} 数据点 {name} 数据不足")
                return
            
            data_bytes = reg_data[start_index:end_index]
            raw_value = int.from_bytes(data_bytes, 'big')
            
            # 根据数据类型解析
            if dtype == "int":
                if length == 2:  # 16位整数
                    value = struct.unpack('>h', data_bytes)[0]
                elif length == 4:  # 32位整数
                    value = struct.unpack('>i', data_bytes)[0]
                else:
                    logging.warning(f"设备 {device_id} 不支持的长度 {length} 的整数")
                    return
                
                if scale is not None:
                    value = value * scale
                
            elif dtype == "float":
                if length == 4:  # 32位浮点数
                    value = struct.unpack('>f', data_bytes)[0]
                else:
                    logging.warning(f"设备 {device_id} 不支持的长度 {length} 的浮点数")
                    return
            
            else:
                logging.warning(f"设备 {device_id} 未知数据类型: {dtype}")
                return
            
            # 记录解析结果
            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            log_msg = f"{timestamp} 设备 {device_id} {name}: {value}"
            
            if scale is not None and dtype == "int":
                log_msg += f" (原始值: {value/scale})"
            
            logging.info(log_msg)
            
            # 保存到缓存
            await self.cache.add_data_point(device_id, name, value, raw_value)
            
            # 触发数据接收事件
            await self.event_manager.trigger_event('data_received', {
                'device_id': device_id,
                'data_point': name,
                'value': value,
                'raw_value': raw_value
            })
        
        except Exception as e:
            logging.error(f"解析设备 {device_id} 数据点 {name} 出错: {str(e)}")
    
    async def _execute_query_tasks(self):
        """执行定时查询任务"""
        while self.running:
            try:
                current_time = asyncio.get_running_loop().time()
                
                for device_id, conn in list(self.connections.items()):
                    for task_id, task_info in self.query_tasks.items():
                        trigger_type = task_info.get('trigger', 'timer')
                        if trigger_type != "timer":
                            continue
                        
                        interval = task_info.get('interval', 0)
                        if interval <= 0:
                            continue
                        
                        task_status = conn['task_status'][task_id]
                        last_query = task_status['last_query']
                        
                        # 检查是否到了查询时间
                        if current_time - last_query >= interval:
                            await self._send_modbus_query(device_id, task_id)
                            task_status['last_query'] = current_time
                
                # 每1秒检查一次
                await asyncio.sleep(1)
            
            except Exception as e:
                logging.error(f"执行查询任务出错: {str(e)}")
                await asyncio.sleep(5)
    
    async def _send_modbus_query(self, device_id, task_id):
        """发送Modbus查询"""
        if device_id not in self.connections:
            return
        
        task_info = self.query_tasks.get(task_id)
        if not task_info:
            return
        
        start_addr = task_info['start_addr']
        length = task_info['length']
        
        try:
            conn = self.connections[device_id]
            writer = conn['writer']
            
            # 构建Modbus RTU查询 (功能码03: 读保持寄存器)
            slave_id = 0x01  # 从机地址
            func_code = 0x03  # MODBUS_READ_HOLDING
            
            # Modbus PDU: 功能码 + 起始地址 + 寄存器数量
            pdu = struct.pack('>BHH', func_code, start_addr, length)
            
            # RTU帧: 地址 + PDU + CRC
            frame = struct.pack('B', slave_id) + pdu
            crc = self._calculate_crc(frame)
            modbus_frame = frame + crc
            
            # 构建完整的数据包 (类型0x04 + 长度 + Modbus帧)
            pkt_type = 0x04  # Modbus查询包
            pkt_len = len(modbus_frame)
            header = struct.pack('>BB', pkt_type, pkt_len)
            packet = header + modbus_frame
            
            # 发送数据
            writer.write(packet)
            await writer.drain()
            
            logging.info(f"向设备 {device_id} 发送Modbus查询: 任务 {task_id}, 地址 {start_addr}, 长度 {length}")
            
            # 更新任务状态
            conn['task_status'][task_id]['last_query'] = asyncio.get_running_loop().time()
        
        except Exception as e:
            logging.error(f"向设备 {device_id} 发送Modbus查询失败: {str(e)}")
            await self._close_connection(device_id)
    
    def _calculate_crc(self, data):
        """计算Modbus CRC16校验"""
        crc = 0xFFFF
        for byte in data:
            crc ^= byte
            for _ in range(8):
                if crc & 0x0001:
                    crc >>= 1
                    crc ^= 0xA001
                else:
                    crc >>= 1
        return struct.pack('<H', crc)
    
    async def _monitor_heartbeats(self):
        """监控心跳状态"""
        while self.running:
            # 每半超时间隔检查一次
            await asyncio.sleep(self.heartbeat_timeout / 2)
            
            current_time = asyncio.get_running_loop().time()
            devices_to_remove = []
            
            for device_id, conn in list(self.connections.items()):
                elapsed = current_time - conn['last_heartbeat']
                
                if elapsed > self.heartbeat_timeout:
                    conn['missed'] += 1
                    logging.warning(f"设备 {device_id} 丢失心跳 #{conn['missed']}")
                    
                    if conn['missed'] >= self.max_missed:
                        logging.error(f"设备 {device_id} 心跳超时，断开连接")
                        devices_to_remove.append(device_id)
                else:
                    conn['missed'] = 0  # 重置计数器
            
            # 移除超时设备
            for device_id in devices_to_remove:
                await self._close_connection(device_id)
    
    async def _close_connection(self, device_id):
        """关闭连接"""
        if device_id in self.connections:
            conn = self.connections.pop(device_id)
            try:
                conn['writer'].close()
                await conn['writer'].wait_closed()
            except:
                pass
            logging.info(f"设备 {device_id} 从连接列表移除")

import aiomysql
import logging
from datetime import datetime

class AsyncDBServer:
    """异步MySQL数据库管理类"""
    def __init__(self, config):
        self.config = config
        self.pool = None
        self.initialized = False
    
    async def initialize(self):
        """初始化数据库连接池"""
        if self.initialized:
            return
            
        try:
            self.pool = await aiomysql.create_pool(
                host=self.config.get('host'),
                port=self.config.get('port', 3306),
                user=self.config.get('user'),
                password=self.config.get('password'),
                db=self.config.get('database'),
                minsize=1,
                maxsize=self.config.get('pool_size', 10),
                autocommit=True
            )
            self.initialized = True
            logging.info("MySQL异步连接池初始化成功")
            await self._create_tables()
        except Exception as e:
            logging.error(f"创建MySQL异步连接池失败: {str(e)}")
    
    async def _create_tables(self):
        """创建必要的数据库表"""
        if not self.initialized:
            return
            
        try:
            async with self.pool.acquire() as conn:
                async with conn.cursor() as cursor:
                    # 创建设备信息表
                    await cursor.execute("""
                        CREATE TABLE IF NOT EXISTS devices (
                            id INT AUTO_INCREMENT PRIMARY KEY,
                            device_id INT NOT NULL,
                            ip_address VARCHAR(20) NOT NULL,
                            port INT NOT NULL,
                            first_seen DATETIME NOT NULL,
                            last_seen DATETIME NOT NULL,
                            status ENUM('online', 'offline') DEFAULT 'online',
                            UNIQUE KEY unique_device (device_id)
                        )
                    """)
                    
                    # 创建设备数据表
                    await cursor.execute("""
                        CREATE TABLE IF NOT EXISTS device_data (
                            id BIGINT AUTO_INCREMENT PRIMARY KEY,
                            device_id INT NOT NULL,
                            data_point VARCHAR(50) NOT NULL,
                            value FLOAT NOT NULL,
                            raw_value BIGINT,
                            timestamp DATETIME NOT NULL,
                            INDEX idx_device (device_id),
                            INDEX idx_timestamp (timestamp)
                        )
                    """)
                    
                    # 创建连接历史表
                    await cursor.execute("""
                        CREATE TABLE IF NOT EXISTS connection_history (
                            id BIGINT AUTO_INCREMENT PRIMARY KEY,
                            device_id INT NOT NULL,
                            event ENUM('connect', 'disconnect', 'heartbeat_timeout') NOT NULL,
                            timestamp DATETIME NOT NULL,
                            details VARCHAR(255)
                        )
                    """)
                    
                    # 创建心跳记录表
                    await cursor.execute("""
                        CREATE TABLE IF NOT EXISTS heartbeat_log (
                            id BIGINT AUTO_INCREMENT PRIMARY KEY,
                            device_id INT NOT NULL,
                            timestamp DATETIME NOT NULL,
                            INDEX idx_device (device_id),
                            INDEX idx_timestamp (timestamp)
                        )
                    """)
                    
                    # 创建原始数据包表
                    await cursor.execute("""
                        CREATE TABLE IF NOT EXISTS raw_packets (
                            id BIGINT AUTO_INCREMENT PRIMARY KEY,
                            device_id INT NOT NULL,
                            packet_type ENUM('register', 'heartbeat', 'data', 'modbus') NOT NULL,
                            raw_data BLOB,
                            timestamp DATETIME NOT NULL,
                            INDEX idx_device (device_id),
                            INDEX idx_timestamp (timestamp)
                        )
                    """)
            logging.info("数据库表创建/验证成功")
        except Exception as e:
            logging.error(f"创建数据库表失败: {str(e)}")
    
    async def save_data_point(self, device_id, data_point, value, raw_value):
        """保存数据点到数据库"""
        if not self.initialized:
            return False
            
        try:
            async with self.pool.acquire() as conn:
                async with conn.cursor() as cursor:
                    await cursor.execute("""
                        INSERT INTO device_data 
                        (device_id, data_point, value, raw_value, timestamp)
                        VALUES (%s, %s, %s, %s, %s)
                    """, (device_id, data_point, value, raw_value, datetime.now()))
            return True
        except Exception as e:
            logging.error(f"保存数据点失败: {str(e)}")
            return False
    
    async def save_raw_packet(self, device_id, packet_type, raw_data):
        """保存原始数据包到数据库"""
        if not self.initialized:
            return False
            
        try:
            async with self.pool.acquire() as conn:
                async with conn.cursor() as cursor:
                    await cursor.execute("""
                        INSERT INTO raw_packets 
                        (device_id, packet_type, raw_data, timestamp)
                        VALUES (%s, %s, %s, %s)
                    """, (device_id, packet_type, raw_data, datetime.now()))
            return True
        except Exception as e:
            logging.error(f"保存原始数据包失败: {str(e)}")
            return False
    
    async def get_device_data(self, device_id, limit=100):
        """获取设备数据"""
        if not self.initialized:
            return []
            
        try:
            async with self.pool.acquire() as conn:
                async with conn.cursor(aiomysql.DictCursor) as cursor:
                    await cursor.execute("""
                        SELECT * FROM device_data 
                        WHERE device_id = %s
                        ORDER BY timestamp DESC
                        LIMIT %s
                    """, (device_id, limit))
                    return await cursor.fetchall()
        except Exception as e:
            logging.error(f"获取设备数据失败: {str(e)}")
            return []
    
    async def export_data_to_csv(self, device_id, start_time, end_time):
        """导出数据到CSV"""
        if not self.initialized:
            return None
            
        try:
            async with self.pool.acquire() as conn:
                async with conn.cursor() as cursor:
                    await cursor.execute("""
                        SELECT 
                            timestamp,
                            data_point,
                            value,
                            raw_value
                        FROM device_data
                        WHERE device_id = %s
                        AND timestamp BETWEEN %s AND %s
                        ORDER BY timestamp
                    """, (device_id, start_time, end_time))
                    
                    # 构建CSV内容
                    csv_content = "timestamp,data_point,value,raw_value\n"
                    async for row in cursor:
                        csv_content += f"{row[0]},{row[1]},{row[2]},{row[3]}\n"
                    
                    return csv_content
        except Exception as e:
            logging.error(f"导出数据到CSV失败: {str(e)}")
            return None
    
    async def close(self):
        """关闭连接池"""
        if self.pool:
            self.pool.close()
            await self.pool.wait_closed()
            logging.info("MySQL连接池已关闭")

import asyncio
import logging
from collections import deque
from datetime import datetime

class DataCache:
    """异步数据缓存管理类"""
    def __init__(self, max_size=10000, flush_interval=10, db_handler=None, event_manager=None):
        self.max_size = max_size
        self.flush_interval = flush_interval
        self.db_handler = db_handler
        self.event_manager = event_manager
        self.data_points = deque(maxlen=max_size)
        self.raw_packets = deque(maxlen=max_size)
        self.running = False
        self.lock = asyncio.Lock()
    
    async def start(self):
        """启动缓存刷新任务"""
        if self.running:
            return
            
        self.running = True
        asyncio.create_task(self._flush_task())
        logging.info("数据缓存刷新任务已启动")
    
    async def stop(self):
        """停止缓存刷新任务"""
        self.running = False
        await self.flush_all()
        logging.info("数据缓存刷新任务已停止")
    
    async def add_data_point(self, device_id, data_point, value, raw_value):
        """添加数据点到缓存"""
        async with self.lock:
            self.data_points.append({
                'device_id': device_id,
                'data_point': data_point,
                'value': value,
                'raw_value': raw_value,
                'timestamp': datetime.now()
            })
    
    async def add_raw_packet(self, device_id, packet_type, raw_data):
        """添加原始数据包到缓存"""
        async with self.lock:
            self.raw_packets.append({
                'device_id': device_id,
                'packet_type': packet_type,
                'raw_data': raw_data,
                'timestamp': datetime.now()
            })
    
    async def flush_all(self):
        """立即刷新所有缓存数据到数据库"""
        async with self.lock:
            # 刷新数据点
            while self.data_points:
                item = self.data_points.popleft()
                if self.db_handler:
                    await self.db_handler.save_data_point(
                        item['device_id'],
                        item['data_point'],
                        item['value'],
                        item['raw_value']
                    )
            
            # 刷新原始数据包
            while self.raw_packets:
                item = self.raw_packets.popleft()
                if self.db_handler:
                    await self.db_handler.save_raw_packet(
                        item['device_id'],
                        item['packet_type'],
                        item['raw_data']
                    )
    
    async def _flush_task(self):
        """定时刷新缓存数据到数据库"""
        while self.running:
            await asyncio.sleep(self.flush_interval)
            await self.flush_all()
            logging.debug(f"缓存数据已刷新，剩余数据点: {len(self.data_points)}，原始包: {len(self.raw_packets)}")

import aiohttp
import asyncio
import logging
from typing import Dict, Callable, Any

class EventManager:
    """异步事件管理框架"""
    def __init__(self):
        self.event_handlers: Dict[str, list[Callable]] = {}
        self.webhooks: Dict[str, str] = {}
    
    def register_webhooks(self, webhooks: Dict[str, str]):
        """注册Webhook配置"""
        self.webhooks = webhooks
    
    def register_event(self, event_name: str, handler: Callable):
        """注册事件处理函数"""
        if event_name not in self.event_handlers:
            self.event_handlers[event_name] = []
        self.event_handlers[event_name].append(handler)
    
    async def trigger_event(self, event_name: str, data: Any):
        """触发事件"""
        # 调用注册的事件处理函数
        if event_name in self.event_handlers:
            for handler in self.event_handlers[event_name]:
                try:
                    if asyncio.iscoroutinefunction(handler):
                        await handler(event_name, data)
                    else:
                        handler(event_name, data)
                except Exception as e:
                    logging.error(f"事件处理函数执行失败: {event_name} - {str(e)}")
        
        # 调用Webhook
        if event_name in self.webhooks:
            webhook_url = self.webhooks[event_name]
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.post(webhook_url, json=data) as response:
                        if response.status >= 400:
                            logging.warning(f"Webhook调用失败: {event_name} - HTTP {response.status}")
            except Exception as e:
                logging.error(f"Webhook调用失败: {event_name} - {str(e)}")
    
    async def call_webhook(self, event_name: str, data: Any):
        """直接调用Webhook"""
        if event_name in self.webhooks:
            webhook_url = self.webhooks[event_name]
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.post(webhook_url, json=data) as response:
                        if response.status >= 400:
                            logging.warning(f"Webhook调用失败: {event_name} - HTTP {response.status}")
                        return response.status
            except Exception as e:
                logging.error(f"Webhook调用失败: {event_name} - {str(e)}")
                return None
        return None

from aiohttp import web
import aiohttp_jinja2
import jinja2
import logging
import csv
import io
from datetime import datetime, timedelta
from .db_handler import AsyncDBServer
from .config_loader import ConfigLoader
from .event_manager import EventManager

async def init_web_app(db_handler, config_loader, event_manager):
    """初始化Web应用"""
    app = web.Application()
    
    # 设置Jinja2模板
    aiohttp_jinja2.setup(
        app, 
        loader=jinja2.FileSystemLoader('templates')
    )
    
    # 添加路由
    app.add_routes([
        web.get('/', index),
        web.get('/devices', devices),
        web.get('/device/{device_id}', device_detail),
        web.get('/data', data),
        web.get('/export', export_data),
        web.get('/connections', connections),
        web.get('/api/devices', api_devices),
        web.get('/api/device/{device_id}/data', api_device_data),
        web.get('/api/raw_packets', api_raw_packets),
        web.post('/webhook/test', test_webhook),
    ])
    
    # 添加共享数据
    app['db'] = db_handler
    app['config_loader'] = config_loader
    app['event_manager'] = event_manager
    
    return app

@aiohttp_jinja2.template('index.html')
async def index(request):
    """首页仪表盘"""
    return {}

@aiohttp_jinja2.template('devices.html')
async def devices(request):
    """设备列表页面"""
    db = request.app['db']
    online_devices = await db.get_connected_devices() if db else []
    return {'devices': online_devices}

@aiohttp_jinja2.template('device_detail.html')
async def device_detail(request):
    """设备详情页面"""
    device_id = int(request.match_info['device_id'])
    db = request.app['db']
    
    device_data = await db.get_device_data(device_id, 100) if db else []
    raw_packets = await db.get_raw_packets(device_id, 50) if db else []
    
    return {
        'device_id': device_id,
        'device_data': device_data,
        'raw_packets': raw_packets
    }

@aiohttp_jinja2.template('data.html')
async def data(request):
    """数据浏览页面"""
    device_id = request.query.get('device_id')
    limit = int(request.query.get('limit', 100))
    
    if device_id:
        device_id = int(device_id)
        db = request.app['db']
        data_points = await db.get_device_data(device_id, limit) if db else []
    else:
        data_points = []
    
    return {'data_points': data_points, 'device_id': device_id}

async def export_data(request):
    """导出数据"""
    device_id = int(request.query.get('device_id', 0))
    if not device_id:
        return web.Response(text="需要提供device_id参数", status=400)
    
    # 默认导出最近24小时数据
    end_time = datetime.now()
    start_time = end_time - timedelta(days=1)
    
    db = request.app['db']
    csv_content = await db.export_data_to_csv(device_id, start_time, end_time)
    
    if csv_content is None:
        return web.Response(text="导出数据失败", status=500)
    
    # 创建CSV响应
    response = web.StreamResponse(
        status=200,
        reason='OK',
        headers={
            'Content-Type': 'text/csv',
            'Content-Disposition': f'attachment; filename="device_{device_id}_data.csv"'
        }
    )
    
    await response.prepare(request)
    await response.write(csv_content.encode('utf-8'))
    await response.write_eof()
    
    return response

@aiohttp_jinja2.template('connections.html')
async def connections(request):
    """连接历史页面"""
    db = request.app['db']
    connections = await db.get_connection_history(100) if db else []
    return {'connections': connections}

async def api_devices(request):
    """API: 获取设备列表"""
    db = request.app['db']
    devices = await db.get_connected_devices() if db else []
    return web.json_response(devices)

async def api_device_data(request):
    """API: 获取设备数据"""
    device_id = int(request.match_info['device_id'])
    limit = int(request.query.get('limit', 100))
    
    db = request.app['db']
    data_points = await db.get_device_data(device_id, limit) if db else []
    return web.json_response(data_points)

async def api_raw_packets(request):
    """API: 获取原始数据包"""
    device_id = request.query.get('device_id')
    limit = int(request.query.get('limit', 50))
    
    if device_id:
        device_id = int(device_id)
    else:
        device_id = None
    
    db = request.app['db']
    raw_packets = await db.get_raw_packets(device_id, limit) if db else []
    return web.json_response(raw_packets)

async def test_webhook(request):
    """测试Webhook"""
    event_manager = request.app['event_manager']
    data = await request.json()
    event_type = data.get('event_type')
    test_data = data.get('data', {})
    
    if not event_type:
        return web.json_response({'status': 'error', 'message': 'Missing event_type'}, status=400)
    
    status = await event_manager.call_webhook(event_type, test_data)
    if status is None:
        return web.json_response({'status': 'error', 'message': 'Webhook not configured'}, status=400)
    
    return web.json_response({'status': 'success', 'http_status': status})

async def start_web_interface(config_loader, event_manager):
    """启动Web界面"""
    # 初始化数据库连接
    mysql_config = config_loader.get_mysql_config()
    db_handler = AsyncDBServer(mysql_config)
    await db_handler.initialize()
    
    # 初始化Web应用
    app = await init_web_app(db_handler, config_loader, event_manager)
    web_port = config_loader.get_web_port()
    
    # 启动Web服务器
    runner = web.AppRunner(app)
    await runner.setup()
    
    site = web.TCPSite(runner, '0.0.0.0', web_port)
    await site.start()
    
    logging.info(f"Web界面已启动，访问 http://localhost:{web_port}")
    
    # 保持运行
    while True:
        await asyncio.sleep(3600)  # 每小时检查一次

import asyncio
import logging
from config_loader import ConfigLoader
from async_dtu_server import AsyncDTUServer
from web_interface import start_web_interface
from event_manager import EventManager

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("dtu_server.log"),
        logging.StreamHandler()
    ]
)

async def main():
    # 加载配置
    config_loader = ConfigLoader("config")
    await config_loader.load_configs()
    
    # 初始化事件管理器
    event_manager = EventManager()
    event_manager.register_webhooks(config_loader.get_webhooks())
    
    # 启动DTU服务器
    dtu_server = AsyncDTUServer(config_loader, event_manager)
    
    # 启动Web界面（在单独任务中）
    web_task = asyncio.create_task(
        start_web_interface(config_loader, event_manager)
    )
    
    try:
        # 打印配置信息
        logging.info("服务器配置:")
        logging.info(f"  主机: {dtu_server.host}")
        logging.info(f"  主端口: {dtu_server.port}")
        logging.info(f"  Web端口: {config_loader.get_web_port()}")
        logging.info(f"  最大连接数: {dtu_server.max_connections}")
        logging.info(f"  连接池大小: {dtu_server.connection_pool_size}")
        
        # 打印查询任务信息
        logging.info("查询任务配置:")
        for task_id, task_info in dtu_server.query_tasks.items():
            logging.info(f"  任务 {task_id}: 地址 {task_info['start_addr']}-{task_info['start_addr']+task_info['length']-1}, "
                        f"长度 {task_info['length']} 寄存器, "
                        f"间隔 {task_info['interval'] if task_info['interval'] > 0 else '心跳触发'}秒, "
                        f"触发方式: {task_info['trigger']}")
        
        # 打印数据点信息
        logging.info("数据点配置:")
        for addr, dp_info in dtu_server.data_points.items():
            logging.info(f"  地址 {addr}: {dp_info['name']} ({dp_info['type']}, {dp_info['length']}字节, "
                        f"精度: {'原始值' if dp_info.get('scale') is None else dp_info['scale']})")
        
        # 启动DTU服务器
        await dtu_server.start()
        
    except asyncio.CancelledError:
        pass
    except KeyboardInterrupt:
        await dtu_server.stop()
        await web_task
        logging.info("服务器已停止")

if __name__ == "__main__":
    asyncio.run(main())