timmy-talking-turd/scripts/dataset_card.py

#!/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())