dragonpilot/selfdrive/test/process_replay

Process replay

Process replay is a regression test designed to identify any changes in the output of a process. This test replays a segment through individual processes and compares the output to a known good replay. Each make is represented in the test with a segment.

If the test fails, make sure that you didn't unintentionally change anything. If there are intentional changes, the reference logs will be updated.

Use test_processes.py to run the test locally. Use FILEREADER_CACHE='1' test_processes.py to cache log files.

Currently the following processes are tested:

• controlsd
• radard
• plannerd
• calibrationd
• dmonitoringd
• locationd
• paramsd
• ubloxd
• torqued

Usage

Usage: test_processes.py [-h] [--whitelist-procs PROCS] [--whitelist-cars CARS] [--blacklist-procs PROCS]
                         [--blacklist-cars CARS] [--ignore-fields FIELDS] [--ignore-msgs MSGS] [--update-refs] [--upload-only]
Regression test to identify changes in a process's output
optional arguments:
  -h, --help            show this help message and exit
  --whitelist-procs PROCS               Whitelist given processes from the test (e.g. controlsd)
  --whitelist-cars WHITELIST_CARS       Whitelist given cars from the test (e.g. HONDA)
  --blacklist-procs BLACKLIST_PROCS     Blacklist given processes from the test (e.g. controlsd)
  --blacklist-cars BLACKLIST_CARS       Blacklist given cars from the test (e.g. HONDA)
  --ignore-fields IGNORE_FIELDS         Extra fields or msgs to ignore (e.g. carState.events)
  --ignore-msgs IGNORE_MSGS             Msgs to ignore (e.g. onroadEvents)
  --update-refs                         Updates reference logs using current commit
  --upload-only                         Skips testing processes and uploads logs from previous test run

Forks

openpilot forks can use this test with their own reference logs, by default test_proccess.py saves logs locally.

To generate new logs:

./test_processes.py

Then, check in the new logs using git-lfs. Make sure to also update the ref_commit file to the current commit.

API

Process replay test suite exposes programmatic APIs for simultaneously running processes or groups of processes on provided logs.

def replay_process_with_name(name: Union[str, Iterable[str]], lr: LogIterable, *args, **kwargs) -> List[capnp._DynamicStructReader]:

def replay_process(
  cfg: Union[ProcessConfig, Iterable[ProcessConfig]], lr: LogIterable, frs: Optional[Dict[str, Any]] = None, 
  fingerprint: Optional[str] = None, return_all_logs: bool = False, custom_params: Optional[Dict[str, Any]] = None, disable_progress: bool = False
) -> List[capnp._DynamicStructReader]:

Example usage:

from openpilot.selfdrive.test.process_replay import replay_process_with_name
from openpilot.tools.lib.logreader import LogReader

lr = LogReader(...)

# provide a name of the process to replay
output_logs = replay_process_with_name('locationd', lr)

# or list of names
output_logs = replay_process_with_name(['ubloxd', 'locationd'], lr)

Supported processes:

• controlsd
• radard
• plannerd
• calibrationd
• dmonitoringd
• locationd
• paramsd
• ubloxd
• torqued
• modeld
• dmonitoringmodeld

Certain processes may require an initial state, which is usually supplied within Params and persisting from segment to segment (e.g CalibrationParams, LiveParameters). The custom_params is dictionary used to prepopulate Params with arbitrary values. The get_custom_params_from_lr helper is provided to fetch meaningful values from log files.

from openpilot.selfdrive.test.process_replay import get_custom_params_from_lr

previous_segment_lr = LogReader(...)
current_segment_lr = LogReader(...)

custom_params = get_custom_params_from_lr(previous_segment_lr, 'last')

output_logs = replay_process_with_name('calibrationd', lr, custom_params=custom_params)

Replaying processes that use VisionIPC (e.g. modeld, dmonitoringmodeld) require additional frs dictionary with camera states as keys and FrameReader objects as values.

from openpilot.tools.lib.framereader import FrameReader

frs = {
  'roadCameraState': FrameReader(...),
  'wideRoadCameraState': FrameReader(...),
  'driverCameraState': FrameReader(...),
}

output_logs = replay_process_with_name(['modeld', 'dmonitoringmodeld'], lr, frs=frs)

To capture stdout/stderr of the replayed process, captured_output_store can be provided.

output_store = dict()
# pass dictionary by reference, it will be filled with standard outputs - even if process replay fails
output_logs = replay_process_with_name(['radard', 'plannerd'], lr, captured_output_store=output_store)

# entries with captured output in format { 'out': '...', 'err': '...' } will be added to provided dictionary for each replayed process
print(output_store['radard']['out']) # radard stdout
print(output_store['radard']['err']) # radard stderr