129 lines
4.5 KiB
Python
129 lines
4.5 KiB
Python
import re
|
|
import csv
|
|
import polars as pl
|
|
from pathlib import Path
|
|
from typing import Union
|
|
|
|
PathLike = Union[str, Path]
|
|
|
|
def parse_log(input_path: PathLike, out_bus1: PathLike, out_bus2: PathLike) -> None:
|
|
"""
|
|
Parses a raw CAN bus log from CANdigger using regex and saves valid frames to separate CSV.
|
|
Corrupted, incomplete, or debug frames are silently discarded.
|
|
"""
|
|
input_file = Path(input_path)
|
|
out1_file = Path(out_bus1)
|
|
out2_file = Path(out_bus2)
|
|
|
|
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)
|
|
|
|
def parse_csv(csv_path: PathLike) -> pl.LazyFrame:
|
|
"""
|
|
Ingests a parsed CSV file, unpacks hex strings into 8 integer columns,
|
|
generates sequential timestamps if missing, and returns a Polars LazyFrame.
|
|
"""
|
|
lf = pl.scan_csv(csv_path, schema_overrides={"ID": pl.String, "Data": pl.String})
|
|
|
|
if "Timestamp" not in lf.columns:
|
|
lf = lf.with_row_index("Timestamp")
|
|
|
|
byte_exprs = []
|
|
for i in range(8):
|
|
expr = (
|
|
pl.col("Data").str.strip_chars().str.split(" ")
|
|
.list.get(i, null_on_oob=True)
|
|
.str.to_integer(base=16, strict=False)
|
|
.alias(f"b{i}").cast(pl.UInt8)
|
|
)
|
|
byte_exprs.append(expr)
|
|
|
|
lf = lf.with_columns(byte_exprs).drop("Data")
|
|
|
|
return lf.with_columns([
|
|
pl.col("DLC").cast(pl.UInt8),
|
|
pl.col("Timestamp").cast(pl.Float64)
|
|
])
|
|
|
|
def parse_parquet(parquet_path: PathLike) -> pl.LazyFrame:
|
|
"""
|
|
Loads a pre-optimized Parquet file directly into a Polars LazyFrame.
|
|
"""
|
|
return pl.scan_parquet(parquet_path)
|
|
|
|
def save_parquet(lf: pl.LazyFrame, output_path: PathLike) -> None:
|
|
"""
|
|
Materializes the LazyFrame query plan and saves it to a Parquet file.
|
|
"""
|
|
lf.sink_parquet(output_path)
|
|
|
|
if __name__ == '__main__':
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="CAN Bus Data Engine & Parser")
|
|
subparsers = parser.add_subparsers(dest="command", required=True, help="Available commands")
|
|
|
|
parser_csv = subparsers.add_parser("csv", help="Parse raw text log into Bus 1 and Bus 2 CSVs")
|
|
parser_csv.add_argument("input", help="Path to the .txt log file from CANdigger")
|
|
parser_csv.add_argument("out_bus1", help="Output CSV filename for Bus 1 (C1)")
|
|
parser_csv.add_argument("out_bus2", help="Output CSV filename for Bus 2 (C2)")
|
|
|
|
parser_parquet = subparsers.add_parser("parquet", help="Convert a parsed CSV into an optimized Parquet file")
|
|
parser_parquet.add_argument("input_csv", help="Path to the input .csv file")
|
|
parser_parquet.add_argument("output_parquet", help="Path to the output .parquet file")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "csv":
|
|
parse_log(args.input, args.out_bus1, args.out_bus2)
|
|
print(f"[+] Successfully csved logs to {args.out_bus1} and {args.out_bus2}")
|
|
|
|
elif args.command == "parquet":
|
|
print(f"[*] Processing {args.input_csv}...")
|
|
lf = parse_csv(args.input_csv)
|
|
save_parquet(lf, args.output_parquet)
|
|
print(f"[+] Successfully optimized and saved to {args.output_parquet}")
|