Metadata-Version: 2.4
Name: app-verification-suite
Version: 2.0.0
Classifier: Programming Language :: Python :: 3
Classifier: Operating System :: OS Independent
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: pydantic
Requires-Dist: depthai<4,>=3.6.1
Requires-Dist: av<18,>=12.0.0
Requires-Dist: boto3
Requires-Dist: opencv-python-headless

# 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:

```bash
pip install app-verification-suite
```

## Quick start

### Record a dataset

```python
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.

```python
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:

```python
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

```python
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

```bash
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

```bash
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:

```bash
python examples/dataset_list.py
```

Rename a finalized local dataset with `DatasetManager`:

```bash
python examples/dataset_rename.py demo renamed-demo
```

Delete a finalized local dataset with `DatasetManager`:

```bash
python examples/dataset_delete.py demo
```

### DatasetManager API

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

```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:

```python
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:

```bash
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`:

```python
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:

```python
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`:

```python
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:

```python
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:

```python
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:

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

## Dataset layout

A finalized dataset currently looks like this:

```text
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
