openpilot v0.11.1 release

date: 2026-06-04T09:49:56
master commit: c0ab3550eca2e9daf197c46b7e4b24aa9637cf2e

selfdrive/ui/tests/body.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python3
+import time
+import cereal.messaging as messaging
+
+if __name__ == "__main__":
+  while True:
+    pm = messaging.PubMaster(['carParams', 'carState'])
+    batt = 1.
+    while True:
+      msg = messaging.new_message('carParams')
+      msg.carParams.brand = "body"
+      msg.carParams.notCar = True
+      pm.send('carParams', msg)
+
+      for b in range(100, 0, -1):
+        msg = messaging.new_message('carState')
+        msg.carState.charging = True
+        msg.carState.fuelGauge = b / 100.
+        pm.send('carState', msg)
+        time.sleep(0.1)
+
+      time.sleep(1)

selfdrive/ui/tests/cycle_offroad_alerts.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+import os
+import sys
+import time
+import json
+
+from openpilot.common.basedir import BASEDIR
+from openpilot.common.params import Params
+from openpilot.selfdrive.selfdrived.alertmanager import set_offroad_alert
+from openpilot.system.updated.updated import parse_release_notes
+
+if __name__ == "__main__":
+  params = Params()
+
+  with open(os.path.join(BASEDIR, "selfdrive/selfdrived/alerts_offroad.json")) as f:
+    offroad_alerts = json.load(f)
+
+  t = 10 if len(sys.argv) < 2 else int(sys.argv[1])
+  while True:
+    print("setting alert update")
+    params.put_bool("UpdateAvailable", True, block=True)
+    params.put("UpdaterNewReleaseNotes", parse_release_notes(BASEDIR), block=True)
+
+    time.sleep(t)
+    params.put_bool("UpdateAvailable", False, block=True)
+
+    # cycle through normal alerts
+    for a in offroad_alerts:
+      print("setting alert:", a)
+      set_offroad_alert(a, True)
+      time.sleep(t)
+      set_offroad_alert(a, False)
+
+    print("no alert")
+    time.sleep(t)

selfdrive/ui/tests/diff/diff.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+import os
+import sys
+import subprocess
+import webbrowser
+import argparse
+from pathlib import Path
+from openpilot.common.basedir import BASEDIR
+
+DIFF_OUT_DIR = Path(BASEDIR) / "selfdrive" / "ui" / "tests" / "diff" / "report"
+HTML_TEMPLATE_PATH = Path(__file__).with_name("diff_template.html")
+
+
+def extract_framehashes(video_path):
+  cmd = ['ffmpeg', '-i', video_path, '-map', '0:v:0', '-vsync', '0', '-f', 'framehash', '-hash', 'md5', '-']
+  result = subprocess.run(cmd, capture_output=True, text=True, check=True)
+  hashes = []
+  for line in result.stdout.splitlines():
+    if not line or line.startswith('#'):
+      continue
+    parts = line.split(',')
+    if len(parts) < 4:
+      continue
+    hashes.append(parts[-1].strip())
+  return hashes
+
+
+def create_diff_video(video1, video2, output_path):
+  """Create a diff video using ffmpeg blend filter with difference mode."""
+  print("Creating diff video...")
+  cmd = ['ffmpeg', '-i', video1, '-i', video2, '-filter_complex', '[0:v]blend=all_mode=difference', '-vsync', '0', '-y', output_path]
+  subprocess.run(cmd, capture_output=True, check=True)
+
+
+def find_differences(video1, video2) -> tuple[list[int], tuple[int, int]]:
+  print(f"Hashing frames from {video1}...")
+  hashes1 = extract_framehashes(video1)
+
+  print(f"Hashing frames from {video2}...")
+  hashes2 = extract_framehashes(video2)
+
+  print(f"Comparing {len(hashes1)} frames...")
+  different_frames = []
+
+  for i, (h1, h2) in enumerate(zip(hashes1, hashes2, strict=False)):
+    if h1 != h2:
+      different_frames.append(i)
+
+  return different_frames, (len(hashes1), len(hashes2))
+
+
+def generate_html_report(videos: tuple[str, str], basedir: str, different_frames: list[int], frame_counts: tuple[int, int], diff_video_name):
+  chunks = []
+  if different_frames:
+    current_chunk = [different_frames[0]]
+    for i in range(1, len(different_frames)):
+      if different_frames[i] == different_frames[i - 1] + 1:
+        current_chunk.append(different_frames[i])
+      else:
+        chunks.append(current_chunk)
+        current_chunk = [different_frames[i]]
+    chunks.append(current_chunk)
+
+  total_frames = max(frame_counts)
+  frame_delta = frame_counts[1] - frame_counts[0]
+  different_total = len(different_frames) + abs(frame_delta)
+
+  result_text = (
+    f"✅ Videos are identical! ({total_frames} frames)"
+    if different_total == 0
+    else f"❌ Found {different_total} different frames out of {total_frames} total ({different_total / total_frames * 100:.1f}%)."
+    + (f" Video {'2' if frame_delta > 0 else '1'} is longer by {abs(frame_delta)} frames." if frame_delta != 0 else "")
+  )
+
+  # Load HTML template and replace placeholders
+  html = HTML_TEMPLATE_PATH.read_text()
+  placeholders = {
+    "VIDEO1_SRC": os.path.join(basedir, os.path.basename(videos[0])),
+    "VIDEO2_SRC": os.path.join(basedir, os.path.basename(videos[1])),
+    "DIFF_SRC": os.path.join(basedir, diff_video_name),
+    "RESULT_TEXT": result_text,
+  }
+  for key, value in placeholders.items():
+    html = html.replace(f"${key}", value)
+
+  return html
+
+
+def main():
+  parser = argparse.ArgumentParser(description='Compare two videos and generate HTML diff report')
+  parser.add_argument('video1', help='First video file')
+  parser.add_argument('video2', help='Second video file')
+  parser.add_argument('output', nargs='?', default='diff.html', help='Output HTML file (default: diff.html)')
+  parser.add_argument("--basedir", type=str, help="Base directory for output", default="")
+  parser.add_argument('--no-open', action='store_true', help='Do not open HTML report in browser')
+
+  args = parser.parse_args()
+
+  if not args.output.lower().endswith('.html'):
+    args.output += '.html'
+
+  os.makedirs(DIFF_OUT_DIR, exist_ok=True)
+
+  print("=" * 60)
+  print("VIDEO DIFF - HTML REPORT")
+  print("=" * 60)
+  print(f"Video 1: {args.video1}")
+  print(f"Video 2: {args.video2}")
+  print(f"Output: {args.output}")
+  print()
+
+  # Create diff video with name derived from output HTML
+  diff_video_name = Path(args.output).stem + '.mp4'
+  diff_video_path = str(DIFF_OUT_DIR / diff_video_name)
+  create_diff_video(args.video1, args.video2, diff_video_path)
+
+  different_frames, frame_counts = find_differences(args.video1, args.video2)
+
+  if different_frames is None:
+    sys.exit(1)
+
+  print()
+  print("Generating HTML report...")
+  html = generate_html_report((args.video1, args.video2), args.basedir, different_frames, frame_counts, diff_video_name)
+
+  with open(DIFF_OUT_DIR / args.output, 'w') as f:
+    f.write(html)
+
+  # Open in browser by default
+  if not args.no_open:
+    print(f"Opening {args.output} in browser...")
+    webbrowser.open(f'file://{os.path.abspath(DIFF_OUT_DIR / args.output)}')
+
+  extra_frames = abs(frame_counts[0] - frame_counts[1])
+  return 0 if (len(different_frames) + extra_frames) == 0 else 1
+
+
+if __name__ == "__main__":
+  sys.exit(main())

selfdrive/ui/tests/diff/diff_template.html

@@ -0,0 +1,80 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+  <meta charset="UTF-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>UI Diff Report</title>
+  <style>
+    h1, h2, h3, h4 {
+      margin: 0.5em 0 0.25em;
+    }
+    #videoTable {
+      width: 100%;
+    }
+    @media (max-width: 800px) {
+      #videoTable tr {
+        display: flex;
+        flex-direction: column;
+      }
+      #videoTable td {
+        width: 100% !important;
+      }
+    }
+  </style>
+</head>
+<body>
+  <h2>UI Diff</h2>
+  <table id='videoTable'></table>
+  <hr>
+  <p><strong>Results:</strong> $RESULT_TEXT</p>
+  <script>
+    const videoDefs = [
+      { id: 'video1', title: 'Video 1', src: '$VIDEO1_SRC', sync: true },
+      { id: 'video2', title: 'Video 2', src: '$VIDEO2_SRC', sync: true },
+      { id: 'diffVideo', title: 'Pixel Diff', src: '$DIFF_SRC', sync: false },
+    ];
+
+    const row = document.getElementById('videoTable').insertRow();
+    videoDefs.forEach(({ id, title, src, sync }) => {
+      const td = row.insertCell();
+      td.width = '33%';
+      td.innerHTML = `<h4>${title}</h4>
+        <video id='${id}' width='100%' autoplay muted ${sync ? "onplay='syncVideos()'" : ''}>
+          <source src='${src}' type='video/mp4'>
+          Your browser does not support the video tag.
+        </video>`;
+    });
+
+    const videos = videoDefs.map(({ id }) => document.getElementById(id));
+
+    const isEnded = (v) => v.ended || (Number.isFinite(v.duration) && v.currentTime >= (v.duration - 0.05));
+    const playAll = () => videos.forEach((v) => v.play());
+
+    function syncVideos() {
+      const t = Math.min(...videos.map((v) => v.currentTime));
+      videos.forEach((v) => { v.currentTime = t; });
+      playAll();
+    }
+
+    function handleEnded(endedVideo) {
+      endedVideo.pause();
+      if (videos.every(isEnded)) {
+        videos.forEach((v) => { v.currentTime = 0; });
+        playAll();
+      }
+    }
+
+    videos.forEach((v) => {
+      v.addEventListener('timeupdate', () => {
+        videos.forEach((other) => {
+          if (other !== v && !isEnded(other) && Math.abs(v.currentTime - other.currentTime) > 0.1) {
+            other.currentTime = v.currentTime;
+            if (other.paused) other.play();
+          }
+        });
+      });
+      v.addEventListener('ended', () => handleEnded(v));
+    });
+  </script>
+</body>
+</html>

selfdrive/ui/tests/diff/replay.py

@@ -0,0 +1,148 @@
+#!/usr/bin/env python3
+import os
+import argparse
+import coverage
+import pyray as rl
+
+from tqdm import tqdm
+from typing import Literal
+from collections.abc import Callable
+from cereal.messaging import PubMaster
+from openpilot.common.api import Api
+from openpilot.common.basedir import BASEDIR
+from openpilot.common.params import Params
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.selfdrive.ui.tests.diff.diff import DIFF_OUT_DIR
+from openpilot.system.updated.updated import parse_release_notes
+from openpilot.system.version import terms_version, training_version
+
+LayoutVariant = Literal["mici", "tizi"]
+
+FPS = 60
+HEADLESS = os.getenv("WINDOWED", "0") != "1"
+
+
+def setup_state():
+  params = Params()
+  params.put("HasAcceptedTerms", terms_version, block=True)
+  params.put("CompletedTrainingVersion", training_version, block=True)
+  params.put("DongleId", "test123456789", block=True)
+  # Combined description for layouts that still use it (BIG home, settings/software)
+  params.put("UpdaterCurrentDescription", "0.10.1 / test-branch / abc1234 / Nov 30", block=True)
+  params.put("UpdaterCurrentReleaseNotes", parse_release_notes(BASEDIR), block=True)
+  # Params for mici home
+  params.put("Version", "0.10.1", block=True)
+  params.put("GitBranch", "test-branch", block=True)
+  params.put("GitCommit", "abc12340ff9131237ba23a1d0fbd8edf9c80e87", block=True)
+  params.put("GitCommitDate", "'1732924800 2024-11-30 00:00:00 +0000'", block=True)
+
+  # Patch Api.get_token to return a static token so the pairing QR code is deterministic across runs
+  Api.get_token = lambda self, payload_extra=None, expiry_hours=0: "test_token"
+
+
+def run_replay(variant: LayoutVariant) -> None:
+  if HEADLESS:
+    rl.set_config_flags(rl.ConfigFlags.FLAG_WINDOW_HIDDEN)
+    os.environ["OFFSCREEN"] = "1"  # Run UI without FPS limit (set before importing gui_app)
+
+  setup_state()
+  os.makedirs(DIFF_OUT_DIR, exist_ok=True)
+
+  from openpilot.selfdrive.ui.ui_state import ui_state, device  # Import within OpenpilotPrefix context so param values are setup correctly
+  from openpilot.system.ui.lib.application import gui_app  # Import here for accurate coverage
+  from openpilot.selfdrive.ui.tests.diff.replay_script import build_script
+
+  gui_app.init_window("ui diff test", fps=FPS)
+
+  # Dynamically import main layout based on variant
+  if variant == "mici":
+    from openpilot.selfdrive.ui.mici.layouts.main import MiciMainLayout as MainLayout
+  else:
+    from openpilot.selfdrive.ui.layouts.main import MainLayout
+  main_layout = MainLayout()
+
+  # Disable interactive timeout — replay clicks use left_down=False so they never reset the timer,
+  # and after 30s of real wall-clock time the settings panel would close automatically.
+  device.set_override_interactive_timeout(99999)
+
+  pm = PubMaster(["deviceState", "pandaStates", "driverStateV2", "selfdriveState"])
+  script = build_script(pm, main_layout, variant)
+  script_index = 0
+
+  send_fn: Callable | None = None
+  frame = 0
+  # Override raylib timing functions to return deterministic values based on frame count instead of real time
+  rl.get_frame_time = lambda: 1.0 / FPS
+  rl.get_time = lambda: frame / FPS
+
+  # Main loop to replay events and render frames
+  with tqdm(total=script[-1][0] + 1, desc="Replaying", unit="frame", disable=bool(os.getenv("CI"))) as pbar:
+    for _ in gui_app.render():
+      # Handle all events for the current frame
+      while script_index < len(script) and script[script_index][0] == frame:
+        _, event = script[script_index]
+        # Call setup function, if any
+        if event.setup:
+          event.setup()
+        # Send mouse events to the application
+        if event.mouse_events:
+          with gui_app._mouse._lock:
+            gui_app._mouse._events.extend(event.mouse_events)
+        # Update persistent send function
+        if event.send_fn is not None:
+          send_fn = event.send_fn
+        # Move to next script event
+        script_index += 1
+
+      # Keep sending cereal messages for persistent states (onroad, alerts)
+      if send_fn:
+        send_fn()
+
+      ui_state.update()
+
+      frame += 1
+      pbar.update(1)
+
+      if script_index >= len(script):
+        break
+
+  gui_app.close()
+
+  print(f"Total frames: {frame}")
+  print(f"Video saved to: {os.environ['RECORD_OUTPUT']}")
+
+
+def main():
+  parser = argparse.ArgumentParser()
+  parser.add_argument('--big', action='store_true', help='Use big UI layout (tizi/tici) instead of mici layout')
+  args = parser.parse_args()
+
+  variant: LayoutVariant = 'tizi' if args.big else 'mici'
+
+  if args.big:
+    os.environ["BIG"] = "1"
+  os.environ["RECORD"] = "1"
+  os.environ["RECORD_QUALITY"] = "0"  # Use CRF 0 ("lossless" encode) for deterministic output across different machines
+  os.environ["RECORD_OUTPUT"] = os.path.join(DIFF_OUT_DIR, os.environ.get("RECORD_OUTPUT", f"{variant}_ui_replay.mp4"))
+
+  print(f"Running {variant} UI replay...")
+  with OpenpilotPrefix():
+    sources = ["openpilot.system.ui"]
+    if variant == "mici":
+      sources.append("openpilot.selfdrive.ui.mici")
+      omit = ["**/*tizi*", "**/*tici*"]  # exclude files containing "tizi" or "tici"
+    else:
+      sources.extend(["openpilot.selfdrive.ui.layouts", "openpilot.selfdrive.ui.onroad", "openpilot.selfdrive.ui.widgets"])
+      omit = ["**/*mici*"]  # exclude files containing "mici"
+    cov = coverage.Coverage(source=sources, omit=omit)
+    with cov.collect():
+      run_replay(variant)
+    cov.save()
+    cov.report()
+    directory = os.path.join(DIFF_OUT_DIR, f"htmlcov-{variant}")
+    cov.html_report(directory=directory)
+    print(f"HTML report: {directory}/index.html")
+
+
+if __name__ == "__main__":
+  main()

selfdrive/ui/tests/diff/replay_script.py

@@ -0,0 +1,526 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+from collections.abc import Callable
+from dataclasses import dataclass
+
+import math
+
+from cereal import car, log, messaging
+from cereal.messaging import PubMaster
+from openpilot.common.basedir import BASEDIR
+from openpilot.common.params import Params
+from openpilot.selfdrive.selfdrived.alertmanager import set_offroad_alert
+from openpilot.selfdrive.ui.lib.prime_state import PrimeType
+from openpilot.selfdrive.ui.tests.diff.replay import FPS, LayoutVariant
+from openpilot.system.updated.updated import parse_release_notes
+
+# Default frames to wait after events
+WAIT_LONG = FPS
+WAIT_SHORT = FPS // 2
+FAST_CLICK = FPS // 6
+
+# Direction vectors for drag gestures
+DIR_LEFT = (-1, 0)
+DIR_RIGHT = (1, 0)
+DIR_UP = (0, -1)
+DIR_DOWN = (0, 1)
+
+AlertSize = log.SelfdriveState.AlertSize
+AlertStatus = log.SelfdriveState.AlertStatus
+
+BRANCH_NAME = "this-is-a-really-super-mega-ultra-max-extreme-ultimate-long-branch-name"
+
+
+@dataclass
+class ScriptEvent:
+  if TYPE_CHECKING:
+    # Only import for type checking to avoid excluding the application code from coverage
+    from openpilot.system.ui.lib.application import MouseEvent
+
+  setup: Callable | None = None  # Setup function to run prior to adding mouse events
+  mouse_events: list[MouseEvent] | None = None  # Mouse events to send to the application on this event's frame
+  send_fn: Callable | None = None  # When set, the main loop uses this as the new persistent sender
+
+
+ScriptEntry = tuple[int, ScriptEvent]  # (frame, event)
+
+
+class Script:
+  def __init__(self, fps: int) -> None:
+    self.fps = fps
+    self.frame = 0
+    self.entries: list[ScriptEntry] = []
+
+  def get_frame_time(self) -> float:
+    return self.frame / self.fps
+
+  def add(self, event: ScriptEvent, before: int = 0, after: int = 0) -> None:
+    """Add event to the script, optionally with the given number of frames to wait before or after the event."""
+    self.frame += before
+    self.entries.append((self.frame, event))
+    self.frame += after
+
+  def end(self) -> None:
+    """Add a final empty event to mark the end of the script."""
+    self.add(ScriptEvent())  # Without this, it will just end on the last event without waiting for any specified delay after it
+
+  def wait(self, frames: int) -> None:
+    """Add a delay for the given number of frames followed by an empty event."""
+    self.add(ScriptEvent(), before=frames)
+
+  def setup(self, fn: Callable, wait_after: int = WAIT_SHORT) -> None:
+    """Add a setup function to be called immediately followed by a delay of the given number of frames."""
+    self.add(ScriptEvent(setup=fn), after=wait_after)
+
+  def set_send(self, fn: Callable, wait_after: int = WAIT_SHORT) -> None:
+    """Set a new persistent send function to be called every frame."""
+    self.add(ScriptEvent(send_fn=fn), after=wait_after)
+
+  def click(self, x: int, y: int, wait_after: int = WAIT_SHORT, wait_between: int = 2) -> None:
+    """Add a click event to the script for the given position and specify frames to wait between mouse events or after the click."""
+    # NOTE: By default we wait a couple frames between mouse events so pressed states will be rendered
+    from openpilot.system.ui.lib.application import MouseEvent, MousePos
+
+    mouse_down = MouseEvent(pos=MousePos(x, y), slot=0, left_pressed=True, left_released=False, left_down=False, t=self.get_frame_time())
+    self.add(ScriptEvent(mouse_events=[mouse_down]), after=wait_between)
+    mouse_up = MouseEvent(pos=MousePos(x, y), slot=0, left_pressed=False, left_released=True, left_down=False, t=self.get_frame_time())
+    self.add(ScriptEvent(mouse_events=[mouse_up]), after=wait_after)
+
+  def drag(self, start_x: int, start_y: int, direction: tuple[int, int], distance: int, duration_frames: int, wait_after: int = WAIT_LONG) -> None:
+    """Add a drag gesture to the script from start position in the specified direction by the given distance over the given number of frames."""
+    from openpilot.system.ui.lib.application import MouseEvent, MousePos
+
+    # Calculate delta and end position based on direction and distance
+    delta_x, delta_y = direction[0] * distance, direction[1] * distance
+    end_x, end_y = start_x + delta_x, start_y + delta_y
+
+    # Mouse down at start
+    mouse_down = MouseEvent(pos=MousePos(start_x, start_y), slot=0, left_pressed=True, left_released=False, left_down=True, t=self.get_frame_time())
+    self.add(ScriptEvent(mouse_events=[mouse_down]), after=1)
+
+    # Interpolate positions over duration_frames
+    for i in range(1, duration_frames):
+      t = i / duration_frames
+      x, y = int(start_x + delta_x * t), int(start_y + delta_y * t)
+      mouse_move = MouseEvent(pos=MousePos(x, y), slot=0, left_pressed=False, left_released=False, left_down=True, t=self.get_frame_time())
+      self.add(ScriptEvent(mouse_events=[mouse_move]), after=1)
+
+    # Mouse up at end
+    mouse_up = MouseEvent(pos=MousePos(end_x, end_y), slot=0, left_pressed=False, left_released=True, left_down=False, t=self.get_frame_time())
+    self.add(ScriptEvent(mouse_events=[mouse_up]), after=wait_after)
+
+
+# --- Setup functions ---
+
+
+def set_prime_state(prime_type: PrimeType) -> None:
+  from openpilot.selfdrive.ui.ui_state import ui_state
+  ui_state.prime_state.set_type(prime_type)
+
+
+def setup_offroad_alerts() -> None:
+  set_offroad_alert("Offroad_TemperatureTooHigh", True, extra_text='99C')
+  set_offroad_alert("Offroad_ExcessiveActuation", True, extra_text='longitudinal')
+  set_offroad_alert("Offroad_IsTakingSnapshot", True)
+
+
+def setup_update_available(available: bool = True) -> None:
+  params = Params()
+  params.put_bool("UpdateAvailable", available, block=True)
+  params.put("UpdaterAvailableBranches", ",".join(["test-branch", "test-branch-2", BRANCH_NAME]), block=True)
+  if available:
+    params.put("UpdaterNewDescription", f"0.10.2 / {BRANCH_NAME} / 0a1b2c3 / Jan 01", block=True)
+    params.put("UpdaterNewReleaseNotes", parse_release_notes(BASEDIR), block=True)
+    params.put("UpdaterTargetBranch", BRANCH_NAME, block=True)
+  else:
+    params.remove("UpdaterNewDescription")
+    params.remove("UpdaterNewReleaseNotes")
+    params.remove("UpdaterTargetBranch")
+
+
+def setup_calibration_params() -> None:
+  params = Params()
+  # live calibration
+  calib = messaging.new_message('liveCalibration')
+  calib.liveCalibration.calStatus = log.LiveCalibrationData.Status.calibrated
+  calib.liveCalibration.rpyCalib = [0.0, math.radians(2.5), math.radians(-1.2)]
+  params.put("CalibrationParams", calib.to_bytes(), block=True)
+  # live delay
+  delay = messaging.new_message('liveDelay')
+  delay.liveDelay.calPerc = 75
+  params.put("LiveDelay", delay.to_bytes(), block=True)
+  # live torque parameters
+  torque = messaging.new_message('liveTorqueParameters')
+  torque.liveTorqueParameters.useParams = True
+  torque.liveTorqueParameters.calPerc = 60
+  params.put("LiveTorqueParameters", torque.to_bytes(), block=True)
+
+
+def setup_developer_params() -> None:
+  CP = car.CarParams()
+  CP.alphaLongitudinalAvailable = True
+  Params().put("CarParamsPersistent", CP.to_bytes(), block=True)
+
+
+# --- Send functions ---
+
+def send_onroad(pm: PubMaster) -> None:
+  ds = messaging.new_message('deviceState')
+  ds.deviceState.started = True
+  ds.deviceState.networkType = log.DeviceState.NetworkType.wifi
+
+  ps = messaging.new_message('pandaStates', 1)
+  ps.pandaStates[0].pandaType = log.PandaState.PandaType.dos
+  ps.pandaStates[0].ignitionLine = True
+
+  pm.send('deviceState', ds)
+  pm.send('pandaStates', ps)
+
+
+def make_network_state_setup(pm: PubMaster, network_type) -> Callable:
+  def _send() -> None:
+    ds = messaging.new_message('deviceState')
+    ds.deviceState.networkType = network_type
+    pm.send('deviceState', ds)
+  return _send
+
+
+def make_alert_setup(pm: PubMaster, size, text1, text2, status) -> Callable:
+  def _send() -> None:
+    alert = messaging.new_message('selfdriveState')
+    ss = alert.selfdriveState
+    ss.alertSize = size
+    ss.alertText1 = text1
+    ss.alertText2 = text2
+    ss.alertStatus = status
+    pm.send('selfdriveState', alert)
+  return _send
+
+
+def test_onroad_alerts(script: Script, pm: PubMaster) -> None:
+  """Go through various alert types and sizes and add them to the script to test alert rendering.
+    Each alert is sent as a separate event with a delay in between."""
+  # Small alert (normal)
+  script.set_send(make_alert_setup(pm, AlertSize.small, "Small Alert", "This is a small alert", AlertStatus.normal))
+  # Medium alert (userPrompt)
+  script.set_send(make_alert_setup(pm, AlertSize.mid, "Medium Alert", "This is a medium alert", AlertStatus.userPrompt))
+  # Full alert (critical)
+  script.set_send(make_alert_setup(pm, AlertSize.full, "DISENGAGE IMMEDIATELY", "Driver Distracted", AlertStatus.critical))
+  # Full alert multiline
+  script.set_send(make_alert_setup(pm, AlertSize.full, "Reverse\nGear", "", AlertStatus.normal))
+  # Full alert long text
+  script.set_send(make_alert_setup(pm, AlertSize.full, "TAKE CONTROL IMMEDIATELY", "Calibration Invalid: Remount Device & Recalibrate", AlertStatus.userPrompt))
+
+
+# --- Script builders ---
+
+def build_mici_script(pm: PubMaster, main_layout, script: Script) -> None:
+  """Build the replay script for the mici layout."""
+  from openpilot.system.ui.lib.application import gui_app
+
+  width, height = gui_app.width, gui_app.height
+  center = (width // 2, height // 2)
+  right = (width * 4 // 5, height // 2)
+  left = (width // 5, height // 2)
+  top = (width // 2, height // 10)
+  bottom = (width // 2, height * 9 // 10)
+
+  DURATION = 5
+  SWIPE_WAIT = FPS * 3 // 4
+
+  def click(times: int = 1, wait_after: int = WAIT_SHORT) -> None:
+    """Click at the center of the screen the given number of times with optional delay after."""
+    for _ in range(times):
+      script.click(*center, wait_after=wait_after)
+
+  def press(x: int, y: int, duration_frames: int = DURATION, wait_after: int = WAIT_SHORT) -> None:
+    """Perform a drag with no movement to simulate a left_down mouse event at the given position for the specified duration and delay after."""
+    script.drag(x, y, (0, 0), 0, duration_frames, wait_after=wait_after)
+
+  def swipe_left(distance: int = right[0] - left[0], duration_frames: int = DURATION, wait_after: int = SWIPE_WAIT) -> None:
+    """Drag from right edge to left (scroll right / slide confirmation)."""
+    script.drag(*right, DIR_LEFT, distance, duration_frames, wait_after)
+
+  def swipe_right(distance: int = right[0] - left[0], duration_frames: int = DURATION, wait_after: int = SWIPE_WAIT) -> None:
+    """Drag from left edge to right (scroll left)."""
+    script.drag(*left, DIR_RIGHT, distance, duration_frames, wait_after)
+
+  def swipe_down(distance: int = bottom[1] - top[1], duration_frames: int = DURATION, wait_after: int = SWIPE_WAIT) -> None:
+    """Drag from top edge to bottom (scroll up / go back)."""
+    script.drag(*top, DIR_DOWN, distance, duration_frames, wait_after)
+
+  def swipe_up(distance: int = bottom[1] - top[1], duration_frames: int = DURATION, wait_after: int = SWIPE_WAIT) -> None:
+    """Drag from bottom edge to top (scroll down)."""
+    script.drag(*bottom, DIR_UP, distance, duration_frames, wait_after)
+
+  ActionFn = Callable[[], None] | None
+  Cases = list[ActionFn]
+
+  def run_actions(*actions: ActionFn, after_each: ActionFn = None) -> None:
+    """Helper function to run a sequence of actions in order for interaction tests, calling after_each callback after each action if provided."""
+    for action in actions:
+      if action is not None:
+        action()
+      if after_each is not None:
+        after_each()
+
+  def explore_setting(*actions: ActionFn) -> None:
+    """Helper function to open a settings item, run the given actions, and go back."""
+    run_actions(click, *actions, swipe_down)  # open, interact, go back
+
+  def scroll_through_cases(cases: Cases) -> None:
+    """Helper function to explore a panel by calling the interaction callbacks for each item/page before swiping to the next one."""
+    run_actions(*cases, after_each=lambda: swipe_left(210, 10))  # swipe to roughly the center of the next toggle after each case
+
+  def interact_keyboard() -> None:
+    """Interact with the keyboard in various ways to test different actions and states.
+    Assumes it's a password keyboard with 8 characters required. Closes by pressing confirm at the end."""
+    KEY = (250, 160)  # key in the middle of the keyboard ('G')
+    SHIFT = (50, 210)
+    NUMBERS = (480, 210)
+    SPACE = (500, 160)
+    BACKSPACE = (490, 30)
+    CONFIRM = (50, 30)
+    # Begin interactions
+    press(*CONFIRM, wait_after=FAST_CLICK)  # confirm while disabled should do nothing
+    swipe_left(duration_frames=FPS // 2)  # swipe to type
+    swipe_up(duration_frames=FPS // 2)  # swipe out of keyboard (nothing typed)
+    # press various keys to test different states:
+    for key in [
+      SHIFT, KEY, KEY, SHIFT, SHIFT, KEY, KEY,  # test casing (upper, lower, caps lock)
+      SPACE, SPACE, BACKSPACE, BACKSPACE,  # test multiple space and backspace
+      NUMBERS, KEY, center, SHIFT, KEY  # test numbers and symbols
+    ]:
+      press(*key, wait_after=FAST_CLICK)
+    # press confirm to close
+    script.wait(WAIT_SHORT)  # wait for confirm to enable
+    press(*CONFIRM)
+
+  toggle_cases: Cases = [
+    lambda: click(times=3, wait_after=FAST_CLICK),  # first toggle is personality, which has 3 states
+    None, None, None, None, None, None,  # skip other toggles to save time
+    lambda: click(times=2, wait_after=FAST_CLICK),  # test final toggle (enable openpilot)
+  ]
+
+  network_cases: Cases = [
+    explore_setting,  # select wifi (just open and close)
+    None, None,
+    lambda: run_actions(click, interact_keyboard),  # tether password keyboard
+  ]
+
+  device_cases: Cases = [
+    None,
+    click,  # update
+    explore_setting,  # pairing (just open and close)
+    lambda: explore_setting(
+      # training guide
+      lambda: swipe_left(width * 2), click,  # first page, click next
+      lambda: swipe_left(width * 2), swipe_down  # second page, go back (TODO: make driver cam preview work)
+    ),
+    None,  # TODO: preview driver camera; enabling this causes MultiplePublishersError later in onroad alert tests
+    lambda: explore_setting(swipe_left),  # terms & conditions (swipe to view QR code)
+    lambda: explore_setting(lambda: swipe_up(height * 3), lambda: swipe_down(height * 3)),  # regulatory info
+    lambda: run_actions(click, lambda: swipe_left(width)),  # reset calibration confirm (goes back automatically)
+    lambda: explore_setting(lambda: swipe_left(width)),  # uninstall
+    lambda: run_actions(
+      lambda: explore_setting(lambda: swipe_left(width)),  # reboot
+      lambda: script.click(430, 120), lambda: swipe_left(width), swipe_down,  # shutdown
+    ),
+  ]
+
+  developer_cases: Cases = [
+    lambda: click(times=2, wait_after=FAST_CLICK),  # toggle ssh mode
+    explore_setting,  # SSH keys keyboard (just open and close)
+    None,  # joystick mode
+    lambda: click(wait_after=FAST_CLICK),  # longitudinal maneuver mode (disabled; should do nothing)
+    lambda: click(times=2, wait_after=FAST_CLICK),  # toggle UI debug mode
+  ]
+
+  settings_cases: Cases = [
+    lambda: scroll_through_cases(toggle_cases),
+    lambda: scroll_through_cases(network_cases),
+    lambda: scroll_through_cases(device_cases),
+    lambda: script.wait(WAIT_SHORT),  # pairing
+    lambda: run_actions(lambda: swipe_up(height * 3), lambda: swipe_down(height * 3)),  # firehose (scroll down and back up)
+    lambda: scroll_through_cases(developer_cases),
+  ]
+
+  # === Homescreen === #
+  script.wait(WAIT_SHORT)
+  swipe_left(width, wait_after=WAIT_SHORT)  # onroad screen
+  swipe_right(width, wait_after=WAIT_SHORT)  # back to home
+
+  # === Offroad Alerts ===
+  def setup_offroad_alerts_and_refresh() -> None:
+    """Setup function to trigger offroad alerts and force a refresh on the alerts layout."""
+    setup_offroad_alerts()
+    params = Params()
+    main_layout._alerts_layout._pending_params = ({"UpdaterNewDescription": params.get("UpdaterNewDescription")} |
+                                                  {alert_data.key: params.get(alert_data.key) for alert_data in main_layout._alerts_layout.sorted_alerts})
+    main_layout._alerts_layout._refresh()
+
+  swipe_right(width, wait_after=WAIT_SHORT)  # open alerts
+  script.setup(setup_offroad_alerts_and_refresh)  # show alerts
+  swipe_up(height)  # scroll alerts
+  swipe_left(width, wait_after=WAIT_SHORT)  # close alerts
+
+  # === Settings === #
+  click()  # open settings
+  scroll_through_cases([lambda case=case: explore_setting(case) for case in settings_cases])  # explore settings
+  swipe_down()  # back to home
+
+  # === Onroad ===
+  script.set_send(lambda: send_onroad(pm))
+  swipe_left(width, wait_after=WAIT_SHORT)  # onroad screen
+  test_onroad_alerts(script, pm)
+  swipe_right(width)  # back to home
+
+  script.end()
+
+
+def build_tizi_script(pm: PubMaster, main_layout, script: Script) -> None:
+  """Build the replay script for the tizi layout."""
+
+  def make_home_refresh_setup(fn: Callable) -> Callable:
+    """Return setup function that calls the given function to modify state and forces an immediate refresh on the home layout."""
+    from openpilot.selfdrive.ui.layouts.main import MainState
+
+    def setup():
+      fn()
+      main_layout._layouts[MainState.HOME].last_refresh = 0
+
+    return setup
+
+  def add_prime_state_setup(prime_type: PrimeType) -> None:
+    script.set_send(lambda: set_prime_state(prime_type))
+
+  def do_onboarding() -> None:
+    """Click through the training guide and close."""
+    from openpilot.selfdrive.ui.layouts.onboarding import STEP_RECTS
+    step = 0
+    for step_rect in STEP_RECTS:
+      if step < len(STEP_RECTS) - 1:
+        script.click(int(step_rect.x), int(step_rect.y), wait_after=FAST_CLICK)
+      else:
+        script.click(950, 900)  # On the last step, click Finish instead of restart
+      step += 1
+
+  def type_keyboard() -> None:
+    """Types 8 characters using the big keyboard to test different layouts and interactions."""
+    KEY = (150, 430)  # e.g. 'Q' key
+    SHIFT = (150, 750)  # also symbols key in number mode
+    NUMBERS = (150, 950)
+    SPACE = (1060, 950)
+    BACKSPACE = (2000, 780)
+    for key in [
+      SHIFT, KEY, KEY, SHIFT, SHIFT, KEY, KEY,  # test casing (upper, lower, caps lock)
+      SPACE, SPACE, BACKSPACE, BACKSPACE,  # test multiple space and backspace
+      NUMBERS, KEY, KEY, SHIFT, KEY, KEY  # test numbers and symbols
+    ]:
+      script.click(*key, wait_after=FAST_CLICK)
+
+  # TODO: Better way of organizing the events
+
+  # === Homescreen ===
+  script.set_send(make_network_state_setup(pm, log.DeviceState.NetworkType.wifi))
+  # Go through different prime state layouts
+  add_prime_state_setup(PrimeType.LITE)
+  add_prime_state_setup(PrimeType.NONE)
+  add_prime_state_setup(PrimeType.UNPAIRED)
+
+  # === Update Available (auto-transitions via HomeLayout refresh) ===
+  script.setup(make_home_refresh_setup(setup_update_available))
+
+  # === Offroad Alerts (auto-transitions via HomeLayout refresh, overrides update) ===
+  script.setup(make_home_refresh_setup(setup_offroad_alerts))
+  script.click(620, 950)  # close alerts
+
+  # === Settings (click sidebar settings button) ===
+  script.click(150, 90)
+
+  # === Settings - Device ===
+  # pair device
+  script.click(2000, 450)  # pair device
+  script.click(110, 110)  # close pairing dialog
+  add_prime_state_setup(PrimeType.NONE)  # changed from unpaired to hide pair device button
+  # calibration
+  script.setup(setup_calibration_params, wait_after=0)
+  script.click(1000, 620)  # expand calibration description
+  script.click(2000, 620)  # reset calibration confirmation
+  script.click(1500, 750)  # confirm reset
+  script.click(1000, 620)  # collapse calibration description
+  # training guide
+  script.click(2000, 800)  # open training guide
+  do_onboarding()
+  # regulatory info
+  script.click(2000, 970)  # regulatory button
+  script.click(2000, 970)  # OK
+
+  # === Settings - Network ===
+  script.click(278, 450)
+  # TODO: mock networks
+  script.click(1880, 100)  # advanced network settings
+
+  # Keyboard (tethering password)
+  script.click(2000, 420, wait_after=FAST_CLICK)  # open tether password keyboard
+  script.click(2000, 950, wait_after=FAST_CLICK)  # click confirm (disabled, should not close)
+  script.click(2000, 115)  # cancel (close without typing)
+  script.click(2000, 420, wait_after=FAST_CLICK)  # open keyboard again
+  type_keyboard()  # test various keyboard layouts and interactions
+  script.click(2050, 250, wait_after=FAST_CLICK)  # toggle show/hide password
+  script.click(2000, 950)  # confirm (close keyboard)
+
+  script.click(630, 80)  # back from advanced network
+
+  # === Settings - Toggles ===
+  script.click(278, 600)
+  script.click(1200, 280)  # expand experimental mode description
+
+  # === Settings - Software ===
+  script.setup(lambda: setup_update_available(False), wait_after=0)  # start with no update available
+  script.click(278, 720)  # software
+  for _ in range(2):
+    script.click(720, 120)  # toggle current release notes
+  script.setup(setup_update_available)  # set update available
+  for _ in range(2):
+    script.click(720, 450)  # toggle new release notes
+  script.click(2000, 630)  # open select branch dialog
+  script.click(1000, 300)  # select 1st option
+  script.click(1600, 900)  # confirm selection
+  script.click(2000, 800)  # uninstall
+  script.click(650, 750)  # cancel uninstall
+
+  # === Settings - Firehose ===
+  script.click(278, 845)
+
+  # === Settings - Developer (set CarParamsPersistent first) ===
+  script.setup(setup_developer_params, wait_after=0)
+  script.click(278, 950)
+  script.click(1930, 470)  # SSH keys (keyboard)
+  script.click(1930, 115)  # click cancel on keyboard
+  script.click(2000, 960)  # toggle alpha long
+  script.click(1500, 875)  # confirm
+
+  # === Close settings ===
+  script.click(250, 160)
+
+  # === Onroad ===
+  script.set_send(lambda: send_onroad(pm))
+  script.click(1000, 500)  # click onroad to toggle sidebar
+  test_onroad_alerts(script, pm)
+
+  # End
+  script.end()
+
+
+def build_script(pm: PubMaster, main_layout, variant: LayoutVariant) -> list[ScriptEntry]:
+  """Build the replay script for the appropriate layout variant and return list of script entries."""
+  print(f"Building {variant} replay script...")
+
+  script = Script(FPS)
+  builder = build_tizi_script if variant == 'tizi' else build_mici_script
+  builder(pm, main_layout, script)
+
+  print(f"Built replay script with {len(script.entries)} events and {script.frame} frames ({script.get_frame_time():.2f} seconds)")
+
+  return script.entries

selfdrive/ui/tests/profile_onroad.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+import os
+import time
+import cProfile
+import pyray as rl
+import numpy as np
+
+from msgq.visionipc import VisionIpcServer, VisionStreamType
+from openpilot.selfdrive.ui.ui_state import ui_state
+from openpilot.selfdrive.ui.mici.layouts.main import MiciMainLayout
+from openpilot.system.ui.lib.application import gui_app
+from openpilot.tools.lib.logreader import LogReader
+
+FPS = 60
+
+
+def chunk_messages_by_time(messages):
+  dt_ns = 1e9 / FPS
+  chunks = []
+  current_services = {}
+  next_time = messages[0].logMonoTime + dt_ns if messages else 0
+
+  for msg in messages:
+    if msg.logMonoTime >= next_time:
+      chunks.append(current_services)
+      current_services = {}
+      next_time += dt_ns * ((msg.logMonoTime - next_time) // dt_ns + 1)
+    current_services[msg.which()] = msg
+
+  if current_services:
+    chunks.append(current_services)
+  return chunks
+
+
+def patch_submaster(message_chunks):
+  def mock_update(timeout=None):
+    sm = ui_state.sm
+    sm.updated = dict.fromkeys(sm.services, False)
+    current_time = time.monotonic()
+    for service, msg in message_chunks[sm.frame].items():
+      if service in sm.data:
+        sm.seen[service] = True
+        sm.updated[service] = True
+
+        msg_builder = msg.as_builder()
+        sm.data[service] = getattr(msg_builder, service)
+        sm.logMonoTime[service] = msg.logMonoTime
+        sm.recv_time[service] = current_time
+        sm.recv_frame[service] = sm.frame
+        sm.valid[service] = True
+    sm.frame += 1
+  ui_state.sm.update = mock_update
+
+
+if __name__ == "__main__":
+  import argparse
+  parser = argparse.ArgumentParser(description='Profile openpilot UI rendering and state updates')
+  parser.add_argument('route', type=str, nargs='?', default="302bab07c1511180/00000006--0b9a7005f1/3",
+                      help='Route to use for profiling')
+  parser.add_argument('--loop', type=int, default=1,
+                      help='Number of times to loop the log (default: 1)')
+  parser.add_argument('--output', type=str, default='cachegrind.out.ui',
+                      help='Output file prefix (default: cachegrind.out.ui)')
+  parser.add_argument('--max-seconds', type=float, default=None,
+                      help='Maximum seconds of messages to process (default: all)')
+  parser.add_argument('--headless', action='store_true',
+                      help='Run in headless mode without GPU (for CI/testing)')
+  args = parser.parse_args()
+
+  print(f"Loading log from {args.route}...")
+  lr = LogReader(args.route, sort_by_time=True)
+  messages = list(lr) * args.loop
+
+  print("Chunking messages...")
+  message_chunks = chunk_messages_by_time(messages)
+  if args.max_seconds:
+    message_chunks = message_chunks[:int(args.max_seconds * FPS)]
+
+  print("Initializing UI with GPU rendering...")
+
+  if args.headless:
+    os.environ['SDL_VIDEODRIVER'] = 'dummy'
+
+  gui_app.init_window("UI Profiling", fps=600)
+  main_layout = MiciMainLayout()
+
+  print("Running...")
+  patch_submaster(message_chunks)
+
+  W, H = 2048, 1216
+  vipc = VisionIpcServer("camerad")
+  vipc.create_buffers(VisionStreamType.VISION_STREAM_ROAD, 5, W, H)
+  vipc.start_listener()
+  yuv_buffer_size = W * H + (W // 2) * (H // 2) * 2
+  yuv_data = np.random.randint(0, 256, yuv_buffer_size, dtype=np.uint8).tobytes()
+  with cProfile.Profile() as pr:
+    for _ in gui_app.render():
+      if ui_state.sm.frame >= len(message_chunks):
+        break
+      if ui_state.sm.frame % 3 == 0:
+        eof = int((ui_state.sm.frame % 3) * 0.05 * 1e9)
+        vipc.send(VisionStreamType.VISION_STREAM_ROAD, yuv_data, ui_state.sm.frame % 3, eof, eof)
+      ui_state.update()
+    pr.dump_stats(f'{args.output}_deterministic.stats')
+
+  rl.close_window()
+  print("\nProfiling complete!")
+  print(f"  run: python -m pstats {args.output}_deterministic.stats")

selfdrive/ui/tests/test_feedbackd.py

@@ -0,0 +1,53 @@
+import pytest
+import cereal.messaging as messaging
+from cereal import car
+from openpilot.common.params import Params
+from openpilot.system.manager.process_config import managed_processes
+
+
+@pytest.mark.skip("tmp disabled")
+class TestFeedbackd:
+  def setup_method(self):
+    self.pm = messaging.PubMaster(['carState', 'rawAudioData'])
+    self.sm = messaging.SubMaster(['audioFeedback'])
+
+  def _send_lkas_button(self, pressed: bool):
+    msg = messaging.new_message('carState')
+    msg.carState.canValid = True
+    msg.carState.buttonEvents = [{'type': car.CarState.ButtonEvent.Type.lkas, 'pressed': pressed}]
+    self.pm.send('carState', msg)
+
+  def _send_audio_data(self, count: int = 5):
+    for _ in range(count):
+      audio_msg = messaging.new_message('rawAudioData')
+      audio_msg.rawAudioData.data = bytes(1600)  # 800 samples of int16
+      audio_msg.rawAudioData.sampleRate = 16000
+      self.pm.send('rawAudioData', audio_msg)
+      self.sm.update(timeout=100)
+
+  @pytest.mark.parametrize("record_feedback", [False, True])
+  def test_audio_feedback(self, record_feedback):
+    Params().put_bool("RecordAudioFeedback", record_feedback, block=True)
+
+    managed_processes["feedbackd"].start()
+    assert self.pm.wait_for_readers_to_update('carState', timeout=5)
+    assert self.pm.wait_for_readers_to_update('rawAudioData', timeout=5)
+
+    self._send_lkas_button(pressed=True)
+    self._send_audio_data()
+    self._send_lkas_button(pressed=False)
+    self._send_audio_data()
+
+    if record_feedback:
+      assert self.sm.updated['audioFeedback'], "audioFeedback should be published when enabled"
+    else:
+      assert not self.sm.updated['audioFeedback'], "audioFeedback should not be published when disabled"
+
+    self._send_lkas_button(pressed=True)
+    self._send_audio_data()
+    self._send_lkas_button(pressed=False)
+    self._send_audio_data()
+
+    assert not self.sm.updated['audioFeedback'], "audioFeedback should not be published after second press"
+
+    managed_processes["feedbackd"].stop()

selfdrive/ui/tests/test_raylib_ui.py

@@ -0,0 +1,8 @@
+import time
+from openpilot.selfdrive.test.helpers import with_processes
+
+
+@with_processes(["ui"])
+def test_raylib_ui():
+  """Test initialization of the UI widgets is successful."""
+  time.sleep(1)

selfdrive/ui/tests/test_soundd.py

@@ -0,0 +1,35 @@
+from cereal import car
+from cereal import messaging
+from cereal.messaging import SubMaster, PubMaster
+from openpilot.selfdrive.ui.soundd import SELFDRIVE_STATE_TIMEOUT, check_selfdrive_timeout_alert
+
+import time
+
+AudibleAlert = car.CarControl.HUDControl.AudibleAlert
+
+
+class TestSoundd:
+  def test_check_selfdrive_timeout_alert(self):
+    sm = SubMaster(['selfdriveState'])
+    pm = PubMaster(['selfdriveState'])
+
+    for _ in range(100):
+      cs = messaging.new_message('selfdriveState')
+      cs.selfdriveState.enabled = True
+
+      pm.send("selfdriveState", cs)
+
+      time.sleep(0.01)
+
+      sm.update(0)
+
+      assert not check_selfdrive_timeout_alert(sm)
+
+    for _ in range(SELFDRIVE_STATE_TIMEOUT * 110):
+      sm.update(0)
+      time.sleep(0.01)
+
+    assert check_selfdrive_timeout_alert(sm)
+
+  # TODO: add test with micd for checking that soundd actually outputs sounds
+

selfdrive/ui/tests/test_translations.py

@@ -0,0 +1,106 @@
+import json
+import re
+import string
+from pathlib import Path
+
+import pytest
+
+from openpilot.selfdrive.ui.translations.potools import parse_po
+from openpilot.system.ui.lib.multilang import LANGUAGES_FILE, TRANSLATIONS_DIR
+
+PERCENT_PLACEHOLDER_RE = re.compile(r"%(?:n|\d+)")
+BAD_ENTITY_RE = re.compile(r'@(\w+);')
+LINE_NUMBER_REF_RE = re.compile(r'^#:\s+.+:\d+(?:\s|$)')
+FORMATTER = string.Formatter()
+PO_DIR = Path(str(TRANSLATIONS_DIR))
+
+with LANGUAGES_FILE.open(encoding='utf-8') as f:
+  TRANSLATION_LANGUAGES = json.load(f)
+
+
+def extract_placeholders(text: str) -> list[str]:
+  placeholders = PERCENT_PLACEHOLDER_RE.findall(text)
+
+  try:
+    parsed = list(FORMATTER.parse(text))
+  except ValueError as e:
+    raise AssertionError(f"invalid brace formatting in {text!r}: {e}") from e
+
+  for _, field_name, format_spec, conversion in parsed:
+    if field_name is None:
+      continue
+
+    token = "{"
+    token += field_name
+    if conversion:
+      token += f"!{conversion}"
+    if format_spec:
+      token += f":{format_spec}"
+    token += "}"
+    placeholders.append(token)
+
+  return sorted(placeholders)
+
+
+def load_po_text(po_path: Path) -> str:
+  return po_path.read_text(encoding='utf-8')
+
+
+@pytest.mark.parametrize("language_code", sorted(TRANSLATION_LANGUAGES.values()))
+def test_translation_file_exists(language_code: str):
+  po_path = PO_DIR / f"app_{language_code}.po"
+  assert po_path.exists(), f"missing translation file: {po_path}"
+
+
+@pytest.mark.parametrize("po_path", sorted(PO_DIR.glob("app_*.po")), ids=lambda p: p.name)
+def test_translation_placeholders_are_preserved(po_path: Path):
+  _, entries = parse_po(po_path)
+  language = po_path.stem.removeprefix("app_")
+
+  for entry in entries:
+    source_placeholders = extract_placeholders(entry.msgid)
+
+    if entry.is_plural:
+      plural_placeholders = extract_placeholders(entry.msgid_plural)
+      message = (
+        f"{language}: source plural placeholders do not match singular for "
+        + f"{entry.msgid!r}: {source_placeholders} vs {plural_placeholders}"
+      )
+      assert plural_placeholders == source_placeholders, message
+
+      for idx, msgstr in sorted(entry.msgstr_plural.items()):
+        if not msgstr:
+          continue
+
+        translated_placeholders = extract_placeholders(msgstr)
+        message = (
+          f"{language}: plural form {idx} changes placeholders for {entry.msgid!r}: "
+          + f"expected {source_placeholders}, got {translated_placeholders}"
+        )
+        assert translated_placeholders == source_placeholders, message
+    else:
+      if not entry.msgstr:
+        continue
+
+      translated_placeholders = extract_placeholders(entry.msgstr)
+      message = (
+        f"{language}: translation changes placeholders for {entry.msgid!r}: "
+        + f"expected {source_placeholders}, got {translated_placeholders}"
+      )
+      assert translated_placeholders == source_placeholders, message
+
+
+@pytest.mark.parametrize("po_path", sorted(PO_DIR.glob("app_*.po")), ids=lambda p: p.name)
+def test_translation_refs_do_not_include_line_numbers(po_path: Path):
+  for line in load_po_text(po_path).splitlines():
+    assert not LINE_NUMBER_REF_RE.match(line), (
+      f"{po_path.name}: line-number source reference found: {line}"
+    )
+
+
+@pytest.mark.parametrize("po_path", sorted(PO_DIR.glob("app_*.po")), ids=lambda p: p.name)
+def test_translation_entities_are_valid(po_path: Path):
+  matches = BAD_ENTITY_RE.findall(load_po_text(po_path))
+  assert not matches, (
+    f"{po_path.name}: found '@...;' entity typo(s): {', '.join(sorted(set(matches)))}"
+  )