process_replay: migrate panda/peripheral/manager states (#30141)

* Migrate pandaStates

* Fix import

* Make pandaStates migration optional, so process_replay still works on logs without carParams

* peripheralState migration if its not available

* Do nothing if log file has peripheralStates

* Add migrate_managerState

---------

Co-authored-by: Shane Smiskol <shane@smiskol.com>
old-commit-hash: 898acb13f2

selfdrive/test/process_replay/migration.py

@@ -2,17 +2,95 @@ from collections import defaultdict
 
 from cereal import messaging
 from openpilot.selfdrive.test.process_replay.vision_meta import meta_from_encode_index
+from openpilot.selfdrive.car.toyota.values import EPS_SCALE
+from openpilot.selfdrive.manager.process_config import managed_processes
+from panda import Panda
 
 
-def migrate_all(lr, old_logtime=False, camera_states=False):
+def migrate_all(lr, old_logtime=False, manager_states=False, panda_states=False, camera_states=False):
   msgs = migrate_sensorEvents(lr, old_logtime)
   msgs = migrate_carParams(msgs, old_logtime)
+  if manager_states:
+    msgs = migrate_managerState(msgs)
+  if panda_states:
+    msgs = migrate_pandaStates(msgs)
+    msgs = migrate_peripheralState(msgs)
   if camera_states:
     msgs = migrate_cameraStates(msgs)
 
   return msgs
 
 
+def migrate_managerState(lr):
+  all_msgs = []
+  for msg in lr:
+    if msg.which() != "managerState":
+      all_msgs.append(msg)
+      continue
+
+    new_msg = msg.as_builder()
+    new_msg.managerState.processes = [{'name': name, 'running': True} for name in managed_processes]
+    all_msgs.append(new_msg.as_reader())
+
+  return all_msgs
+
+
+def migrate_pandaStates(lr):
+  all_msgs = []
+  # TODO: safety param migration should be handled automatically
+  safety_param_migration = {
+    "TOYOTA PRIUS 2017": EPS_SCALE["TOYOTA PRIUS 2017"] | Panda.FLAG_TOYOTA_STOCK_LONGITUDINAL,
+    "TOYOTA RAV4 2017": EPS_SCALE["TOYOTA RAV4 2017"] | Panda.FLAG_TOYOTA_ALT_BRAKE,
+    "KIA EV6 2022": Panda.FLAG_HYUNDAI_EV_GAS | Panda.FLAG_HYUNDAI_CANFD_HDA2,
+  }
+
+  # Migrate safety param base on carState
+  CP = next((m.carParams for m in lr if m.which() == 'carParams'), None)
+  assert CP is not None, "carParams message not found"
+  if CP.carFingerprint in safety_param_migration:
+    safety_param = safety_param_migration[CP.carFingerprint]
+  elif len(CP.safetyConfigs):
+    safety_param = CP.safetyConfigs[0].safetyParam
+    if CP.safetyConfigs[0].safetyParamDEPRECATED != 0:
+      safety_param = CP.safetyConfigs[0].safetyParamDEPRECATED
+  else:
+    safety_param = CP.safetyParamDEPRECATED
+
+  for msg in lr:
+    if msg.which() == 'pandaStateDEPRECATED':
+      new_msg = messaging.new_message('pandaStates', 1)
+      new_msg.valid = msg.valid
+      new_msg.logMonoTime = msg.logMonoTime
+      new_msg.pandaStates[0] = msg.pandaStateDEPRECATED
+      new_msg.pandaStates[0].safetyParam = safety_param
+      all_msgs.append(new_msg.as_reader())
+    elif msg.which() == 'pandaStates':
+      new_msg = msg.as_builder()
+      new_msg.pandaStates[-1].safetyParam = safety_param
+      all_msgs.append(new_msg.as_reader())
+    else:
+      all_msgs.append(msg)
+
+  return all_msgs
+
+
+def migrate_peripheralState(lr):
+  if any(msg.which() == "peripheralState" for msg in lr):
+    return lr
+
+  all_msg = []
+  for msg in lr:
+    all_msg.append(msg)
+    if msg.which() not in ["pandaStates", "pandaStateDEPRECATED"]:
+      continue
+
+    new_msg = messaging.new_message("peripheralState")
+    new_msg.logMonoTime = msg.logMonoTime
+    all_msg.append(new_msg.as_reader())
+
+  return all_msg
+
+
 def migrate_cameraStates(lr):
   all_msgs = []
   frame_to_encode_id = defaultdict(dict)

selfdrive/test/process_replay/process_replay.py

@@ -653,7 +653,10 @@ def replay_process(
   else:
     cfgs = [cfg]
 
-  all_msgs = migrate_all(lr, old_logtime=True, camera_states=any(len(cfg.vision_pubs) != 0 for cfg in cfgs))
+  all_msgs = migrate_all(lr, old_logtime=True,
+                         manager_states=True,
+                         panda_states=any("pandaStates" in cfg.pubs for cfg in cfgs),
+                         camera_states=any(len(cfg.vision_pubs) != 0 for cfg in cfgs))
   process_logs = _replay_multi_process(cfgs, all_msgs, frs, fingerprint, custom_params, captured_output_store, disable_progress)
 
   if return_all_logs: