#!/usr/bin/env python3 """Generate the published dataset card from the append-only corpus manifest (issue #27). A dataset card that is typed by hand drifts from the corpus it describes, and a card that estimates its own numbers is worse than no card at all. This generator therefore only reports what the manifest actually contains: - counts of active samples, contributors, and withdrawals, - the contributor-level split policy, checked before a single line is rendered, - class balance over the active samples only, - every fact the pipeline does not collect (devices, contexts, licenses) printed as ``pending``. Nothing is inferred. A person's pseudonymous subject key never reaches the page, and a withdrawn sample contributes no label to the balance tables. The output is a pure function of the manifest, so the committed card can be regenerated and compared byte for byte. Usage: python3 scripts/dataset_card.py --corpus /srv/timmy-corpus python3 scripts/dataset_card.py --corpus /srv/timmy-corpus --out docs/DATASET_CARD.md python3 scripts/dataset_card.py --corpus /srv/timmy-corpus --check docs/DATASET_CARD.md python3 scripts/dataset_card.py --corpus /srv/timmy-corpus --json Exit codes: 0 the card rendered (and matched, with --check); 1 the card on disk is out of date; 2 the corpus cannot be described honestly and was refused. """ from __future__ import annotations import argparse import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from corpus_manifest import ( # noqa: E402 COLORS, QUALITIES, REVIEWERS, SAMPLE, SPLITS, TOMBSTONE, CorpusError, active_records, load_manifest, ) from calibration_report import expected_split # noqa: E402 # The facts no current pipeline step records. They are reported as pending rather than guessed. PENDING_FACTS = ("contexts", "devices", "licenses") PENDING_LABELS = { "devices": "Devices", "contexts": "Contexts", "licenses": "Licenses", } PENDING_REASONS = { "devices": "not collected by ingest — no camera, model, or OS fields are recorded", "contexts": "not collected by ingest — no setting, time-of-day, or location fields are recorded", "licenses": "no contributor grant is recorded yet, so no license is asserted", } class CardError(Exception): """The corpus cannot be described as asked.""" def assert_contributor_held_out(records: list[dict]) -> None: """Refuse a corpus where one contributor straddles splits or a split cannot be re-derived. The rule is the ingestion rule, so a corpus cannot quietly relabel its own splits: a split must follow from the pseudonymous subject key alone, and one person's photos must all land in one split. Otherwise every held-out number downstream is measuring a person the model already met. """ splits_by_subject: dict[str, set[str]] = {} for record in records: if record.get("recordType", SAMPLE) != SAMPLE: continue subject = record.get("subjectKey") if not isinstance(subject, str) or not subject: raise CardError("every sample must carry the pseudonymous subject key") split = record.get("split") if split not in SPLITS: raise CardError(f"split must be one of {', '.join(SPLITS)}: {split!r}") splits_by_subject.setdefault(subject, set()).add(split) for subject, splits in sorted(splits_by_subject.items()): if len(splits) > 1: raise CardError( f"contributor {subject} leaks across {', '.join(sorted(splits))}: " "one contributor's photos must stay in one split" ) for record in records: if record.get("recordType", SAMPLE) != SAMPLE: continue try: required = expected_split(record["subjectKey"]) except Exception as exc: # noqa: BLE001 - the rule reports its own refusal raise CardError(str(exc)) from exc if record["split"] != required: raise CardError( f"split {record['split']!r} for contributor {record['subjectKey']} is not reproducible: " f"the subject key derives {required!r}" ) def summarize(records: list[dict]) -> dict: """Count what the manifest holds. Pure function of the records, in a stable order.""" assert_contributor_held_out(records) active = active_records(records) # A withdrawal can appear as a tombstone or as a revoked sample record (both are legal history); # it is one withdrawn contribution either way, so count distinct contribution ids. withdrawn: dict[str, dict] = {} for record in records: is_tombstone = record.get("recordType") == TOMBSTONE is_revoked_sample = ( record.get("recordType", SAMPLE) == SAMPLE and record.get("lifecycle", {}).get("state") == "revoked" ) if not (is_tombstone or is_revoked_sample): continue contribution = record.get("contributionId") if not isinstance(contribution, str) or not contribution: raise CardError("a withdrawn record must carry the contribution id it revokes") lifecycle = record.get("lifecycle", {}) entry = withdrawn.setdefault( contribution, {"contributionId": contribution, "withdrawnAt": None, "reason": None, "artifacts": 0}, ) stamp = lifecycle.get("updatedAt") if stamp and (entry["withdrawnAt"] is None or stamp > entry["withdrawnAt"]): entry["withdrawnAt"] = stamp if lifecycle.get("reason"): entry["reason"] = lifecycle["reason"] if is_tombstone: entry["artifacts"] = len(record.get("deletedArtifacts") or []) by_split = {split: {"contributors": set(), "samples": 0} for split in SPLITS} for record in active: by_split[record["split"]]["contributors"].add(record["subjectKey"]) by_split[record["split"]]["samples"] += 1 bristol = {str(day): 0 for day in range(1, 8)} color = {name: 0 for name in COLORS} quality = {name: 0 for name in QUALITIES} review = {name: 0 for name in REVIEWERS} for record in active: labels = record.get("labels", {}) bristol[str(labels.get("bristolType"))] = bristol.get(str(labels.get("bristolType")), 0) + 1 if labels.get("color") in color: color[labels["color"]] += 1 if labels.get("quality") in quality: quality[labels["quality"]] += 1 status = record.get("review", {}).get("status") if status in review: review[status] += 1 stamps = [record.get("ingestedAt") for record in records if record.get("ingestedAt")] stamps += [entry["withdrawnAt"] for entry in withdrawn.values() if entry["withdrawnAt"]] withdrawals_detail = [ { "contributionId": entry["contributionId"], "withdrawnAt": entry["withdrawnAt"], "reason": entry["reason"], "artifacts": entry["artifacts"], } for entry in withdrawn.values() ] return { "activeSamples": len(active), "contributors": len({record["subjectKey"] for record in active}), "withdrawals": len(withdrawn), "bySplit": {split: {"contributors": len(entry["contributors"]), "samples": entry["samples"]} for split, entry in by_split.items()}, "bristol": bristol, "colors": color, "qualities": quality, "review": review, "lastRecordAt": max(stamps) if stamps else None, "withdrawalsDetail": sorted(withdrawals_detail, key=lambda entry: (entry["withdrawnAt"] or "", entry["contributionId"] or "")), "pending": sorted(PENDING_FACTS), } def _count_table(title: str, rows: list[tuple[str, int]]) -> list[str]: lines = [f"**{title}**", "", f"| {title} | Active samples |", "|---|---:|"] lines += [f"| {label} | {count} |" for label, count in rows] lines.append("") return lines def render_card(summary: dict, *, corpus: str) -> str: """Render the card. Every value here comes from the manifest, or is marked pending.""" present = summary["activeSamples"] > 0 or summary["withdrawals"] > 0 lines = [ "# Dataset card — Timmy consented stool-image corpus", "", "Generated by [`scripts/dataset_card.py`](../scripts/dataset_card.py) from the append-only", "corpus manifest. Do not edit this page by hand: regenerate it, or `--check` fails. Every count", "below is read back from `manifest.jsonl`; a fact the pipeline does not collect is printed as", "`pending` rather than estimated, so nothing here is invented.", "", "| Field | Value |", "|---|---|", f"| Corpus | `{corpus}` |", f"| Corpus present on this machine | {'yes' if present else 'no'} |", f"| Last manifest record | {summary['lastRecordAt'] or 'pending — no record yet'} |", f"| Active samples | {summary['activeSamples']} |", f"| Contributors | {summary['contributors']} |", f"| Withdrawals | {summary['withdrawals']} |", ] for fact in PENDING_FACTS: lines.append(f"| {PENDING_LABELS[fact]} | pending — {PENDING_REASONS[fact]} |") lines.append(f"| Regenerate | `python3 scripts/dataset_card.py --corpus {corpus}` |") lines += [ "", "## What this corpus is for", "", "Consented photos of stool, labelled by a human reviewer, for research on a stool-form", "classifier. The corpus may not be used to diagnose disease, to identify a person, or to", "train a model that speaks to a patient with clinical authority. A model's output stays", "provisional until the person logging confirms it: see [PRODUCT-DECISIONS.md](PRODUCT-DECISIONS.md).", "", ] if summary["activeSamples"] == 0: lines += [ "This corpus holds no consented images yet. Every count of zero below is a real count of", "an empty manifest, not a placeholder, and the pending facts stay pending until the", "pipeline collects them.", "", ] lines += [ "## Contributor-level split policy", "", "Splits are derived from the salted pseudonymous subject key, never assigned per photo: the", "first byte of the key decides the split, `< 205` train, `< 230` validation, otherwise test", "(the ingestion rule in `scripts/ingest_training_photo.py`). One contributor's photos all land", "in one split, and this generator refuses a manifest where a subject key appears in two splits", "or where a recorded split is not the one its subject key derives. Both edges are stated here", "so the policy on this page is the policy in the code.", "", "| Split | Contributors | Active samples |", "|---|---:|---:|", ] for split in SPLITS: entry = summary["bySplit"][split] lines.append(f"| `{split}` | {entry['contributors']} | {entry['samples']} |") lines += ["", "## Class balance (active samples only)", ""] lines += _count_table("Type", [(f"Type {day}", summary["bristol"][str(day)]) for day in range(1, 8)]) lines += _count_table("Color", [(f"`{name}`", count) for name, count in summary["colors"].items()]) lines += _count_table("Quality", [(f"`{name}`", count) for name, count in summary["qualities"].items()]) lines += _count_table("Review status", [(f"`{name}`", count) for name, count in summary["review"].items()]) lines += [ "A withdrawn sample is excluded from every table above. Labels are what a reviewer saw, not a", "diagnosis: [LABELING_GUIDE.md](LABELING_GUIDE.md) says what a label may and may not assert.", "", "## Exclusions and withdrawals", "", "A withdrawal appends a tombstone, deletes the derived image and any raw copy, and stays in the", "history. Contribution ids and reasons are published here because a deletion receipt is", "checkable; subject keys and labels are never published.", "", ] if summary["withdrawalsDetail"]: lines += [ "| Contribution | Withdrawn at | Reason | Artifacts removed |", "|---|---|---|---:|", ] for entry in summary["withdrawalsDetail"]: lines.append( f"| `{entry['contributionId']}` | {entry['withdrawnAt'] or 'pending'} | " f"{entry['reason'] or 'pending'} | {entry['artifacts']} |" ) else: lines.append("No withdrawals are recorded.") lines += [ "", "## Limitations", "", "- Composition (devices, contexts, licenses) is pending, so the corpus is not yet a", " representative sample of anything and must not be described as one.", "- Review status is a human decision, not a clinical gold standard; `user-confirmed` means the", " person who took the photo agreed with the proposal.", "- A held-out split measures generalization across contributors, not across a population.", "", ] return "\n".join(lines) def load_corpus(corpus: Path) -> list[dict]: try: return load_manifest(corpus / "manifest.jsonl") except (CorpusError, json.JSONDecodeError) as exc: raise CardError(f"the manifest at {corpus / 'manifest.jsonl'} cannot be read: {exc}") from exc def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Generate the Timmy dataset card from the corpus manifest.") parser.add_argument("--corpus", required=True, help="dataset root holding manifest.jsonl") parser.add_argument("--out", help="write the card here instead of stdout") parser.add_argument("--check", help="fail if the card at this path is not what would be generated") parser.add_argument("--json", action="store_true", help="print the summary object instead of the card") parser.add_argument("--corpus-label", help="path to name on the card (defaults to --corpus)") args = parser.parse_args(argv) corpus = Path(args.corpus) try: records = load_corpus(corpus) summary = summarize(records) except CardError as exc: print(f"refusing to render a dataset card: {exc}", file=sys.stderr) return 2 if args.json: print(json.dumps(summary, indent=2, sort_keys=True)) return 0 card = render_card(summary, corpus=args.corpus_label or args.corpus) if args.check: target = Path(args.check) existing = target.read_text(encoding="utf-8") if target.exists() else "" if existing != card: print( f"{target} is out of date: regenerate it with " f"`python3 scripts/dataset_card.py --corpus {args.corpus_label or args.corpus} --out {target}`", file=sys.stderr, ) return 1 print(f"{target} matches the corpus manifest") return 0 if args.out: target = Path(args.out) target.parent.mkdir(parents=True, exist_ok=True) target.write_text(card, encoding="utf-8") print(f"wrote {target} ({summary['activeSamples']} active samples, {summary['contributors']} contributors)") return 0 sys.stdout.write(card) return 0 if __name__ == "__main__": sys.exit(main())