#!/usr/bin/env python3 """Standalone NMD-VCell external candidate-evidence preflight validator.""" import argparse import csv import json import math import re from datetime import datetime, timezone from pathlib import Path SHA256 = re.compile(r"^[0-9a-f]{64}$") PLACEHOLDERS = {"", "pending", "tbd", "na", "n/a", "unknown", "replace_me"} CORE = [ "record_id", "gene", "event_date", "study_id", "record_status", "evidence_scope", "biological_context", "species", "model_system", "perturbation_type", "comparator", "endpoint_name", "endpoint_unit", "endpoint_direction", "result_class", "effect_measure", "effect_estimate", "ci_level", "ci_lower", "ci_upper", "n_hypotheses_tested", "n_units", "n_independent_replicates", "null_evidence_framework", "analysis_design", "determinism_class", "context_match_to_nmd_vcell", "independence_from_nmd_vcell", ] PROVENANCE = [ "data_reference", "data_sha256", "code_reference", "code_version", "code_sha256", "environment_reference", "source_reference", "source_reference_type", ] ENUMS = { "record_status": {"candidate_import"}, "result_class": {"no_material_signal", "supportive_signal", "mixed", "inconclusive"}, "null_evidence_framework": { "equivalence_test", "confidence_interval_against_predeclared_margin", "noninferiority_test", "absence_not_tested", }, "context_match_to_nmd_vcell": { "direct_same_endpoint_and_context", "partial_context_match", "cross_context_only", "unknown", }, "independence_from_nmd_vcell": { "independent_external_project", "partially_overlapping_inputs", "same_project_reanalysis", "unknown", }, } def blank(value): return str(value or "").strip().lower() in PLACEHOLDERS def number(row, field, errors, required=False): raw = str(row.get(field) or "").strip() if not raw and not required: return None try: value = float(raw) if not math.isfinite(value): raise ValueError return value except ValueError: errors.append(f"{field}: expected finite number") return None def check(row, row_number): errors = [f"{field}: required" for field in CORE if blank(row.get(field))] for field, allowed in ENUMS.items(): if row.get(field, "") not in allowed: errors.append(f"{field}: invalid value") estimate = number(row, "effect_estimate", errors, True) ci_level = number(row, "ci_level", errors, True) ci_lower = number(row, "ci_lower", errors, True) ci_upper = number(row, "ci_upper", errors, True) for field in ("n_hypotheses_tested", "n_units", "n_independent_replicates"): try: if int(row.get(field, "")) < 1: raise ValueError except ValueError: errors.append(f"{field}: expected integer >= 1") if ci_level is not None and not 0 < ci_level <= 1: errors.append("ci_level: expected proportion in (0, 1]") if None not in (ci_lower, estimate, ci_upper) and not ci_lower <= estimate <= ci_upper: errors.append("effect_estimate must lie inside its confidence interval") if errors: return {"row_number": row_number, "record_id": row.get("record_id", ""), "gene": row.get("gene", ""), "verdict": "SCHEMA_FAIL", "automatic_status_change": False, "errors": errors} gaps = [field for field in PROVENANCE if blank(row.get(field))] for field in ("data_sha256", "code_sha256"): if row.get(field) and not SHA256.fullmatch(row[field]): gaps.append(f"{field} (invalid SHA-256)") if row.get("author_attestation") != "true": gaps.append("author_attestation") if gaps: return {"row_number": row_number, "record_id": row["record_id"], "gene": row["gene"], "verdict": "PROVENANCE_PENDING", "rule_review": "ETR-02", "automatic_status_change": False, "blocking_gaps": gaps, "claim_ceiling": "status_update_only"} if (row["context_match_to_nmd_vcell"] != "direct_same_endpoint_and_context" or row["independence_from_nmd_vcell"] != "independent_external_project"): return {"row_number": row_number, "record_id": row["record_id"], "gene": row["gene"], "verdict": "CONTEXT_REVIEW_REQUIRED", "rule_review": "ETR-03", "automatic_status_change": False, "claim_ceiling": "context_conflict_only"} if row["result_class"] == "no_material_signal": lower = number(row, "negligible_effect_lower", errors) upper = number(row, "negligible_effect_upper", errors) established = ( row["null_evidence_framework"] != "absence_not_tested" and None not in (lower, upper, ci_lower, ci_upper) and lower < upper and ci_lower >= lower and ci_upper <= upper ) if not established: return {"row_number": row_number, "record_id": row["record_id"], "gene": row["gene"], "verdict": "NULL_EVIDENCE_NOT_ESTABLISHED", "automatic_status_change": False, "claim_ceiling": "auditable_result_without_negative_claim", "warnings": ["A nonsignificant P value alone is not evidence of absence."]} return {"row_number": row_number, "record_id": row["record_id"], "gene": row["gene"], "verdict": "ELIGIBLE_CONTEXT_SPECIFIC_NEGATIVE_EVIDENCE", "rule_review": "ETR-01", "automatic_status_change": False, "claim_ceiling": "negative_in_observed_context_only"} verdict = ("ELIGIBLE_SUPPORTIVE_EVIDENCE_FOR_MANUAL_REVIEW" if row["result_class"] == "supportive_signal" else "AUDITABLE_BUT_INCONCLUSIVE") return {"row_number": row_number, "record_id": row["record_id"], "gene": row["gene"], "verdict": verdict, "rule_review": "manual_governance_review", "automatic_status_change": False, "claim_ceiling": "observed_endpoint_and_context_only"} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", required=True, type=Path) parser.add_argument("--output", type=Path) args = parser.parse_args() with args.input.open("r", encoding="utf-8-sig", newline="") as handle: rows = list(csv.DictReader(handle, delimiter="\t")) records = [check(row, index + 2) for index, row in enumerate(rows)] blocking = {"SCHEMA_FAIL", "PROVENANCE_PENDING", "CONTEXT_REVIEW_REQUIRED", "NULL_EVIDENCE_NOT_ESTABLISHED"} report = { "contract_version": "1.0.0", "generated_at_utc": datetime.now(timezone.utc).isoformat(), "input_file": str(args.input), "row_count": len(rows), "automatic_status_change": False, "manual_governance_required": True, "blocking_record_count": sum(item["verdict"] in blocking for item in records), "records": records, } rendered = json.dumps(report, indent=2) + "\n" if args.output: args.output.write_text(rendered, encoding="utf-8") else: print(rendered, end="") return 3 if not rows else (2 if report["blocking_record_count"] else 0) if __name__ == "__main__": raise SystemExit(main())