PyTorch复现CBOW词向量:从数据管道到训练避坑全记录
2026/10/16 9:53:36
为了方便查看swc中port、runnable等信息,将数据提取到表格中。
1、gen_swc_stats.py
#!/usr/bin/env python3 """Generate Excel report: Complete SWC Port / Signal / Runnable mapping.""" import os import re import xml.etree.ElementTree as ET from collections import defaultdict from openpyxl import Workbook from openpyxl.styles import Font, Alignment, PatternFill, Border, Side from openpyxl.utils import get_column_letter COMP_DIR = r"d:\workspace\0_Demo\\davinci\ImportFiles\Swcs" OUTPUT = r"d:\workspace\0_Demo\tools\J1939_SWC_Statistics.xlsx" NS = {"ar": "http://autosar.org/schema/r4.0"} OUR_ECUS = ["ABS_EBS_ESC", "ADMC", "ADSC", "CGW", "ECAS", "EHSP_G", "EHSP_P", "EPB", "IAPU", "PMS"] def parse_swc(filepath): """Parse one SWC ARXML and return structured data.""" tree = ET.parse(filepath) root = tree.getroot() ports = [] # {name, direction:'Tx'|'Rx', signal_name, if_ref} runnables = [] # {name, period_sec, data_refs: [{signal_name, direction}]} signal_port_map = defaultdict(list) # signal_name -> [{port_name, direction}] signal_runnable_map = defaultdict(list) # signal_name -> [{runnable_name, direction}] # -- Parse ports -- for pp in root.iter("{http://autosar.org/schema/r4.0}P-PORT-PROTOTYPE"): sn = pp.find("ar:SHORT-NAME", NS) if_ref = pp.find("ar:PROVIDED-INTERFACE-TREF", NS) if sn is not None: port_name = sn.text signal_name = port_name[2:] if port_name.startswith("Pp") else port_name ports.append({"name": port_name, "direction": "Tx", "signal_name": signal_name, "if_ref": if_ref.text if if_ref is not None else ""}) signal_port_map[signal_name].append({"port_name": port_name, "direction": "Tx"}) for rp in root.iter("{http://autosar.org/schema/r4.0}R-PORT-PROTOTYPE"): sn = rp.find("ar:SHORT-NAME", NS) if_ref = rp.find("ar:REQUIRED-INTERFACE-TREF", NS) if sn is not None: port_name = sn.text signal_name = port_name[2:] if port_name.startswith("Rp") else port_name ports.append({"name": port_name, "direction": "Rx", "signal_name": signal_name, "if_ref": if_ref.text if if_ref is not None else ""}) signal_port_map[signal_name].append({"port_name": port_name, "direction": "Rx"}) # -- Parse runnables -- for re_el in root.iter("{http://autosar.org/schema/r4.0}RUNNABLE-ENTITY"): rn_sn = re_el.find("ar:SHORT-NAME", NS) rn_name = rn_sn.text if rn_sn is not None else "N/A" interval = re_el.find("ar:MINIMUM-START-INTERVAL", NS) period = interval.text if interval is not None else "N/A" # Extract data references data_refs = [] # RX: DATA-RECEIVE-POINT-BY-ARGUMENTS for drp in re_el.findall("ar:DATA-RECEIVE-POINT-BY-ARGUMENTS", NS): vdp_ref = drp.find("ar:VARIABLE-DATA-PROTOTYPE-REF", NS) if vdp_ref is not None and vdp_ref.text: # /UserApplication/SwcJ1939/PortInterfaces/PiSIGNAL/DeSIGNAL sig = _extract_signal_from_ref(vdp_ref.text) data_refs.append({"signal_name": sig, "direction": "Rx"}) signal_runnable_map[sig].append({"runnable_name": rn_name, "direction": "Rx"}) # TX: DATA-SEND-POINTS for dsp in re_el.findall("ar:DATA-SEND-POINTS", NS): vdp_ref = dsp.find("ar:VARIABLE-DATA-PROTOTYPE-REF", NS) if vdp_ref is not None and vdp_ref.text: sig = _extract_signal_from_ref(vdp_ref.text) data_refs.append({"signal_name": sig, "direction": "Tx"}) signal_runnable_map[sig].append({"runnable_name": rn_name, "direction": "Tx"}) runnables.append({"name": rn_name, "period_sec": period, "data_refs": data_refs}) return ports, runnables, signal_port_map, signal_runnable_map def _extract_signal_from_ref(ref_text): """Extract signal name from a VDP reference like: /UserApplication/SwcJ1939/PortInterfaces/PiSignalName/DeSignalName """ m = re.search(r"/Pi(\w+)/De", ref_text) return m.group(1) if m else ref_text # ── Parse all ECUs ──────────────────────────────────────────────────────────── all_data = {} for ecu in OUR_ECUS: filepath = os.path.join(COMP_DIR, f"CtAp{ecu}.arxml") if os.path.exists(filepath): ports, runnables, spm, srm = parse_swc(filepath) all_data[ecu] = { "ports": ports, "runnables": runnables, "signal_port_map": spm, "signal_runnable_map": srm, } # ── Styles ──────────────────────────────────────────────────────────────────── hdr_font = Font(name="Microsoft YaHei", size=11, bold=True, color="FFFFFF") hdr_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") sub_fill = PatternFill(start_color="D9E2F3", end_color="D9E2F3", fill_type="solid") sub_font = Font(name="Microsoft YaHei", size=10, bold=True) title_font = Font(name="Microsoft YaHei", size=14, bold=True, color="1F4E79") data_font = Font(name="Consolas", size=10) small_font = Font(name="Consolas", size=9) tx_font = Font(name="Consolas", size=10, color="2F5496") rx_font = Font(name="Consolas", size=10, color="C00000") boundary = Border( left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin"), ) ctr = Alignment(horizontal="center", vertical="center") wrap = Alignment(horizontal="left", vertical="center", wrap_text=True) wb = Workbook() sheet_idx = 0 # ═══════════════════════════════════════════════════════════════════════════════ # Sheet 1: Summary # ═══════════════════════════════════════════════════════════════════════════════ ws = wb.active ws.title = "1-Summary" ws.merge_cells("A1:H1") ws["A1"] = "J1939 SWC Statistics - CHA_CAN_GEELY_CV_G3_L3_V4.1_20250623_ADMC.arxml" ws["A1"].font = title_font ws["A1"].alignment = Alignment(horizontal="center", vertical="center") ws.row_dimensions[1].height = 32 headers = ["ECU", "SWC Name", "P-Port (TX)", "R-Port (RX)", "Total Port", "Runnable", "TX Runnable", "RX Runnable"] for ci, h in enumerate(headers, 1): c = ws.cell(row=3, column=ci, value=h) c.font = hdr_font; c.fill = hdr_fill; c.alignment = ctr; c.border = boundary ws.row_dimensions[3].height = 30 for ri, ecu in enumerate(OUR_ECUS, 4): d = all_data.get(ecu) if not d: continue pports = [p for p in d["ports"] if p["direction"] == "Tx"] rports = [p for p in d["ports"] if p["direction"] == "Rx"] rn_tx = [r for r in d["runnables"] if any(dr["direction"] == "Tx" for dr in r["data_refs"])] rn_rx = [r for r in d["runnables"] if any(dr["direction"] == "Rx" for dr in r["data_refs"])] vals = [ecu, f"CtAp{ecu}", len(pports), len(rports), len(d["ports"]), len(d["runnables"]), len(rn_tx), len(rn_rx)] for ci, v in enumerate(vals, 1): c = ws.cell(row=ri, column=ci, value=v) c.font = data_font; c.alignment = ctr; c.border = boundary # Totals tr = 4 + len(OUR_ECUS) tot_fill = PatternFill(start_color="FFF2CC", end_color="FFF2CC", fill_type="solid") for ci in range(1, 9): ws.cell(row=tr, column=ci).fill = tot_fill; ws.cell(row=tr, column=ci).border = boundary ws.cell(row=tr, column=ci).font = Font(name="Consolas", size=10, bold=True) ws.cell(row=tr, column=ci).alignment = ctr ws.cell(row=tr, column=1, value="TOTAL") ws.cell(row=tr, column=3, value=sum(len([p for p in all_data[e]["ports"] if p["direction"]=="Tx"]) for e in OUR_ECUS)) ws.cell(row=tr, column=4, value=sum(len([p for p in all_data[e]["ports"] if p["direction"]=="Rx"]) for e in OUR_ECUS)) ws.cell(row=tr, column=5, value=sum(len(all_data[e]["ports"]) for e in OUR_ECUS)) ws.cell(row=tr, column=6, value=sum(len(all_data[e]["runnables"]) for e in OUR_ECUS)) ws.cell(row=tr, column=7, value=sum(len([r for r in all_data[e]["runnables"] if any(dr["direction"]=="Tx" for dr in r["data_refs"])]) for e in OUR_ECUS)) ws.cell(row=tr, column=8, value=sum(len([r for r in all_data[e]["runnables"] if any(dr["direction"]=="Rx" for dr in r["data_refs"])]) for e in OUR_ECUS)) for i, w in enumerate([16, 22, 13, 13, 13, 13, 13, 13], 1): ws.column_dimensions[get_column_letter(i)].width = w for ri in range(4, tr): if ri % 2 == 0: for ci in range(1, 9): ws.cell(row=ri, column=ci).fill = PatternFill(start_color="F2F2F2", end_color="F2F2F2", fill_type="solid") # ═══════════════════════════════════════════════════════════════════════════════ # Sheet 2: Port Details (one section per ECU, Port Name + Signal Name) # ═══════════════════════════════════════════════════════════════════════════════ ws2 = wb.create_sheet("2-Port Details") row = 1 for ecu in OUR_ECUS: d = all_data.get(ecu) if not d: continue # ECU header ws2.merge_cells(start_row=row, start_column=1, end_row=row, end_column=7) c = ws2.cell(row=row, column=1, value=f"{ecu} | P-Port(TX): {len([p for p in d['ports'] if p['direction']=='Tx'])} | R-Port(RX): {len([p for p in d['ports'] if p['direction']=='Rx'])}") c.font = Font(name="Microsoft YaHei", size=11, bold=True, color="FFFFFF") c.fill = PatternFill(start_color="2F5496", end_color="2F5496", fill_type="solid") c.alignment = Alignment(horizontal="left", vertical="center") for cc in range(2, 8): ws2.cell(row=row, column=cc).fill = PatternFill(start_color="2F5496", end_color="2F5496", fill_type="solid") ws2.row_dimensions[row].height = 22 row += 1 # Sub-header for ci, h in enumerate(["#", "Port Name", "Direction", "Signal Name", "Port Interface", "Port Type", "Route"], 1): c = ws2.cell(row=row, column=ci, value=h) c.font = sub_font; c.fill = sub_fill; c.alignment = ctr; c.border = boundary row += 1 for i, p in enumerate(d["ports"], 1): vals = [i, p["name"], "TX" if p["direction"]=="Tx" else "RX", p["signal_name"], f"Pi{p['signal_name']}", "P-PORT (Provider)" if p["direction"]=="Tx" else "R-PORT (Require)", "S/R → SWC" if p["direction"]=="Tx" else "SWC ← S/R"] for ci, v in enumerate(vals, 1): c = ws2.cell(row=row, column=ci, value=v) c.font = small_font c.alignment = ctr if ci <= 3 else wrap c.border = boundary if ci == 3: c.font = tx_font if p["direction"]=="Tx" else rx_font row += 1 row += 1 # blank ws2.column_dimensions["A"].width = 5 ws2.column_dimensions["B"].width = 38 ws2.column_dimensions["C"].width = 10 ws2.column_dimensions["D"].width = 38 ws2.column_dimensions["E"].width = 38 ws2.column_dimensions["F"].width = 22 ws2.column_dimensions["G"].width = 14 # ═══════════════════════════════════════════════════════════════════════════════ # Sheet 3: Runnable Details # ═══════════════════════════════════════════════════════════════════════════════ ws3 = wb.create_sheet("3-Runnable Details") row = 1 for ecu in OUR_ECUS: d = all_data.get(ecu) if not d: continue ws3.merge_cells(start_row=row, start_column=1, end_row=row, end_column=7) c = ws3.cell(row=row, column=1, value=f"{ecu} | Runnables: {len(d['runnables'])}") c.font = Font(name="Microsoft YaHei", size=11, bold=True, color="FFFFFF") c.fill = PatternFill(start_color="2F5496", end_color="2F5496", fill_type="solid") c.alignment = Alignment(horizontal="left", vertical="center") for cc in range(2, 8): ws3.cell(row=row, column=cc).fill = PatternFill(start_color="2F5496", end_color="2F5496", fill_type="solid") ws3.row_dimensions[row].height = 22 row += 1 for ci, h in enumerate(["#", "Runnable Name", "Direction", "Period (s)", "Period (ms)", "Signal Count", "Signals"], 1): c = ws3.cell(row=row, column=ci, value=h) c.font = sub_font; c.fill = sub_fill; c.alignment = ctr; c.border = boundary row += 1 for i, rn in enumerate(d["runnables"], 1): # Determine primary direction dirs = set(dr["direction"] for dr in rn["data_refs"]) direction = "Tx/Rx" if len(dirs) > 1 else (list(dirs)[0] if dirs else "N/A") try: period_ms = int(float(rn["period_sec"]) * 1000) except ValueError: period_ms = rn["period_sec"] signals = list(set(dr["signal_name"] for dr in rn["data_refs"])) vals = [i, rn["name"], direction, rn["period_sec"], period_ms, len(signals), ", ".join(sorted(signals))] for ci, v in enumerate(vals, 1): c = ws3.cell(row=row, column=ci, value=v) c.font = small_font c.alignment = ctr if ci <= 6 else wrap c.border = boundary if ci == 3: c.font = tx_font if direction=="Tx" else (rx_font if direction=="Rx" else small_font) row += 1 row += 1 ws3.column_dimensions["A"].width = 5 ws3.column_dimensions["B"].width = 42 ws3.column_dimensions["C"].width = 10 ws3.column_dimensions["D"].width = 12 ws3.column_dimensions["E"].width = 12 ws3.column_dimensions["F"].width = 12 ws3.column_dimensions["G"].width = 90 # ═══════════════════════════════════════════════════════════════════════════════ # Sheet 4: Signal -> Port -> Runnable Mapping (master mapping) # ═══════════════════════════════════════════════════════════════════════════════ ws4 = wb.create_sheet("4-Signal-Port-Runnable Map") ws4.merge_cells("A1:G1") ws4["A1"] = "Signal -> Port -> Runnable Full Mapping (All ECUs)" ws4["A1"].font = title_font ws4["A1"].alignment = Alignment(horizontal="center", vertical="center") ws4.row_dimensions[1].height = 30 for ci, h in enumerate(["ECU", "Signal Name", "TX Port", "RX Port", "TX Runnable", "RX Runnable", "Cycle (ms)"], 1): c = ws4.cell(row=3, column=ci, value=h) c.font = hdr_font; c.fill = hdr_fill; c.alignment = ctr; c.border = boundary ws4.row_dimensions[3].height = 22 row = 4 for ecu in OUR_ECUS: d = all_data.get(ecu) if not d: continue # Build per-signal view across all ports and runnables all_signals = set() for p in d["ports"]: all_signals.add(p["signal_name"]) for rn in d["runnables"]: for dr in rn["data_refs"]: all_signals.add(dr["signal_name"]) for sig in sorted(all_signals): tx_port = next((p["name"] for p in d["ports"] if p["signal_name"]==sig and p["direction"]=="Tx"), "") rx_port = next((p["name"] for p in d["ports"] if p["signal_name"]==sig and p["direction"]=="Rx"), "") tx_rns = [] rx_rns = [] for rn in d["runnables"]: for dr in rn["data_refs"]: if dr["signal_name"] == sig: if dr["direction"] == "Tx": tx_rns.append(rn["name"]) else: rx_rns.append(rn["name"]) # Get cycle time from first runnable cycle = "" for rn in d["runnables"]: for dr in rn["data_refs"]: if dr["signal_name"] == sig: try: cycle = f"{int(float(rn['period_sec'])*1000)}" except ValueError: cycle = rn["period_sec"] break if cycle: break # Deduplicate tx_rns_unique = list(dict.fromkeys(tx_rns)) rx_rns_unique = list(dict.fromkeys(rx_rns)) vals = [ecu, sig, tx_port, rx_port, "\n".join(tx_rns_unique), "\n".join(rx_rns_unique), cycle] for ci, v in enumerate(vals, 1): c = ws4.cell(row=row, column=ci, value=v) c.font = small_font c.alignment = ctr if ci <= 2 else wrap c.border = boundary if ci == 3: c.font = tx_font if tx_port else small_font elif ci == 4: c.font = rx_font if rx_port else small_font row += 1 ws4.column_dimensions["A"].width = 16 ws4.column_dimensions["B"].width = 36 ws4.column_dimensions["C"].width = 38 ws4.column_dimensions["D"].width = 38 ws4.column_dimensions["E"].width = 42 ws4.column_dimensions["F"].width = 42 ws4.column_dimensions["G"].width = 10 # ═══════════════════════════════════════════════════════════════════════════════ # Sheet 5: Cycle Time Distribution # ═══════════════════════════════════════════════════════════════════════════════ ws5 = wb.create_sheet("5-Cycle Time") ws5.merge_cells("A1:F1") ws5["A1"] = "Runnable Cycle Time Distribution" ws5["A1"].font = title_font ws5["A1"].alignment = Alignment(horizontal="center", vertical="center") ws5.row_dimensions[1].height = 30 for ci, h in enumerate(["Cycle (ms)", "Cycle (s)", "ECU Count", "Runnable Count", "ECUs", "Runnable Names"], 1): c = ws5.cell(row=3, column=ci, value=h) c.font = hdr_font; c.fill = hdr_fill; c.alignment = ctr; c.border = boundary cycle_groups = defaultdict(lambda: {"ecus": defaultdict(list), "total": 0}) for ecu in OUR_ECUS: d = all_data.get(ecu) if not d: continue for rn in d["runnables"]: try: ms = int(float(rn["period_sec"]) * 1000) except ValueError: ms = 0 cycle_groups[ms]["ecus"][ecu].append(rn["name"]) cycle_groups[ms]["total"] += 1 row = 4 for ms in sorted(cycle_groups.keys()): g = cycle_groups[ms] ecu_list = ", ".join(f"{e}({len(v)})" for e, v in sorted(g["ecus"].items())) all_names = [] for e, v in sorted(g["ecus"].items()): all_names.extend(v) vals = [ms, ms / 1000.0, len(g["ecus"]), g["total"], ecu_list, "\n".join(sorted(all_names))] for ci, v in enumerate(vals, 1): c = ws5.cell(row=row, column=ci, value=v) c.font = small_font c.alignment = ctr if ci <= 4 else wrap c.border = boundary row += 1 ws5.column_dimensions["A"].width = 12 ws5.column_dimensions["B"].width = 12 ws5.column_dimensions["C"].width = 12 ws5.column_dimensions["D"].width = 14 ws5.column_dimensions["E"].width = 40 ws5.column_dimensions["F"].width = 100 # ── Save ────────────────────────────────────────────────────────────────────── wb.save(OUTPUT) print(f"[OK] Excel: {OUTPUT}") print(f" Sheets: 1-Summary | 2-Port Details | 3-Runnable Details | 4-Signal-Port-Runnable Map | 5-Cycle Time") tot_pp = sum(len([p for p in all_data[e]["ports"] if p["direction"]=="Tx"]) for e in OUR_ECUS) tot_rp = sum(len([p for p in all_data[e]["ports"] if p["direction"]=="Rx"]) for e in OUR_ECUS) tot_rn = sum(len(all_data[e]["runnables"]) for e in OUR_ECUS) print(f" Total P-Port(TX): {tot_pp} | R-Port(RX): {tot_rp} | Runnables: {tot_rn}")提示:
只需要指定这两个参数内容即可:
COMP_DIR :arxml数据文件所在的文件夹路径
OUTPUT :Excel输出路径
最终输出的数据表如下: