process_replay: most messages valid check (#32521)

* check_most_messages_valid impl

* Add to both regen and test_processes

* Refactor

* Bring back carOutput

* Use Counter

* Use get(k, 0)

selfdrive/test/process_replay/process_replay.py

@@ -6,7 +6,7 @@ import json
 import heapq
 import signal
 import platform
-from collections import OrderedDict
+from collections import Counter, OrderedDict
 from dataclasses import dataclass, field
 from typing import Any
 from collections.abc import Callable, Iterable
@@ -816,3 +816,18 @@ def check_openpilot_enabled(msgs: LogIterable) -> bool:
       max_enabled_count = max(max_enabled_count, cur_enabled_count)
 
   return max_enabled_count > int(10. / DT_CTRL)
+
+
+def check_most_messages_valid(msgs: LogIterable, threshold: float = 0.9) -> bool:
+  msgs_counts = Counter(msg.which() for msg in msgs)
+  msgs_valid_counts = Counter(msg.which() for msg in msgs if msg.valid)
+
+  most_valid_for_service = {}
+  for msg_type in msgs_counts.keys():
+    valid_share = msgs_valid_counts.get(msg_type, 0) / msgs_counts[msg_type]
+    ok = valid_share >= threshold
+    if not ok:
+      print(f"WARNING: Service {msg_type} has {valid_share * 100:.2f}% valid messages, which is below threshold of {threshold * 100:.2f}%")
+    most_valid_for_service[msg_type] = ok
+
+  return all(most_valid_for_service.values())

selfdrive/test/process_replay/regen.py

@@ -9,7 +9,7 @@ from typing import Any
 from collections.abc import Iterable
 
 from openpilot.selfdrive.test.process_replay.process_replay import CONFIGS, FAKEDATA, ProcessConfig, replay_process, get_process_config, \
-                                                                   check_openpilot_enabled, get_custom_params_from_lr
+                                                                   check_openpilot_enabled, check_most_messages_valid, get_custom_params_from_lr
 from openpilot.selfdrive.test.process_replay.vision_meta import DRIVER_CAMERA_FRAME_SIZES
 from openpilot.selfdrive.test.update_ci_routes import upload_route
 from openpilot.tools.lib.route import Route
@@ -129,6 +129,8 @@ def regen_and_save(
 
   if not check_openpilot_enabled(output_logs):
     raise Exception("Route did not engage for long enough")
+  if not check_most_messages_valid(output_logs):
+    raise Exception("Route has too many invalid messages")
 
   if upload:
     upload_route(rel_log_dir)

selfdrive/test/process_replay/test_processes.py

@@ -11,7 +11,8 @@ from openpilot.common.git import get_commit
 from openpilot.selfdrive.car.car_helpers import interface_names
 from openpilot.tools.lib.openpilotci import get_url, upload_file
 from openpilot.selfdrive.test.process_replay.compare_logs import compare_logs, format_diff
-from openpilot.selfdrive.test.process_replay.process_replay import CONFIGS, PROC_REPLAY_DIR, FAKEDATA, check_openpilot_enabled, replay_process
+from openpilot.selfdrive.test.process_replay.process_replay import CONFIGS, PROC_REPLAY_DIR, FAKEDATA, replay_process, \
+                                                                   check_openpilot_enabled, check_most_messages_valid
 from openpilot.tools.lib.filereader import FileReader
 from openpilot.tools.lib.logreader import LogReader
 from openpilot.tools.lib.helpers import save_log
@@ -108,6 +109,8 @@ def test_process(cfg, lr, segment, ref_log_path, new_log_path, ignore_fields=Non
   if cfg.proc_name == "controlsd":
     if not check_openpilot_enabled(log_msgs):
       return f"Route did not enable at all or for long enough: {new_log_path}", log_msgs
+  if not check_most_messages_valid(log_msgs):
+    return f"Route did not have enough valid messages: {new_log_path}", log_msgs
 
   if cfg.proc_name != 'ubloxd' or segment != 'regen3BB55FA5E20|2024-05-21--06-59-03--0':
     seen_msgs = {m.which() for m in log_msgs}