""" Tan mingyan 2026/3/9 AD 驱动示例（bus_ad.py）—— 补充参考 """ # midware/drivers/bus_can.py from .base_driver import BaseBusDriver from typing import Optional, Dict, Any from loguru import logger import time # 导入全局配置字典（需确保main.py已提前加载CSV配置） from midware.config.base_config import DEVICE_CONFIG_DICT, get_device_config import midware.config.base_config as base_config from .base_driver import BaseBusDriver from .renode_agent import renode from midware.config.base_config import DEVICE_CONFIG_DICT class BusADDriver(BaseBusDriver): """uart总线驱动实现（适配）""" def _init_config(self): self.config.setdefault("channel", 0) # CAN通道 self.config.setdefault("baudrate", 500000) # 500kbps self.config.setdefault("frame_format", "standard") # 标准帧/扩展帧 logger.info(f"UART驱动配置初始化完成：{self.config}") def connect(self) -> bool: try: # 实际逻辑：打开CAN通道 time.sleep(0.05) self.is_connected = True logger.info("CAN总线连接成功") return True except Exception as e: logger.error(f"CAN总线连接失败：{str(e)}") return False def disconnect(self) -> bool: try: self.is_connected = False logger.info("CAN总线断开成功") return True except Exception as e: logger.error(f"CAN总线断开失败：{str(e)}") return False def send(self, data: bytes, **kwargs) -> bool: if not self.is_connected: logger.error("CAN驱动未连接，发送失败") return False ''' try: frame_id = kwargs.get("frame_id", 0x123) # CAN帧ID is_extended = kwargs.get("is_extended", False) # 是否扩展帧 logger.info( f"CAN发送数据｜帧ID：0x{frame_id:X}｜扩展帧：{is_extended}｜" f"数据：{data.hex()}｜长度：{len(data)}字节" ) # 实际逻辑：调用AD驱动发送接口 # 获取优先级，设备参数，地址等参数 return True except Exception as e: logger.error(f"CAN数据发送失败：{str(e)}") return False ''' # ========== 核心步骤1：从kwargs提取设备名称 ========== #dev_name = kwargs.get("dev_name")[0] dev_name = kwargs.get("dev_name")#2026/4/10 if not dev_name: logger.error("AD发送失败：kwargs中未传入dev_name（设备名称）") return False # ========== 核心步骤2：从全局配置字典查找设备参数 ========== # 检查全局配置是否加载 if not base_config.DEVICE_CONFIG_DICT: logger.error("AD发送失败：全局配置DEVICE_CONFIG_DICT未加载") return False # 查找当前设备的配置 #device_config = base_config.DEVICE_CONFIG_DICT.get(dev_name) device_config = get_device_config().get(dev_name) if not device_config: logger.error(f"AD发送失败：设备 {dev_name} 未在DEVICE_CONFIG_DICT中配置") return False # 提取关键配置项（兼容配置项缺失的情况） try: # 内存基地址（转为16进制整数，如"0x80000000" → 0x80000000） # mem_base_addr = int(str(device_config.get("base_addr", "0x0")), 16)#mem_base_addr = int(device_config.get("base_addr", "0x0"), 16)#内存基地址 mem_base_addr = int(device_config.get("base_addr", 0)) # 内存偏移地址 #mem_offset = int(str(device_config.get("offset_addr", "0x0")), 16)#内存偏移地址 mem_offset = int(device_config.get("offset_addr", 0)) # 发送缓存区大小（字节） send_buffer_size = int(str(device_config.get("send_cache", 1024)))#发送缓存区大小 # 接收缓存区大小（字节） recv_buffer_size = int(device_config.get("recv_cache", 1024))#接收缓存区大小 except (ValueError, TypeError) as e: logger.error(f"AD发送失败：{dev_name} 配置参数解析错误 → {str(e)}") return False # ========== 核心步骤3：使用配置项处理发送逻辑 ========== # 1. 校验数据长度不超过发送缓存区大小 if len(data) > send_buffer_size: logger.error( f"AD发送失败：{dev_name} 数据长度 {len(data)} 超过发送缓存区大小 {send_buffer_size}" ) return False # 2. 模拟AD指令发送（结合内存地址） logger.info( f"AD发送指令｜设备：{dev_name}｜" f"内存基地址：0x{mem_base_addr:X}｜偏移地址：0x{mem_offset:X}｜" f"发送缓存区：{send_buffer_size}B｜接收缓存区：{recv_buffer_size}B｜" f"数据：{data.hex()}｜长度：{len(data)}字节" ) # 实际业务逻辑： # - 将数据写入指定内存地址（mem_base_addr + mem_offset） # - 配置接收缓存区大小 # - 调用AD采集卡SDK发送指令 # ad_card.write_mem(mem_base_addr + mem_offset, data, send_buffer_size) #AD0-AD6的寄存器写入策略 if(dev_name == "正电压采集热敏量采集"): for i in range(4): logger.info(f"AD发送指令：{i}") for j in range(14): logger.info(f"AD发送指令：{j}") target_addr = mem_base_addr + mem_offset + i * 32 + j*8 canshu = data[2*(i*14+j)]*256 + data[2*(i*14+j)+1]# 2字节 sysbus_cmd = f"sysbus WriteDoubleWord {target_addr:#010x} {canshu:#006x}" print(sysbus_cmd) pass ## 外设名称+函数名+字符串参数 adc1 LoadRegistersFromHexString "55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA55AA"' ad_cmd = f'{dev_name} LoadRegistersFromHexString "{data.hex()}"' print(ad_cmd) logger.info(f"AD发送指令：{ad_cmd}") response = renode.send_sync(ad_cmd,10)#2026/4/10 ad_cmd2 = f'{dev_name} ReadAllChannelsToString02' logger.info(f"AD发送指令：{ad_cmd2}") response = renode.send_sync(ad_cmd2,10)#2026/4/10 return True def recv(self, timeout: float = 0.1, **kwargs) -> Optional[bytes]: if not self.is_connected: logger.error("ADC驱动未连接，接收失败") return None try: # 模拟接收 #遍历所有AD外设，循环发送读取指令：例如 adc1 ReadAllChannelsToString02 #接收renode回复 time.sleep(0.51) ''' mock_data = b"\x11\x22\x33\x44" logger.debug(f"AD接收数据：{mock_data.hex()}") #2026/4/10 dev_name = kwargs.get("dev_name") if not dev_name: logger.error("AD接收失败：kwargs中未传入dev_name（设备名称）") return None cfg = DEVICE_CONFIG_DICT[dev_name] base = int(cfg["内存基地址"], 16) offset = int(cfg["内存偏移地址"], 16) addr = base + offset # ====================== # 调用 Renode 读取数据 # ====================== cmd = f"read 0x{addr:X}" hex_str = renode.send_sync(cmd) #return bytes.fromhex(hex_str.strip()) logger.info(f"AD接收数据：{mock_data.hex()}") return mock_data ''' return None except Exception as e: logger.error(f"AD数据接收失败：{str(e)}") return None