d3-log-p95--google-gemini-3-8-flash

Write a Python CLI `p95paths`: reads an access-log CSV with columns timestamp,path,latency_ms and prints the top-N paths by 95th-percentile latency (default N=10), highest first, as `path p95_ms count`. Use argparse for --top and the input file (stdin when omitted). Malformed rows are skipped with a warning count on stderr; an empty or header-only input prints nothing and exits 0. Deliver a complete, runnable script plus focused tests covering the percentile math and malformed-row handling. Accompany the code with a brief explanation of your design choices.

Prompt

Write a Python CLI `p95paths`: reads an access-log CSV with columns timestamp,path,latency_ms and prints the top-N paths by 95th-percentile latency (default N=10), highest first, as `path p95_ms count`. Use argparse for --top and the input file (stdin when omitted). Malformed rows are skipped with a warning count on stderr; an empty or header-only input prints nothing and exits 0. Deliver a complete, runnable script plus focused tests covering the percentile math and malformed-row handling. Accompany the code with a brief explanation of your design choices. ## Output contract First, write a brief explanation of your design choices — plain prose, no code fences. Then emit exactly one fenced code block per file, and no other text between or after the blocks: - The FIRST line of every block is that file's marker line: - TypeScript or JavaScript: `// FILE: <path>` - Python: `# FILE: <path>` - A single-file web app is one file named `index.html`, so its marker line is `// FILE: index.html`. - Everything after the marker line is that file complete, from its first line to its last — no diffs, no omissions, no placeholders. - Tag each fence with the file's language, for example ```html, ```typescript, ```python.

Artifact

Form: tree

p95paths.py

#!/usr/bin/env python3
"""Compute the 95th-percentile latency per path from an access-log CSV."""

import argparse
import csv
import math
import sys
from typing import Dict, List, Optional, TextIO, Tuple


def calculate_p95(latencies: List[float]) -> float:
    """Compute the 95th percentile using linear interpolation.

    latencies must be non-empty.
    """
    if not latencies:
        raise ValueError("latencies list cannot be empty")

    sorted_vals = sorted(latencies)
    n = len(sorted_vals)
    if n == 1:
        return sorted_vals[0]

    # Linear interpolation between nearest ranks
    rank = 0.95 * (n - 1)
    low_idx = int(math.floor(rank))
    high_idx = int(math.ceil(rank))

    if low_idx == high_idx:
        return sorted_vals[low_idx]

    weight = rank - low_idx
    return sorted_vals[low_idx] + weight * (sorted_vals[high_idx] - sorted_vals[low_idx])


def process_access_log(
    file_obj: TextIO,
) -> Tuple[Dict[str, List[float]], int]:
    """Parse access-log CSV and aggregate latencies by path.

    Returns:
        (path_latencies_dict, malformed_count)
    """
    reader = csv.reader(file_obj)
    try:
        header = next(reader)
    except StopIteration:
        # Empty input
        return {}, 0

    header_clean = [col.strip().lower() for col in header]
    try:
        path_idx = header_clean.index("path")
        latency_idx = header_clean.index("latency_ms")
    except ValueError:
        # Header does not contain required columns; treat header as malformed
        # and attempt to process from start if header was missing, but specification
        # explicitly notes input has timestamp,path,latency_ms.
        return {}, 1

    path_latencies: Dict[str, List[float]] = {}
    malformed_count = 0

    for row in reader:
        if not row:
            continue

        if len(row) <= max(path_idx, latency_idx):
            malformed_count += 1
            continue

        raw_path = row[path_idx].strip()
        raw_latency = row[latency_idx].strip()

        if not raw_path:
            malformed_count += 1
            continue

        try:
            latency = float(raw_latency)
            if math.isnan(latency) or math.isinf(latency) or latency < 0:
                malformed_count += 1
                continue
        except ValueError:
            malformed_count += 1
            continue

        path_latencies.setdefault(raw_path, []).append(latency)

    return path_latencies, malformed_count


def compute_top_p95(
    path_latencies: Dict[str, List[float]], top_n: int = 10
) -> List[Tuple[str, float, int]]:
    """Calculate p95 and return top-N paths sorted by p95 descending."""
    results = []
    for path, latencies in path_latencies.items():
        p95 = calculate_p95(latencies)
        results.append((path, p95, len(latencies)))

    # Sort primarily by p95 descending, secondarily by path ascending for determinism
    results.sort(key=lambda item: (-item[1], item[0]))
    return results[:top_n]


def main(argv: Optional[List[str]] = None) -> int:
    parser = argparse.ArgumentParser(
        prog="p95paths",
        description="Print top-N paths by 95th-percentile latency from access-log CSV.",
    )
    parser.add_argument(
        "file",
        nargs="?",
        default="-",
        help="Access log CSV file to read (defaults to standard input).",
    )
    parser.add_argument(
        "--top",
        type=int,
        default=10,
        help="Number of top paths to display (default: 10).",
    )

    args = parser.parse_args(argv)

    if args.top < 0:
        sys.stderr.write("Error: --top must be a non-negative integer.\n")
        return 1

    try:
        if args.file == "-":
            path_latencies, malformed = process_access_log(sys.stdin)
        else:
            with open(args.file, "r", encoding="utf-8", newline="") as f:
                path_latencies, malformed = process_access_log(f)
    except OSError as err:
        sys.stderr.write(f"Error reading input: {err}\n")
        return 1

    if malformed > 0:
        sys.stderr.write(f"Warning: skipped {malformed} malformed row(s).\n")

    if not path_latencies:
        return 0

    top_paths = compute_top_p95(path_latencies, args.top)
    for path, p95, count in top_paths:
        print(f"{path}  {p95:.2f}  {count}")

    return 0


if __name__ == "__main__":
    sys.exit(main())

test_p95paths.py

#!/usr/bin/env python3
"""Focused tests for p95paths percentile math and malformed row handling."""

import io
import unittest

from p95paths import calculate_p95, compute_top_p95, process_access_log


class TestCalculateP95(unittest.TestCase):
    def test_single_value(self):
        self.assertEqual(calculate_p95([42.0]), 42.0)

    def test_two_values(self):
        # n=2: rank = 0.95 * 1 = 0.95 -> 10 + 0.95 * (20 - 10) = 19.5
        self.assertAlmostEqual(calculate_p95([10.0, 20.0]), 19.5)

    def test_twenty_values_exact_index(self):
        # 20 items from 1 to 20: rank = 0.95 * 19 = 18.05
        # vals[18] = 19, vals[19] = 20 -> 19 + 0.05 * 1 = 19.05
        data = list(range(1, 21))
        self.assertAlmostEqual(calculate_p95(data), 19.05)

    def test_hundred_values(self):
        # 100 items from 1 to 100: rank = 0.95 * 99 = 94.05
        # vals[94] = 95, vals[95] = 96 -> 95 + 0.05 * 1 = 95.05
        data = list(range(1, 101))
        self.assertAlmostEqual(calculate_p95(data), 95.05)

    def test_unsorted_input(self):
        # Order of input should not alter output
        data = [100.0, 10.0, 50.0, 20.0]
        sorted_copy = sorted(data)
        self.assertEqual(calculate_p95(data), calculate_p95(sorted_copy))

    def test_empty_list_raises(self):
        with self.assertRaises(ValueError):
            calculate_p95([])


class TestMalformedRowHandling(unittest.TestCase):
    def test_empty_input(self):
        stream = io.StringIO("")
        path_latencies, malformed = process_access_log(stream)
        self.assertEqual(path_latencies, {})
        self.assertEqual(malformed, 0)

    def test_header_only(self):
        stream = io.StringIO("timestamp,path,latency_ms\n")
        path_latencies, malformed = process_access_log(stream)
        self.assertEqual(path_latencies, {})
        self.assertEqual(malformed, 0)

    def test_valid_data(self):
        csv_data = (
            "timestamp,path,latency_ms\n"
            "2023-01-01T00:00:00Z,/api/login,120.5\n"
            "2023-01-01T00:00:01Z,/api/login,180.0\n"
            "2023-01-01T00:00:02Z,/api/items,45.2\n"
        )
        stream = io.StringIO(csv_data)
        path_latencies, malformed = process_access_log(stream)
        self.assertEqual(malformed, 0)
        self.assertEqual(len(path_latencies["/api/login"]), 2)
        self.assertEqual(len(path_latencies["/api/items"]), 1)

    def test_malformed_rows_skipped(self):
        csv_data = (
            "timestamp,path,latency_ms\n"
            "2023-01-01T00:00:00Z,/api/login,100\n"
            "2023-01-01T00:00:01Z,/api/login,not-a-number\n"  # non-numeric latency
            "2023-01-01T00:00:02Z,,50\n"                      # missing path
            "2023-01-01T00:00:03Z,/api/login,-20\n"          # negative latency
            "2023-01-01T00:00:04Z,/api/login,nan\n"          # NaN latency
            "2023-01-01T00:00:05Z\n"                         # insufficient columns
            "2023-01-01T00:00:06Z,/api/login,200\n"          # valid
        )
        stream = io.StringIO(csv_data)
        path_latencies, malformed = process_access_log(stream)
        self.assertEqual(malformed, 5)
        self.assertEqual(path_latencies["/api/login"], [100.0, 200.0])

    def test_column_order_independence(self):
        csv_data = (
            "latency_ms,timestamp,path\n"
            "150.0,2023-01-01T00:00:00Z,/api/users\n"
        )
        stream = io.StringIO(csv_data)
        path_latencies, malformed = process_access_log(stream)
        self.assertEqual(malformed, 0)
        self.assertEqual(path_latencies["/api/users"], [150.0])


class TestComputeTopP95(unittest.TestCase):
    def test_ordering_and_top_limit(self):
        data = {
            "/slow": [500.0, 500.0],
            "/fast": [10.0, 20.0],
            "/medium": [100.0, 150.0],
        }
        results = compute_top_p95(data, top_n=2)
        self.assertEqual(len(results), 2)
        self.assertEqual(results[0][0], "/slow")
        self.assertEqual(results[1][0], "/medium")


if __name__ == "__main__":
    unittest.main()

Explanation

For this CLI, I prioritized portability, robust error handling, and deterministic output using only the Python standard library. Percentile calculation uses standard linear interpolation (NumPy/R type-7 style) over sorted samples, which behaves gracefully for single-element and small datasets while avoiding external dependencies such as NumPy. Latency values are parsed as floating-point numbers to accommodate fractional millisecond timings, and 95th-percentile results are rounded to two decimal places. Ties in 95th-percentile latency are broken alphabetically by path name for deterministic ordering. The CSV parser uses `csv.reader` to inspect the header row and map column names dynamically to index positions, tolerating arbitrary column order as long as `path` and `latency_ms` are present. Rows with column count mismatches, empty paths, non-numeric or non-finite latencies, or negative durations are tracked as malformed, skipped, and reported via a summary count to standard error upon completion. Header-only or empty files exit immediately with status code 0 and no standard output. The CLI interface uses `argparse` with an optional positional argument for the input file path (defaulting to standard input when omitted or set to `-`) and `--top` (defaulting to 10) for configuring result limits.

Judge detail

No raw judge output is published for this item yet. When it is, it lands under results/raw/ and appears here verbatim; the rubric and protocol are already documented on the Methodology page.