# -*- coding: utf-8 -*- ''' Excel 配置解析（excel_config.py）：基于openpyxl实现 Excel 文件解析，提取外设名称、UDP 端口、协议类型、基地址等信息，存入DEVICE_CONFIG_DICT。 ''' import openpyxl from loguru import logger import re import pandas as pd import os import sys import io # 设置标准输出和标准错误的编码为UTF-8 # sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') # sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8') def hex_to_int (hex_str): """处理十六进制字符串转整数（兼容带逗号 / 空格的格式，如 0x2080，0000、0x0000 1234） :param hex_str: 十六进制字符串 :return: 转换后的整数，失败返回 0 """ if not isinstance (hex_str, str): return 0 # 非字符串直接返回0 # 移除逗号、空格等非十六进制字符 clean_hex = re.sub (r'[,\s]', '', hex_str.strip ()) try: return int (clean_hex, 16) if clean_hex else 0 except (ValueError, TypeError): logger.warning (f"十六进制转换失败：{hex_str}，默认返回 0") return 0 def load_csv_config (file_path): """加载 Excel/CSV 外设配置文件，解析核心配置项 :param file_path: Excel/CSV文件路径（xlsx/xls/csv） :return: 外设配置字典，key = 外设名称，value = 配置项字典 """ device_config = {} wb = None temp_excel = "file.xlsx" is_temp_file = False # 标记是否生成了临时文件 try: # 处理CSV文件：转换为Excel后解析 if file_path.endswith('.csv'): df = pd.read_csv(file_path) df.to_excel(temp_excel, index=False) file_path = temp_excel is_temp_file = True # 标记临时文件已生成 # 打开工作簿（兼容直接传入Excel文件的情况） wb = openpyxl.load_workbook(file_path, data_only=True) ws = wb.active logger.info(f"成功打开配置文件：{file_path}，工作表：{ws.title}") # 获取表头行，匹配核心配置列（兼容列名大小写 / 空格） header = [cell.value.strip() if cell.value else "" for cell in ws[1]] col_mapping = { "名称": header.index([h for h in header if h.lower() == "名称"][0]) if [h for h in header if h.lower() == "名称"] else -1, "编号": header.index([h for h in header if h.lower() == "编号"][0]) if [h for h in header if h.lower() == "编号"] else -1, "协议类型": header.index([h for h in header if h.lower() in ["协议", "协议类型"]][0]) if [h for h in header if h.lower() in ["协议", "协议类型"]] else -1, "内存基地址": header.index([h for h in header if h.lower() in ["内存基地址", "基地址"]][0]) if [h for h in header if h.lower() in ["内存基地址", "基地址"]] else -1, "内存偏移地址": header.index([h for h in header if h.lower() in ["内存偏移地址", "偏移地址"]][0]) if [h for h in header if h.lower() in ["内存偏移地址", "偏移地址"]] else -1, "UDP端口": header.index([h for h in header if h.lower() in ["UDP端口", "端口地址", "udp 地址", "UDP port","port"]][0]) if [h for h in header if h.lower() in ["UDP端口", "端口地址", "udp 地址", "UDP port","port"]] else -1, "UDP远程端口": header.index([h for h in header if h.lower() in ["UDP远程端口", "远程端口地址", "远程udp 地址", "remote_port","remote port"]][0]) if [h for h in header if h.lower() in ["UDP端口", "端口地址", "udp 地址", "remote_port","remote port"]] else -1, "TCP端口": header.index([h for h in header if h.lower() in ["TCP端口", "端口地址", "TCP 地址", "tcp_local_port"]][0]) if [h for h in header if h.lower() in ["TCP端口", "tcp_local_port"]] else -1, "TCP总线端口": header.index([h for h in header if h.lower() in ["TCP远程端口", "远程端口地址", "远程tcp 地址", "tcp_bus_port"]][0]) if [h for h in header if h.lower() in ["TCP端口", "tcp_bus_port"]] else -1, "说明": header.index([h for h in header if h.lower() == "说明"][0]) if [h for h in header if h.lower() == "说明"] else -1, "发送缓存区大小": header.index([h for h in header if h.lower() in ["发送缓存区大小", "发送缓存"]][0]) if [h for h in header if h.lower() in ["发送缓存区大小", "发送缓存"]] else -1, "接收缓存区大小": header.index([h for h in header if h.lower() in ["接收缓存区大小", "接收缓存"]][0]) if [h for h in header if h.lower() in ["接收缓存区大小", "接收缓存"]] else -1, } # 校验核心列（名称、协议、基地址必须存在） required_cols = ["名称", "协议类型", "内存基地址"] missing_cols = [col for col in required_cols if col_mapping[col] == -1] if missing_cols: logger.error(f"配置文件缺少核心列：{','.join(missing_cols)}，请检查表头") return device_config # 关键优化：将迭代器转为列表，提前加载所有行数据（避免后续关闭wb后访问） rows_data = list(ws.iter_rows(min_row=2, values_only=True)) # 遍历数据行（从第 2 行开始） for row_num, row in enumerate(rows_data, start=2): if not row[col_mapping["名称"]]: # 外设名称为空则跳过 logger.warning(f"第 {row_num} 行：外设名称为空，跳过该行") continue dev_name = row[col_mapping["名称"]].strip() print(f'原始值：{row[col_mapping["编号"]]}') print(f'类型：{type(row[col_mapping["编号"]])}') print(str(row[col_mapping["编号"]])) str_value = str(row[col_mapping["编号"]]) print(f'isdigit?:{str_value.isdigit()}') # 解析单行配置，缺省项赋默认值 dev_info = { "protocol": row[col_mapping["协议类型"]].strip().upper() if row[col_mapping["协议类型"]] else "", "number": int(row[col_mapping["编号"]]) if (row[col_mapping["编号"]] is not None and str(row[col_mapping["编号"]]).isdigit()) else 99,#2026/4/15 "base_addr": hex_to_int(row[col_mapping["内存基地址"]]), "offset_addr": hex_to_int(row[col_mapping["内存偏移地址"]]), "udp_port": int(row[col_mapping["UDP端口"]]) if (row[col_mapping["UDP端口"]] and str(row[col_mapping["UDP端口"]]).isdigit()) else 8888, "remote_port": int(row[col_mapping["UDP远程端口"]]) if (row[col_mapping["UDP远程端口"]] and str(row[col_mapping["UDP远程端口"]]).isdigit()) else 9999, "tcp_local_port": int(row[col_mapping["TCP端口"]]) if (row[col_mapping["TCP端口"]] and str(row[col_mapping["TCP端口"]]).isdigit()) else 1, "tcp_bus_port": int(row[col_mapping["TCP总线端口"]]) if (row[col_mapping["TCP总线端口"]] and str(row[col_mapping["TCP总线端口"]]).isdigit()) else 1, "desc": row[col_mapping["说明"]].strip() if row[col_mapping["说明"]] else "无", "send_cache": int(row[col_mapping["发送缓存区大小"]]) if (row[col_mapping["发送缓存区大小"]] and str(row[col_mapping["发送缓存区大小"]]).isdigit()) else 512, "recv_cache": int(row[col_mapping["接收缓存区大小"]]) if (row[col_mapping["接收缓存区大小"]] and str(row[col_mapping["接收缓存区大小"]]).isdigit()) else 512, "enable": True # 默认启用该外设 } # 校验协议类型合法性 if dev_info["protocol"] not in ["UART", "CAN", "1553B", "AD", "OC"]: logger.warning(f"第 {row_num} 行 [{dev_name}]：协议类型 {dev_info['protocol']} 不合法，仅支持 UART/CAN/1553B/AD/OC，跳过该行") continue device_config[dev_name] = dev_info logger.debug( f"解析外设配置：{dev_name} | {dev_info['protocol']} | 基地址 {hex(dev_info['base_addr'])} | " f"端口号 {dev_info['udp_port']} | 发送缓存 {dev_info['send_cache']} | 接收缓存 {dev_info['recv_cache']} | 说明 {dev_info['desc']}" ) logger.info(f"配置文件解析完成，共加载 {len(device_config)} 个有效外设配置") print(f"配置文件解析完成，共加载 {len(device_config)} 个有效外设配置") except FileNotFoundError: logger.error(f"配置文件不存在：{file_path}") print(f"配置文件不存在：{file_path}") except Exception as e: logger.error(f"配置文件解析失败：{str(e)}", exc_info=True) print(f"配置文件解析失败：{str(e)}") finally: # 第一步：先关闭工作簿（确保所有数据已读取完成） if wb: try: wb.close() logger.debug("Excel工作簿已正常关闭") except Exception as e: logger.warning(f"关闭Excel工作簿失败：{str(e)}") # 第二步：删除临时文件（仅当生成了临时文件时执行） if is_temp_file and os.path.exists(temp_excel): try: # 延迟删除（避免文件句柄未释放） import time time.sleep(0.1) os.remove(temp_excel) logger.debug("临时Excel文件已删除") except Exception as e: logger.warning(f"删除临时Excel文件失败：{str(e)}") return device_config # 测试代码（单独运行该文件时执行） if __name__ == "__main__": test_config = load_csv_config("config_files/device_config.csv") print(test_config)