Skip to main content
Commit31bf16ed5a2e
15 lines · 547 BBlameRaw
import csv
import math

def summarize(values):
    values = sorted(float(value) for value in values)
    if not values:
        return {"count": 0, "mean": None, "p95": None}
    if any(not math.isfinite(value) or value < 0 for value in values):
        raise ValueError("Metrics must be finite and nonnegative")
    return {"count": len(values), "mean": sum(values) / len(values),
            "p95": values[math.ceil(len(values) * 0.95) - 1]}

def summarize_csv(handle):
    return summarize(row["latency_ms"] for row in csv.DictReader(handle))