Metadata-Version: 2.4
Name: app-verification-suite
Version: 0.0.5
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~=3.6.1
Requires-Dist: av==17.0.1
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
- 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:

```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",
)

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

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

## Example scripts

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

### Use the S3 example path

Both example scripts also support:

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

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