Add stub to decode J1939 and calculate useful values

src/j1939_decoder.py

@@ -17,40 +17,83 @@ 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)
-    mask = get_j1939_mask()
+
+    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)
 
-    da = pl.when(pf < 240).then(ps).otherwise(pl.lit(255, dtype=pl.UInt8))
+    j1939_meta = pl.when(is_j1939).then(
-    pgn = pl.when(pf < 240).then((id_int // 256) % 65536).otherwise((id_int // 256) % 262144).cast(pl.UInt32)
+        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)
 
-    j1939_struct = pl.struct([
+    b = [pl.col(f"b{i}").fill_null("00").str.to_integer(base=16, strict=False) for i in range(8)]
-        priority.cast(pl.UInt8).alias("Priority"),
+    pgn_j = pl.when(is_j1939).then(pgn).otherwise(None)
-        pf.cast(pl.UInt8).alias("PF"),
+
-        ps.cast(pl.UInt8).alias("PS"),
+    data_1 = pl.when(pgn_j.is_in([61444, 65331])).then(1 * 1)
-        sa.cast(pl.UInt8).alias("SA"),
+    data_2 = pl.when(pgn_j.is_in([61444, 65331])).then(1 * 1)
-        da.cast(pl.UInt8).alias("DA"),
+    data_3 = pl.when(pgn_j == 65265).then(1 * 1)
-        pgn.alias("PGN")
+    data_4 = pl.when(pgn_j == 65226).then(1 * 1)
+    data_5 = pl.when(pgn_j == 65271).then(1 * 1)
+
+    data_col = pl.coalesce(
+        pl.when(data_1.is_not_null()).then(pl.lit("VALUE 1")),
+        pl.when(data_2.is_not_null()).then(pl.lit("VALUE 2")),
+        pl.when(data_3.is_not_null()).then(pl.lit("VALUE 3")),
+        pl.when(data_4.is_not_null()).then(pl.lit("VALUE 4")),
+        pl.when(data_5.is_not_null()).then(pl.lit("VALUE 5")),
+    )
+    value_col = pl.coalesce(data_1, data_2, data_3, data_4, data_5)
+
+    return df.with_columns([
+        j1939_meta.alias("j1939_metadata"),
+        data_col.alias("data"),
+        value_col.alias("value")
     ])
 
-    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 = argparse.ArgumentParser(description="CAN Bus Full Decoder")
-    parser.add_argument("input_parquet", help="Path to the parsed .parquet file")
+    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()
 
-    print(f"[*] Loading {args.input_parquet}")
+    df = pl.scan_parquet(args.input_parquet).collect()
-    lf = pl.scan_parquet(args.input_parquet)
+    decoded_df = decode_j1939_frames(df)
-
+    decoded_df.write_parquet(args.output_parquet)
-    print("[*] Decoding J1939 IDs into a nested Struct column")
-    lf_decoded = decode_j1939_metadata(lf)
-
-    print(f"[*] Saving decoded data to {args.output_parquet}")
-    lf_decoded.sink_parquet(args.output_parquet)