Create ingest/latency.py

ingest/latency.py

@@ -0,0 +1,46 @@
+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