Add parquet optimizer for faster data analysis in the future

parser.py

@@ -1,21 +1,21 @@
 import re
 import csv
+import polars as pl
 from pathlib import Path
+from typing import Union
 
-def parse_can_log(input_path: str | Path, out_bus1: str | Path, out_bus2: str | Path) -> None:
+PathLike = Union[str, Path]
+
+def parse_log(input_path: PathLike, out_bus1: PathLike, out_bus2: PathLike) -> None:
     """
-    Parses a CAN bus log file from CANdigger and saves valid frames to separate CSV files for Bus 1 and Bus 2.
+    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)
 
-    # 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, \
@@ -61,21 +61,68 @@ def parse_can_log(input_path: str | Path, out_bus1: str | Path, out_bus2: str |
                     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__':
-    # Example usage:
+    import argparse
-    # 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__':
+    parser = argparse.ArgumentParser(description="CAN Bus Data Engine & Parser")
-        import argparse
+    subparsers = parser.add_subparsers(dest="command", required=True, help="Available commands")
 
-        parser = argparse.ArgumentParser(description="Parse CAN bus logs to separate CSV files")
+    parser_csv = subparsers.add_parser("csv", help="Parse raw text log into Bus 1 and Bus 2 CSVs")
-        parser.add_argument("input", help="Path to the input .txt log file")
+    parser_csv.add_argument("input", help="Path to the .txt log file from CANdigger")
-        parser.add_argument("out_bus1", help="Output CSV filename for Bus 1 (C1)")
+    parser_csv.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)")
+    parser_csv.add_argument("out_bus2", help="Output CSV filename for Bus 2 (C2)")
 
-        args = parser.parse_args()
+    parser_parquet = subparsers.add_parser("parquet", help="Convert a parsed CSV into an optimized Parquet file")
-        parse_can_log(args.input, args.out_bus1, args.out_bus2)
+    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")
 
-    pass
+    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}")