From 90fa258075de0dd61ba358be22ee3c698cf21802 Mon Sep 17 00:00:00 2001 From: Pooja Kumbharkar Date: Wed, 14 Oct 2026 08:48:51 +0530 Subject: [PATCH 1/5] performance-tools: Added benchnarking scripts for digital signage Changes: benchmark-scripts - Added scripts for digital signage Signed-off-by: Pooja Kumbharkar --- benchmark-scripts/benchmark_aig.py | 161 +++++++++++++++++++++ benchmark-scripts/benchmark_pid.py | 198 ++++++++++++++++++++++++++ benchmark-scripts/benchmark_qdrant.py | 169 ++++++++++++++++++++++ benchmark-scripts/benchmark_web_ui.py | 133 +++++++++++++++++ 4 files changed, 661 insertions(+) create mode 100644 benchmark-scripts/benchmark_aig.py create mode 100644 benchmark-scripts/benchmark_pid.py create mode 100644 benchmark-scripts/benchmark_qdrant.py create mode 100644 benchmark-scripts/benchmark_web_ui.py diff --git a/benchmark-scripts/benchmark_aig.py b/benchmark-scripts/benchmark_aig.py new file mode 100644 index 0000000..1867584 --- /dev/null +++ b/benchmark-scripts/benchmark_aig.py @@ -0,0 +1,161 @@ +"""Benchmark ASe search and AIG image generation through the HTTP API.""" + +import argparse +import base64 +import csv +import json +import os +import sys +import time +from datetime import datetime, timezone +from pathlib import Path + +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, 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 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 + subprocess.run(["docker", "compose", "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["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", "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) + print(json.dumps(rows, indent=2)) + print(f"Saved {args.output / name}.json and .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", "NPU"), default="GPU") + parser.add_argument("--iterations", type=int, default=10) + parser.add_argument("--duration", type=int, default=180, 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, ValueError, OSError, RuntimeError) as error: + print(f"AIG benchmark failed: {error}", file=sys.stderr) + sys.exit(1) \ No newline at end of file diff --git a/benchmark-scripts/benchmark_pid.py b/benchmark-scripts/benchmark_pid.py new file mode 100644 index 0000000..a6a4f3c --- /dev/null +++ b/benchmark-scripts/benchmark_pid.py @@ -0,0 +1,198 @@ +"""Sample the running PID pipeline through its MQTT output and Docker metrics.""" + +import argparse +import csv +import json +import os +import re +import select +import subprocess +import sys +import time +from datetime import datetime, timezone +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from statistics import mean + + +SERVICE = "dlstreamer-pipeline-server" +BROKER = "mqtt-broker" +TOPIC = "yolo_od_results" +FPS_PATTERN = re.compile(r"FpsCounter\(last [^)]*\): total=([\d.]+) fps") + + +def docker(*args): + return subprocess.check_output(["docker", *args], text=True, stderr=subprocess.STDOUT).strip() + + +def average(values): + return round(mean(values), 3) if values else None + + +def parse_size(value): + match = re.match(r"\s*([\d.]+)\s*([KMGT]?i?B)", value) + if not match: + return None + unit = match.group(2) + factors = {"B": 1, "KB": 1000, "MB": 1000**2, "GB": 1000**3, + "KiB": 1024, "MiB": 1024**2, "GiB": 1024**3, "TiB": 1024**4} + return round(float(match.group(1)) * factors[unit] / 1024**2, 2) + + +def parse_frame(line, arrival_ns=None): + payload = json.loads(line) + metadata = payload["metadata"] + status = metadata.get("pipeline", {}).get("status", {}) + detections = metadata.get("gva_meta", []) + frame_time_ns = metadata.get("time") + arrival_delay_ms = None + if (isinstance(frame_time_ns, int) and not isinstance(frame_time_ns, bool) + and arrival_ns is not None and 0 <= arrival_ns - frame_time_ns < 60_000_000_000): + arrival_delay_ms = (arrival_ns - frame_time_ns) / 1_000_000 + return { + "frame_id": metadata.get("frame_id"), + "frame_timestamp_ns": frame_time_ns, + "arrival_timestamp_ns": arrival_ns, + "detections": len(detections), + "frame_fps": status.get("frame_fps"), + "pipeline_latency_reported": status.get("avg_pipeline_latency"), + "mqtt_arrival_delay_ms": arrival_delay_ms, + } + + +def gpu_compute_usage(path): + if path is None: + return None + data = json.loads(path.read_text()) + samples = [] + for entry in data: + value = entry.get("CCS %") + if value is not None: + samples.append(float(value)) + return average(samples) + + +def run(duration, output, restart, gpu_metrics=None): + for container in (SERVICE, BROKER): + if docker("inspect", "--format", "{{.State.Running}}", container) != "true": + raise RuntimeError(f"{container} is not running; start the stack with make up") + + frames = [] + cpu = [] + memory = [] + started_at = None + if restart: + subprocess.run(["docker", "compose", "restart", SERVICE], check=True) + started_at = datetime.fromisoformat( + docker("inspect", "--format", "{{.State.StartedAt}}", SERVICE).replace("Z", "+00:00") + ).timestamp() + subscriber = subprocess.Popen( + ["docker", "exec", BROKER, "mosquitto_sub", "-h", "localhost", "-t", TOPIC], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + ) + try: + sample_start_wall = time.time() + sample_start = time.monotonic() + deadline = sample_start + duration + next_stats = sample_start + first_frame_at = None + buffered = b"" + with ThreadPoolExecutor(max_workers=1) as pool: + stats_future = None + while time.monotonic() < deadline: + if subscriber.poll() is not None: + raise RuntimeError(f"MQTT subscriber exited: {subscriber.stderr.read().decode().strip()}") + now = time.monotonic() + if stats_future is not None and stats_future.done(): + stats = json.loads(stats_future.result()) + cpu.append(float(stats["CPUPerc"].strip("%"))) + used = parse_size(stats["MemUsage"].split("/")[0]) + if used is not None: + memory.append(used) + stats_future = None + if now >= next_stats and stats_future is None: + stats_future = pool.submit(docker, "stats", "--no-stream", "--format", "{{json .}}", SERVICE) + next_stats = now + 1 + ready, _, _ = select.select([subscriber.stdout], [], [], min(0.2, max(0, deadline - time.monotonic()))) + if ready: + chunk = os.read(subscriber.stdout.fileno(), 65536) + arrived_ns = time.time_ns() + buffered += chunk + while b"\n" in buffered: + line, buffered = buffered.split(b"\n", 1) + try: + frame = parse_frame(line, arrived_ns) + except (ValueError, KeyError, TypeError): + continue + if first_frame_at is None: + first_frame_at = arrived_ns / 1_000_000_000 + frames.append(frame) + if stats_future is not None: + stats = json.loads(stats_future.result()) + cpu.append(float(stats["CPUPerc"].strip("%"))) + used = parse_size(stats["MemUsage"].split("/")[0]) + if used is not None: + memory.append(used) + if not frames: + raise RuntimeError("No PID frames received on MQTT; check the pipeline and topic") + finally: + subscriber.terminate() + try: + subscriber.communicate(timeout=3) + except subprocess.TimeoutExpired: + subscriber.kill() + subscriber.communicate() + + log_lines = docker("logs", "--since", datetime.fromtimestamp( + started_at if started_at else sample_start_wall, timezone.utc + ).isoformat(), SERVICE).splitlines() + log_fps = [float(match.group(1)) for line in log_lines if (match := FPS_PATTERN.search(line))] + latency = [float(frame["pipeline_latency_reported"]) for frame in frames + if isinstance(frame["pipeline_latency_reported"], (int, float))] + result = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "duration_seconds": duration, + "frames": len(frames), + "fps_dlstreamer": average(log_fps), + "fps_mqtt_reported": average([float(frame["frame_fps"]) for frame in frames + if isinstance(frame["frame_fps"], (int, float))]), + "detections_per_frame": average([frame["detections"] for frame in frames]), + "detections_total": sum(frame["detections"] for frame in frames), + "cpu_percent": average(cpu), + "memory_mib": average(memory), + "pipeline_latency_reported": average(latency), + "mqtt_arrival_delay_ms": average([frame["mqtt_arrival_delay_ms"] for frame in frames + if frame["mqtt_arrival_delay_ms"] is not None]), + "startup_to_first_frame_ms": round((first_frame_at - started_at) * 1000, 3) if started_at else None, + "gpu_compute_percent": gpu_compute_usage(gpu_metrics), + } + output.mkdir(parents=True, exist_ok=True) + name = datetime.now().strftime("pid_benchmark_%Y%m%d_%H%M%S") + (output / f"{name}.json").write_text(json.dumps(result, indent=2) + "\n") + with (output / f"{name}.csv").open("w", newline="") as destination: + writer = csv.DictWriter(destination, fieldnames=result.keys()) + writer.writeheader() + writer.writerow(result) + (output / f"{name}_frames.json").write_text(json.dumps(frames, indent=2) + "\n") + with (output / f"{name}_frames.csv").open("w", newline="") as destination: + writer = csv.DictWriter(destination, fieldnames=frames[0].keys()) + writer.writeheader() + writer.writerows(frames) + print(json.dumps(result, indent=2)) + print(f"Saved {output / name}.json, .csv, _frames.json and _frames.csv") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--duration", type=int, default=180) + parser.add_argument("--output", type=Path, default=Path("results")) + parser.add_argument("--no-restart", action="store_true", help="sample without restarting PID (no startup metric)") + parser.add_argument("--gpu-metrics", type=Path, help="qmassa parsed JSON from the same measurement window") + args = parser.parse_args() + if args.duration < 1: + parser.error("--duration must be positive") + try: + run(args.duration, args.output, not args.no_restart, args.gpu_metrics) + except (subprocess.CalledProcessError, OSError, RuntimeError, ValueError, TypeError) as error: + print(f"PID benchmark failed: {error}", file=sys.stderr) + sys.exit(1) \ No newline at end of file diff --git a/benchmark-scripts/benchmark_qdrant.py b/benchmark-scripts/benchmark_qdrant.py new file mode 100644 index 0000000..625b0df --- /dev/null +++ b/benchmark-scripts/benchmark_qdrant.py @@ -0,0 +1,169 @@ +"""Benchmark Qdrant collection scaling and ASe predefined-ad indexing.""" + +import argparse +import base64 +import csv +import json +import math +import random +import re +import secrets +import statistics +import subprocess +import sys +import time +import uuid +from datetime import datetime, timezone +from pathlib import Path + +import requests + + +DEFAULT_ASE_URL = "https://localhost:5000/aig-api" +IMAGE_PATH = Path(__file__).resolve().parents[1] / "web-ui/pre-defined-ads/brocoli_carrot.jpg" + + +def memory_mib(): + stats = json.loads(subprocess.check_output( + ["docker", "stats", "--no-stream", "--format", "{{json .}}", "ase-qdrant"], text=True + )) + value = stats["MemUsage"].split("/")[0].strip() + match = re.fullmatch(r"([\d.]+)\s*([KMGT]?i?B)", value) + if match is None: + raise ValueError(f"Unrecognized Qdrant memory usage: {value}") + amount, unit = match.groups() + factors = {"KiB": 1 / 1024, "MiB": 1, "GiB": 1024, "TiB": 1024**2, + "kB": 1000 / 1024**2, "MB": 1000**2 / 1024**2, "GB": 1000**3 / 1024**2} + return round(float(amount) * factors[unit], 3) + + +def api(session, method, url, **kwargs): + response = session.request(method, url, timeout=120, **kwargs) + response.raise_for_status() + data = response.json() + if data.get("status") != "ok": + raise RuntimeError(f"Qdrant error: {data}") + return data["result"] + + +def benchmark(args): + address = subprocess.check_output( + ["docker", "inspect", "--format", "{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}", + "ase-qdrant"], text=True + ).strip() + if not address: + raise RuntimeError("ase-qdrant is not running") + base = f"http://{address}:6333" + qdrant = requests.Session() + qdrant.trust_env = False + config = api(qdrant, "GET", f"{base}/collections/{args.collection}") + vectors = config["config"]["params"]["vectors"] + if "size" not in vectors: + raise RuntimeError("Only a single unnamed vector collection is supported") + dimensions = vectors["size"] + rng = random.Random(42) + query = [rng.random() for _ in range(dimensions)] + name = f"benchmark-{uuid.uuid4().hex[:12]}" + url = f"{base}/collections/{name}" + rows = [] + created = False + try: + api(qdrant, "PUT", url, json={"vectors": {"size": dimensions, "distance": vectors["distance"]}}) + created = True + next_id = 1 + for size in args.sizes: + memory_before = memory_mib() + upsert_start = time.perf_counter() + while next_id <= size: + points = [{"id": point_id, "vector": [rng.random() for _ in range(dimensions)]} + for point_id in range(next_id, min(next_id + args.batch_size, size + 1))] + api(qdrant, "PUT", f"{url}/points?wait=true", json={"points": points}) + next_id += len(points) + upsert_ms = (time.perf_counter() - upsert_start) * 1000 + info = api(qdrant, "GET", url) + memory_after = memory_mib() + samples = [] + query_loop_start = time.perf_counter() + for _ in range(args.queries): + started = time.perf_counter() + api(qdrant, "POST", f"{url}/points/query", json={ + "query": query, "limit": 1, "with_payload": False, + }) + samples.append((time.perf_counter() - started) * 1000) + query_loop_seconds = time.perf_counter() - query_loop_start + rows.append({"kind": "vector_search", "collection": name, + "points": info["points_count"], "indexed_vectors": info.get("indexed_vectors_count"), + "queries": args.queries, "avg_query_ms": round(statistics.mean(samples), 3), + "p95_query_ms": round(sorted(samples)[math.ceil(0.95 * len(samples)) - 1], 3), + "queries_per_sec": round(args.queries / query_loop_seconds, 3), + "upsert_ms": round(upsert_ms, 3), "qdrant_memory_before_mib": memory_before, + "qdrant_memory_after_mib": memory_after, + "qdrant_memory_delta_mib": round(memory_after - memory_before, 3)}) + finally: + if created: + api(qdrant, "DELETE", url) + + ase = requests.Session() + ase.verify = not args.insecure and args.ase_url.rstrip("/") != DEFAULT_ASE_URL + if not ase.verify: + requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning) + point_id = 1_000_000_000_000 + secrets.randbelow(1_000_000_000_000) + live_url = f"{base}/collections/{args.collection}/points/{point_id}" + if qdrant.get(live_url, timeout=10).status_code != 404: + raise RuntimeError("Temporary ad ID is already in use; no ad was written") + ad_url = f"{args.ase_url.rstrip('/')}/ase/predef/" + payload = {"id": point_id, "description": f"Qdrant benchmark temporary ad {point_id}", + "source": "benchmark", "imgb64": base64.b64encode(IMAGE_PATH.read_bytes()).decode("ascii")} + try: + memory_before = memory_mib() + started = time.perf_counter() + response = ase.post(ad_url, json=payload, timeout=120) + index_ms = (time.perf_counter() - started) * 1000 + response.raise_for_status() + if response.json().get("message") != "Success": + raise RuntimeError(f"ASe add failed: {response.text[:200]}") + memory_after = memory_mib() + rows.append({"kind": "predefined_ad_index", "collection": args.collection, + "points": api(qdrant, "GET", f"{base}/collections/{args.collection}")["points_count"], + "index_http_ms": round(index_ms, 3), "image_kb": round(len(IMAGE_PATH.read_bytes()) / 1024, 3), + "qdrant_memory_before_mib": memory_before, + "qdrant_memory_after_mib": memory_after, + "qdrant_memory_delta_mib": round(memory_after - memory_before, 3)}) + finally: + if qdrant.get(live_url, timeout=10).status_code == 200: + response = ase.delete(f"{ad_url}{point_id}", timeout=120) + response.raise_for_status() + + args.output.mkdir(parents=True, exist_ok=True) + name = f"qdrant_benchmark_{datetime.now().strftime('%Y%m%d_%H%M%S')}" + fields = ("kind", "collection", "points", "indexed_vectors", "queries", "avg_query_ms", + "p95_query_ms", "queries_per_sec", "upsert_ms", "index_http_ms", "image_kb", + "qdrant_memory_before_mib", "qdrant_memory_after_mib", "qdrant_memory_delta_mib") + (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) + print(json.dumps(rows, indent=2)) + print(f"Saved {args.output / name}.json and .csv") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--collection", default="ase-collection") + parser.add_argument("--sizes", type=int, nargs="+", default=[100, 1000]) + parser.add_argument("--queries", type=int, default=30) + parser.add_argument("--batch-size", type=int, default=100) + parser.add_argument("--ase-url", default=DEFAULT_ASE_URL) + parser.add_argument("--insecure", action="store_true", help="disable TLS verification for custom ASe endpoints") + parser.add_argument("--output", type=Path, default=Path("results")) + options = parser.parse_args() + if options.queries < 1 or options.batch_size < 1 or any(size < 1 for size in options.sizes): + parser.error("queries, batch size and sizes must be positive") + if options.sizes != sorted(set(options.sizes)): + parser.error("sizes must be unique and sorted in ascending order") + try: + benchmark(options) + except (requests.RequestException, subprocess.CalledProcessError, RuntimeError, ValueError, OSError) as error: + print(f"Qdrant benchmark failed: {error}", file=sys.stderr) + sys.exit(1) \ No newline at end of file diff --git a/benchmark-scripts/benchmark_web_ui.py b/benchmark-scripts/benchmark_web_ui.py new file mode 100644 index 0000000..79ee00e --- /dev/null +++ b/benchmark-scripts/benchmark_web_ui.py @@ -0,0 +1,133 @@ +"""Collect per-frame Web UI product selection timing from the running container.""" + +import argparse +import csv +import json +import os +import select +import subprocess +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from statistics import mean + + +CALLBACK_MARKER = "WEB_UI_SELECTION " +SELECTOR_MARKER = "WEB_UI_PRODUCT_SELECTION " +FIELDS = ("timestamp", "frame_id", "recency_frames", "confidence_threshold", + "label_normalization_ms", "recency_filtering_ms", "confidence_gating_ms", + "callback_total_ms", "product_key_normalization_ms", "csv_lookup_ms", + "product_deduplication_ms", "selector_total_ms", "candidate_count", + "total_selection_logic_ms") + + +def parse_log_line(line): + for marker in (CALLBACK_MARKER, SELECTOR_MARKER): + if marker in line: + try: + payload = json.loads(line.split(marker, 1)[1]) + except json.JSONDecodeError: + return None + return marker, line.split(" ", 1)[0], payload + return None + + +def combine_events(events): + frames = [] + selections = {} + for marker, timestamp, payload in events: + frame_id = payload.get("frame_id") + if marker == CALLBACK_MARKER: + row = {field: payload.get(field) for field in FIELDS} + row["timestamp"] = timestamp + frames.append(row) + elif frame_id is not None: + selections[frame_id] = payload + for row in frames: + payload = selections.get(row["frame_id"]) + if payload is not None: + row.update({field: payload.get(field) for field in + ("product_key_normalization_ms", "csv_lookup_ms", + "product_deduplication_ms", "selector_total_ms", "candidate_count")}) + row["total_selection_logic_ms"] = round( + row["callback_total_ms"] + row["selector_total_ms"], 6) + return frames + + +def run(duration, output): + if subprocess.check_output( + ["docker", "inspect", "--format", "{{.State.Running}}", "web-ui"], text=True + ).strip() != "true": + raise RuntimeError("web-ui is not running; start the stack with make up") + since = datetime.now(timezone.utc).isoformat() + started = time.monotonic() + process = subprocess.Popen( + ["docker", "logs", "--follow", "--timestamps", "--since", since, "web-ui"], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + ) + events = [] + buffer = b"" + try: + while time.monotonic() - started < duration: + ready, _, _ = select.select([process.stdout], [], [], 0.5) + if ready: + chunk = os.read(process.stdout.fileno(), 65536) + if not chunk: + raise RuntimeError("web-ui log stream ended during capture") + buffer += chunk + while b"\n" in buffer: + line, buffer = buffer.split(b"\n", 1) + event = parse_log_line(line.decode("utf-8", errors="replace")) + if event is not None: + events.append(event) + finally: + process.terminate() + try: + process.communicate(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + process.communicate() + + frames = combine_events(events) + if not frames: + raise RuntimeError("No WEB_UI_SELECTION frames found; restart Web UI with instrumented code") + summary = {"timestamp": datetime.now(timezone.utc).isoformat(), + "capture_seconds": round(time.monotonic() - started, 3), + "frames": len(frames), + "frames_selected": sum(row["selector_total_ms"] is not None for row in frames), + "recency_frames": frames[0]["recency_frames"], + "confidence_threshold": frames[0]["confidence_threshold"]} + for field in FIELDS: + if field.endswith("_ms"): + values = [row[field] for row in frames if isinstance(row[field], (int, float))] + summary[f"avg_{field}"] = round(mean(values), 6) if values else None + + output.mkdir(parents=True, exist_ok=True) + name = f"web_ui_benchmark_{datetime.now().strftime('%Y%m%d_%H%M%S')}" + (output / f"{name}.json").write_text(json.dumps(summary, indent=2) + "\n") + with (output / f"{name}.csv").open("w", newline="") as destination: + writer = csv.DictWriter(destination, fieldnames=summary.keys()) + writer.writeheader() + writer.writerow(summary) + (output / f"{name}_frames.json").write_text(json.dumps(frames, indent=2) + "\n") + with (output / f"{name}_frames.csv").open("w", newline="") as destination: + writer = csv.DictWriter(destination, fieldnames=FIELDS) + writer.writeheader() + writer.writerows(frames) + print(json.dumps(summary, indent=2)) + print(f"Saved {output / name}.json, .csv, _frames.json and _frames.csv") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--duration", type=int, default=180) + parser.add_argument("--output", type=Path, default=Path("results")) + options = parser.parse_args() + if options.duration < 1: + parser.error("--duration must be positive") + try: + run(options.duration, options.output) + except (OSError, subprocess.CalledProcessError, RuntimeError) as error: + print(f"Web UI benchmark failed: {error}", file=sys.stderr) + sys.exit(1) \ No newline at end of file From de676b5763659866e70cfe074fc5730d5dee172a Mon Sep 17 00:00:00 2001 From: Pooja Kumbharkar Date: Thu, 15 Oct 2026 15:29:19 +0530 Subject: [PATCH 2/5] performance-tools: Updated benchmarking ds scripts Signed-off-by: Pooja Kumbharkar --- benchmark-scripts/benchmark_aig.py | 10 +- benchmark-scripts/benchmark_pid.py | 13 +- benchmark-scripts/benchmark_qdrant.py | 2 +- benchmark-scripts/benchmark_web_ui.py | 50 +++++- benchmark-scripts/consolidate_e2e.py | 150 ++++++++++++++++ benchmark-scripts/docker_service_stats.py | 210 ++++++++++++++++++++++ 6 files changed, 420 insertions(+), 15 deletions(-) create mode 100644 benchmark-scripts/consolidate_e2e.py create mode 100644 benchmark-scripts/docker_service_stats.py diff --git a/benchmark-scripts/benchmark_aig.py b/benchmark-scripts/benchmark_aig.py index 1867584..fa26c9e 100644 --- a/benchmark-scripts/benchmark_aig.py +++ b/benchmark-scripts/benchmark_aig.py @@ -5,6 +5,7 @@ import csv import json import os +import subprocess import sys import time from datetime import datetime, timezone @@ -121,11 +122,12 @@ def run(args): 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", "http_elapsed_ms", "sla_ms", "sla_met", + 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)) @@ -145,9 +147,9 @@ def run(args): 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", "NPU"), default="GPU") + parser.add_argument("--device", choices=("CPU", "GPU"), default="GPU") parser.add_argument("--iterations", type=int, default=10) - parser.add_argument("--duration", type=int, default=180, help="minimum capture duration in seconds") + 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")) @@ -156,6 +158,6 @@ def run(args): parser.error("iterations and timeout must be positive; duration cannot be negative") try: run(options) - except (requests.RequestException, ValueError, OSError, RuntimeError) as error: + except (requests.RequestException, subprocess.CalledProcessError, ValueError, OSError, RuntimeError) as error: print(f"AIG benchmark failed: {error}", file=sys.stderr) sys.exit(1) \ No newline at end of file diff --git a/benchmark-scripts/benchmark_pid.py b/benchmark-scripts/benchmark_pid.py index a6a4f3c..24261f9 100644 --- a/benchmark-scripts/benchmark_pid.py +++ b/benchmark-scripts/benchmark_pid.py @@ -72,7 +72,7 @@ def gpu_compute_usage(path): return average(samples) -def run(duration, output, restart, gpu_metrics=None): +def run(duration, output, restart, gpu_metrics=None, device="CPU"): for container in (SERVICE, BROKER): if docker("inspect", "--format", "{{.State.Running}}", container) != "true": raise RuntimeError(f"{container} is not running; start the stack with make up") @@ -151,6 +151,7 @@ def run(duration, output, restart, gpu_metrics=None): if isinstance(frame["pipeline_latency_reported"], (int, float))] result = { "timestamp": datetime.now(timezone.utc).isoformat(), + "device": device, "duration_seconds": duration, "frames": len(frames), "fps_dlstreamer": average(log_fps), @@ -184,15 +185,17 @@ def run(duration, output, restart, gpu_metrics=None): if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--duration", type=int, default=180) + parser.add_argument("--duration", type=int, default=300) parser.add_argument("--output", type=Path, default=Path("results")) - parser.add_argument("--no-restart", action="store_true", help="sample without restarting PID (no startup metric)") + parser.add_argument("--device", choices=("CPU", "GPU", "NPU"), default="CPU") + parser.add_argument("--no-restart", action="store_true", + help="sample without restarting PID (no startup metric)") parser.add_argument("--gpu-metrics", type=Path, help="qmassa parsed JSON from the same measurement window") args = parser.parse_args() if args.duration < 1: parser.error("--duration must be positive") try: - run(args.duration, args.output, not args.no_restart, args.gpu_metrics) - except (subprocess.CalledProcessError, OSError, RuntimeError, ValueError, TypeError) as error: + run(args.duration, args.output, not args.no_restart, args.gpu_metrics, args.device) + except (subprocess.CalledProcessError, OSError, RuntimeError, ValueError, TypeError, KeyError) as error: print(f"PID benchmark failed: {error}", file=sys.stderr) sys.exit(1) \ No newline at end of file diff --git a/benchmark-scripts/benchmark_qdrant.py b/benchmark-scripts/benchmark_qdrant.py index 625b0df..c454b90 100644 --- a/benchmark-scripts/benchmark_qdrant.py +++ b/benchmark-scripts/benchmark_qdrant.py @@ -20,7 +20,7 @@ DEFAULT_ASE_URL = "https://localhost:5000/aig-api" -IMAGE_PATH = Path(__file__).resolve().parents[1] / "web-ui/pre-defined-ads/brocoli_carrot.jpg" +IMAGE_PATH = Path(__file__).resolve().parents[1] / "../web-ui/pre-defined-ads/brocoli_carrot.jpg" def memory_mib(): diff --git a/benchmark-scripts/benchmark_web_ui.py b/benchmark-scripts/benchmark_web_ui.py index 79ee00e..54b66c0 100644 --- a/benchmark-scripts/benchmark_web_ui.py +++ b/benchmark-scripts/benchmark_web_ui.py @@ -15,15 +15,22 @@ CALLBACK_MARKER = "WEB_UI_SELECTION " SELECTOR_MARKER = "WEB_UI_PRODUCT_SELECTION " +AD_TRACE_MARKER = "WEB_UI_AD_TRACE " +AD_SERVED_MARKER = "WEB_UI_AD_SERVED " +AD_RENDERED_MARKER = "WEB_UI_AD_RENDERED " +MARKERS = (CALLBACK_MARKER, SELECTOR_MARKER, AD_TRACE_MARKER, AD_SERVED_MARKER, AD_RENDERED_MARKER) FIELDS = ("timestamp", "frame_id", "recency_frames", "confidence_threshold", - "label_normalization_ms", "recency_filtering_ms", "confidence_gating_ms", + "mqtt_to_web_ui_ms", "label_normalization_ms", "recency_filtering_ms", "confidence_gating_ms", "callback_total_ms", "product_key_normalization_ms", "csv_lookup_ms", "product_deduplication_ms", "selector_total_ms", "candidate_count", "total_selection_logic_ms") +AD_FIELDS = ("timestamp", "ad_id", "frame_id", "item", "path", "queue_wait_ms", "ase_query_ms", + "aig_generation_ms", "ad_generation_total_ms", "response_size_kb", "frame_to_ad_ready_ms", + "ready_to_served_ms", "browser_fetch_ms", "browser_render_ms", "frame_to_render_ms") def parse_log_line(line): - for marker in (CALLBACK_MARKER, SELECTOR_MARKER): + for marker in MARKERS: if marker in line: try: payload = json.loads(line.split(marker, 1)[1]) @@ -42,7 +49,7 @@ def combine_events(events): row = {field: payload.get(field) for field in FIELDS} row["timestamp"] = timestamp frames.append(row) - elif frame_id is not None: + elif marker == SELECTOR_MARKER and frame_id is not None: selections[frame_id] = payload for row in frames: payload = selections.get(row["frame_id"]) @@ -55,6 +62,28 @@ def combine_events(events): return frames +def combine_ad_events(events): + ads = {} + served = {} + rendered = {} + for marker, timestamp, payload in events: + ad_id = payload.get("ad_id") + if marker == AD_TRACE_MARKER: + ads[ad_id] = {field: payload.get(field) for field in AD_FIELDS} + ads[ad_id]["timestamp"] = timestamp + elif marker == AD_SERVED_MARKER: + served.setdefault(ad_id, payload) + elif marker == AD_RENDERED_MARKER: + rendered.setdefault(ad_id, payload) + for ad_id, row in ads.items(): + row.update({key: value for key, value in served.get(ad_id, {}).items() if key in AD_FIELDS}) + row.update({key: value for key, value in rendered.get(ad_id, {}).items() if key in AD_FIELDS}) + parts = [row[key] for key in ("frame_to_ad_ready_ms", "ready_to_served_ms", + "browser_fetch_ms", "browser_render_ms")] + if all(isinstance(part, (int, float)) for part in parts): + row["frame_to_render_ms"] = round(sum(parts), 3) + return list(ads.values()) + def run(duration, output): if subprocess.check_output( ["docker", "inspect", "--format", "{{.State.Running}}", "web-ui"], text=True @@ -90,6 +119,7 @@ def run(duration, output): process.communicate() frames = combine_events(events) + ads = combine_ad_events(events) if not frames: raise RuntimeError("No WEB_UI_SELECTION frames found; restart Web UI with instrumented code") summary = {"timestamp": datetime.now(timezone.utc).isoformat(), @@ -102,6 +132,11 @@ def run(duration, output): if field.endswith("_ms"): values = [row[field] for row in frames if isinstance(row[field], (int, float))] summary[f"avg_{field}"] = round(mean(values), 6) if values else None + summary["ads"] = len(ads) + for field in AD_FIELDS: + if field.endswith("_ms"): + values = [row[field] for row in ads if isinstance(row[field], (int, float))] + summary[f"avg_ad_{field}"] = round(mean(values), 3) if values else None output.mkdir(parents=True, exist_ok=True) name = f"web_ui_benchmark_{datetime.now().strftime('%Y%m%d_%H%M%S')}" @@ -115,13 +150,18 @@ def run(duration, output): writer = csv.DictWriter(destination, fieldnames=FIELDS) writer.writeheader() writer.writerows(frames) + (output / f"{name}_ads.json").write_text(json.dumps(ads, indent=2) + "\n") + with (output / f"{name}_ads.csv").open("w", newline="") as destination: + writer = csv.DictWriter(destination, fieldnames=AD_FIELDS) + writer.writeheader() + writer.writerows(ads) print(json.dumps(summary, indent=2)) - print(f"Saved {output / name}.json, .csv, _frames.json and _frames.csv") + print(f"Saved {output / name}.json, .csv, _frames.json/.csv and _ads.json/.csv") if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--duration", type=int, default=180) + parser.add_argument("--duration", type=int, default=300) parser.add_argument("--output", type=Path, default=Path("results")) options = parser.parse_args() if options.duration < 1: diff --git a/benchmark-scripts/consolidate_e2e.py b/benchmark-scripts/consolidate_e2e.py new file mode 100644 index 0000000..ba8bab5 --- /dev/null +++ b/benchmark-scripts/consolidate_e2e.py @@ -0,0 +1,150 @@ +"""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_*.json") + aig = load(aig_path) or [] + qdrant_path = latest(results, "qdrant_benchmark_*.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 selection", "Web UI", web_path, ad_values("queue_wait_ms"), + note="Includes polling and ad cooldown waits")) + 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() diff --git a/benchmark-scripts/docker_service_stats.py b/benchmark-scripts/docker_service_stats.py new file mode 100644 index 0000000..f69734a --- /dev/null +++ b/benchmark-scripts/docker_service_stats.py @@ -0,0 +1,210 @@ +"""Sample Docker CPU and memory usage for running Compose services.""" + +import argparse +import csv +import json +import re +import signal +import subprocess +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from statistics import mean +from threading import Event + + +MEMORY_PATTERN = re.compile(r"\s*([\d.]+)\s*([KMGT]?i?B)") +MEMORY_FACTORS = { + "B": 1 / 1024**2, + "KB": 1000 / 1024**2, + "MB": 1000**2 / 1024**2, + "GB": 1000**3 / 1024**2, + "TB": 1000**4 / 1024**2, + "KiB": 1 / 1024, + "MiB": 1, + "GiB": 1024, + "TiB": 1024**2, +} + + +def average(values): + values = [float(value) for value in values if isinstance(value, (int, float))] + return round(mean(values), 3) if values else None + + +def parse_percent(value): + try: + return float(value.strip().rstrip("%")) + except (AttributeError, TypeError, ValueError): + return None + + +def parse_memory_mib(value): + match = MEMORY_PATTERN.match(value or "") + if not match: + return None + return round(float(match.group(1)) * MEMORY_FACTORS[match.group(2)], 3) + + +def running_services(): + output = subprocess.check_output( + ["docker", "compose", "ps", "--format", "json"], + text=True, stderr=subprocess.STDOUT, + ) + containers = [] + for line in output.splitlines(): + try: + container = json.loads(line) + except json.JSONDecodeError: + continue + if container.get("State", "").lower() == "running" and container.get("ID"): + containers.append(container) + return containers + + +def sample_services(): + containers = running_services() + if not containers: + return [] + + services_by_id = {container["ID"][:12]: container.get("Service", "unknown") + for container in containers} + services_by_name = {container.get("Name"): container.get("Service", "unknown") + for container in containers} + command = ["docker", "stats", "--no-stream", "--format", "{{json .}}"] + command.extend(container["ID"] for container in containers) + output = subprocess.check_output(command, text=True, stderr=subprocess.STDOUT) + timestamp = datetime.now(timezone.utc).isoformat() + samples = [] + for line in output.splitlines(): + try: + stats = json.loads(line) + except json.JSONDecodeError: + continue + container_id = (stats.get("ID") or stats.get("Container") or "")[:12] + name = stats.get("Name") or "" + service = services_by_id.get(container_id) or services_by_name.get(name) + if not service: + continue + memory_used, _, memory_limit = (stats.get("MemUsage", "").partition("/")) + samples.append({ + "timestamp": timestamp, + "service": service, + "container": name, + "cpu_percent": parse_percent(stats.get("CPUPerc")), + "memory_percent": parse_percent(stats.get("MemPerc")), + "memory_usage_mib": parse_memory_mib(memory_used), + "memory_limit_mib": parse_memory_mib(memory_limit), + }) + return samples + + +def host_metric_averages(path): + if not path or not path.is_file(): + return { + "cpu_utilization_percent": None, + "gpu_utilization_percent": None, + "memory_utilization_percent": None, + } + try: + payload = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError): + return { + "cpu_utilization_percent": None, + "gpu_utilization_percent": None, + "memory_utilization_percent": None, + } + return { + "cpu_utilization_percent": average([row[1] for row in payload.get("cpu_utilization", []) + if len(row) > 1]), + "gpu_utilization_percent": average([row[1] for row in payload.get("gpu_utilization", []) + if len(row) > 1]), + "memory_utilization_percent": average([row[4] for row in payload.get("memory", []) + if len(row) > 4]), + } + + +def write_results(output_prefix, samples, interval, host_metrics): + output_prefix.parent.mkdir(parents=True, exist_ok=True) + csv_path = Path(f"{output_prefix}.csv") + json_path = Path(f"{output_prefix}.json") + fields = ("timestamp", "service", "container", "cpu_percent", "memory_percent", + "memory_usage_mib", "memory_limit_mib") + with csv_path.open("w", newline="") as destination: + writer = csv.DictWriter(destination, fieldnames=fields) + writer.writeheader() + writer.writerows(samples) + + by_service = {} + for sample in samples: + by_service.setdefault(sample["service"], []).append(sample) + service_averages = {} + for service, service_samples in sorted(by_service.items()): + service_averages[service] = { + "samples": len(service_samples), + "containers": sorted({sample["container"] for sample in service_samples}), + "average_cpu_percent": average([sample["cpu_percent"] for sample in service_samples]), + "average_memory_percent": average([sample["memory_percent"] for sample in service_samples]), + "average_memory_usage_mib": average([sample["memory_usage_mib"] for sample in service_samples]), + "average_memory_limit_mib": average([sample["memory_limit_mib"] for sample in service_samples]), + } + result = { + "sample_interval_seconds": interval, + "services": service_averages, + "host_metric_averages": host_metric_averages(host_metrics), + } + json_path.write_text(json.dumps(result, indent=2) + "\n") + print(f"Saved Docker service averages to {json_path} and samples to {csv_path}") + return bool(samples) + + +def run(output_prefix, interval, duration, host_metrics, ready_file): + stop = Event() + signal.signal(signal.SIGTERM, lambda *_: stop.set()) + signal.signal(signal.SIGINT, lambda *_: stop.set()) + samples = [] + deadline = time.monotonic() + duration if duration else None + + try: + sample_start = time.monotonic() + try: + samples.extend(sample_services()) + except (OSError, subprocess.CalledProcessError) as error: + print(f"Docker stats sample failed: {error}", file=sys.stderr) + if ready_file: + ready_file.write_text("ready\n") + while not stop.is_set() and (deadline is None or time.monotonic() < deadline): + sample_start = time.monotonic() + try: + samples.extend(sample_services()) + except (OSError, subprocess.CalledProcessError) as error: + print(f"Docker stats sample failed: {error}", file=sys.stderr) + wait = max(0, interval - (time.monotonic() - sample_start)) + if wait and (deadline is None or time.monotonic() + wait < deadline): + stop.wait(wait) + return write_results(output_prefix, samples, interval, host_metrics) + finally: + if ready_file: + ready_file.unlink(missing_ok=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output-prefix", type=Path, required=True) + parser.add_argument("--host-metrics", type=Path) + parser.add_argument("--ready-file", type=Path) + parser.add_argument("--interval", type=float, default=1.0) + parser.add_argument("--duration", type=float, default=0, + help="stop after this many seconds; defaults to signal-controlled sampling") + args = parser.parse_args() + if args.interval <= 0 or args.duration < 0: + parser.error("interval must be positive and duration cannot be negative") + if not run(args.output_prefix, args.interval, args.duration, args.host_metrics, args.ready_file): + print("Docker stats produced no samples for running Compose services", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file From b62dcc7153b873008144471b9359292cfd36391f Mon Sep 17 00:00:00 2001 From: Pooja Kumbharkar Date: Fri, 16 Oct 2026 16:28:13 +0530 Subject: [PATCH 3/5] benchmark-scripts: Renamed the digital signage scripts Changes: benchmark-scripts - Renamed the digital signage benchmarking scripts to benchmarkinmg_signage_*.py and added summary for aig and qdrant services. Signed-off-by: Pooja Kumbharkar --- ...chmark_aig.py => benchmark_signage_aig.py} | 37 +++++++++++++++++-- ...y => benchmark_signage_consolidate_e2e.py} | 15 ++++++-- ...s.py => benchmark_signage_docker_stats.py} | 0 ...chmark_pid.py => benchmark_signage_pid.py} | 0 ..._qdrant.py => benchmark_signage_qdrant.py} | 29 ++++++++++++++- ..._web_ui.py => benchmark_signage_web_ui.py} | 31 ++++++++++++++-- 6 files changed, 100 insertions(+), 12 deletions(-) rename benchmark-scripts/{benchmark_aig.py => benchmark_signage_aig.py} (78%) rename benchmark-scripts/{consolidate_e2e.py => benchmark_signage_consolidate_e2e.py} (89%) rename benchmark-scripts/{docker_service_stats.py => benchmark_signage_docker_stats.py} (100%) rename benchmark-scripts/{benchmark_pid.py => benchmark_signage_pid.py} (100%) rename benchmark-scripts/{benchmark_qdrant.py => benchmark_signage_qdrant.py} (84%) rename benchmark-scripts/{benchmark_web_ui.py => benchmark_signage_web_ui.py} (82%) diff --git a/benchmark-scripts/benchmark_aig.py b/benchmark-scripts/benchmark_signage_aig.py similarity index 78% rename from benchmark-scripts/benchmark_aig.py rename to benchmark-scripts/benchmark_signage_aig.py index fa26c9e..fb5041c 100644 --- a/benchmark-scripts/benchmark_aig.py +++ b/benchmark-scripts/benchmark_signage_aig.py @@ -10,6 +10,7 @@ import time from datetime import datetime, timezone from pathlib import Path +from statistics import mean import requests @@ -33,7 +34,7 @@ def parse_server_timing(header): def request_ad(session, url, payload, kind, call, timeout, cold_start=False): start = time.perf_counter() - response = session.post(url, json=payload, timeout=timeout) + 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": @@ -65,6 +66,30 @@ def request_ad(session, url, payload, kind, call, timeout, cold_start=False): } +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 @@ -136,8 +161,14 @@ def run(args): writer = csv.DictWriter(destination, fieldnames=fields) writer.writeheader() writer.writerows(rows) - print(json.dumps(rows, indent=2)) - print(f"Saved {args.output / name}.json and .csv") + 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__": diff --git a/benchmark-scripts/consolidate_e2e.py b/benchmark-scripts/benchmark_signage_consolidate_e2e.py similarity index 89% rename from benchmark-scripts/consolidate_e2e.py rename to benchmark-scripts/benchmark_signage_consolidate_e2e.py index ba8bab5..f07a830 100644 --- a/benchmark-scripts/consolidate_e2e.py +++ b/benchmark-scripts/benchmark_signage_consolidate_e2e.py @@ -48,9 +48,9 @@ def consolidate(results): 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_*.json") + aig_path = latest(results, "aig_benchmark_*[0-9].json") aig = load(aig_path) or [] - qdrant_path = latest(results, "qdrant_benchmark_*.json") + qdrant_path = latest(results, "qdrant_benchmark_*[0-9].json") qdrant = load(qdrant_path) or [] def ad_values(field, path=None): @@ -70,8 +70,15 @@ def ad_values(field, path=None): 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 selection", "Web UI", web_path, ad_values("queue_wait_ms"), - note="Includes polling and ad cooldown waits")) + 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"))) diff --git a/benchmark-scripts/docker_service_stats.py b/benchmark-scripts/benchmark_signage_docker_stats.py similarity index 100% rename from benchmark-scripts/docker_service_stats.py rename to benchmark-scripts/benchmark_signage_docker_stats.py diff --git a/benchmark-scripts/benchmark_pid.py b/benchmark-scripts/benchmark_signage_pid.py similarity index 100% rename from benchmark-scripts/benchmark_pid.py rename to benchmark-scripts/benchmark_signage_pid.py diff --git a/benchmark-scripts/benchmark_qdrant.py b/benchmark-scripts/benchmark_signage_qdrant.py similarity index 84% rename from benchmark-scripts/benchmark_qdrant.py rename to benchmark-scripts/benchmark_signage_qdrant.py index c454b90..14000a3 100644 --- a/benchmark-scripts/benchmark_qdrant.py +++ b/benchmark-scripts/benchmark_signage_qdrant.py @@ -144,8 +144,33 @@ def benchmark(args): writer = csv.DictWriter(destination, fieldnames=fields) writer.writeheader() writer.writerows(rows) - print(json.dumps(rows, indent=2)) - print(f"Saved {args.output / name}.json and .csv") + summary = summarize(rows) + (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") + + +def summarize(rows): + """Flatten per-size search results into one row; search cost scales with collection size, + so sizes stay separate instead of being averaged together.""" + summary = {"timestamp": datetime.now(timezone.utc).isoformat()} + searches = [row for row in rows if row["kind"] == "vector_search"] + summary["sizes"] = " ".join(str(row["points"]) for row in searches) + summary["queries_per_size"] = searches[0]["queries"] if searches else None + for row in searches: + for field in ("avg_query_ms", "p95_query_ms", "queries_per_sec", "upsert_ms", + "qdrant_memory_delta_mib"): + summary[f"search_{row['points']}_{field}"] = row.get(field) + for row in rows: + if row["kind"] == "predefined_ad_index": + summary["predefined_ad_index_http_ms"] = row.get("index_http_ms") + summary["predefined_ad_index_image_kb"] = row.get("image_kb") + summary["predefined_ad_index_memory_delta_mib"] = row.get("qdrant_memory_delta_mib") + return summary if __name__ == "__main__": diff --git a/benchmark-scripts/benchmark_web_ui.py b/benchmark-scripts/benchmark_signage_web_ui.py similarity index 82% rename from benchmark-scripts/benchmark_web_ui.py rename to benchmark-scripts/benchmark_signage_web_ui.py index 54b66c0..a0a201a 100644 --- a/benchmark-scripts/benchmark_web_ui.py +++ b/benchmark-scripts/benchmark_signage_web_ui.py @@ -16,13 +16,16 @@ CALLBACK_MARKER = "WEB_UI_SELECTION " SELECTOR_MARKER = "WEB_UI_PRODUCT_SELECTION " AD_TRACE_MARKER = "WEB_UI_AD_TRACE " +COOLDOWN_MARKER = "WEB_UI_AD_COOLDOWN " AD_SERVED_MARKER = "WEB_UI_AD_SERVED " AD_RENDERED_MARKER = "WEB_UI_AD_RENDERED " -MARKERS = (CALLBACK_MARKER, SELECTOR_MARKER, AD_TRACE_MARKER, AD_SERVED_MARKER, AD_RENDERED_MARKER) +MARKERS = (CALLBACK_MARKER, SELECTOR_MARKER, AD_TRACE_MARKER, COOLDOWN_MARKER, + AD_SERVED_MARKER, AD_RENDERED_MARKER) FIELDS = ("timestamp", "frame_id", "recency_frames", "confidence_threshold", "mqtt_to_web_ui_ms", "label_normalization_ms", "recency_filtering_ms", "confidence_gating_ms", "callback_total_ms", "product_key_normalization_ms", "csv_lookup_ms", - "product_deduplication_ms", "selector_total_ms", "candidate_count", + "product_deduplication_ms", "selector_total_ms", "candidate_count", "queue_wait_ms", + "cooldown_remaining_ms", "cooldown_suppressed", "total_selection_logic_ms") AD_FIELDS = ("timestamp", "ad_id", "frame_id", "item", "path", "queue_wait_ms", "ase_query_ms", "aig_generation_ms", "ad_generation_total_ms", "response_size_kb", "frame_to_ad_ready_ms", @@ -43,6 +46,8 @@ def parse_log_line(line): def combine_events(events): frames = [] selections = {} + queue_waits = {} + cooldowns = {} for marker, timestamp, payload in events: frame_id = payload.get("frame_id") if marker == CALLBACK_MARKER: @@ -51,14 +56,25 @@ def combine_events(events): frames.append(row) elif marker == SELECTOR_MARKER and frame_id is not None: selections[frame_id] = payload + elif marker == AD_TRACE_MARKER and frame_id is not None: + queue_waits[frame_id] = payload.get("queue_wait_ms") + elif marker == COOLDOWN_MARKER and frame_id is not None: + queue_waits[frame_id] = payload.get("queue_wait_ms") + cooldowns[frame_id] = payload for row in frames: - payload = selections.get(row["frame_id"]) + frame_id = row["frame_id"] + payload = selections.get(frame_id) if payload is not None: row.update({field: payload.get(field) for field in ("product_key_normalization_ms", "csv_lookup_ms", "product_deduplication_ms", "selector_total_ms", "candidate_count")}) row["total_selection_logic_ms"] = round( row["callback_total_ms"] + row["selector_total_ms"], 6) + row["queue_wait_ms"] = queue_waits.get(frame_id) + cooldown = cooldowns.get(frame_id) + row["cooldown_suppressed"] = cooldown is not None if frame_id is not None else None + if cooldown is not None: + row["cooldown_remaining_ms"] = cooldown.get("cooldown_remaining_ms") return frames @@ -126,6 +142,7 @@ def run(duration, output): "capture_seconds": round(time.monotonic() - started, 3), "frames": len(frames), "frames_selected": sum(row["selector_total_ms"] is not None for row in frames), + "cooldown_suppressed_frames": sum(row["cooldown_suppressed"] is True for row in frames), "recency_frames": frames[0]["recency_frames"], "confidence_threshold": frames[0]["confidence_threshold"]} for field in FIELDS: @@ -137,6 +154,14 @@ def run(duration, output): if field.endswith("_ms"): values = [row[field] for row in ads if isinstance(row[field], (int, float))] summary[f"avg_ad_{field}"] = round(mean(values), 3) if values else None + # Predefined ads skip AIG, so a combined average mixes two different pipelines. + for path in ("predefined", "dynamic"): + subset = [row for row in ads if row.get("path") == path] + summary[f"ads_{path}"] = len(subset) + for field in AD_FIELDS: + if field.endswith("_ms"): + values = [row[field] for row in subset if isinstance(row[field], (int, float))] + summary[f"avg_{path}_{field}"] = round(mean(values), 3) if values else None output.mkdir(parents=True, exist_ok=True) name = f"web_ui_benchmark_{datetime.now().strftime('%Y%m%d_%H%M%S')}" From 4052cb9df2b784cc4790cdd1a479718baa2e32a0 Mon Sep 17 00:00:00 2001 From: Pooja Kumbharkar Date: Fri, 16 Oct 2026 16:56:14 +0530 Subject: [PATCH 4/5] benchmark-scripts: Added license header for digital signage Changes: benchmark-scripts - Added license header for digital signage scripts Signed-off-by: Pooja Kumbharkar --- benchmark-scripts/benchmark_signage_aig.py | 10 +++++++++- .../benchmark_signage_consolidate_e2e.py | 7 +++++++ .../benchmark_signage_docker_stats.py | 7 +++++++ benchmark-scripts/benchmark_signage_pid.py | 15 +++++++++++++-- benchmark-scripts/benchmark_signage_qdrant.py | 7 +++++++ benchmark-scripts/benchmark_signage_web_ui.py | 7 +++++++ 6 files changed, 50 insertions(+), 3 deletions(-) diff --git a/benchmark-scripts/benchmark_signage_aig.py b/benchmark-scripts/benchmark_signage_aig.py index fb5041c..dbbff0e 100644 --- a/benchmark-scripts/benchmark_signage_aig.py +++ b/benchmark-scripts/benchmark_signage_aig.py @@ -1,3 +1,10 @@ +''' +* Copyright (C) 2026 Intel Corporation. +* +* SPDX-License-Identifier: Apache-2.0 +''' + + """Benchmark ASe search and AIG image generation through the HTTP API.""" import argparse @@ -103,7 +110,8 @@ def run(args): else "TLS verification disabled by --insecure", file=sys.stderr) if args.cold_start: import subprocess - subprocess.run(["docker", "compose", "restart", "aig-server"], check=True) + # 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: diff --git a/benchmark-scripts/benchmark_signage_consolidate_e2e.py b/benchmark-scripts/benchmark_signage_consolidate_e2e.py index f07a830..34784ab 100644 --- a/benchmark-scripts/benchmark_signage_consolidate_e2e.py +++ b/benchmark-scripts/benchmark_signage_consolidate_e2e.py @@ -1,3 +1,10 @@ +''' +* 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 diff --git a/benchmark-scripts/benchmark_signage_docker_stats.py b/benchmark-scripts/benchmark_signage_docker_stats.py index f69734a..85a9e15 100644 --- a/benchmark-scripts/benchmark_signage_docker_stats.py +++ b/benchmark-scripts/benchmark_signage_docker_stats.py @@ -1,3 +1,10 @@ +''' +* Copyright (C) 2026 Intel Corporation. +* +* SPDX-License-Identifier: Apache-2.0 +''' + + """Sample Docker CPU and memory usage for running Compose services.""" import argparse diff --git a/benchmark-scripts/benchmark_signage_pid.py b/benchmark-scripts/benchmark_signage_pid.py index 24261f9..ad59903 100644 --- a/benchmark-scripts/benchmark_signage_pid.py +++ b/benchmark-scripts/benchmark_signage_pid.py @@ -1,3 +1,10 @@ +''' +* Copyright (C) 2026 Intel Corporation. +* +* SPDX-License-Identifier: Apache-2.0 +''' + + """Sample the running PID pipeline through its MQTT output and Docker metrics.""" import argparse @@ -82,7 +89,8 @@ def run(duration, output, restart, gpu_metrics=None, device="CPU"): memory = [] started_at = None if restart: - subprocess.run(["docker", "compose", "restart", SERVICE], check=True) + # Restart by container name, not `compose restart`, so the script does not depend on the caller's Compose project. + subprocess.run(["docker", "restart", SERVICE], check=True) started_at = datetime.fromisoformat( docker("inspect", "--format", "{{.State.StartedAt}}", SERVICE).replace("Z", "+00:00") ).timestamp() @@ -149,6 +157,8 @@ def run(duration, output, restart, gpu_metrics=None, device="CPU"): log_fps = [float(match.group(1)) for line in log_lines if (match := FPS_PATTERN.search(line))] latency = [float(frame["pipeline_latency_reported"]) for frame in frames if isinstance(frame["pipeline_latency_reported"], (int, float))] + # Frame 0 proves the subscription beat the restarted pipeline; without it the startup number would be inflated. + captured_opening_frame = frames[0]["frame_id"] == 0 result = { "timestamp": datetime.now(timezone.utc).isoformat(), "device": device, @@ -164,7 +174,8 @@ def run(duration, output, restart, gpu_metrics=None, device="CPU"): "pipeline_latency_reported": average(latency), "mqtt_arrival_delay_ms": average([frame["mqtt_arrival_delay_ms"] for frame in frames if frame["mqtt_arrival_delay_ms"] is not None]), - "startup_to_first_frame_ms": round((first_frame_at - started_at) * 1000, 3) if started_at else None, + "startup_to_first_frame_ms": round((first_frame_at - started_at) * 1000, 3) + if started_at and captured_opening_frame else None, "gpu_compute_percent": gpu_compute_usage(gpu_metrics), } output.mkdir(parents=True, exist_ok=True) diff --git a/benchmark-scripts/benchmark_signage_qdrant.py b/benchmark-scripts/benchmark_signage_qdrant.py index 14000a3..404507f 100644 --- a/benchmark-scripts/benchmark_signage_qdrant.py +++ b/benchmark-scripts/benchmark_signage_qdrant.py @@ -1,3 +1,10 @@ +''' +* Copyright (C) 2026 Intel Corporation. +* +* SPDX-License-Identifier: Apache-2.0 +''' + + """Benchmark Qdrant collection scaling and ASe predefined-ad indexing.""" import argparse diff --git a/benchmark-scripts/benchmark_signage_web_ui.py b/benchmark-scripts/benchmark_signage_web_ui.py index a0a201a..7730562 100644 --- a/benchmark-scripts/benchmark_signage_web_ui.py +++ b/benchmark-scripts/benchmark_signage_web_ui.py @@ -1,3 +1,10 @@ +''' +* Copyright (C) 2026 Intel Corporation. +* +* SPDX-License-Identifier: Apache-2.0 +''' + + """Collect per-frame Web UI product selection timing from the running container.""" import argparse From 6e4eca0289543958a224d1ee36c15a8483fd33fb Mon Sep 17 00:00:00 2001 From: Pooja Kumbharkar Date: Mon, 19 Oct 2026 09:14:00 +0530 Subject: [PATCH 5/5] performance-tools: Updated copyright header for digital signage files Changes: benchmark-scripts-Updated the copyright header for digital signage scripts Signed-off-by: Pooja Kumbharkar --- benchmark-scripts/benchmark_signage_aig.py | 10 +++++----- benchmark-scripts/benchmark_signage_consolidate_e2e.py | 10 +++++----- benchmark-scripts/benchmark_signage_docker_stats.py | 10 +++++----- benchmark-scripts/benchmark_signage_pid.py | 10 +++++----- benchmark-scripts/benchmark_signage_qdrant.py | 10 +++++----- benchmark-scripts/benchmark_signage_web_ui.py | 10 +++++----- 6 files changed, 30 insertions(+), 30 deletions(-) diff --git a/benchmark-scripts/benchmark_signage_aig.py b/benchmark-scripts/benchmark_signage_aig.py index dbbff0e..1ae463c 100644 --- a/benchmark-scripts/benchmark_signage_aig.py +++ b/benchmark-scripts/benchmark_signage_aig.py @@ -1,8 +1,8 @@ -''' -* Copyright (C) 2026 Intel Corporation. -* -* SPDX-License-Identifier: Apache-2.0 -''' +# +# 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.""" diff --git a/benchmark-scripts/benchmark_signage_consolidate_e2e.py b/benchmark-scripts/benchmark_signage_consolidate_e2e.py index 34784ab..9ad13ce 100644 --- a/benchmark-scripts/benchmark_signage_consolidate_e2e.py +++ b/benchmark-scripts/benchmark_signage_consolidate_e2e.py @@ -1,8 +1,8 @@ -''' -* Copyright (C) 2026 Intel Corporation. -* -* SPDX-License-Identifier: Apache-2.0 -''' +# +# 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.""" diff --git a/benchmark-scripts/benchmark_signage_docker_stats.py b/benchmark-scripts/benchmark_signage_docker_stats.py index 85a9e15..826dcb4 100644 --- a/benchmark-scripts/benchmark_signage_docker_stats.py +++ b/benchmark-scripts/benchmark_signage_docker_stats.py @@ -1,8 +1,8 @@ -''' -* Copyright (C) 2026 Intel Corporation. -* -* SPDX-License-Identifier: Apache-2.0 -''' +# +# Apache v2 license +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 +# """Sample Docker CPU and memory usage for running Compose services.""" diff --git a/benchmark-scripts/benchmark_signage_pid.py b/benchmark-scripts/benchmark_signage_pid.py index ad59903..30a60d1 100644 --- a/benchmark-scripts/benchmark_signage_pid.py +++ b/benchmark-scripts/benchmark_signage_pid.py @@ -1,8 +1,8 @@ -''' -* Copyright (C) 2026 Intel Corporation. -* -* SPDX-License-Identifier: Apache-2.0 -''' +# +# Apache v2 license +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 +# """Sample the running PID pipeline through its MQTT output and Docker metrics.""" diff --git a/benchmark-scripts/benchmark_signage_qdrant.py b/benchmark-scripts/benchmark_signage_qdrant.py index 404507f..05ac9ee 100644 --- a/benchmark-scripts/benchmark_signage_qdrant.py +++ b/benchmark-scripts/benchmark_signage_qdrant.py @@ -1,8 +1,8 @@ -''' -* Copyright (C) 2026 Intel Corporation. -* -* SPDX-License-Identifier: Apache-2.0 -''' +# +# Apache v2 license +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 +# """Benchmark Qdrant collection scaling and ASe predefined-ad indexing.""" diff --git a/benchmark-scripts/benchmark_signage_web_ui.py b/benchmark-scripts/benchmark_signage_web_ui.py index 7730562..0823ce9 100644 --- a/benchmark-scripts/benchmark_signage_web_ui.py +++ b/benchmark-scripts/benchmark_signage_web_ui.py @@ -1,8 +1,8 @@ -''' -* Copyright (C) 2026 Intel Corporation. -* -* SPDX-License-Identifier: Apache-2.0 -''' +# +# Apache v2 license +# Copyright (C) 2026 Intel Corporation +# SPDX-License-Identifier: Apache-2.0 +# """Collect per-frame Web UI product selection timing from the running container."""