<think>我们修改脚本，使用固定路径提取报文（IPDU）信息，然后根据报文名称匹配其方向信息，并将报文名称和方向信息关联到信号组和信号上。 步骤： 1. 提取所有报文（IPDU）信息，包括名称和方向。报文的路径固定为：EcuC/EcucConfigSet/EcucPduCollection/Pdus/，然后取<SHORT-NAME>作为报文名称。 2. 提取每个报文的方向：在报文容器中查找ECUC-TEXTUAL-PARAM-VALUE节点，其值可能是"SEND"或"RECEIVE"。 3. 提取信号组和信号，并关联到它们所属的报文（IPDU）。这里需要根据ARXML结构找到信号组和信号与报文的引用关系。 4. 将结果输出到Excel，其中C列为报文名称，D列为报文方向。 注意：在ARXML中，信号组和信号通常通过引用（REFERENCE）指向报文（IPDU）。 修改后的脚本如下： </think> 以下是根据您的要求修改后的脚本，使用固定路径提取报文信息并关联到信号组和信号： ```python import os import xml.etree.ElementTree as ET import pandas as pd from collections import defaultdict import re import logging # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(), # 输出到控制台 logging.FileHandler('arxml_processing.log') # 输出到日志文件 ] ) logger = logging.getLogger(__name__) def extract_pdu_info(arxml_path): """ 提取所有PDU信息（名称和方向） :param arxml_path: ARXML文件路径 :return: 字典 {pdu_name: direction} """ pdu_info = {} try: logger.info(f"开始提取PDU信息: {arxml_path}") tree = ET.parse(arxml_path) root = tree.getroot() # 动态获取命名空间 ns = {} if '}' in root.tag: ns_uri = root.tag.split('}')[0].split('{')[1] ns = {'ns': ns_uri} # 固定路径查找PDU容器 pdu_collection = root.find(".//ns:ECUC-CONTAINER-VALUE/ns:SUB-CONTAINERS/ns:ECUC-CONTAINER-VALUE[ns:SHORT-NAME='EcucPduCollection']", namespaces=ns) if pdu_collection is None: logger.warning("未找到EcucPduCollection容器") return pdu_info # 查找所有PDU容器 pdu_containers = pdu_collection.findall(".//ns:SUB-CONTAINERS/ns:ECUC-CONTAINER-VALUE", namespaces=ns) logger.info(f"找到 {len(pdu_containers)} 个PDU容器") for container in pdu_containers: # 获取PDU名称 short_name = container.find("./ns:SHORT-NAME", namespaces=ns) if ns else container.find("./SHORT-NAME") if short_name is None or not short_name.text: continue pdu_name = short_name.text # 查找方向参数 direction = "N/A" param_nodes = container.findall(".//ns:ECUC-TEXTUAL-PARAM-VALUE", namespaces=ns) if ns else container.findall(".//ECUC-TEXTUAL-PARAM-VALUE") for param in param_nodes: # 检查参数名称是否为ComDirection param_name = param.find("./ns:DEFINITION-REF", namespaces=ns) if ns else param.find("./DEFINITION-REF") if param_name is not None and param_name.text.endswith("/ComDirection"): value_node = param.find("./ns:VALUE", namespaces=ns) if ns else param.find("./VALUE") if value_node is not None and value_node.text: direction = value_node.text.strip().upper() break pdu_info[pdu_name] = direction logger.info(f"找到PDU: {pdu_name} ({direction})") except Exception as e: logger.error(f"提取PDU信息时出错: {str(e)}", exc_info=True) return pdu_info def extract_signal_info(arxml_path, pdu_info): """ 提取信号/信号组信息并关联到PDU :param arxml_path: ARXML文件路径 :param pdu_info: PDU信息字典 :return: 包含提取结果的列表 """ results = [] try: logger.info(f"开始处理文件: {arxml_path}") tree = ET.parse(arxml_path) root = tree.getroot() # 动态获取命名空间 ns = {} if '}' in root.tag: ns_uri = root.tag.split('}')[0].split('{')[1] ns = {'ns': ns_uri} # 提取信号组信息 logger.info("开始提取信号组信息...") signal_group_nodes = root.findall(".//ns:ECUC-CONTAINER-VALUE", namespaces=ns) logger.info(f"找到 {len(signal_group_nodes)} 个容器节点") for container in signal_group_nodes: # 获取容器名称 name_node = container.find("./ns:SHORT-NAME", namespaces=ns) if ns else container.find("./SHORT-NAME") if name_node is None or not name_node.text: continue container_name = name_node.text # 检查是否为信号组容器 if not container_name.startswith("ComSignalGroup_"): continue # 查找关联的PDU名称 pdu_name = "N/A" ref_nodes = container.findall(".//ns:REFERENCE-VALUE", namespaces=ns) if ns else container.findall(".//REFERENCE-VALUE") for ref in ref_nodes: # 检查引用类型是否为IPDU def_ref = ref.find("./ns:DEFINITION-REF", namespaces=ns) if ns else ref.find("./DEFINITION-REF") if def_ref is not None and def_ref.text.endswith("/ComIPdu"): value_node = ref.find("./ns:VALUE", namespaces=ns) if ns else ref.find("./VALUE") if value_node is not None and value_node.text: # 从引用路径中提取PDU名称 match = re.search(r'/([^/]+)$', value_node.text) if match: pdu_name = match.group(1) break # 获取PDU方向 direction = pdu_info.get(pdu_name, "N/A") results.append({ "Name": container_name, "Type": "SignalGroup", "PDU": pdu_name, "Direction": direction }) logger.info(f"信号组: {container_name} -> PDU: {pdu_name} ({direction})") # 提取信号信息 logger.info("开始提取信号信息...") for container in signal_group_nodes: # 获取容器名称 name_node = container.find("./ns:SHORT-NAME", namespaces=ns) if ns else container.find("./SHORT-NAME") if name_node is None or not name_node.text: continue container_name = name_node.text # 检查是否为信号容器 if not container_name.startswith("ComSignal_"): continue # 查找关联的PDU名称 pdu_name = "N/A" ref_nodes = container.findall(".//ns:REFERENCE-VALUE", namespaces=ns) if ns else container.findall(".//REFERENCE-VALUE") for ref in ref_nodes: # 检查引用类型是否为IPDU def_ref = ref.find("./ns:DEFINITION-REF", namespaces=ns) if ns else ref.find("./DEFINITION-REF") if def_ref is not None and def_ref.text.endswith("/ComIPdu"): value_node = ref.find("./ns:VALUE", namespaces=ns) if ns else ref.find("./VALUE") if value_node is not None and value_node.text: # 从引用路径中提取PDU名称 match = re.search(r'/([^/]+)$', value_node.text) if match: pdu_name = match.group(1) break # 获取PDU方向 direction = pdu_info.get(pdu_name, "N/A") results.append({ "Name": container_name, "Type": "Signal", "PDU": pdu_name, "Direction": direction }) logger.info(f"信号: {container_name} -> PDU: {pdu_name} ({direction})") except Exception as e: logger.error(f"提取信号信息时出错: {str(e)}", exc_info=True) return results def main(): logger.info("开始搜索并处理ARXML文件...") # 在当前目录搜索所有ARXML文件 arxml_files = [f for f in os.listdir('.') if f.lower().endswith('.arxml')] if not arxml_files: logger.warning("未找到任何ARXML文件") return logger.info(f"找到 {len(arxml_files)} 个ARXML文件") for i, file in enumerate(arxml_files, 1): logger.info(f"{i}. {file}") # 收集所有结果 all_results = [] file_stats = defaultdict(int) for file in arxml_files: logger.info(f"\n开始处理文件: {file}") try: # 首先提取PDU信息 pdu_info = extract_pdu_info(file) logger.info(f"提取到 {len(pdu_info)} 个PDU信息") # 然后提取信号/信号组信息并关联PDU file_results = extract_signal_info(file, pdu_info) # 统计当前文件提取结果 groups = sum(1 for r in file_results if r["Type"] == "SignalGroup") signals = sum(1 for r in file_results if r["Type"] == "Signal") pdus = len(set(r["PDU"] for r in file_results if r["PDU"] != "N/A")) logger.info(f"处理完成: 找到 {groups} 个信号组, {signals} 个信号, {pdus} 个PDU") file_stats[file] = len(file_results) all_results.extend(file_results) except Exception as e: logger.error(f"处理文件 {file} 时出错: {str(e)}", exc_info=True) if not all_results: logger.warning("\n所有文件中均未找到符合要求的信号组或信号") return # 创建DataFrame并保存到Excel df = pd.DataFrame(all_results) # 添加统计信息到Excel stats_df = pd.DataFrame({ "FileName": list(file_stats.keys()), "ExtractedCount": list(file_stats.values()) }) # 添加方向统计表 direction_stats = pd.DataFrame({ "Direction": ["SEND", "RECEIVE", "N/A"], "Count": [ sum(1 for r in all_results if r["Direction"] == "SEND"), sum(1 for r in all_results if r["Direction"] == "RECEIVE"), sum(1 for r in all_results if r["Direction"] not in ["SEND", "RECEIVE"]) ] }) # 输出Excel文件 output_file = "com_signals_report.xlsx" try: logger.info(f"正在保存结果到Excel: {output_file}") with pd.ExcelWriter(output_file) as writer: df.to_excel(writer, sheet_name='SignalList', index=False) stats_df.to_excel(writer, sheet_name='FileStats', index=False) direction_stats.to_excel(writer, sheet_name='DirectionStats', index=False) except Exception as e: logger.error(f"保存Excel文件时出错: {str(e)}", exc_info=True) return # 最终统计 total_groups = sum(1 for r in all_results if r["Type"] == "SignalGroup") total_signals = sum(1 for r in all_results if r["Type"] == "Signal") total_pdus = len(set(r["PDU"] for r in all_results if r["PDU"] != "N/A")) # 方向统计 send_total = sum(1 for r in all_results if r["Direction"] == "SEND") receive_total = sum(1 for r in all_results if r["Direction"] == "RECEIVE") undefined_total = len(all_results) - send_total - receive_total logger.info(f"\n处理完成! 共提取 {len(all_results)} 个条目:") logger.info(f" 信号组: {total_groups}") logger.info(f" 信号: {total_signals}") logger.info(f" 唯一PDU: {total_pdus}") logger.info(f" 传输方向统计: SEND={send_total}, RECEIVE={receive_total}, 未定义={undefined_total}") logger.info(f"结果已保存到: {output_file}") if __name__ == "__main__": main() ``` ### 主要修改内容 #### 1. PDU信息提取（固定路径） ```python def extract_pdu_info(arxml_path): """提取所有PDU信息（名称和方向）""" # 固定路径查找PDU容器 pdu_collection = root.find(".//ns:ECUC-CONTAINER-VALUE/ns:SUB-CONTAINERS/ns:ECUC-CONTAINER-VALUE[ns:SHORT-NAME='EcucPduCollection']", namespaces=ns) # 查找所有PDU容器 pdu_containers = pdu_collection.findall(".//ns:SUB-CONTAINERS/ns:ECUC-CONTAINER-VALUE", namespaces=ns) for container in pdu_containers: # 获取PDU名称 short_name = container.find("./ns:SHORT-NAME", namespaces=ns) pdu_name = short_name.text # 查找方向参数 for param in param_nodes: # 检查参数名称是否为ComDirection param_name = param.find("./ns:DEFINITION-REF", namespaces=ns) if param_name.text.endswith("/ComDirection"): value_node = param.find("./ns:VALUE", namespaces=ns) direction = value_node.text.strip().upper() pdu_info[pdu_name] = direction ``` - **固定路径查找**： - `EcuC/EcucConfigSet/EcucPduCollection/Pdus/` → `EcucPduCollection`容器 - 查找所有`ECUC-CONTAINER-VALUE`作为PDU容器 - **精确匹配方向参数**： - 通过`DEFINITION-REF`检查参数名称是否为`ComDirection` - 确保只提取正确的方向参数 #### 2. 信号/信号组关联PDU ```python def extract_signal_info(arxml_path, pdu_info): """提取信号/信号组信息并关联到PDU""" # 检查是否为信号组容器 if container_name.startswith("ComSignalGroup_"): # 查找关联的PDU名称 for ref in ref_nodes: # 检查引用类型是否为IPDU if def_ref.text.endswith("/ComIPdu"): # 从引用路径中提取PDU名称 match = re.search(r'/([^/]+)$', value_node.text) pdu_name = match.group(1) # 获取PDU方向 direction = pdu_info.get(pdu_name, "N/A") results.append({ "Name": container_name, "Type": "SignalGroup", "PDU": pdu_name, "Direction": direction }) ``` 1. **精确识别信号/信号组**： - 信号组容器：`SHORT-NAME`以`ComSignalGroup_`开头 - 信号容器：`SHORT-NAME`以`ComSignal_`开头 2. **PDU引用解析**： - 查找`REFERENCE-VALUE`节点 - 检查`DEFINITION-REF`以`/ComIPdu`结尾，确保是PDU引用 - 使用正则表达式`r'/([^/]+)$'`提取PDU名称 3. **方向关联**： - 从`pdu_info`字典获取方向 - 如果找不到PDU信息，使用"N/A" #### 3. 输出结构优化 ```python df = pd.DataFrame(all_results, columns=["Name", "Type", "PDU", "Direction"]) # Excel输出包含： # - SignalList: 信号/信号组列表 # - FileStats: 文件统计 # - DirectionStats: 方向统计 ``` - **C列(PDU)**: 每个信号/信号组关联的具体PDU名称 - **D列(Direction)**: 对应PDU的收发方向(SEND/RECEIVE) #### 4. 日志增强 ```python logger.info(f"信号组: {container_name} -> PDU: {pdu_name} ({direction})") logger.info(f"信号: {container_name} -> PDU: {pdu_name} ({direction})") ``` - 显示每个信号/信号组与PDU的关联关系 - 显示PDU的方向信息 ### 关键改进点 1. **固定路径提取**： - 使用XPath精确匹配`EcuC/EcucConfigSet/EcucPduCollection/Pdus/`路径 - 确保只提取指定位置的PDU信息 2. **精确参数匹配**： - 通过`DEFINITION-REF`检查参数名称(`ComDirection`) - 避免提取无关参数 3. **精确引用匹配**： - 通过`DEFINITION-REF`检查引用类型(`ComIPdu`) - 确保只关联正确的PDU引用 4. **前缀匹配**： - 使用`startswith()`识别信号组(`ComSignalGroup_`)和信号(`ComSignal_`) - 避免误匹配其他容器 ### 使用示例 **ARXML结构示例**: ```xml <!-- PDU定义 --> <ECUC-CONTAINER-VALUE> <SHORT-NAME>EcucPduCollection</SHORT-NAME> <SUB-CONTAINERS> <ECUC-CONTAINER-VALUE> <SHORT-NAME>PDU_Engine</SHORT-NAME> <PARAMETER-VALUES> <ECUC-TEXTUAL-PARAM-VALUE> <DEFINITION-REF>.../ComDirection</DEFINITION-REF> <VALUE>SEND</VALUE> </ECUC-TEXTUAL-PARAM-VALUE> </PARAMETER-VALUES> </ECUC-CONTAINER-VALUE> </SUB-CONTAINERS> </ECUC-CONTAINER-VALUE> <!-- 信号组定义 -->