app-verification-suite 0.0.3

No project description provided

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
• 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
• 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.
• 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 and finalizes the dataset locally before returning. If the storage backend performs a follow-up upload, the returned Future lets you wait for it.
• 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",
)

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")
    # Run the application here.
    upload_future = runtime.stop_recording()
    upload_future.result()

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.

Example scripts

The repository includes two 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

Use the S3 example path

Both example scripts also support:

python examples/dataset_recording.py --storage-backend s3 --dataset-id demo
python examples/dataset_replay.py --storage-backend s3 --dataset-id demo

They read S3 configuration from:

• S3_BUCKET_NAME
• S3_ACCESS_KEY_ID
• S3_ENDPOINT
• S3_REGION
• S3_SECRET_ACCESS_KEY

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