Remove J1939 stub for data calculation

- To be done separately

.python-version

@@ -0,0 +1 @@
+3.14

LICENSE

@@ -633,8 +633,8 @@ the "copyright" line and a pointer to where the full notice is found.
 Copyright (C) <year>  <name of author>
 
 This program is free software: you can redistribute it and/or modify
-    it under the terms of the GNU Affero General Public License as published by
-    the Free Software Foundation, either version 3 of the License, or
+it under the terms of the GNU Affero General Public License as published
+by the Free Software Foundation, either version 3 of the License, or
 (at your option) any later version.
 
 This program is distributed in the hope that it will be useful,

parser.py

@@ -1,81 +0,0 @@
-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

pyproject.toml

@@ -0,0 +1,7 @@
+[project]
+name = "CANveyor"
+version = "0.1.0"
+description = "J1939 CAN bus parser that works in pair with CANdigger"
+readme = "README.md"
+requires-python = ">=3.14"
+dependencies = ["polars", "pathlib", "typing"]

src/j1939_decoder.py

@@ -0,0 +1,75 @@
+import argparse
+import polars as pl
+
+def get_j1939_mask() -> pl.Expr:
+    """
+    Returns a Polars expression representing the strict J1939 filtering rules.
+    """
+    id_int = pl.col("ID").str.to_integer(base=16).cast(pl.UInt32)
+    return (
+        (id_int > 0x7FF) &
+        ((id_int % 33554432 // 16777216) == 0) &
+        (pl.col("DLC") <= 8)
+    )
+
+def decode_j1939_metadata(lf: pl.LazyFrame) -> pl.LazyFrame:
+    """
+    Decodes J1939 fields and bundles them into a Struct column.
+    """
+    id_int = pl.col("ID").str.to_integer(base=16).cast(pl.UInt32)
+
+    id_shifted_8 = id_int // 256
+    id_shifted_16 = id_int // 65536
+
+    priority = ((id_int // 67108864) % 8).cast(pl.UInt8)
+    pf = (id_shifted_16 % 256).cast(pl.UInt8)
+    ps = (id_shifted_8 % 256).cast(pl.UInt8)
+    sa = (id_int % 256).cast(pl.UInt8)
+
+    da = pl.when(pf < 240).then(ps).otherwise(pl.lit(255, dtype=pl.UInt8))
+
+    pgn = pl.when(pf < 240).then(id_shifted_8 % 65536).otherwise(id_shifted_8 % 262144).cast(pl.UInt32)
+
+    return lf.with_columns(
+        pl.struct([
+            priority.alias("Priority"),
+            pf.alias("PF"),
+            ps.alias("PS"),
+            sa.alias("SA"),
+            da.alias("DA"),
+            pgn.alias("PGN")
+        ]).alias("j1939_metadata")
+    )
+
+def decode_j1939_frames(df: pl.DataFrame) -> pl.DataFrame:
+    id_int = pl.col("ID").str.to_integer(base=16).cast(pl.UInt32)
+
+    is_j1939 = (id_int > 0x7FF) & ((id_int % 33554432 // 16777216) == 0) & (pl.col("DLC") <= 8)
+
+    priority = (id_int // 67108864) % 8
+    pf = (id_int // 65536) % 256
+    ps = (id_int // 256) % 256
+    sa = id_int % 256
+    da = pl.when(pf < 240).then(ps).otherwise(255)
+    pgn = pl.when(pf < 240).then((id_int // 256) % 65536).otherwise((id_int // 256) % 262144)
+
+    j1939_meta = pl.when(is_j1939).then(
+        pl.struct([
+            priority.cast(pl.UInt8).alias("Priority"),
+            pf.cast(pl.UInt8).alias("PF"),
+            ps.cast(pl.UInt8).alias("PS"),
+            sa.cast(pl.UInt8).alias("SA"),
+            da.cast(pl.UInt8).alias("DA"),
+            pgn.cast(pl.UInt32).alias("PGN")
+        ])
+    ).otherwise(None)
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="J1939 decoder")
+    parser.add_argument("input_parquet", help="Path to the raw .parquet file")
+    parser.add_argument("output_parquet", help="Path to save the decoded .parquet file")
+    args = parser.parse_args()
+
+    df = pl.scan_parquet(args.input_parquet).collect()
+    decoded_df = decode_j1939_frames(df)
+    decoded_df.write_parquet(args.output_parquet)

src/parser.py

@@ -0,0 +1,115 @@
+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 hex 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)
+            .alias(f"b{i}")
+        )
+        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)
+    ])
+
+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"[+] Saved csv file to {args.out_bus1} and {args.out_bus2}")
+
+    elif args.command == "parquet":
+        print(f"[*] Processing {args.input_csv}...")
+        lf = parse_csv(args.input_csv)
+        lf.sink_parquet(args.output_parquet)
+        print(f"[+] Saved parquet file to {args.output_parquet}")