openpilot_comma/tools/rerun/run.py

#!/usr/bin/env python3

import sys
import argparse
import multiprocessing
import rerun as rr
import rerun.blueprint as rrb
from functools import partial
from collections import defaultdict

from cereal.services import SERVICE_LIST
from openpilot.tools.rerun.camera_reader import probe_packet_info, CameraReader, CameraConfig, CameraType
from openpilot.tools.lib.logreader import LogReader
from openpilot.tools.lib.route import Route, SegmentRange


NUM_CPUS = multiprocessing.cpu_count()
DEMO_ROUTE = "a2a0ccea32023010|2023-07-27--13-01-19"
RR_TIMELINE_NAME = "Timeline"
RR_WIN = "openpilot logs"


"""
Relevant upstream Rerun issues:
- loading videos directly: https://github.com/rerun-io/rerun/issues/6532
"""

class Rerunner:
  def __init__(self, route, segment_range, camera_config):
    self.lr = LogReader(route_or_segment_name)

    # hevc files don't have start_time. We get it from qcamera.ts
    start_time = 0
    dat = probe_packet_info(route.qcamera_paths()[0])
    for d in dat:
      if d.startswith("pts_time="):
        start_time = float(d.split('=')[1])
        break

    qcam, fcam, ecam, dcam = camera_config
    self.camera_readers = {}
    if qcam:
      self.camera_readers[CameraType.qcam] = CameraReader(route.qcamera_paths(), start_time, segment_range.seg_idxs)
    if fcam:
      self.camera_readers[CameraType.fcam] = CameraReader(route.camera_paths(), start_time, segment_range.seg_idxs)
    if ecam:
      self.camera_readers[CameraType.ecam] = CameraReader(route.ecamera_paths(), start_time, segment_range.seg_idxs)
    if dcam:
      self.camera_readers[CameraType.dcam] = CameraReader(route.dcamera_paths(), start_time, segment_range.seg_idxs)

  def _create_blueprint(self):
    blueprint = None
    service_views = []

    for topic in sorted(SERVICE_LIST.keys()):
      View = rrb.TimeSeriesView if topic != "thumbnail" else rrb.Spatial2DView
      service_views.append(View(name=topic, origin=f"/{topic}/", visible=False))
      rr.log(topic, rr.SeriesLine(name=topic), timeless=True)

    center_view = [rrb.Vertical(*service_views, name="streams")]
    if len(self.camera_readers):
      center_view.append(rrb.Vertical(*[rrb.Spatial2DView(name=cam_type, origin=cam_type) for cam_type in self.camera_readers.keys()], name="cameras"))

    blueprint = rrb.Blueprint(
      rrb.Horizontal(
        *center_view
      ),
      rrb.SelectionPanel(expanded=False),
      rrb.TimePanel(expanded=False),
    )
    return blueprint

  @staticmethod
  def _parse_msg(msg, parent_key=''):
    stack = [(msg, parent_key)]
    while stack:
      current_msg, current_parent_key = stack.pop()
      if isinstance(current_msg, list):
        for index, item in enumerate(current_msg):
          new_key = f"{current_parent_key}/{index}"
          if isinstance(item, (int, float)):
            yield new_key, item
          elif isinstance(item, dict):
            stack.append((item, new_key))
      elif isinstance(current_msg, dict):
        for key, value in current_msg.items():
          new_key = f"{current_parent_key}/{key}"
          if isinstance(value, (int, float)):
            yield new_key, value
          elif isinstance(value, dict):
            stack.append((value, new_key))
          elif isinstance(value, list):
            for index, item in enumerate(value):
              if isinstance(item, (int, float)):
                yield f"{new_key}/{index}", item
      else:
        pass  # Not a plottable value

  @staticmethod
  @rr.shutdown_at_exit
  def _process_log_msgs(blueprint, lr):
    rr.init(RR_WIN)
    rr.connect()
    rr.send_blueprint(blueprint)

    log_msgs = defaultdict(lambda: defaultdict(list))
    for msg in lr:
      msg_type = msg.which()

      if msg_type == "thumbnail":
        continue

      for entity_path, dat in Rerunner._parse_msg(msg.to_dict()[msg_type], msg_type):
        log_msgs[entity_path]["times"].append(msg.logMonoTime)
        log_msgs[entity_path]["data"].append(dat)

    for entity_path, log_msg in log_msgs.items():
      rr.send_columns(
        entity_path,
        times=[rr.TimeNanosColumn(RR_TIMELINE_NAME, log_msg["times"])],
        components=[rr.components.ScalarBatch(log_msg["data"])]
      )

    return []

  @staticmethod
  @rr.shutdown_at_exit
  def _process_cam_readers(blueprint, cam_type, h, w, fr):
    rr.init(RR_WIN)
    rr.connect()
    rr.send_blueprint(blueprint)

    for ts, frame in fr:
      rr.set_time_nanos(RR_TIMELINE_NAME, int(ts * 1e9))
      rr.log(cam_type, rr.Image(bytes=frame, width=w, height=h, pixel_format=rr.PixelFormat.NV12))

  def load_data(self):
    rr.init(RR_WIN, spawn=True)

    startup_blueprint = self._create_blueprint()
    self.lr.run_across_segments(NUM_CPUS, partial(self._process_log_msgs, startup_blueprint), desc="Log messages")
    for cam_type, cr in self.camera_readers.items():
      cr.run_across_segments(NUM_CPUS, partial(self._process_cam_readers, startup_blueprint, cam_type, cr.h, cr.w), desc=cam_type)

    rr.send_blueprint(self._create_blueprint())


if __name__ == '__main__':
  parser = argparse.ArgumentParser(description="A helper to run rerun on openpilot routes",
                                  formatter_class=argparse.ArgumentDefaultsHelpFormatter)
  parser.add_argument("--demo", action="store_true", help="Use the demo route instead of providing one")
  parser.add_argument("--qcam", action="store_true", help="Show low-res road camera")
  parser.add_argument("--fcam", action="store_true", help="Show driving camera")
  parser.add_argument("--ecam", action="store_true", help="Show wide camera")
  parser.add_argument("--dcam", action="store_true", help="Show driver monitoring camera")
  parser.add_argument("route_or_segment_name", nargs='?', help="The route or segment name to plot")
  args = parser.parse_args()

  if not args.demo and not args.route_or_segment_name:
    parser.print_help()
    sys.exit()

  camera_config = CameraConfig(args.qcam, args.fcam, args.ecam, args.dcam)
  route_or_segment_name = DEMO_ROUTE if args.demo else args.route_or_segment_name.strip()

  sr = SegmentRange(route_or_segment_name)
  r = Route(sr.route_name)

  hevc_requested = any(camera_config[1:])
  if len(sr.seg_idxs) > 1 and hevc_requested:
    print("You're requesting more than 1 segment with hevc videos, " + \
          "please be aware that might take a lot of memory " + \
          "since rerun isn't yet well supported for high resolution video logging")
    response = input("Do you wish to continue? (Y/n): ")
    if response.strip().lower() != "y":
      sys.exit()

  rerunner = Rerunner(r, sr, camera_config)
  rerunner.load_data()