Decode J1939 metadata for compliant frames and add a struct with J1939

metadata

src/j1939_decoder.py

@@ -2,13 +2,55 @@ 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 & 0x02000000) == 0) &
+        ((id_int % 33554432 // 16777216) == 0) &
         (pl.col("DLC") <= 8)
     )
 
+def decode_j1939(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)
+    mask = get_j1939_mask()
 
+    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(pl.lit(255, dtype=pl.UInt8))
+    pgn = pl.when(pf < 240).then((id_int // 256) % 65536).otherwise((id_int // 256) % 262144).cast(pl.UInt32)
+
+    j1939_struct = 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.alias("PGN")
+    ])
+
+    return lf.with_columns(
+        pl.when(mask).then(j1939_struct).otherwise(None).alias("j1939_metadata")
+    )
 
 if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="CAN Log J1939 Decoder")
+    parser.add_argument("input_parquet", help="Path to the parsed .parquet file")
+    parser.add_argument("output_parquet", help="Path to save the decoded .parquet file")
+    args = parser.parse_args()
+
+    print(f"[*] Loading {args.input_parquet}")
+    lf = pl.scan_parquet(args.input_parquet)
+
+    print("[*] Decoding J1939 IDs into a nested Struct column")
+    lf_decoded = decode_j1939(lf)
+
+    print(f"[*] Saving decoded data to {args.output_parquet}")
+    lf_decoded.sink_parquet(args.output_parquet)