latency-summary.py

#!/usr/bin/env python3
"""Calculate latency statistics from a CSV column or one numeric value per line."""

import argparse
import csv
import json
import math
import statistics
import sys
from pathlib import Path


def percentile(values, fraction):
    if len(values) == 1:
        return values[0]
    position = (len(values) - 1) * fraction
    lower = math.floor(position)
    upper = math.ceil(position)
    if lower == upper:
        return values[lower]
    return values[lower] + (values[upper] - values[lower]) * (position - lower)


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("file", nargs="?", type=Path, help="CSV/text file; stdin when omitted")
    parser.add_argument("--column", help="CSV column name; otherwise expects one value per line")
    parser.add_argument("--delimiter", default=",")
    parser.add_argument("--json", action="store_true", dest="as_json")
    return parser.parse_args()


def main():
    args = parse_args()
    if len(args.delimiter) != 1:
        raise SystemExit("--delimiter must be one character")
    handle = args.file.open(newline="", encoding="utf-8-sig") if args.file else sys.stdin
    values = []
    skipped = 0
    try:
        source = csv.DictReader(handle, delimiter=args.delimiter) if args.column else ((line.strip(),) for line in handle)
        if args.column and (not source.fieldnames or args.column not in source.fieldnames):
            raise SystemExit(f"Column not found: {args.column}")
        for row in source:
            raw = row.get(args.column, "") if args.column else row[0]
            try:
                value = float(raw)
                if math.isfinite(value):
                    values.append(value)
                else:
                    skipped += 1
            except (TypeError, ValueError):
                skipped += 1
    finally:
        if args.file:
            handle.close()

    if not values:
        raise SystemExit("No numeric values found")
    values.sort()
    result = {
        "count": len(values), "skipped": skipped, "min": values[0],
        "mean": statistics.fmean(values), "p50": percentile(values, 0.50),
        "p90": percentile(values, 0.90), "p95": percentile(values, 0.95),
        "p99": percentile(values, 0.99), "max": values[-1],
    }
    if args.as_json:
        print(json.dumps(result, indent=2))
    else:
        for key, value in result.items():
            print(f"{key}: {value:.3f}" if isinstance(value, float) else f"{key}: {value}")


if __name__ == "__main__":
    main()