From 6e386fcb4bc813fecd8b0f317d2662151f7afc92 Mon Sep 17 00:00:00 2001
From: Erick Ahmed <engineering@erickahmed.com>
Date: Fri, 10 Jul 2026 01:24:57 +0200
Subject: [PATCH] Add stub to decode J1939 and calculate useful values

---
 src/j1939_decoder.py | 62 +++++++++++++++++++++++++++++++++++++-------
 1 file changed, 52 insertions(+), 10 deletions(-)

diff --git a/src/j1939_decoder.py b/src/j1939_decoder.py
index c8985fe..239d100 100644
--- a/src/j1939_decoder.py
+++ b/src/j1939_decoder.py
@@ -41,17 +41,59 @@ def decode_j1939_metadata(lf: pl.LazyFrame) -> pl.LazyFrame:
         ]).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)
+
+    b = [pl.col(f"b{i}").fill_null("00").str.to_integer(base=16, strict=False) for i in range(8)]
+    pgn_j = pl.when(is_j1939).then(pgn).otherwise(None)
+
+    data_1 = pl.when(pgn_j.is_in([61444, 65331])).then(1 * 1)
+    data_2 = pl.when(pgn_j.is_in([61444, 65331])).then(1 * 1)
+    data_3 = pl.when(pgn_j == 65265).then(1 * 1)
+    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")
+    ])
+
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description="CAN Log J1939 Decoder")
-    parser.add_argument("input_parquet", help="Path to the parsed .parquet file")
+    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()
 
-    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_metadata(lf)
-
-    print(f"[*] Saving decoded data to {args.output_parquet}")
-    lf_decoded.sink_parquet(args.output_parquet)
+    df = pl.scan_parquet(args.input_parquet).collect()
+    decoded_df = decode_j1939_frames(df)
+    decoded_df.write_parquet(args.output_parquet)