process replay: use zst (#33138)

* test

* update refs to zst

* update

selfdrive/test/process_replay/ref_commit

@@ -1 +1 @@
-7eb60abe8030b97f79f858315d67d080704733f7
+6ca7313b9c1337fad1334d9e087cb4984fdce74d

selfdrive/test/process_replay/regen.py

@@ -74,7 +74,7 @@ def setup_data_readers(
         assert device_type != "neo", "Driver camera not supported on neo segments. Use dummy dcamera."
         frs['driverCameraState'] = FrameReader(r.dcamera_paths()[sidx])
   else:
-    lr = LogReader(f"cd:/{route.replace('|', '/')}/{sidx}/rlog.bz2")
+    lr = LogReader(f"{route}/{sidx}/r")
     frs = {}
     if needs_road_cam:
       frs['roadCameraState'] = FrameReader(f"cd:/{route.replace('|', '/')}/{sidx}/fcamera.hevc")
@@ -118,7 +118,7 @@ def regen_and_save(
 
   log_dir = os.path.join(outdir, time.strftime("%Y-%m-%d--%H-%M-%S--0", time.gmtime()))
   rel_log_dir = os.path.relpath(log_dir)
-  rpath = os.path.join(log_dir, "rlog.bz2")
+  rpath = os.path.join(log_dir, "rlog.zst")
 
   os.makedirs(log_dir)
   save_log(rpath, output_logs, compress=True)

selfdrive/test/process_replay/test_processes.py

@@ -198,11 +198,11 @@ if __name__ == "__main__":
         if cfg.proc_name not in tested_procs:
           continue
 
-        cur_log_fn = os.path.join(FAKEDATA, f"{segment}_{cfg.proc_name}_{cur_commit}.bz2")
+        cur_log_fn = os.path.join(FAKEDATA, f"{segment}_{cfg.proc_name}_{cur_commit}.zst")
         if args.update_refs:  # reference logs will not exist if routes were just regenerated
           ref_log_path = get_url(*segment.rsplit("--", 1))
         else:
-          ref_log_fn = os.path.join(FAKEDATA, f"{segment}_{cfg.proc_name}_{ref_commit}.bz2")
+          ref_log_fn = os.path.join(FAKEDATA, f"{segment}_{cfg.proc_name}_{ref_commit}.zst")
           ref_log_path = ref_log_fn if os.path.exists(ref_log_fn) else BASE_URL + os.path.basename(ref_log_fn)
 
         dat = None if args.upload_only else log_data[segment]

tools/lib/logreader.py

@@ -30,8 +30,10 @@ RawLogIterable = Iterable[bytes]
 def save_log(dest, log_msgs, compress=True):
   dat = b"".join(msg.as_builder().to_bytes() for msg in log_msgs)
 
-  if compress:
+  if compress and dest.endswith(".bz2"):
     dat = bz2.compress(dat)
+  elif compress and dest.endswith(".zst"):
+    dat = zstd.compress(dat, 10)
 
   with open(dest, "wb") as f:
     f.write(dat)