feat(performance-tools): Added benchnarking scripts for digital signage

benchmark-scripts/benchmark_signage_aig.py

@@ -0,0 +1,202 @@
+#
+# Apache v2 license
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+#
+
+
+"""Benchmark ASe search and AIG image generation through the HTTP API."""
+
+import argparse
+import base64
+import csv
+import json
+import os
+import subprocess
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from statistics import mean
+
+import requests
+
+
+SEARCH_STAGES = ("query_build", "vector_search", "ranking_retrieval")
+GENERATION_STAGES = ("model_load", "inference", "overlay", "jpeg_encode", "model_unload", "model_cleanup")
+DEFAULT_BASE_URL = "https://localhost:5000/aig-api"
+
+
+def parse_server_timing(header):
+    timings = {}
+    for entry in header.split(","):
+        name, _, timing = entry.strip().partition(";dur=")
+        if timing:
+            try:
+                timings[f"{name}_ms"] = float(timing)
+            except ValueError:
+                pass
+    return timings
+
+
+def request_ad(session, url, payload, kind, call, timeout, cold_start=False):
+    start = time.perf_counter()
+    response = session.post(url, json=payload, headers={"X-Benchmark-Timing": "true"}, timeout=timeout)
+    elapsed_ms = (time.perf_counter() - start) * 1000
+    response.raise_for_status()
+    if kind == "predefined":
+        ads = response.json()
+        if not isinstance(ads, list):
+            raise ValueError("ASe did not return a list of ads")
+        image_size = sum(len(base64.b64decode(ad["imgb64"])) for ad in ads if ad.get("imgb64"))
+        result_count = len(ads)
+        stages = SEARCH_STAGES
+    else:
+        if not response.headers.get("Content-Type", "").startswith("image/jpeg"):
+            raise ValueError("AIG did not return a JPEG image")
+        image_size = len(response.content)
+        result_count = 1
+        stages = GENERATION_STAGES
+    timings = parse_server_timing(response.headers.get("Server-Timing", ""))
+    sla_ms = 5000 if kind == "predefined" else (60000 if call == 1 and cold_start else 15000)
+    return {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "kind": kind,
+        "call": call,
+        "http_elapsed_ms": round(elapsed_ms, 3),
+        "sla_ms": sla_ms,
+        "sla_met": elapsed_ms < sla_ms,
+        "response_size_kb": round(len(response.content) / 1024, 3),
+        "image_size_kb": round(image_size / 1024, 3),
+        "result_count": result_count,
+        **{f"{stage}_ms": timings.get(f"{stage}_ms") for stage in stages},
+    }
+
+
+def summarize(rows, device, capture_elapsed_ms):
+    """Average each call type so predefined (ASe) and dynamic (AIG) runs compare directly."""
+    def average(values):
+        values = [value for value in values if isinstance(value, (int, float))]
+        return round(mean(values), 3) if values else None
+
+    summary = {"timestamp": datetime.now(timezone.utc).isoformat(), "device": device,
+               "capture_elapsed_ms": capture_elapsed_ms}
+    for kind, stages in (("predefined", SEARCH_STAGES), ("dynamic", GENERATION_STAGES)):
+        subset = [row for row in rows if row["kind"] == kind]
+        summary[f"{kind}_calls"] = len(subset)
+        summary[f"{kind}_sla_met"] = sum(bool(row["sla_met"]) for row in subset)
+        for field in ("http_elapsed_ms", "response_size_kb", "image_size_kb",
+                      *(f"{stage}_ms" for stage in stages)):
+            summary[f"avg_{kind}_{field}"] = average([row.get(field) for row in subset])
+    # The first dynamic call carries model load, so warm calls are the comparable number.
+    dynamic = [row for row in rows if row["kind"] == "dynamic"]
+    first = [row["http_elapsed_ms"] for row in dynamic if row["call"] == 1]
+    summary["dynamic_first_call_http_elapsed_ms"] = first[0] if first else None
+    summary["avg_dynamic_warm_http_elapsed_ms"] = average(
+        [row["http_elapsed_ms"] for row in dynamic if row["call"] > 1])
+    return summary
+
+
+def run(args):
+    base_url = args.base_url.rstrip("/")
+    local_self_signed = (base_url == DEFAULT_BASE_URL
+                         and not os.environ.get("REQUESTS_CA_BUNDLE")
+                         and not os.environ.get("CURL_CA_BUNDLE"))
+    session = requests.Session()
+    session.verify = not (args.insecure or local_self_signed)
+    if not session.verify:
+        requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)
+        print("TLS verification disabled for local AIG proxy" if local_self_signed and not args.insecure
+              else "TLS verification disabled by --insecure", file=sys.stderr)
+    if args.cold_start:
+        import subprocess
+        # Restart by container name, not `compose restart`, so the script does not depend on the caller's Compose project.
+        subprocess.run(["docker", "restart", "aig-server"], check=True)
+    deadline = time.monotonic() + min(args.timeout, 120)
+    while True:
+        try:
+            status = session.get(f"{base_url}/aig/hstatus/1", timeout=3)
+            status.raise_for_status()
+            break
+        except requests.exceptions.SSLError as error:
+            raise RuntimeError(
+                "TLS certificate verification failed; use --insecure for the local self-signed "
+                "certificate or set REQUESTS_CA_BUNDLE to a trusted CA file"
+            ) from error
+        except requests.RequestException:
+            if time.monotonic() >= deadline:
+                raise RuntimeError("AIG did not become ready")
+            time.sleep(1)
+    search_payload = {"query": args.query, "n_results": 1, "use_default_ad_onempty": False}
+    generation_payload = {
+        "description": args.description,
+        "device": args.device,
+        "promo_details": {"promo_text": args.slogan},
+        "logo_details": {"logo_percentage": 15},
+    }
+    rows = []
+    benchmark_start = time.perf_counter()
+    for call in range(1, args.iterations + 1):
+        scheduled_at = benchmark_start + (call - 1) * args.duration / args.iterations
+        if scheduled_at > time.perf_counter():
+            time.sleep(scheduled_at - time.perf_counter())
+        start = time.perf_counter()
+        search = request_ad(session, f"{base_url}/ase/predef/query/ad", search_payload,
+                            "predefined", call, args.timeout)
+        if search["result_count"] == 0:
+            raise RuntimeError(f"No predefined image matched {args.query!r}; choose an indexed query")
+        rows.append(search)
+        generation = request_ad(session, f"{base_url}/aig/minf/", generation_payload,
+                                "dynamic", call, args.timeout, args.cold_start)
+        rows.append(generation)
+        for row in (search, generation):
+            row["workflow_elapsed_ms"] = round((time.perf_counter() - start) * 1000, 3)
+    if benchmark_start + args.duration > time.perf_counter():
+        time.sleep(benchmark_start + args.duration - time.perf_counter())
+    capture_elapsed_ms = round((time.perf_counter() - benchmark_start) * 1000, 3)
+    for row in rows:
+        row["device"] = args.device
+        row["capture_elapsed_ms"] = capture_elapsed_ms
+
+    args.output.mkdir(parents=True, exist_ok=True)
+    name = f"aig_benchmark_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+    fields = ("timestamp", "kind", "call", "device", "http_elapsed_ms", "sla_ms", "sla_met",
+              "response_size_kb", "image_size_kb", "result_count", "workflow_elapsed_ms",
+              "capture_elapsed_ms",
+              *(f"{stage}_ms" for stage in SEARCH_STAGES + GENERATION_STAGES))
+    (args.output / f"{name}.json").write_text(json.dumps(rows, indent=2) + "\n")
+    with (args.output / f"{name}.csv").open("w", newline="") as destination:
+        writer = csv.DictWriter(destination, fieldnames=fields)
+        writer.writeheader()
+        writer.writerows(rows)
+    summary = summarize(rows, args.device, capture_elapsed_ms)
+    (args.output / f"{name}_summary.json").write_text(json.dumps(summary, indent=2) + "\n")
+    with (args.output / f"{name}_summary.csv").open("w", newline="") as destination:
+        writer = csv.DictWriter(destination, fieldnames=summary.keys())
+        writer.writeheader()
+        writer.writerow(summary)
+    print(json.dumps(summary, indent=2))
+    print(f"Saved {args.output / name}.json, .csv and _summary.json/.csv")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base-url", default=DEFAULT_BASE_URL)
+    parser.add_argument("--insecure", action="store_true", help="disable TLS verification for a custom HTTPS endpoint")
+    parser.add_argument("--query", default="broccoli and carrot")
+    parser.add_argument("--description", default="A studio photograph of a freshly brewed cup of coffee on a clean table")
+    parser.add_argument("--slogan", default="Fresh coffee")
+    parser.add_argument("--device", choices=("CPU", "GPU"), default="GPU")
+    parser.add_argument("--iterations", type=int, default=10)
+    parser.add_argument("--duration", type=int, default=300, help="minimum capture duration in seconds")
+    parser.add_argument("--timeout", type=int, default=400)
+    parser.add_argument("--cold-start", action="store_true", help="restart only aig-server before measuring")
+    parser.add_argument("--output", type=Path, default=Path("results"))
+    options = parser.parse_args()
+    if options.iterations < 1 or options.timeout < 1 or options.duration < 0:
+        parser.error("iterations and timeout must be positive; duration cannot be negative")
+    try:
+        run(options)
+    except (requests.RequestException, subprocess.CalledProcessError, ValueError, OSError, RuntimeError) as error:
+        print(f"AIG benchmark failed: {error}", file=sys.stderr)
+        sys.exit(1)

benchmark-scripts/benchmark_signage_consolidate_e2e.py

@@ -0,0 +1,164 @@
+#
+# Apache v2 license
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+#
+
+
+"""Consolidate PID, Web UI, AIG, Qdrant and host-metrics results into an end-to-end stage table."""
+
+import argparse
+import csv
+import json
+import math
+import statistics
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+FIELDS = ("stage", "source", "samples", "avg_ms", "p50_ms", "p95_ms", "max_ms", "value", "note", "file")
+
+
+def latest(directory, pattern):
+    files = sorted(directory.glob(pattern), key=lambda path: path.stat().st_mtime)
+    return files[-1] if files else None
+
+
+def load(path):
+    return json.loads(path.read_text()) if path else None
+
+
+def stats(values):
+    values = sorted(float(value) for value in values if isinstance(value, (int, float)))
+    if not values:
+        return {"samples": 0}
+    return {"samples": len(values), "avg_ms": round(statistics.mean(values), 3),
+            "p50_ms": round(statistics.median(values), 3),
+            "p95_ms": round(values[math.ceil(0.95 * len(values)) - 1], 3),
+            "max_ms": round(values[-1], 3)}
+
+
+def row(stage, source, path, values=None, value=None, note=""):
+    entry = {"stage": stage, "source": source, "note": note, "file": path.name if path else None}
+    entry.update(stats(values or []) if value is None else {"value": value})
+    return entry
+
+
+def consolidate(results):
+    rows = []
+    pid_path = latest(results, "pid_benchmark_*[0-9].json")
+    pid = load(pid_path)
+    pid_frames_path = pid_path.with_name(pid_path.stem + "_frames.json") if pid_path else None
+    pid_frames = load(pid_frames_path) if pid_frames_path and pid_frames_path.exists() else []
+    web_path = latest(results, "web_ui_benchmark_*[0-9].json")
+    web_frames = load(web_path.with_name(web_path.stem + "_frames.json")) if web_path else []
+    ads_path = web_path.with_name(web_path.stem + "_ads.json") if web_path else None
+    ads = load(ads_path) if ads_path and ads_path.exists() else []
+    aig_path = latest(results, "aig_benchmark_*[0-9].json")
+    aig = load(aig_path) or []
+    qdrant_path = latest(results, "qdrant_benchmark_*[0-9].json")
+    qdrant = load(qdrant_path) or []
+
+    def ad_values(field, path=None):
+        return [ad[field] for ad in ads if path is None or ad.get("path") == path]
+
+    if pid:
+        rows.append(row("1. Video frame -> PID inference", "PID", pid_path,
+                        [frame.get("pipeline_latency_reported") for frame in pid_frames],
+                        note="DLSPS avg_pipeline_latency; empty when DLSPS does not report it"))
+        rows.append(row("1. PID throughput (FPS)", "PID", pid_path, value=pid.get("fps_dlstreamer"),
+                        note="Throughput, not latency"))
+        rows.append(row("2. PID -> MQTT -> benchmark subscriber", "PID", pid_path,
+                        [frame.get("mqtt_arrival_delay_ms") for frame in pid_frames]))
+    if web_path:
+        rows.append(row("2-3. PID frame timestamp -> Web UI receive", "Web UI", web_path,
+                        [frame.get("mqtt_to_web_ui_ms") for frame in web_frames],
+                        note="Publish and broker delivery combined; DLSPS does not expose publish time"))
+        rows.append(row("4. Web UI filtering (callback)", "Web UI", web_path,
+                        [frame.get("callback_total_ms") for frame in web_frames]))
+        rows.append(row("4b. Queue wait before ad decision", "Web UI", web_path,
+                [frame.get("queue_wait_ms") for frame in web_frames],
+                note="Candidate enqueue to generator dequeue; excludes cooldown suppression"))
+        rows.append(row("4c. Cooldown remaining at suppression", "Web UI", web_path,
+                [frame.get("cooldown_remaining_ms") for frame in web_frames],
+                note="Remaining cooldown when a candidate is dropped; it is not deferred"))
+        rows.append(row("4d. Cooldown-suppressed candidates", "Web UI", web_path,
+                value=sum(frame.get("cooldown_suppressed") is True for frame in web_frames),
+                note="Candidates dropped during cooldown instead of queued for later"))
+        rows.append(row("5. Product selection", "Web UI", web_path,
+                        [frame.get("selector_total_ms") for frame in web_frames]))
+        rows.append(row("6. ASe query (live)", "Web UI", web_path, ad_values("ase_query_ms")))
+        rows.append(row("7. AIG generation on ASe miss (live)", "Web UI", web_path,
+                        ad_values("aig_generation_ms", "dynamic")))
+        rows.append(row("8a. Ad ready -> served to browser", "Web UI", web_path, ad_values("ready_to_served_ms"),
+                        note="Includes the browser's 2 s polling interval"))
+        rows.append(row("8b. Browser fetch", "Browser", web_path, ad_values("browser_fetch_ms")))
+        rows.append(row("9. Browser render", "Browser", web_path, ad_values("browser_render_ms"),
+                        note="Requires an open Web UI page; paint-opportunity estimate, not measured display time"))
+        rows.append(row("E2E frame -> ad ready", "Web UI", web_path, ad_values("frame_to_ad_ready_ms")))
+        rows.append(row("E2E frame -> ad rendered", "Web UI + Browser", web_path, ad_values("frame_to_render_ms")))
+        rows.append(row("Ads generated (predefined / dynamic)", "Web UI", web_path,
+                        value=f"{len(ad_values('ad_id', 'predefined'))} / {len(ad_values('ad_id', 'dynamic'))}"))
+    if aig:
+        searches = [entry for entry in aig if entry["kind"] == "predefined"]
+        generations = [entry for entry in aig if entry["kind"] == "dynamic"]
+        rows.append(row("6. ASe query (synthetic)", "AIG benchmark", aig_path,
+                        [entry["http_elapsed_ms"] for entry in searches]))
+        rows.append(row("6. ASe vector search (server)", "AIG benchmark", aig_path,
+                        [entry.get("vector_search_ms") for entry in searches]))
+        rows.append(row("7. AIG first call (cold if run with BENCHMARK_COLD_START)", "AIG benchmark", aig_path,
+                        [entry["http_elapsed_ms"] for entry in generations if entry["call"] == 1]))
+        rows.append(row("7. AIG warm calls", "AIG benchmark", aig_path,
+                        [entry["http_elapsed_ms"] for entry in generations if entry["call"] > 1]))
+        rows.append(row("7. AIG inference (server)", "AIG benchmark", aig_path,
+                        [entry.get("inference_ms") for entry in generations]))
+    for entry in qdrant:
+        if entry["kind"] == "vector_search":
+            rows.append(row(f"Qdrant search @ {entry['points']} points", "Qdrant benchmark", qdrant_path,
+                            value=entry["avg_query_ms"], note=f"{entry['queries_per_sec']} queries/s"))
+        else:
+            rows.append(row("Qdrant predefined-ad indexing", "Qdrant benchmark", qdrant_path,
+                            value=entry["index_http_ms"]))
+
+    for prefix in ("e2e", "pid", "web_ui", "aig", "qdrant"):
+        metrics_path = latest(results, f"{prefix}_metrics*.json")
+        metrics = load(metrics_path)
+        if not metrics:
+            continue
+        for key, label, column in (("cpu_utilization", "Host CPU %", 1), ("gpu_utilization", "GPU busy %", 1),
+                                   ("npu_utilization", "NPU %", 1), ("memory", "Host memory %", 4),
+                                   ("power", "Package power W", 1)):
+            values = [sample[column] for sample in metrics.get(key, []) if len(sample) > column]
+            if values:
+                rows.append(row(f"Resources during {prefix}: {label}", "performance-tools", metrics_path,
+                                value=round(statistics.mean(values), 2)))
+    return rows
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--results", type=Path, default=Path("results"))
+    parser.add_argument("--output", type=Path, default=None)
+    args = parser.parse_args()
+    rows = consolidate(args.results)
+    if not rows:
+        print(f"No benchmark results found in {args.results}", file=sys.stderr)
+        sys.exit(1)
+    output = args.output or args.results
+    output.mkdir(parents=True, exist_ok=True)
+    name = f"e2e_consolidated_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+    (output / f"{name}.json").write_text(json.dumps(
+        {"timestamp": datetime.now(timezone.utc).isoformat(), "stages": rows}, indent=2) + "\n")
+    with (output / f"{name}.csv").open("w", newline="") as destination:
+        writer = csv.DictWriter(destination, fieldnames=FIELDS)
+        writer.writeheader()
+        writer.writerows(rows)
+    for entry in rows:
+        summary = entry.get("value", entry.get("avg_ms"))
+        print(f"{entry['stage']:<62} {str(summary):>12}  samples={entry.get('samples', '-')}")
+    print(f"Saved {output / name}.json and .csv")
+
+
+if __name__ == "__main__":
+    main()