1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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))