J01: Project 1: a tested telemetry analysis command

Prerequisites and observable goals#

Level: First integrated project. One assigned project, not a collection of separately counted milestones.

Purpose and boundary#

Build a local tool that reads synthetic latency records from CSV and emits one JSON summary. This resembles analyzing an engine subsystem or service endpoint, but uses no customer data. The goal is trustworthy data handling and a reproducible interface, not a dashboard, cloud pipeline, or sophisticated statistical model.

Each row has exactly three columns in this order: timestamp_s, latency_ms, and status. The two numbers must be finite and nonnegative; status must be ok or error. An empty dataset is invalid. Timestamps identify sample times but are not used to weight observations or calculate throughput. The tool does not silently discard error-status rows: all valid rows contribute latency measurements, and the error fraction is reported separately.

The output object contains count, unit, mean, median, p95 under the nearest-rank convention, and error fraction. With sorted values $x_{(1)}\le\cdots\le x_{(n)}$, nearest-rank p95 is $x_{(\lceil0.95n\rceil)}$. This explicit definition prevents disagreements caused by different quantile conventions from masquerading as code defects.

Functional requirements and nonrequirements#

Successful execution prints valid JSON to standard output and exits zero. Invalid files, malformed records, or invalid numeric domains print a useful diagnostic to standard error and exit with status 2. Row-level errors identify the CSV record’s reported line number under this reader; this is not a full byte-offset parser. The input fixture uses ordinary single-line records. No partial successful summary is emitted after an invalid row.

This bounded implementation loads valid records into memory, so memory usage is $O(n)$. It is not a streaming approximate-quantile tool, does not authenticate uploads, and does not accept unbounded remote data. It neither infers causation nor supplies confidence intervals from a tiny fixture. Those limitations belong in the project report.

Milestones and interfaces#

First implement a pure summary function on already validated sample objects and test small known lists. Next implement strict CSV parsing into an immutable record representation. Then implement command-line argument handling and separate output channels. Finally run integration tests in a temporary directory and document the schema, quantile convention, commands, and limits.

A useful architecture is path -> read_samples -> list[Sample] -> summarize -> JSON. Parsing owns file and format errors. Summarization owns mathematical definitions. The CLI owns user-facing messages and exit status. Avoid embedding printing inside the pure computation function, because that makes reuse and unit testing unnecessarily difficult.

Acceptance criteria and debugging guide#

The supplied fixture contains ten observations and must report count 10, mean 15 ms, median 12 ms, nearest-rank p95 40 ms, and error fraction 0.1. Tests must cover zero latency, an invalid header, missing or extra fields, NaN, negative latency, unknown status, and empty data. Run the actual command in an integration test, not only the underlying function.

When a result differs, inspect the parsed records and sorted values before changing formulas. A wrong p95 may reflect a rank convention; a wrong mean may reflect accidental filtering; a wrong count may reflect a malformed row silently skipped. Keep invalid input handling deterministic so users do not mistake a partial report for a complete one.

Complete solutions: J01#

Compare the first step where your work differs. A corrected answer is not yet an independent success.

Open every solution

Complete reference implementation#

The helper nearest_rank is printed in C14 and included in the companion. The rest of the complete project is:

"""J01: synthetic latency CSV -> validated, reproducible JSON report."""
import argparse
import csv
import json
import math
import statistics
import sys
from dataclasses import dataclass
from pathlib import Path
from .testing_lab import nearest_rank

@dataclass(frozen=True)
class Sample:
    timestamp_s: float
    latency_ms: float
    status: str

def read_samples(path: Path) -> list[Sample]:
    out = []
    with path.open(newline="", encoding="utf-8") as stream:
        reader = csv.DictReader(stream)
        expected = ["timestamp_s", "latency_ms", "status"]
        if reader.fieldnames != expected:
            raise ValueError("header must be timestamp_s,latency_ms,status")
        for line, row in enumerate(reader, 2):
            try:
                if None in row or any(v is None for v in row.values()):
                    raise ValueError("wrong number of fields")
                timestamp, latency = float(row["timestamp_s"]), float(row["latency_ms"])
                if not all(math.isfinite(x) and x >= 0 for x in (timestamp, latency)):
                    raise ValueError("finite nonnegative numbers required")
                if row["status"] not in {"ok", "error"}:
                    raise ValueError("status must be ok or error")
                out.append(Sample(timestamp, latency, row["status"]))
            except (TypeError, ValueError) as error:
                raise ValueError(f"line {line}: {error}") from error
    if not out:
        raise ValueError("at least one sample is required")
    return out

def summarize(samples: list[Sample]) -> dict:
    if not samples:
        raise ValueError("cannot summarize an empty sequence")
    values = [s.latency_ms for s in samples]
    if any(not math.isfinite(x) or x < 0 for x in values):
        raise ValueError("invalid latency")
    if any(s.status not in {"ok", "error"} for s in samples):
        raise ValueError("invalid status")
    return {"count": len(samples), "unit": "ms",
            "mean": statistics.mean(values), "median": statistics.median(values),
            "p95_nearest_rank": nearest_rank(values, .95),
            "error_fraction": sum(s.status == "error" for s in samples)/len(samples)}

def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("csv", type=Path)
    args = parser.parse_args(argv)
    try:
        report = summarize(read_samples(args.csv))
    except (OSError, ValueError, csv.Error) as error:
        print(str(error), file=sys.stderr)
        return 2
    print(json.dumps(report, sort_keys=True, indent=2, allow_nan=False))
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

Run and verify#

From the companion root:

python -m solutions.telemetry_cli fixtures/latency.csv
python -m unittest discover -s tests -v

The expected successful report contains these values (JSON key order is not a semantic requirement):

{
  "count": 10, "unit": "ms", "mean": 15.0,
  "median": 12.0, "p95_nearest_rank": 40.0,
  "error_fraction": 0.1
}

The reference tests exercised the function and CLI, including malformed data. They used synthetic fixtures and temporary files. They did not test multi-gigabyte inputs, remote uploads, every CSV dialect, or spreadsheet import behavior.

Correctness, complexity, and alternatives#

Parsing either yields one valid sample for every input row or raises before a summary is printed. Thus successful count corresponds to the complete validated input, not a silently filtered subset. The summary functions then apply their stated definitions to that same population. Sorting for exact quantiles costs $O(n\log n)$ time with $O(n)$ storage in this reference. A streaming mean uses $O(1)$ extra state, but exact arbitrary quantiles generally require retaining or externally processing more information. Approximate streaming quantiles are a different specification and need error guarantees.

One alternative is to reject error-status rows from the latency population, but that changes the question being answered. It can make a failing service appear faster by excluding slow failed requests. Another is to summarize each status separately, clearly reporting group counts. That is a reasonable unassigned extension, not a correction required by this project’s contract.

Evaluation rubric#

Score setup, reasoning, implementation, verification, and communication separately from 0 to 3. A score of 3 in setup means the schema and population are explicit; in reasoning, statistical conventions and complexity are explained; in implementation, all valid and invalid paths match the contract; in verification, independent expected values and CLI failures are tested; in communication, a fresh reader can reproduce the result and see limitations. Passing tests with an unexplained p95 convention is not independent operational competence.

Unassigned extensions#

A grouped report by subsystem, a bounded streaming mode, or a comparison of paired benchmark runs can deepen the project. Each needs a revised schema and acceptance tests. Do not count these suggestions as completed projects or silently replace the original exact quantile with an approximation.

Retention and readiness#

Return to C14 when revising the output contract and A04 before drawing statistical conclusions. Readiness means explaining every output field and reproducing both a successful run and an intentional failure without consulting the solution.