#!/usr/bin/env python3 """Offline teaching tool; no network, proxy credentials, or provider measurements. Run without arguments for synthetic data; --self-test checks arithmetic/invariants. Optional input: a JSON object with planned_jobs, deadline_ms, and attempts. Only use closed observation windows; valid is an externally checked content flag. """ import argparse import copy import json import math from pathlib import Path DEMO = { "planned_jobs": ["J%d" % i for i in range(1, 11)], "deadline_ms": 5000, "attempts": [ {"job_id": "J1", "no": 1, "status": 200, "valid": True, "elapsed_ms": 400}, {"job_id": "J2", "no": 1, "status": 200, "valid": True, "elapsed_ms": 800}, {"job_id": "J3", "no": 1, "status": 200, "valid": False, "elapsed_ms": 300}, {"job_id": "J4", "no": 1, "status": 503, "valid": False, "elapsed_ms": 1000}, {"job_id": "J4", "no": 2, "status": 200, "valid": True, "elapsed_ms": 3400}, {"job_id": "J5", "no": 1, "status": None, "valid": False, "elapsed_ms": 5000}, {"job_id": "J6", "no": 1, "status": 200, "valid": False, "elapsed_ms": 600}, {"job_id": "J6", "no": 2, "status": 200, "valid": True, "elapsed_ms": 6200}, {"job_id": "J7", "no": 1, "status": 200, "valid": True, "elapsed_ms": 2000}, {"job_id": "J8", "no": 1, "status": 200, "valid": True, "elapsed_ms": 4000}, {"job_id": "J9", "no": 1, "status": 407, "valid": False, "elapsed_ms": 200}, ], } def score(data): planned = data["planned_jobs"] if not isinstance(planned, list) or not planned or not all(isinstance(x, str) and x for x in planned): raise ValueError("planned_jobs must be a nonempty list of unique string IDs") if len(set(planned)) != len(planned): raise ValueError("duplicate planned job") deadline = data["deadline_ms"] if type(deadline) not in (int, float) or not math.isfinite(deadline) or deadline <= 0: raise ValueError("deadline_ms must be finite and positive") attempts = data["attempts"] if not isinstance(attempts, list): raise ValueError("attempts must be a list") groups = {j: [] for j in planned} seen = set() for a in attempts: if a["job_id"] not in groups: raise ValueError("attempt for unplanned job") n, elapsed, status = a["no"], a["elapsed_ms"], a["status"] if type(n) is not int or n < 1 or (a["job_id"], n) in seen: raise ValueError("invalid or duplicate attempt number") if type(elapsed) not in (int, float) or not math.isfinite(elapsed) or elapsed < 0: raise ValueError("elapsed_ms must be finite and nonnegative") if status is not None and (type(status) is not int or not 100 <= status <= 599): raise ValueError("status must be HTTP status or null") if type(a["valid"]) is not bool or (a["valid"] and (status is None or not 200 <= status < 300)): raise ValueError("valid requires a boolean and a successful HTTP response") seen.add((a["job_id"], n)); groups[a["job_id"]].append(a) attempted = [sorted(v, key=lambda a: a["no"]) for v in groups.values() if v] for group in attempted: if [a["no"] for a in group] != list(range(1, len(group) + 1)): raise ValueError("missing attempt numbers; reconcile logs first") if [a["elapsed_ms"] for a in group] != sorted(a["elapsed_ms"] for a in group): raise ValueError("example expects sequential attempts and cumulative time") finishes = sorted(min(a["elapsed_ms"] for a in g if a["valid"]) for g in attempted if any(a["valid"] for a in g)) timely = sum(t <= deadline for t in finishes) def metric(n, d): return {"numerator": n, "denominator": d, "percent": round(100*n/d, 2) if d else None} return { "planned_jobs": len(planned), "attempted_jobs": len(attempted), "attempts": len(attempts), "missing_job_ids": [j for j, v in groups.items() if not v], "coverage": metric(len(attempted), len(planned)), "http_200_per_attempt": metric(sum(a["status"] == 200 for a in attempts), len(attempts)), "valid_per_attempt": metric(sum(a["valid"] for a in attempts), len(attempts)), "first_attempt_valid_per_attempted_job": metric(sum(g[0]["valid"] for g in attempted), len(attempted)), "eventual_valid_per_attempted_job": metric(len(finishes), len(attempted)), "timely_valid_per_planned_job": metric(timely, len(planned)), "eventual_valid_per_planned_job": metric(len(finishes), len(planned)), "p95_successful_job_elapsed_ms_nearest_rank": finishes[math.ceil(.95*len(finishes))-1] if finishes else None, "successful_job_sample_size": len(finishes), } def self_test(): s = score(DEMO) assert (s["planned_jobs"], s["attempted_jobs"], s["attempts"]) == (10, 9, 11) expected = {"coverage":(9,10), "http_200_per_attempt":(8,11), "valid_per_attempt":(6,11), "first_attempt_valid_per_attempted_job":(4,9), "eventual_valid_per_attempted_job":(6,9), "timely_valid_per_planned_job":(5,10), "eventual_valid_per_planned_job":(6,10)} for k, (n,d) in expected.items(): assert s[k] == {"numerator":n,"denominator":d,"percent":round(100*n/d,2)} assert s["p95_successful_job_elapsed_ms_nearest_rank"] == 6200 empty = copy.deepcopy(DEMO); empty["attempts"] = [] assert score(empty)["http_200_per_attempt"]["percent"] is None assert score(empty)["timely_valid_per_planned_job"]["percent"] == 0 boundary = copy.deepcopy(DEMO); boundary["attempts"][7]["elapsed_ms"] = 5000 assert score(boundary)["timely_valid_per_planned_job"]["numerator"] == 6 for mutate in [lambda d:d["attempts"].append(d["attempts"][0]), lambda d:d["attempts"][0].update(job_id="UNKNOWN"), lambda d:d["attempts"][0].update(valid="false"), lambda d:d["attempts"][0].update(elapsed_ms=float('nan')), lambda d:d["attempts"].pop(3)]: bad=copy.deepcopy(DEMO); mutate(bad) try: score(bad) except ValueError: pass else: raise AssertionError("invalid input accepted") print("PASS: denominators, deadlines, missing jobs, empty attempts and invalid inputs") if __name__ == "__main__": parser=argparse.ArgumentParser(description=__doc__) parser.add_argument("input", nargs="?", help="optional closed-window JSON; default is synthetic DEMO") parser.add_argument("--self-test", action="store_true") args=parser.parse_args() if args.self_test: self_test() else: data=json.loads(Path(args.input).read_text()) if args.input else DEMO print(json.dumps({"data_kind":"user_supplied_unverified" if args.input else "synthetic_not_provider_measurement", "score":score(data)}, indent=2))