整车通讯一致性测试方案：实战教程

#!/usr/bin/env python3# -*- coding: utf-8 -*-"""整车CAN报文发送与网关转发测试工具功能说明：- 读取BLF日志文件和Excel测试用例（报文定义sheet）- 按ID+网段统计实际报文信息- 执行漏发、多发、帧格式、DLC、周期等全面检查- 生成包含测试概要、综合检查、多发检查的Excel报告依赖库：  python-can - CAN总线数据处理  openpyxl   - Excel文件操作使用方法：  python can_gateway_test.py [--blf ] [--xlsx ]若不指定参数，自动选取当前目录下唯一的.blf和.xlsx文件。"""import argparseimport sysfrom pathlib import Pathfrom datetime import datetimefrom collections import defaultdictfrom statistics import median, stdev, meantry:    import can    from can import BLFReaderexcept ImportError:    print("请安装 python-can 库: pip install python-can")    sys.exit(1)try:    from openpyxl import Workbook, load_workbook    from openpyxl.styles import PatternFill, Font, Alignment    from openpyxl.utils import get_column_letterexcept ImportError:    print("请安装 openpyxl 库: pip install openpyxl")    sys.exit(1)# ========================== 固定通道映射 ==========================CHANNEL_MAP = {    "0": "BCAN",    "1": "PCAN",    "2": "CCAN",    "3": "ICAN",}# ========================== 辅助函数 ==========================def hex_id_str(can_id: int) -> str:    """将CAN ID转为三位十六进制字符串"""    return f"0x{can_id:03X}"def parse_a_b_c(cell_value: str) -> tuple:    """解析形如 'FaSong/CAN/BCM' 的单元格"""    if not cell_value:        return None, None, None    parts = cell_value.split('/')    if len(parts) != 3:        return None, None, None    return parts[0].strip(), parts[1].strip(), parts[2].strip()def calc_period_stats(timestamps_ms: list):    """计算周期统计量：中位数、平均值、标准差"""    if len(timestamps_ms) < 2:        return None, None, None, []    periods = [timestamps_ms[i] - timestamps_ms[i-1]                for i in range(1, len(timestamps_ms))]    med = median(periods)    avg = mean(periods)    std = stdev(periods) if len(periods) > 1 else 0.0    return med, avg, std, periodsdef check_period(periods, expected_cycle_ms):    """检查周期是否符合规范"""    if expected_cycle_ms is None or len(periods) == 0:        return True, None, "无周期样本或期望周期为空"
    max_dev = max(abs(p - expected_cycle_ms) for p in periods)    avg_period = mean(periods)    abs_avg_err = abs(avg_period - expected_cycle_ms)
    # 周期越短，容差越小    if expected_cycle_ms < 10:        if abs_avg_err > 1.0:            return False, max_dev, f"周期平均值偏差{abs_avg_err:.2f}ms > 1ms"    else:        if abs_avg_err > expected_cycle_ms * 0.05:            return False, max_dev, f"周期平均值偏差{abs_avg_err:.2f}ms > 期望周期*5%"
    if max_dev > expected_cycle_ms * 0.5:        return False, max_dev, f"最大偏差{max_dev:.2f}ms > 期望周期*50%"
    return True, max_dev, None# ========================== 主测试类 ==========================class CANGatewayTester:    def __init__(self, blf_path: Path, xlsx_path: Path):        self.blf_path = blf_path        self.xlsx_path = xlsx_path        self.test_items = []        self.actual_stats = {}        self.definition_keys = set()    def load_testcases(self):        """加载Excel中的报文定义"""        wb = load_workbook(self.xlsx_path, data_only=True)        if "报文定义" not in wb.sheetnames:            raise ValueError(f"Excel缺少'报文定义'sheet")
        ws = wb["报文定义"]        headers = [cell.value for cell in ws[1]]        col_idx = {h: i+1 for i, h in enumerate(headers) if h}
        for row in ws.iter_rows(min_row=2, values_only=True):            if not row[0]:                continue
            can_id = int(row[col_idx["ID"]-1], 16) if isinstance(row[col_idx["ID"]-1], str) else int(row[col_idx["ID"]-1])            id_str = hex_id_str(can_id)
            for seg in ["BCAN", "PCAN", "CCAN", "ICAN"]:                cell_val = row[col_idx[seg]-1]                if not cell_val:                    continue                A, B, C = parse_a_b_c(str(cell_val))                if A:                    self.test_items.append({                        "id": can_id, "id_str": id_str, "segment": seg,                        "expected_direction": A, "expected_frame_type": B,                         "expected_ecu": C,                    })                    self.definition_keys.add((id_str, seg))    def parse_blf(self):        """解析BLF文件，统计报文信息"""        self.actual_stats = defaultdict(lambda: {            "timestamps_ms": [], "frame_types": set(),            "dlc_values": [], "count": 0,        })
        with BLFReader(self.blf_path) as reader:            for msg in reader:                segment = CHANNEL_MAP.get(str(msg.channel), f"UNKNOWN_{msg.channel}")                id_str = hex_id_str(msg.arbitration_id)                key = (id_str, segment)
                stat = self.actual_stats[key]                stat["timestamps_ms"].append(msg.timestamp * 1000.0)                stat["frame_types"].add("CANFD" if getattr(msg, "is_fd", False) else "CAN")                stat["dlc_values"].append(msg.dlc)                stat["count"] += 1
        for key, stat in self.actual_stats.items():            stat["timestamps_ms"].sort()            med, avg, std, periods = calc_period_stats(stat["timestamps_ms"])            stat.update({"period_median_ms": med, "period_avg_ms": avg,                         "period_stddev_ms": std, "periods": periods})    def check_and_generate_report(self):        """执行全面检查并生成报告"""        comprehensive_rows = []        extra_rows = []
        # 综合检查        for item in self.test_items:            key = (item["id_str"], item["segment"])            actual = self.actual_stats.get(key)
            row_info = {                "ID": item["id_str"], "网段": item["segment"],                "发送路径": f"{item['expected_direction']}/{item['expected_ecu']}",            }
            if not actual:                row_info.update({"问题类型": "漏发", "结果": "FAIL",                                 "备注": "周期型报文未在BLF中出现"})            else:                problems = []                # 帧格式检查                if item["expected_frame_type"] == "CAN" and "CANFD" in actual["frame_types"]:                    problems.append("帧格式异常")                # DLC检查                if item.get("expected_dlc") and any(d != item["expected_dlc"] for d in actual["dlc_values"]):                    problems.append("DLC异常")                # 周期检查                if item.get("expected_cycle_ms") and actual["periods"]:                    ok, _, err = check_period(actual["periods"], item["expected_cycle_ms"])                    if not ok:                        problems.append("周期异常")
                row_info["问题类型"] = "、".join(problems) if problems else "无"                row_info["结果"] = "FAIL" if problems else "PASS"
            comprehensive_rows.append(row_info)
        # 多发检查        for (id_str, seg), stat in self.actual_stats.items():            if (id_str, seg) not in self.definition_keys:                extra_rows.append({                    "ID": id_str, "网段": seg, "问题类型": "多发",                    "结果": "FAIL", "备注": "BLF中出现但用例未定义"                })
        self.write_report(comprehensive_rows, extra_rows)    def write_report(self, comp_rows, extra_rows):        """生成Excel测试报告"""        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")        report_name = f"{timestamp}_整车报文测试报告.xlsx"        wb = Workbook()
        # 测试概要        ws_summary = wb.active        ws_summary.title = "测试概要"        summary_data = [            ("生成时间", datetime.now().strftime("%Y-%m-%d %H:%M:%S")),            ("综合检查条目", len(comp_rows)),            ("PASS", sum(1 for r in comp_rows if r["结果"] == "PASS")),            ("FAIL", sum(1 for r in comp_rows if r["结果"] == "FAIL")),            ("多发条目", len(extra_rows)),        ]        for row_idx, (k, v) in enumerate(summary_data, 1):            ws_summary.cell(row=row_idx, column=1, value=k)            ws_summary.cell(row=row_idx, column=2, value=v)
        # 综合检查Sheet        ws_comp = wb.create_sheet("综合检查")        ws_comp.append(["ID", "网段", "问题类型", "发送路径", "结果", "备注"])        for row in comp_rows:            ws_comp.append([row["ID"], row["网段"], row["问题类型"],                           row["发送路径"], row["结果"], row.get("备注", "")])
        # 多发检查Sheet        ws_extra = wb.create_sheet("多发检查")        ws_extra.append(["ID", "网段", "问题类型", "结果", "备注"])        for row in extra_rows:            ws_extra.append([row["ID"], row["网段"], row["问题类型"],                            row["结果"], row["备注"]])
        wb.save(report_name)        print(f"报告已生成: {report_name}")# ========================== 主入口 ==========================def main():    parser = argparse.ArgumentParser(description="整车CAN报文一致性测试")    parser.add_argument("--blf", type=str, help="BLF文件路径")    parser.add_argument("--xlsx", type=str, help="Excel测试用例路径")    args = parser.parse_args()
    cur_dir = Path.cwd()
    # 自动查找文件    blf_path = Path(args.blf) if args.blf else next((f for f in cur_dir.iterdir() if f.suffix == '.blf'), None)    xlsx_path = Path(args.xlsx) if args.xlsx else next((f for f in cur_dir.iterdir() if f.suffix == '.xlsx'), None)
    if not blf_path or not xlsx_path:        print("请确保当前目录有BLF和Excel文件，或使用参数指定")        sys.exit(1)
    tester = CANGatewayTester(blf_path, xlsx_path)    tester.load_testcases()    tester.parse_blf()    tester.check_and_generate_report()if __name__ == "__main__":    main()