Files
CANveyor/parser.py
T

82 lines
2.7 KiB
Python

import re
import csv
from pathlib import Path
def parse_can_log(input_path: str | Path, out_bus1: str | Path, out_bus2: str | Path) -> None:
"""
Parses a CAN bus log file from CANdigger and saves valid frames to separate CSV files for Bus 1 and Bus 2.
Corrupted, incomplete, or debug frames are silently discarded.
"""
input_file = Path(input_path)
out1_file = Path(out_bus1)
out2_file = Path(out_bus2)
# Group 1: Bus (C1 or C2)
# Group 2: ID (1 to 8 hex chars)
# Group 3: DLC (1 to 2 hex chars)
start_pattern = re.compile(r'(C[12]):([0-9A-Fa-f]{1,8})\s+([0-9A-Fa-f]{1,2})\s+')
byte_pattern = re.compile(r'^[0-9A-Fa-f]{2}$')
with input_file.open('r', encoding='utf-8') as f_in, \
out1_file.open('w', newline='', encoding='utf-8') as f_out1, \
out2_file.open('w', newline='', encoding='utf-8') as f_out2:
writer1 = csv.writer(f_out1)
writer2 = csv.writer(f_out2)
writer1.writerow(['ID', 'DLC', 'Data'])
writer2.writerow(['ID', 'DLC', 'Data'])
for line in f_in:
for match in start_pattern.finditer(line):
bus = match.group(1)
can_id = match.group(2).upper()
dlc_str = match.group(3)
try:
dlc = int(dlc_str, 16)
except ValueError:
continue
if dlc > 8:
continue
remaining = line[match.end():]
tokens = remaining.split()
data_bytes = []
for token in tokens:
if byte_pattern.match(token):
data_bytes.append(token.upper())
else:
break
if len(data_bytes) == dlc:
data_str = ' '.join(data_bytes)
row = [can_id, dlc, data_str]
if bus == 'C1':
writer1.writerow(row)
elif bus == 'C2':
writer2.writerow(row)
if __name__ == '__main__':
# Example usage:
# parse_can_log('can_traffic.txt', 'bus1_output.csv', 'bus2_output.csv')
# or on CLI:
# python3 can_parser.py can_traffic.txt' bus1_output.csv bus2_output.csv
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description="Parse CAN bus logs to separate CSV files")
parser.add_argument("input", help="Path to the input .txt log file")
parser.add_argument("out_bus1", help="Output CSV filename for Bus 1 (C1)")
parser.add_argument("out_bus2", help="Output CSV filename for Bus 2 (C2)")
args = parser.parse_args()
parse_can_log(args.input, args.out_bus1, args.out_bus2)
pass