Add panel CSV schema and validator

schemas/panel_csv.schema.json

@@ -0,0 +1,14 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "title": "Panel CSV Schema",
+  "description": "Required and optional columns for candidate panel CSV files.",
+  "type": "object",
+  "required": ["candidate_id", "sequence"],
+  "optional": ["source", "ensemble", "activity", "safety", "synthesis", "novelty",
+               "charge_bias", "rich_selectivity", "selectivity_proxy", "hemolysis_risk",
+               "selected", "selection_reason", "family", "mechanism", "batch"],
+  "properties": {
+    "candidate_id": {"type": "string", "description": "Unique candidate identifier"},
+    "sequence": {"type": "string", "description": "Amino-acid sequence (standard single-letter codes)"}
+  }
+}

scripts/validate_panel_csv.py

@@ -0,0 +1,71 @@
+"""Validate panel CSV files against expected columns and content."""
+from __future__ import annotations
+
+import csv
+import json
+import sys
+from pathlib import Path
+
+
+REQUIRED = {"candidate_id", "sequence"}
+
+
+def validate_panel_csv(csv_path: str) -> dict:
+    p = Path(csv_path)
+    if not p.exists():
+        return {"error": f"Panel CSV not found: {csv_path}"}
+
+    with p.open("r", newline="") as f:
+        reader = csv.DictReader(f)
+        if not reader.fieldnames:
+            return {"error": "Empty or unreadable CSV"}
+        headers = set(reader.fieldnames)
+        rows = list(reader)
+
+    missing = REQUIRED - headers
+    if missing:
+        return {"error": f"Missing required columns: {missing}"}
+
+    row_errors = []
+    for i, row in enumerate(rows, 1):
+        errs = []
+        cid = row.get("candidate_id", "").strip()
+        seq = row.get("sequence", "").strip()
+        if not cid:
+            errs.append("Missing candidate_id")
+        if not seq:
+            errs.append("Missing sequence")
+        if seq and not all(aa in "ACDEFGHIKLMNPQRSTVWY" for aa in seq.upper()):
+            errs.append(f"Invalid amino acid in sequence: {seq[:20]}")
+        if errs:
+            row_errors.append({"row": i, "candidate_id": cid or "?", "errors": errs})
+
+    return {
+        "csv": csv_path,
+        "columns": sorted(headers),
+        "missing_required": sorted(missing),
+        "total_rows": len(rows),
+        "valid_rows": len(rows) - len(row_errors),
+        "row_errors": len(row_errors),
+        "per_row_errors": row_errors,
+        "all_valid": len(row_errors) == 0,
+    }
+
+
+def main() -> int:
+    import argparse
+    parser = argparse.ArgumentParser(description="Validate panel CSV")
+    parser.add_argument("--csv", required=True)
+    args = parser.parse_args()
+
+    result = validate_panel_csv(args.csv)
+    if "error" in result:
+        print(f"Error: {result['error']}", file=sys.stderr)
+        return 2
+
+    print(json.dumps(result, indent=2))
+    return 0 if result["all_valid"] else 3
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_validate_panel_csv.py

@@ -0,0 +1,69 @@
+"""Tests for panel CSV validator."""
+import csv
+import json
+import subprocess
+import sys
+from pathlib import Path
+import pytest
+
+from scripts.validate_panel_csv import validate_panel_csv
+
+
+def _make_csv(tmp, rows):
+    p = Path(tmp) / "panel.csv"
+    with p.open("w", newline="") as f:
+        if rows:
+            w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
+            w.writeheader()
+            w.writerows(rows)
+    return str(p)
+
+
+def test_valid_panel(tmp_path):
+    p = _make_csv(tmp_path, [{"candidate_id": "C1", "sequence": "ACDEFGHIKL"}])
+    r = validate_panel_csv(p)
+    assert r["all_valid"] is True
+
+
+def test_missing_required_column(tmp_path):
+    p = _make_csv(tmp_path, [{"candidate_id": "C1"}])
+    r = validate_panel_csv(p)
+    assert "error" in r or r["row_errors"] > 0
+
+
+def test_missing_candidate_id(tmp_path):
+    p = _make_csv(tmp_path, [{"candidate_id": "", "sequence": "AAAA"}])
+    r = validate_panel_csv(p)
+    assert r["row_errors"] > 0
+
+
+def test_invalid_amino_acid(tmp_path):
+    p = _make_csv(tmp_path, [{"candidate_id": "C1", "sequence": "ACB"}] )
+    r = validate_panel_csv(p)
+    assert r["row_errors"] > 0
+
+
+def test_missing_file():
+    r = validate_panel_csv("/nonexistent.csv")
+    assert "error" in r
+
+
+def test_empty_csv(tmp_path):
+    p = Path(tmp_path) / "empty.csv"
+    p.write_text("candidate_id,sequence\n")
+    r = validate_panel_csv(str(p))
+    assert r["total_rows"] == 0
+
+
+def test_cli_valid(tmp_path):
+    p = _make_csv(tmp_path, [{"candidate_id": "C1", "sequence": "ACDEFGHIKL"}])
+    res = subprocess.run([sys.executable, "scripts/validate_panel_csv.py", "--csv", p],
+                         capture_output=True, text=True)
+    assert res.returncode == 0
+
+
+def test_cli_invalid(tmp_path):
+    p = _make_csv(tmp_path, [{"candidate_id": "", "sequence": ""}])
+    res = subprocess.run([sys.executable, "scripts/validate_panel_csv.py", "--csv", p],
+                         capture_output=True, text=True)
+    assert res.returncode == 3