lil more

Author: adeebshihadeh

tools/replay/consoleui.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 import curses
+import logging
 import signal
 import threading
 import time
@@ -12,6 +13,24 @@
 from openpilot.tools.replay.seg_mgr import ReplayFlags
 from openpilot.tools.replay.timeline import FindFlag, TimelineType
 
+
+class QueueHandler(logging.Handler):
+  """Logging handler that queues messages for display in curses UI."""
+  def __init__(self, queue: list, lock: threading.Lock):
+    super().__init__()
+    self._queue = queue
+    self._lock = lock
+
+  def emit(self, record: logging.LogRecord) -> None:
+    # Map logging levels to our color scheme
+    level = 0  # DEBUG
+    if record.levelno >= logging.ERROR:
+      level = 2  # Critical/Red
+    elif record.levelno >= logging.WARNING:
+      level = 1  # Warning/Yellow
+    with self._lock:
+      self._queue.append((level, self.format(record)))
+
 BORDER_SIZE = 3
 
 KEYBOARD_SHORTCUTS = [
@@ -88,12 +107,11 @@ def __init__(self, replay: Replay):
     # Set up signal handler for clean exit
     signal.signal(signal.SIGINT, lambda s, f: setattr(self, '_exit', True))
 
-    # Set up log callback
-    self.replay._seg_mgr.set_log_callback(self._on_log)
-
-  def _on_log(self, level: int, msg: str) -> None:
-    with self._lock:
-      self._logs.append((level, msg))
+    # Set up logging handler to capture logs for display
+    self._log_handler = QueueHandler(self._logs, self._lock)
+    self._log_handler.setFormatter(logging.Formatter('%(message)s'))
+    logging.getLogger("replay").addHandler(self._log_handler)
+    logging.getLogger("replay").setLevel(logging.DEBUG)
 
   def _init_curses(self, stdscr) -> None:
     self._stdscr = stdscr

tools/replay/replay.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+import logging
 import threading
 import time
 from enum import IntFlag
@@ -13,6 +14,8 @@
 from openpilot.tools.replay.seg_mgr import SegmentManager, ReplayFlags
 from openpilot.tools.replay.timeline import Timeline, FindFlag
 
+log = logging.getLogger("replay")
+
 DEMO_ROUTE = "a2a0ccea32023010|2023-07-27--13-01-19"
 
 
@@ -66,10 +69,10 @@ def __init__(self, route: str, allow: list[str] = None, block: list[str] = None,
 
   def __del__(self):
     if hasattr(self, '_stream_thread') and self._stream_thread is not None and self._stream_thread.is_alive():
-      print("shutdown: in progress...")
+      log.info("shutdown: in progress...")
       self._interrupt_stream(lambda: setattr(self, '_exit', True) or False)
       self._stream_thread.join()
-      print("shutdown: done")
+      log.info("shutdown: done")
 
   def _setup_services(self, allow: list[str], block: list[str]) -> None:
     active_services = []
@@ -83,7 +86,7 @@ def _setup_services(self, allow: list[str], block: list[str]) -> None:
       else:
         self._sockets[name] = False
 
-    print(f"active services: {', '.join(active_services)}")
+    log.info(f"active services: {', '.join(active_services)}")
     if not self._sm:
       self._pm = messaging.PubMaster(active_services)
 
@@ -159,7 +162,7 @@ def install_event_filter(self, filter_fn: Callable) -> None:
     self._event_filter = filter_fn
 
   def load(self) -> bool:
-    print(f"loading route {self._seg_mgr._route_name}")
+    log.info(f"loading route {self._seg_mgr._route_name}")
     if not self._seg_mgr.load():
       return False
 
@@ -175,7 +178,7 @@ def start(self, seconds: int = 0) -> None:
   def pause(self, pause: bool) -> None:
     if self._user_paused != pause:
       def update():
-        print(f"{'paused...' if pause else 'resuming'} at {self.current_seconds:.2f} s")
+        log.info(f"{'paused...' if pause else 'resuming'} at {self.current_seconds:.2f} s")
         self._user_paused = pause
         return not pause
       self._interrupt_stream(update)
@@ -191,10 +194,10 @@ def seek_to(self, seconds: float, relative: bool) -> None:
     target_segment = int(target_time / 60)
 
     if not self._seg_mgr.has_segment(target_segment):
-      print(f"Invalid seek to {target_time:.2f} s (segment {target_segment})")
+      log.warning(f"Invalid seek to {target_time:.2f} s (segment {target_segment})")
       return
 
-    print(f"Seeking to {int(target_time)} s, segment {target_segment}")
+    log.info(f"Seeking to {int(target_time)} s, segment {target_segment}")
     if self.on_seeking:
       self.on_seeking(target_time)
 
@@ -272,7 +275,7 @@ def _start_stream(self, segment) -> None:
           params.put("CarParams", car_params_bytes)
           params.put("CarParamsPersistent", car_params_bytes)
         except Exception as e:
-          print(f"failed to write CarParams: {e}")
+          log.warning(f"failed to write CarParams: {e}")
         break
 
     # Start camera server
@@ -314,7 +317,7 @@ def _stream_thread_fn(self) -> None:
             first_idx = i
             break
         else:
-          print("waiting for events...")
+          log.info("waiting for events...")
           self._events_ready = False
           continue
 
@@ -328,7 +331,7 @@ def _stream_thread_fn(self) -> None:
         if last_idx >= len(events) and not self.has_flag(ReplayFlags.NO_LOOP):
           segments = list(self._seg_mgr._segments.keys())
           if segments and event_data.is_segment_loaded(max(segments)):
-            print("reaches the end of route, restart from beginning")
+            log.info("reaches the end of route, restart from beginning")
             self._stream_lock.release()
             self.seek_to(self._min_seconds, relative=False)
             self._stream_lock.acquire()
@@ -395,7 +398,7 @@ def _publish_message(self, evt) -> None:
         msg_bytes = evt.as_builder().to_bytes()
         self._pm.send(which, msg_bytes)
       except Exception as e:
-        print(f"stop publishing {which} due to error: {e}")
+        log.warning(f"stop publishing {which} due to error: {e}")
         self._sockets[which] = False
 
   def _publish_frame(self, evt, which: str) -> None:

tools/replay/seg_mgr.py

@@ -1,10 +1,13 @@
 #!/usr/bin/env python3
+import logging
 import threading
 from dataclasses import dataclass, field
 from enum import Enum, IntFlag, auto
 from typing import Callable, Optional
 
 from openpilot.selfdrive.test.process_replay.migration import migrate_all
+
+log = logging.getLogger("replay")
 from openpilot.tools.lib.logreader import LogReader
 from openpilot.tools.lib.route import Route, Segment
 from openpilot.tools.lib.framereader import FrameReader
@@ -69,7 +72,6 @@ def __init__(self, route_name: str, flags: int = 0, data_dir: str = "", auto_sou
 
     self.segment_cache_limit = MIN_SEGMENTS_CACHE
     self._on_segment_merged_callback: Optional[Callable[[], None]] = None
-    self._log_callback: Optional[Callable[[int, str], None]] = None  # (level, msg)
 
   def __del__(self):
     if hasattr(self, '_cv'):
@@ -87,7 +89,7 @@ def load(self) -> bool:
     try:
       self._route = Route(self._route_name, data_dir=self._data_dir)
     except Exception as e:
-      print(f"failed to load route: {self._route_name}: {e}")
+      log.error(f"failed to load route: {self._route_name}: {e}")
       return False
 
     # Initialize segment slots for all available segments
@@ -97,10 +99,10 @@ def load(self) -> bool:
         self._segments[seg_num] = None
 
     if not self._segments:
-      print(f"no valid segments in route: {self._route_name}")
+      log.error(f"no valid segments in route: {self._route_name}")
       return False
 
-    print(f"loaded route {self._route_name} with {len(self._segments)} valid segments")
+    log.info(f"loaded route {self._route_name} with {len(self._segments)} valid segments")
     self._thread = threading.Thread(target=self._manage_segment_cache, daemon=True)
     self._thread.start()
     return True
@@ -116,13 +118,6 @@ def set_current_segment(self, seg_num: int) -> None:
   def set_callback(self, callback: Callable[[], None]) -> None:
     self._on_segment_merged_callback = callback
 
-  def set_log_callback(self, callback: Callable[[int, str], None]) -> None:
-    self._log_callback = callback
-
-  def _log(self, level: int, msg: str) -> None:
-    if self._log_callback:
-      self._log_callback(level, msg)
-
   def set_filters(self, filters: list[bool]) -> None:
     self._filters = filters
 
@@ -188,14 +183,14 @@ def _load_segments_in_range(self, seg_nums: list[int], cur_seg_num: int) -> bool
           continue
 
       # Load segment (blocking - downloads and parses)
-      self._log(0, f"loading segment {seg_num}...")
+      log.info(f"loading segment {seg_num}...")
       seg_data = self._load_segment(seg_num)
       with self._cv:
         self._segments[seg_num] = seg_data
         self._needs_update = True
         self._cv.notify_all()
       loaded_any = True
-      self._log(0, f"segment {seg_num} loaded with {len(seg_data.events)} events")
+      log.info(f"segment {seg_num} loaded with {len(seg_data.events)} events")
 
       # Only load one segment at a time to be responsive
       return loaded_any
@@ -224,7 +219,7 @@ def _load_segment(self, seg_num: int) -> SegmentData:
         events = list(migrate_all(lr))
         seg_data.events = sorted(events, key=lambda x: x.logMonoTime)
       except Exception as e:
-        print(f"failed to load log for segment {seg_num}: {e}")
+        log.warning(f"failed to load log for segment {seg_num}: {e}")
         seg_data.load_state = LoadState.FAILED
         return seg_data
 
@@ -243,7 +238,7 @@ def _load_segment(self, seg_num: int) -> SegmentData:
       if segment.qcamera_path and (self._flags & ReplayFlags.QCAMERA):
         seg_data.frame_readers['qcam'] = FrameReader(segment.qcamera_path)
     except Exception as e:
-      print(f"failed to load frames for segment {seg_num}: {e}")
+      log.warning(f"failed to load frames for segment {seg_num}: {e}")
       # Don't fail the whole segment, just skip frames
 
     seg_data.load_state = LoadState.LOADED
@@ -289,5 +284,5 @@ def _merge_segments(self, seg_nums: list[int]) -> bool:
       self._event_data = merged_event_data
       self._merged_segments = segments_to_merge
 
-    self._log(0, f"merged segments: {sorted(segments_to_merge)}")
+    log.debug(f"merged segments: {sorted(segments_to_merge)}")
     return True