tools: use log migration in cpp tools (#38571)

* Universal PyProcess module runner. Migration cli

* print to stderr

* Migration in jotpluggler

* Do -

* Improvements

* Explicit stdout

* Existing master pyprocess changes

* PyTools

* Fix runtime warning

(cherry picked from commit b9c815d56a2827796ad73ca8f09d186243dd8c17)

openpilot/selfdrive/test/process_replay/migration.py

@@ -4,6 +4,8 @@ from typing import cast
 import capnp
 import functools
 import traceback
+import sys
+import argparse
 
 from openpilot.cereal import messaging, log
 from opendbc.car.structs import car
@@ -17,7 +19,7 @@ from openpilot.selfdrive.modeld.fill_model_msg import fill_xyz_poly, fill_lane_l
 from openpilot.selfdrive.test.process_replay.vision_meta import meta_from_encode_index
 from openpilot.selfdrive.controls.lib.drive_helpers import CONTROL_N, get_accel_from_plan, should_stop
 from openpilot.system.manager.process_config import managed_processes
-from openpilot.tools.lib.logreader import LogIterable
+from openpilot.tools.lib.logreader import LogIterable, LogReader, save_log
 
 MessageWithIndex = tuple[int, capnp.lib.capnp._DynamicStructReader]
 MigrationOps = tuple[list[tuple[int, capnp.lib.capnp._DynamicStructReader]], list[capnp.lib.capnp._DynamicStructReader], list[int]]
@@ -387,7 +389,7 @@ def migrate_cameraStates(msgs):
 
     encode_id = frame_to_encode_id[msg.which()].get(camera_state.frameId)
     if encode_id is None:
-      print(f"Missing encoded frame for camera feed {msg.which()} with frameId: {camera_state.frameId}")
+      print(f"Missing encoded frame for camera feed {msg.which()} with frameId: {camera_state.frameId}", file=sys.stderr)
       if len(frame_to_encode_id[msg.which()]) != 0:
         del_ops.append(index)
         continue
@@ -395,7 +397,7 @@ def migrate_cameraStates(msgs):
       # fallback mechanism for logs without encodeIdx (e.g. logs from before 2022 with driver recording disabled)
       # try to fake encode_id by subtracting lowest frameId
       encode_id = camera_state.frameId - min_frame_id[msg.which()]
-      print(f"Faking encodeId to {encode_id} for camera feed {msg.which()} with frameId: {camera_state.frameId}")
+      print(f"Faking encodeId to {encode_id} for camera feed {msg.which()} with frameId: {camera_state.frameId}", file=sys.stderr)
 
     new_msg = messaging.new_message(msg.which())
     new_camera_state = getattr(new_msg, new_msg.which())
@@ -527,3 +529,14 @@ def migrate_driverMonitoringState(msgs):
     ops.append((index, as_reader(new_msg)))
 
   return ops, [], []
+
+
+if __name__ == '__main__':
+  parser = argparse.ArgumentParser(description="Migrate logs")
+  parser.add_argument("input_path", help="Segment identifier or path to file")
+  parser.add_argument("output_path", help="Path to output file")
+  args = parser.parse_args()
+
+  mlr = migrate_all(LogReader(args.input_path))
+  print(f"Saving migrated log to {args.output_path}", file=sys.stderr)
+  save_log(args.output_path , mlr)

openpilot/selfdrive/test/process_replay/process_replay.py

@@ -27,7 +27,6 @@ from openpilot.common.realtime import DT_CTRL
 from openpilot.system.camerad.cameras.nv12_info import get_nv12_info
 from openpilot.system.manager.process_config import managed_processes
 from openpilot.selfdrive.test.process_replay.vision_meta import meta_from_camera_state, available_streams
-from openpilot.selfdrive.test.process_replay.migration import migrate_all
 from openpilot.selfdrive.test.process_replay.capture import ProcessOutputCapture
 from openpilot.tools.lib.logreader import LogIterable
 from openpilot.tools.lib.framereader import FrameReader
@@ -637,6 +636,8 @@ def replay_process(
   else:
     cfgs = list(cfg)
 
+  from openpilot.selfdrive.test.process_replay.migration import migrate_all
+
   all_msgs = migrate_all(lr,
                          manager_states=True,
                          panda_states=any("pandaStates" in cfg.pubs for cfg in cfgs),

openpilot/tools/cabana/routes.cc

@@ -6,7 +6,7 @@
 #include <thread>
 
 #include "json11/json11.hpp"
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_tools.h"
 
 namespace routes {
 
@@ -91,7 +91,7 @@ std::vector<RouteInfo> parseRoutes(const std::string &json, bool preserved) {
 
 void fetchDevices(DevicesCallback callback) {
   std::thread([callback = std::move(callback)]() {
-    std::string result = PyDownloader::getDevices();
+    std::string result = PyTools::getDevices();
     auto [success, error_code] = checkApiResponse(result);
     callback(success ? parseDevices(result) : std::vector<DeviceInfo>{}, success, error_code);
   }).detach();
@@ -106,7 +106,7 @@ void fetchRoutes(const std::string &dongle_id, int period_days, RoutesCallback c
   }
 
   std::thread([dongle_id, start_ms, end_ms, preserved, callback = std::move(callback)]() {
-    std::string result = PyDownloader::getDeviceRoutes(dongle_id, start_ms, end_ms, preserved);
+    std::string result = PyTools::getDeviceRoutes(dongle_id, start_ms, end_ms, preserved);
     auto [success, error_code] = checkApiResponse(result);
     callback(success ? parseRoutes(result, preserved) : std::vector<RouteInfo>{}, success, error_code);
   }).detach();

openpilot/tools/cabana/routes.h

@@ -21,7 +21,7 @@ struct RouteInfo {
 using DevicesCallback = std::function<void(std::vector<DeviceInfo> devices, bool success, int error_code)>;
 using RoutesCallback = std::function<void(std::vector<RouteInfo> routes, bool success, int error_code)>;
 
-// Parse a PyDownloader JSON response into (success, error_code).
+// Parse a PyTools JSON response into (success, error_code).
 std::pair<bool, int> checkApiResponse(const std::string &result);
 
 int64_t nowUnixMs();

openpilot/tools/cabana/ui/dialogs/routesdialog.cc

@@ -6,7 +6,7 @@
 #include <thread>
 
 #include "json11/json11.hpp"
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_tools.h"
 #include "tools/cabana/ui/theme.h"
 
 #include "imgui.h"
@@ -157,7 +157,7 @@ void RoutesDialog::signIn(const std::string &provider) {
   s_.auth_error.clear();
   auth_abort_ = std::make_shared<std::atomic<bool>>(false);
   std::thread([this, alive = std::weak_ptr<bool>(alive_), abort = auth_abort_, provider]() {
-    const std::string result = PyDownloader::authenticate(provider, abort.get());
+    const std::string result = PyTools::authenticate(provider, abort.get());
     utils::runOnMainThread(utils::guarded(alive.lock(), [this, abort, result]() {
       if (*abort) return;
       auth_abort_.reset();

openpilot/tools/cabana/ui/mainwin.cc

@@ -25,7 +25,7 @@
 #include "tools/cabana/ui/util.h"
 #include "tools/cabana/utils/export.h"
 #include "tools/cabana/utils/util.h"
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_tools.h"
 #include "tools/replay/util.h"
 
 namespace {

openpilot/tools/jotpluggler/app.cc

@@ -1821,6 +1821,7 @@ int run(const Options &options) {
     .route_name = options.route_name,
     .data_dir = options.data_dir,
     .dbc_override = {},
+    .migrate = options.migrate,
     .stream_source = StreamSourceConfig{.kind = is_local_stream_address(options.stream_address)
                                                  ? StreamSourceKind::CerealLocal
                                                  : StreamSourceKind::CerealRemote,

openpilot/tools/jotpluggler/app.h

@@ -34,6 +34,7 @@ struct Options {
   bool show = false;
   bool sync_load = false;
   bool stream = false;
+  bool migrate = true;
   double stream_buffer_seconds = 30.0;
 };
 
@@ -435,6 +436,7 @@ std::vector<RouteSeries> decode_can_messages(const std::vector<CanMessageData> &
 RouteData load_route_data(const std::string &route_name,
                           const std::string &data_dir = {},
                           const std::string &dbc_name = {},
+                          bool migrate = true,
                           const RouteLoadProgressCallback &progress = {});
 RouteIdentifier parse_route_identifier(std::string_view route_name);
 void rebuild_gps_trace(RouteData *route_data);
@@ -508,6 +510,7 @@ struct AppSession {
   std::string route_name;
   std::string data_dir;
   std::string dbc_override;
+  bool migrate = true;
   StreamSourceConfig stream_source;
   double stream_buffer_seconds = 30.0;
   SessionDataMode data_mode = SessionDataMode::Route;
@@ -885,7 +888,7 @@ public:
   AsyncRouteLoader(const AsyncRouteLoader &) = delete;
   AsyncRouteLoader &operator=(const AsyncRouteLoader &) = delete;
 
-  void start(const std::string &route_name, const std::string &data_dir, const std::string &dbc_name);
+  void start(const std::string &route_name, const std::string &data_dir, const std::string &dbc_name, bool migrate);
   RouteLoadSnapshot snapshot() const;
   bool consume(RouteData *route_data, std::string *error_text);
 

openpilot/tools/jotpluggler/layout.cc

@@ -547,7 +547,7 @@ bool save_layout(AppSession *session, UiState *state, const std::string &layout_
 
 void rebuild_session_route_data(AppSession *session, UiState *state,
                                 const RouteLoadProgressCallback &progress) {
-  apply_route_data(session, state, load_route_data(session->route_name, session->data_dir, session->dbc_override, progress));
+  apply_route_data(session, state, load_route_data(session->route_name, session->data_dir, session->dbc_override, session->migrate, progress));
 }
 
 void stop_stream_session(AppSession *session, UiState *state, bool preserve_data) {
@@ -627,7 +627,7 @@ void start_async_route_load(AppSession *session, UiState *state) {
     return;
   }
   apply_route_data(session, state, RouteData{});
-  session->route_loader->start(session->route_name, session->data_dir, session->dbc_override);
+  session->route_loader->start(session->route_name, session->data_dir, session->dbc_override, session->migrate);
   state->status_text = session->route_name.empty() ? "Ready" : "Loading route " + session->route_name;
 }
 

openpilot/tools/jotpluggler/main.cc

@@ -22,6 +22,7 @@ void print_usage(const char *argv0) {
       << "  --output <png>\n"
       << "  --show\n"
       << "  --sync-load\n"
+      << "  --no-migration\n"
       << "\n"
       << "Examples:\n"
       << "  " << argv0 << "\n"
@@ -94,6 +95,8 @@ int main(int argc, char *argv[]) {
       options.show = true;
     } else if (arg == "--sync-load") {
       options.sync_load = true;
+    } else if (arg == "--no-migration") {
+      options.migrate = false;
     } else if (arg == "--help" || arg == "-h") {
       print_usage(argv[0]);
       return 0;

openpilot/tools/jotpluggler/runtime.cc

@@ -261,13 +261,14 @@ struct AsyncRouteLoader::Impl {
     join();
   }
 
-  void start(const std::string &route_name_value, const std::string &data_dir_value, const std::string &dbc_name_value) {
+  void start(const std::string &route_name_value, const std::string &data_dir_value, const std::string &dbc_name_value, bool migrate_value) {
     join();
     {
       std::lock_guard<std::mutex> lock(mutex);
       route_name = route_name_value;
       data_dir = data_dir_value;
       dbc_name = dbc_name_value;
+      migrate = migrate_value;
       result.reset();
       error_text.clear();
     }
@@ -283,7 +284,7 @@ struct AsyncRouteLoader::Impl {
 
     worker = std::thread([this]() {
       try {
-        RouteData route_data = load_route_data(route_name, data_dir, dbc_name, [this](const RouteLoadProgress &progress) {
+        RouteData route_data = load_route_data(route_name, data_dir, dbc_name, migrate, [this](const RouteLoadProgress &progress) {
           total_segments.store(progress.total_segments > 0 ? progress.total_segments : progress.segment_count);
           segments_downloaded.store(progress.segments_downloaded);
           segments_parsed.store(progress.segments_parsed);
@@ -347,6 +348,7 @@ struct AsyncRouteLoader::Impl {
   std::string route_name;
   std::string data_dir;
   std::string dbc_name;
+  bool migrate = true;
   std::string error_text;
   std::atomic<bool> active{false};
   std::atomic<bool> completed{false};
@@ -362,8 +364,8 @@ AsyncRouteLoader::AsyncRouteLoader(bool enable_terminal_progress)
 
 AsyncRouteLoader::~AsyncRouteLoader() = default;
 
-void AsyncRouteLoader::start(const std::string &route_name, const std::string &data_dir, const std::string &dbc_name) {
-  impl_->start(route_name, data_dir, dbc_name);
+void AsyncRouteLoader::start(const std::string &route_name, const std::string &data_dir, const std::string &dbc_name, bool migrate) {
+  impl_->start(route_name, data_dir, dbc_name, migrate);
 }
 
 RouteLoadSnapshot AsyncRouteLoader::snapshot() const {

openpilot/tools/jotpluggler/sketch_layout.cc

@@ -25,7 +25,7 @@
 #include "common/util.h"
 #include "json11/json11.hpp"
 #include "tools/replay/logreader.h"
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_tools.h"
 
 namespace fs = std::filesystem;
 
@@ -331,7 +331,7 @@ std::map<int, SegmentLogs> load_segments_from_json(const json11::Json &json) {
 
 std::map<int, SegmentLogs> load_segments_from_server(const RouteSelection &route) {
   const std::string selector = route.selector == LogSelector::RLog ? "r" : route.selector == LogSelector::QLog ? "q" : "a";
-  const std::string result = PyDownloader::resolveRouteFiles(route.canonical_name, route.begin_segment, route.end_segment, selector);
+  const std::string result = PyTools::resolveRouteFiles(route.canonical_name, route.begin_segment, route.end_segment, selector);
   if (result.empty()) throw std::runtime_error("Failed to fetch route files for " + route.canonical_name);
 
   std::string parse_error;
@@ -1577,6 +1577,7 @@ LoadedRouteArtifacts load_route_series_parallel(
     const dbc::Database *can_dbc,
     LogSelector selector,
     bool skip_raw_can,
+    bool migrate,
     LoadStats *stats) {
   struct SegmentResult {
     SeriesAccumulator series;
@@ -1621,8 +1622,16 @@ LoadedRouteArtifacts load_route_series_parallel(
         continue;
       }
 
+      std::string migrated_data;
+      if (migrate) {
+        migrated_data = PyTools::migrateLog(log_path);
+      }
+
       LogReader reader;
-      if (!reader.load(log_path, nullptr, true)) {
+      const bool loaded = !migrated_data.empty()
+        ? reader.load(migrated_data.data(), migrated_data.size())
+        : reader.load(log_path, nullptr, true);
+      if (!loaded) {
         segment_stats.failed = true;
         std::lock_guard<std::mutex> lock(error_mutex);
         if (first_error.empty()) {
@@ -1889,6 +1898,7 @@ SketchLayout load_sketch_layout(const fs::path &layout_path) {
 RouteData load_route_data(const std::string &route_name,
                           const std::string &data_dir,
                           const std::string &dbc_name,
+                          bool migrate,
                           const RouteLoadProgressCallback &progress) {
   if (route_name.empty()) return RouteData{};
 
@@ -1916,7 +1926,7 @@ RouteData load_route_data(const std::string &route_name,
 
   const SchemaIndex &schema = SchemaIndex::instance();
   LoadedRouteArtifacts artifacts = load_route_series_parallel(segments, schema, can_dbc ? &*can_dbc : nullptr,
-                                                             route.selector, can_dbc.has_value(), &stats);
+                                                             route.selector, can_dbc.has_value(), migrate, &stats);
   RouteData route_data = build_route_data(std::move(artifacts.series),
                                           std::move(artifacts.can_messages),
                                           std::move(artifacts.logs),

openpilot/tools/replay/SConscript

@@ -7,7 +7,7 @@ base_frameworks = ['VideoToolbox', 'CoreMedia', 'CoreFoundation', 'CoreVideo'] i
 base_libs = [common, messaging, cereal, visionipc, 'm', 'pthread']
 
 replay_lib_src = ["replay.cc", "consoleui.cc", "camera.cc", "filereader.cc", "logreader.cc", "framereader.cc",
-                  "route.cc", "util.cc", "seg_mgr.cc", "timeline.cc", "py_downloader.cc"]
+                  "route.cc", "util.cc", "seg_mgr.cc", "timeline.cc", "py_process.cc", "py_tools.cc"]
 if arch != "Darwin":
   replay_lib_src.append("#openpilot/system/loggerd/encoder/v4l_decoder.cc")
 replay_lib = replay_env.Library("replay", replay_lib_src, LIBS=base_libs, FRAMEWORKS=base_frameworks)

openpilot/tools/replay/consoleui.cc

@@ -9,7 +9,7 @@
 #include "common/ratekeeper.h"
 #include "common/util.h"
 #include "common/version.h"
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_tools.h"
 
 namespace {
 

openpilot/tools/replay/filereader.cc

@@ -4,12 +4,12 @@
 #include <unistd.h>
 
 #include "common/util.h"
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_tools.h"
 
 std::string FileReader::read(const std::string &file, std::atomic<bool> *abort) {
   const bool is_remote = (file.find("https://") == 0) || (file.find("http://") == 0);
   if (is_remote) {
-    std::string local_path = PyDownloader::download(file, cache_to_local_, abort);
+    std::string local_path = PyTools::download(file, cache_to_local_, abort);
     if (local_path.empty()) return {};
     return util::read_file(local_path);
   }
@@ -19,7 +19,7 @@ std::string FileReader::read(const std::string &file, std::atomic<bool> *abort)
   const std::string magic(header, stream.gcount());
   if (util::ends_with(file, ".bz2") || util::ends_with(file, ".zst") ||
       util::starts_with(magic, "BZh") || magic == "\x28\xB5\x2F\xFD") {
-    std::string local_path = PyDownloader::decompress(file, abort);
+    std::string local_path = PyTools::decompress(file, abort);
     if (local_path.empty()) return {};
     std::string data = util::read_file(local_path);
     unlink(local_path.c_str());

openpilot/tools/replay/framereader.cc

@@ -7,7 +7,7 @@
 
 #include "common/util.h"
 #include "common/yuv.h"
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_tools.h"
 #include "tools/replay/util.h"
 #include "common/hardware/hw.h"
 
@@ -75,7 +75,7 @@ FrameReader::~FrameReader() {
 bool FrameReader::load(CameraType type, const std::string &url, bool no_hw_decoder, std::atomic<bool> *abort, bool local_cache) {
   std::string local_file_path;
   if (url.find("https://") == 0 || url.find("http://") == 0) {
-    local_file_path = PyDownloader::download(url, local_cache, abort);
+    local_file_path = PyTools::download(url, local_cache, abort);
     if (local_file_path.empty()) return false;
   } else {
     local_file_path = url;

openpilot/tools/replay/logreader.cc

@@ -4,7 +4,7 @@
 #include <chrono>
 #include <utility>
 #include "tools/replay/filereader.h"
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_tools.h"
 #include "tools/replay/util.h"
 #include "common/util.h"
 

openpilot/tools/replay/py_process.cc

@@ -1,10 +1,9 @@
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_process.h"
 
 #include <csignal>
-#include <fcntl.h>
 #include <cstdio>
 #include <cstring>
-#include <mutex>
+#include <fcntl.h>
 #include <spawn.h>
 #ifdef __APPLE__
 #include <crt_externs.h>
@@ -16,29 +15,15 @@
 
 #include "tools/replay/util.h"
 
-namespace {
+namespace PyProcess {
 
-constexpr const char *AUTH_MODULE = "openpilot.tools.lib.auth";
-constexpr const char *DOWNLOADER_MODULE = "openpilot.tools.lib.file_downloader";
-
-static std::mutex handler_mutex;
-static DownloadProgressHandler progress_handler = nullptr;
-
-void reportProgress(const char *line) {
-  uint64_t cur = 0, total = 0;
-  if (sscanf(line, "PROGRESS:%llu:%llu", (unsigned long long *)&cur, (unsigned long long *)&total) != 2) return;
-  std::lock_guard<std::mutex> lk(handler_mutex);
-  if (progress_handler && total > 0) progress_handler(cur, total, true);
-}
-
-// Run a Python command and capture stdout. Stderr is scanned for PROGRESS lines and otherwise passed
-// through to the parent's stderr. Returns stdout content. If abort is signaled, kills the child process.
-std::string runPython(const char *module, const std::vector<std::string> &args, std::atomic<bool> *abort = nullptr) {
+std::string runModule(const std::string &module, const std::vector<std::string> &args,
+                      std::atomic<bool> *abort, bool trim, const StderrLineCallback &stderr_line_cb) {
   // Build argv for the Python module
   std::vector<const char *> argv;
   argv.push_back("python3");
   argv.push_back("-m");
-  argv.push_back(module);
+  argv.push_back(module.c_str());
   for (const auto &a : args) {
     argv.push_back(a.c_str());
   }
@@ -56,17 +41,17 @@ std::string runPython(const char *module, const std::vector<std::string> &args,
   };
   int stdout_pipe[2], stderr_pipe[2];
   if (open_pipe(stdout_pipe) != 0) {
-    rWarning("py_downloader: pipe() failed");
+    rWarning("py_process: pipe() failed");
     return {};
   }
   if (open_pipe(stderr_pipe) != 0) {
-    rWarning("py_downloader: pipe() failed");
+    rWarning("py_process: pipe() failed");
     close(stdout_pipe[0]); close(stdout_pipe[1]);
     return {};
   }
 
   // Avoid copying the large replay address space and running atfork handlers on
-  // every segment download: both can stall rendering even from a worker thread.
+  // every spawn: both can stall rendering even from a worker thread.
   std::vector<std::string> environment;
 #ifdef __APPLE__
   char **parent_environment = *_NSGetEnviron();
@@ -103,7 +88,7 @@ std::string runPython(const char *module, const std::vector<std::string> &args,
   if (attributes_initialized) posix_spawnattr_destroy(&attributes);
   if (actions_initialized) posix_spawn_file_actions_destroy(&actions);
   if (error) {
-    rWarning("py_downloader: posix_spawnp() failed: %s", strerror(error));
+    rWarning("py_process: posix_spawnp() failed: %s", strerror(error));
     close(stdout_pipe[0]); close(stdout_pipe[1]);
     close(stderr_pipe[0]); close(stderr_pipe[1]);
     return {};
@@ -113,8 +98,8 @@ std::string runPython(const char *module, const std::vector<std::string> &args,
   close(stdout_pipe[1]);
   close(stderr_pipe[1]);
 
-  // stderr carries the progress lines, so a thread reads it while the loop below waits on stdout
-  std::thread stderr_thread([fd = stderr_pipe[0]]() {
+  // stderr is read in a thread so it can be inspected while the loop below waits on stdout
+  std::thread stderr_thread([fd = stderr_pipe[0], cb = stderr_line_cb]() {
     FILE *f = fdopen(fd, "r");
     if (!f) {
       close(fd);
@@ -123,8 +108,8 @@ std::string runPython(const char *module, const std::vector<std::string> &args,
     char *line = nullptr;
     size_t cap = 0;
     while (getline(&line, &cap, f) > 0) {
-      if (strncmp(line, "PROGRESS:", 9) == 0) {
-        reportProgress(line);
+      if (cb) {
+        cb(line);
       } else {
         fputs(line, stderr);
       }
@@ -182,81 +167,23 @@ std::string runPython(const char *module, const std::vector<std::string> &args,
                 WIFSIGNALED(status);
   if (failed) {
     if (expected_sigterm) {
-      // Route/camera teardown cancels outstanding downloader subprocesses.
-      // Keep that expected shutdown path quiet.
+      // Caller signaled abort; expected shutdown path.
     } else if (WIFEXITED(status) && WEXITSTATUS(status) != 0) {
-      rWarning("py_downloader: process exited with code %d", WEXITSTATUS(status));
+      rWarning("py_process: %s exited with code %d", module.c_str(), WEXITSTATUS(status));
     } else if (WIFSIGNALED(status)) {
-      rWarning("py_downloader: process killed by signal %d", WTERMSIG(status));
-    }
-    std::lock_guard<std::mutex> lk(handler_mutex);
-    if (progress_handler) {
-      progress_handler(0, 0, false);
+      rWarning("py_process: %s killed by signal %d", module.c_str(), WTERMSIG(status));
     }
     return {};
   }
 
   // Trim trailing newline
-  while (!stdout_data.empty() && (stdout_data.back() == '\n' || stdout_data.back() == '\r')) {
-    stdout_data.pop_back();
+  if (trim) {
+    while (!stdout_data.empty() && (stdout_data.back() == '\n' || stdout_data.back() == '\r')) {
+      stdout_data.pop_back();
+    }
   }
 
   return stdout_data;
 }
 
-}  // namespace
-
-void installDownloadProgressHandler(DownloadProgressHandler handler) {
-  std::lock_guard<std::mutex> lk(handler_mutex);
-  progress_handler = handler;
-}
-
-namespace PyDownloader {
-
-std::string download(const std::string &url, bool use_cache, std::atomic<bool> *abort) {
-  std::vector<std::string> args = {"download", url};
-  if (!use_cache) {
-    args.push_back("--no-cache");
-  }
-  return runPython(DOWNLOADER_MODULE, args, abort);
-}
-
-std::string decompress(const std::string &path, std::atomic<bool> *abort) {
-  return runPython(DOWNLOADER_MODULE, {"decompress", path}, abort);
-}
-
-std::string getRouteFiles(const std::string &route) {
-  return runPython(DOWNLOADER_MODULE, {"route-files", route});
-}
-
-std::string resolveRouteFiles(const std::string &route, int begin, int end, const std::string &selector) {
-  return runPython(DOWNLOADER_MODULE, {"resolve-route-files", route, "--begin", std::to_string(begin),
-                                     "--end", std::to_string(end), "--selector", selector});
-}
-
-std::string authenticate(const std::string &provider, std::atomic<bool> *abort) {
-  return runPython(AUTH_MODULE, {provider, "--json"}, abort);
-}
-
-std::string getDevices() {
-  return runPython(DOWNLOADER_MODULE, {"devices"});
-}
-
-std::string getDeviceRoutes(const std::string &dongle_id, int64_t start_ms, int64_t end_ms, bool preserved) {
-  std::vector<std::string> args = {"device-routes", dongle_id};
-  if (preserved) {
-    args.push_back("--preserved");
-  } else {
-    if (start_ms > 0) {
-      args.push_back("--start");
-      args.push_back(std::to_string(start_ms));
-    }
-    if (end_ms > 0) {
-      args.push_back("--end");
-      args.push_back(std::to_string(end_ms));
-    }
-  }
-  return runPython(DOWNLOADER_MODULE, args);
-}
-
-}  // namespace PyDownloader
+}  // namespace PyProcess

openpilot/tools/replay/py_process.h

@@ -0,0 +1,21 @@
+#pragma once
+
+#include <atomic>
+#include <functional>
+#include <string>
+#include <vector>
+
+namespace PyProcess {
+
+// Called with each line of the child's stderr (newline-terminated).
+using StderrLineCallback = std::function<void(const char *line)>;
+
+// Run a Python module (`python3 -m <module> <args...>`) and capture stdout.
+// Stderr lines are passed to stderr_line_cb if set, otherwise through to the parent's stderr.
+// If trim is true (default), trailing newlines are removed from the returned stdout content.
+// Returns empty string on failure. If abort is signaled, kills the child process.
+std::string runModule(const std::string &module, const std::vector<std::string> &args,
+                      std::atomic<bool> *abort = nullptr, bool trim = true,
+                      const StderrLineCallback &stderr_line_cb = nullptr);
+
+}  // namespace PyProcess

openpilot/tools/replay/py_tools.cc

@@ -0,0 +1,102 @@
+#include "tools/replay/py_tools.h"
+
+#include <cstdio>
+#include <cstring>
+#include <mutex>
+#include <vector>
+
+#include "tools/replay/py_process.h"
+
+namespace {
+
+constexpr const char *AUTH_MODULE = "openpilot.tools.lib.auth";
+constexpr const char *DOWNLOADER_MODULE = "openpilot.tools.lib.file_downloader";
+constexpr const char *MIGRATION_MODULE = "openpilot.selfdrive.test.process_replay.migration";
+
+static std::mutex handler_mutex;
+static DownloadProgressHandler progress_handler = nullptr;
+
+void reportProgress(const char *line) {
+  uint64_t cur = 0, total = 0;
+  if (sscanf(line, "PROGRESS:%llu:%llu", (unsigned long long *)&cur, (unsigned long long *)&total) != 2) return;
+  std::lock_guard<std::mutex> lk(handler_mutex);
+  if (progress_handler && total > 0) progress_handler(cur, total, true);
+}
+
+// Run a Python module, report download progress from PROGRESS lines on stderr,
+// and notify the progress handler on failure.
+std::string runModuleWithProgress(const std::string &module, const std::vector<std::string> &args, std::atomic<bool> *abort = nullptr) {
+  std::string result = PyProcess::runModule(module, args, abort, true, [](const char *line) {
+    if (strncmp(line, "PROGRESS:", 9) == 0) {
+      reportProgress(line);
+    } else {
+      fputs(line, stderr);
+    }
+  });
+  if (result.empty()) {
+    std::lock_guard<std::mutex> lk(handler_mutex);
+    if (progress_handler) progress_handler(0, 0, false);
+  }
+  return result;
+}
+
+}  // namespace
+
+void installDownloadProgressHandler(DownloadProgressHandler handler) {
+  std::lock_guard<std::mutex> lk(handler_mutex);
+  progress_handler = handler;
+}
+
+namespace PyTools {
+
+std::string download(const std::string &url, bool use_cache, std::atomic<bool> *abort) {
+  std::vector<std::string> args = {"download", url};
+  if (!use_cache) {
+    args.push_back("--no-cache");
+  }
+  return runModuleWithProgress(DOWNLOADER_MODULE, args, abort);
+}
+
+std::string decompress(const std::string &path, std::atomic<bool> *abort) {
+  return runModuleWithProgress(DOWNLOADER_MODULE, {"decompress", path}, abort);
+}
+
+std::string getRouteFiles(const std::string &route) {
+  return runModuleWithProgress(DOWNLOADER_MODULE, {"route-files", route});
+}
+
+std::string resolveRouteFiles(const std::string &route, int begin, int end, const std::string &selector) {
+  return runModuleWithProgress(DOWNLOADER_MODULE, {"resolve-route-files", route, "--begin", std::to_string(begin),
+                                                  "--end", std::to_string(end), "--selector", selector});
+}
+
+std::string authenticate(const std::string &provider, std::atomic<bool> *abort) {
+  return runModuleWithProgress(AUTH_MODULE, {provider, "--json"}, abort);
+}
+
+std::string getDevices() {
+  return runModuleWithProgress(DOWNLOADER_MODULE, {"devices"});
+}
+
+std::string getDeviceRoutes(const std::string &dongle_id, int64_t start_ms, int64_t end_ms, bool preserved) {
+  std::vector<std::string> args = {"device-routes", dongle_id};
+  if (preserved) {
+    args.push_back("--preserved");
+  } else {
+    if (start_ms > 0) {
+      args.push_back("--start");
+      args.push_back(std::to_string(start_ms));
+    }
+    if (end_ms > 0) {
+      args.push_back("--end");
+      args.push_back(std::to_string(end_ms));
+    }
+  }
+  return runModuleWithProgress(DOWNLOADER_MODULE, args);
+}
+
+std::string migrateLog(const std::string &log_path) {
+  return PyProcess::runModule(MIGRATION_MODULE, {log_path, "/dev/stdout"}, nullptr, false);
+}
+
+}  // namespace PyTools

openpilot/tools/replay/py_tools.h

@@ -7,7 +7,7 @@
 typedef std::function<void(uint64_t cur, uint64_t total, bool success)> DownloadProgressHandler;
 void installDownloadProgressHandler(DownloadProgressHandler handler);
 
-namespace PyDownloader {
+namespace PyTools {
 
 // Downloads url to local cache, returns local file path. Reports progress via installDownloadProgressHandler.
 std::string download(const std::string &url, bool use_cache = true, std::atomic<bool> *abort = nullptr);
@@ -30,4 +30,7 @@ std::string getDevices();
 // Returns JSON string of device routes
 std::string getDeviceRoutes(const std::string &dongle_id, int64_t start_ms = 0, int64_t end_ms = 0, bool preserved = false);
 
-}  // namespace PyDownloader
+// Migrates a log segment through the Python migration CLI. Returns the migrated bytes (empty on failure).
+std::string migrateLog(const std::string &log_path);
+
+}  // namespace PyTools

openpilot/tools/replay/route.cc

@@ -6,7 +6,7 @@
 
 #include "json11/json11.hpp"
 #include "common/hardware/hw.h"
-#include "tools/replay/py_downloader.h"
+#include "tools/replay/py_tools.h"
 #include "tools/replay/replay.h"
 #include "tools/replay/util.h"
 
@@ -104,7 +104,7 @@ bool Route::loadFromAutoSource() {
 }
 
 bool Route::loadFromServer() {
-  std::string result = PyDownloader::getRouteFiles(route_.str);
+  std::string result = PyTools::getRouteFiles(route_.str);
   if (result.empty()) {
     err_ = RouteLoadError::NetworkError;
     rWarning("Failed to fetch route files from server");