detections/e2e/tests/test_performance.py

import json
import os

import pytest


def _percentile_ms(sorted_ms, p):
    n = len(sorted_ms)
    if n == 0:
        return 0.0
    if n == 1:
        return float(sorted_ms[0])
    k = (n - 1) * (p / 100.0)
    lo = int(k)
    hi = min(lo + 1, n - 1)
    w = k - lo
    return sorted_ms[lo] * (1 - w) + sorted_ms[hi] * w


@pytest.mark.timeout(240)
def test_nft_perf_01_single_image_latency_p95(
    warm_engine, image_detect, image_small
):
    threshold_ms = float(os.environ.get("E2E_SINGLE_IMAGE_P95_MS", "15000"))
    times_ms = []
    for _ in range(10):
        _, elapsed_ms = image_detect(image_small, "img.jpg", timeout=20)
        times_ms.append(elapsed_ms)

    sorted_ms = sorted(times_ms)
    p50 = _percentile_ms(sorted_ms, 50)
    p95 = _percentile_ms(sorted_ms, 95)
    p99 = _percentile_ms(sorted_ms, 99)
    print(
        "nft_perf_01_csv,run_ms,"
        + ",".join(f"{x:.2f}" for x in sorted_ms)
        + f",p50,{p50:.2f},p95,{p95:.2f},p99,{p99:.2f}"
    )
    assert p95 < threshold_ms


@pytest.mark.timeout(120)
def test_nft_perf_03_tiling_overhead_large_image(
    warm_engine, image_detect, image_small, image_large
):
    _, small_ms = image_detect(image_small, "small.jpg", timeout=20)
    _, large_ms = image_detect(
        image_large, "large.jpg",
        config=json.dumps(
            {
                "camera_config": {
                    "focal_length": 24,
                    "sensor_width": 23.5,
                    "current_zoom": 1,
                    "current_angle": 90,
                    "current_height": 400,
                }
            }
        ),
        timeout=20,
    )
    assert large_ms < 30_000.0
    print(
        f"nft_perf_03_csv,baseline_small_ms,{small_ms:.2f},large_ms,{large_ms:.2f}"
    )
    assert large_ms > small_ms - 500.0