App Verification Suite

App Verification Suite is a Python library for recording OAK app inputs into datasets, replaying those datasets back into normal DepthAI pipelines, and optionally recording or verifying structured app outputs.

What the library does

• switches an app between live and replay modes
• records camera and IMU inputs into reusable datasets
• optionally prepends pre-trigger camera, IMU, calibration, and structured-output data to recordings
• replays recorded datasets through the same DepthAI pipelines
• captures calibration needed for replay
• optionally records structured expected outputs
• optionally verifies replayed structured outputs
• stores datasets locally or in S3

Public API at a glance

Top-level exports from app_verification_suite include:

• AppVerificationRuntime
• EncoderConfig
• DatasetManager
• LocalFilesystemStorageBackend
• S3StorageBackend
• structured-output models such as AppData, AppDataField, and AppVerificationResult

The main entrypoint is AppVerificationRuntime.

Core concepts

• AppVerificationRuntime is the main entrypoint and runs in either live mode or replay mode.
• You create cameras and IMUs through the runtime so the same application pipeline can work in both modes.
• In live mode, you can start and stop recording while the application is already running. This makes it practical to capture datasets during normal runtime, for example when certain events occur such as low-confidence detections.
• Live runtimes can keep a pre-trigger buffer. When an event occurs, start_recording(..., pre_trigger_sec=N) prepends up to N seconds of already-buffered camera, IMU, calibration, and structured-output data to the new dataset.
• In replay mode, the runtime serves recorded inputs back into the pipeline so those datasets can be analyzed and replayed offline to improve the application.
• Replay is strict: camera and IMU requests must match what was recorded.
• stop_recording() ends the active recording session, writes dataset artifacts locally, then starts backend finalization in the background. The returned Future lets you wait for upload/publish completion and observe failures.
• Replay uses the calibration stored in the dataset.

Installation

Install from PyPI:

pip install app-verification-suite

Quick start

Record a dataset

import depthai as dai

from app_verification_suite import (
    AppVerificationRuntime,
    LocalFilesystemStorageBackend,
)

storage = LocalFilesystemStorageBackend("datasets")
runtime = AppVerificationRuntime.live(
    storage_backend=storage,
    application_id="my-app",
    application_version="0.1.0",
    pre_trigger_buffer_sec=5,
)

with dai.Pipeline() as pipeline:
    camera = runtime.create_camera(
        pipeline=pipeline,
        socket=dai.CameraBoardSocket.CAM_A,
        sensor_fps=30.0,
    )
    preview = camera.requestOutput(size=(640, 480), fps=30.0)
    # Link preview into the rest of your pipeline.

    pipeline.build()
    pipeline.start()

    runtime.start_recording(dataset_id="demo", pre_trigger_sec=2)
    # Run the application here.
    upload_future = runtime.stop_recording(dataset_id="demo")
    upload_future.result()

Configure recording encoding

Recordings use lossless MJPEG at quality 100 by default. To reduce file size at the cost of image quality, pass an EncoderConfig when creating the live runtime. These settings apply to all recorded camera outputs.

from app_verification_suite import AppVerificationRuntime, EncoderConfig

runtime = AppVerificationRuntime.live(
    storage_backend=storage,
    encoder_config=EncoderConfig(lossless=False, quality=75),
)

MJPEG quality accepts integers from 0 to 100; lower values reduce quality. Set lossless=False when choosing a quality below 100.

For H264 or H265 recording, select "h264_baseline", "h264_main", "h264_high" or "h265_main". For example:

runtime = AppVerificationRuntime.live(
    storage_backend=storage,
    encoder_config=EncoderConfig(
        profile="h264_main",
        bitrate=32_000_000,
        rate_control="cbr",
        keyframe_frequency=30,
    ),
)

For H264/H265:

• bitrate is in bits per second; omit it or use 0 for automatic bitrate.
• rate_control accepts "cbr" or "vbr". Use "cbr" with an explicit nonzero bitrate.
• keyframe_frequency sets the interval between keyframes in frames. Shorter intervals can reduce the wait before footage starts and preserve more pre-trigger footage.
• Omitted settings use encoder defaults. Lossless encoding, MJPEG quality and B-frames are not supported.

H264/H265 footage starts at an independent keyframe, so recording may begin after start_recording() returns and pre-trigger footage may be shorter than requested. Stopping before the first usable frame leaves a dataset with no footage to replay.

Replay a dataset

import depthai as dai

from app_verification_suite import (
    AppVerificationRuntime,
    LocalFilesystemStorageBackend,
)

storage = LocalFilesystemStorageBackend("datasets")
runtime = AppVerificationRuntime.replay(
    dataset_id="demo",
    storage_backend=storage,
    loop=False,
)

with dai.Pipeline() as pipeline:
    camera = runtime.create_camera(
        pipeline=pipeline,
        socket=dai.CameraBoardSocket.CAM_A,
        sensor_fps=30.0,
    )
    preview = camera.requestOutput(size=(640, 480), fps=30.0)
    # Link preview into the rest of your pipeline.

Replay requests must match the recorded manifest, including socket, sensor settings, and requestOutput(...) parameters. In replay mode, requestOutput(...) requires fps.

Camera request matching preserves the original socket, including AUTO. Replayed frames carry the actual source camera's instance number, recorded separately in the required source_instance_num frame metadata field. Replay rejects frame metadata with missing, null or invalid source identity. Datasets recorded without this field are no longer supported; replay does not infer identity from the request or frame geometry.

Example scripts

The repository includes several runnable examples:

Record locally

python examples/dataset_recording.py --dataset-id demo

• uses LocalFilesystemStorageBackend("datasets")
• writes the dataset under datasets/demo/
• records a simple structured app metric from the CAM_A preview into dataset-root expected.jsonl
• opens a visualizer at http://127.0.0.1:8082
• stop by pressing q

Replay locally

python examples/dataset_replay.py --dataset-id demo

• reads the dataset from datasets/demo/
• replays the same CAM_A metric through AppVerifier, prints per-frame pass/fail, and summarizes results when expected.jsonl is present
• warns and falls back to plain replay/visualization when expected.jsonl is missing
• opens the same visualizer view for replayed streams

Manage datasets

List finalized local datasets:

python examples/dataset_list.py

Rename a finalized local dataset with DatasetManager:

python examples/dataset_rename.py demo renamed-demo

Delete a finalized local dataset with DatasetManager:

python examples/dataset_delete.py demo

DatasetManager API

Use DatasetManager directly when you want to administer finalized datasets from Python:

from app_verification_suite import DatasetManager, LocalFilesystemStorageBackend

storage = LocalFilesystemStorageBackend("datasets")
datasets = DatasetManager(storage)

ids = datasets.list()
manifest = datasets.get_manifest("demo")
size_bytes = datasets.get_size_bytes("demo")

datasets.rename("demo", "renamed-demo")
datasets.delete("renamed-demo")

For direct access to a complete local dataset, use the separate availability query and whole-dataset materialization methods:

local_path = datasets.get_local_dataset_path("demo")  # Path or None
dataset_path = datasets.materialize_dataset("demo")  # Path

get_local_dataset_path() checks local availability without downloads or copying. It may list remote storage and returns None for absent or incomplete local data. materialize_dataset() uses the backend's configured local location and raises FileNotFoundError for missing required files. Neither method takes a destination. With LocalFilesystemStorageBackend, both return the original dataset directory, including extra artifacts outside manifest references, without copying content. The manager uses the backend's list_files(dataset_id) inventory and ensure_local_files(dataset_id, relative_paths) to fetch all stored files, including extra artifacts. Listing returns sorted dataset-relative paths, recursively, without interpreting the manifest; empty directories are ignored. With S3StorageBackend, missing files are downloaded into local_cache_root. Existing cached bytes are reused, and unrelated local files are retained. Failed transfers may leave successful downloads cached for retry. Repeated materialization may contact storage; it does not refresh existing cached bytes or promise a transactional snapshot of remote contents.

The manager checks that every listed file exists locally and required recording artifacts appear in the storage inventory. A stale cached file cannot conceal a required artifact missing from storage. Manifest-only and selected-range replay caches remain incomplete; complete existing caches are recognized after restarting the backend without completeness bookkeeping. Invalid recordings remain cached but are rejected by the manager. Backends without optional list_files support return None for the query and raise NotImplementedError for materialization.

Both operations validate the manifest and required file availability, including the calibration event index for datasets with recorded inputs and its referenced calibration files. Manifest and read errors propagate. These checks do not decode media or certify full media integrity. Availability is checked again on each call; list() still discovers datasets by manifest presence, independently of completeness.

You can read the returned directory directly into an archive. The path describes availability when checked and grants no lease or deletion protection; callers must protect the source while using it.

Use the S3 example path

The dataset example scripts also support S3 storage:

python examples/dataset_recording.py --storage-backend s3 --dataset-id demo
python examples/dataset_replay.py --storage-backend s3 --dataset-id demo
python examples/dataset_list.py --storage-backend s3
python examples/dataset_rename.py --storage-backend s3 demo renamed-demo
python examples/dataset_delete.py --storage-backend s3 demo

They read S3 configuration from:

• S3_BUCKET_NAME
• S3_ACCESS_KEY_ID
• S3_ENDPOINT
• S3_REGION
• S3_SECRET_ACCESS_KEY

Pre-trigger recording

Configure the maximum retained history on the live runtime with pre_trigger_buffer_sec, then choose how much of that history each recording should include with pre_trigger_sec:

runtime = AppVerificationRuntime.live(
    storage_backend=storage,
    pre_trigger_buffer_sec=5,
)

# Later, when your app decides an event is worth saving:
runtime.start_recording(dataset_id="event-001", pre_trigger_sec=2)

pre_trigger_sec must be non-negative and no larger than pre_trigger_buffer_sec. The runtime applies the same trigger point across video, IMU, calibration events, and AppDataRecorder output.

IMU usage

Create IMU inputs through the runtime so recording and replay stay symmetric:

imu = runtime.create_imu(
    pipeline=pipeline,
    enabled_sensors=[
        (dai.IMUSensor.ACCELEROMETER_RAW, 400),
        (dai.IMUSensor.GYROSCOPE_RAW, 400),
    ],
    batch_report_threshold=5,
    max_batch_reports=10,
)

imu.out.link(...)

Structured output recording and verification

Use the same AppData / AppDataField message shape in both modes. Your app should emit AppData on a normal pipeline output, then the runtime attaches either a recorder or a verifier to that output stream.

AppDataField uses plain equality by default, and its verifier is ignored during recording. Field values can be any JSON-serializable value, including nested structures such as dict or list.

Record expected app output

In live mode, attach a recorder to an output that emits AppData:

metric_producer = pipeline.create(AppDataMetricProducer).build(cam_a_out)
runtime.create_app_data_recorder(pipeline, metric_producer.out)

While recording is active, the recorder writes dataset-root expected.jsonl.

Verify replayed app output

In replay mode, attach a verifier to the same kind of AppData output:

metric_producer = pipeline.create(AppDataMetricProducer).build(cam_a_out)
verifier = runtime.create_verifier(pipeline, metric_producer.out)
verifier.out.link(...)

The verifier loads expected.jsonl, matches rows by sequence_num, and emits AppVerificationResult messages.

Emitting AppData

The producer output should send AppData messages like this:

app_data = AppData()
app_data.setSequenceNum(frame.getSequenceNum())
app_data.fields = [
    AppDataField(name="label", value=label),
    AppDataField(name="metadata", value={"source": "cam_a", "ok": True}),
]
self.out.send(app_data)

Custom verifiers

For custom comparisons during replay, set a custom verifier on the field emitted by your replay-side AppData producer:

AppDataField(
    name="score",
    value=score,
    verifier=lambda actual, expected: abs(actual - expected) <= 0.05,
)

Dataset layout

A finalized dataset currently looks like this:

datasets/<dataset_id>/
  manifest.json
  expected.jsonl                # optional
  calibrations/
    calibration-events.jsonl
    000000.json
    000001.json                 # optional later calibration updates
    ...
  inputs/
    <video_input_id>/
      000000.mp4
      000000.frames.jsonl
      000001.mp4
      000001.frames.jsonl
      ...
    <imu_input_id>/
      000000.jsonl

manifest.json currently stores separate top-level video_inputs and imu_inputs lists.

Current scope and limitations

• DepthAI v3 only
• recording currently requires at least one video output before start_recording(...)
• replay request validation is strict by design