ingest/latency.py

import time

def measure_latency(adapter, query="test"):
    """
    Non-invasive latency probe.
    One request max, no retries.
    """
    start = time.time()
    try:
        adapter.search(query)
        elapsed = time.time() - start
        return round(elapsed, 2)
    except Exception:
        return None


def latency_badges(adapters):
    rows = []

    for a in adapters:
        if not a.is_live:
            rows.append({
                "Agency": a.name,
                "Latency (s)": "—",
                "Badge": "🔒 Stub"
            })
            continue

        latency = measure_latency(a)

        if latency is None:
            badge = "⚠️ Error"
        elif latency < 1.5:
            badge = "🟢 Fast"
        elif latency < 4:
            badge = "🟡 Moderate"
        else:
            badge = "🔴 Slow"

        rows.append({
            "Agency": a.name,
            "Latency (s)": latency,
            "Badge": badge
        })

    return rows