scripts/validate_reshape.py

"""Validate a generated Stars pair against a known-good target MasterSheet.

Checkpoint-1 validation: the goal is *structural* correctness on the ITT
variables (raw timepoints + computed %Basal) and the derived metadata, not a
byte-exact reproduction of the full multi-file MasterSheet. So this compares
only the samples and variables present in BOTH files and reports:

  * sample overlap (how many generated samples are found in the target),
  * value agreement on shared (variable, sample) cells, within a tolerance,
  * metadata decode agreement (Genotype / Sex / Diet).

Usage:
    python scripts/validate_reshape.py <gen_data.csv> <gen_metadata.csv> \\
        [--target-data data_MasterSheet.csv] [--target-meta metadata_MasterSheet.csv] \\
        [--tol 0.5]

Exit code is 0 when no value/metadata mismatches are found among shared cells.
"""

import argparse
import sys

import pandas as pd

# ITT-relevant variable rows we care about for Checkpoint 1.
ITT_VAR_PREFIXES = ("T", "%Basal")


def _load_data(path: str) -> pd.DataFrame:
    """Load a Stars data matrix (index = Sample_ID / variable names, cols = samples)."""
    return pd.read_csv(path, index_col=0)


def _data_variables(index) -> list[str]:
    """All data-variable rows (everything except the Subject_title identifier row)."""
    return [str(v) for v in index if str(v) != "Subject_title"]


def validate(gen_data: str, gen_meta: str, tgt_data: str, tgt_meta: str, tol: float) -> int:
    gd = _load_data(gen_data)
    td = _load_data(tgt_data)
    gm = pd.read_csv(gen_meta)
    tm = pd.read_csv(tgt_meta)

    gen_samples = [c for c in gd.columns]
    shared_samples = [s for s in gen_samples if s in set(td.columns)]
    print(f"Generated samples : {len(gen_samples)}")
    print(f"Target samples    : {len(td.columns)}")
    print(f"Shared samples    : {len(shared_samples)}")
    missing = [s for s in gen_samples if s not in set(td.columns)]
    if missing:
        print(f"  generated samples NOT in target ({len(missing)}): {missing[:8]}"
              f"{' …' if len(missing) > 8 else ''}")

    variables = [v for v in _data_variables(gd.index) if v in set(td.index)]
    print(f"Shared data variables: {len(variables)} {variables}")

    # --- Value agreement on shared cells ---
    cells = 0
    mismatches = []
    for var in variables:
        for s in shared_samples:
            g = pd.to_numeric(gd.loc[var, s], errors="coerce")
            t = pd.to_numeric(td.loc[var, s], errors="coerce")
            if pd.isna(g) and pd.isna(t):
                continue
            cells += 1
            if pd.isna(g) != pd.isna(t) or (pd.notna(g) and abs(g - t) > tol):
                mismatches.append((var, s, g, t))

    print(f"\nValue cells compared: {cells}")
    print(f"Value mismatches    : {len(mismatches)} (tol={tol})")
    for var, s, g, t in mismatches[:10]:
        print(f"  {var:>12} / {s:<16} gen={g} target={t}")

    # --- Metadata decode agreement ---
    # Drop duplicate Sample_IDs in the target (the source has genuine collisions,
    # e.g. KO-M-C-11 appears twice) so the lookup returns a scalar, not a Series.
    tm_idx = tm.drop_duplicates(subset="Sample_ID").set_index("Sample_ID")
    meta_cols = [c for c in ("Genotype", "Sex", "Diet") if c in gm.columns and c in tm.columns]
    meta_checked = 0
    meta_mismatch = []
    for _, row in gm.iterrows():
        sid = row["Sample_ID"]
        if sid not in tm_idx.index:
            continue
        for c in meta_cols:
            meta_checked += 1
            if str(row[c]).strip() != str(tm_idx.loc[sid, c]).strip():
                meta_mismatch.append((sid, c, row[c], tm_idx.loc[sid, c]))

    print(f"\nMetadata cells compared: {meta_checked}")
    print(f"Metadata mismatches    : {len(meta_mismatch)}")
    for sid, c, g, t in meta_mismatch[:10]:
        print(f"  {sid:<16} {c}: gen={g} target={t}")

    ok = not mismatches and not meta_mismatch and len(shared_samples) > 0
    print(f"\n{'PASS' if ok else 'FAIL'}: "
          f"{len(shared_samples)} shared samples, "
          f"{cells - len(mismatches)}/{cells} value cells match, "
          f"{meta_checked - len(meta_mismatch)}/{meta_checked} metadata cells match")
    return 0 if ok else 1


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("gen_data", help="generated data_*.csv")
    ap.add_argument("gen_meta", help="generated metadata_*.csv")
    ap.add_argument("--target-data", required=True, help="target data_MasterSheet.csv")
    ap.add_argument("--target-meta", required=True, help="target metadata_MasterSheet.csv")
    ap.add_argument("--tol", type=float, default=0.5, help="numeric tolerance (default 0.5)")
    args = ap.parse_args()
    return validate(args.gen_data, args.gen_meta, args.target_data, args.target_meta, args.tol)


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