diff --git a/schemas/panel_csv.schema.json b/schemas/panel_csv.schema.json new file mode 100644 index 00000000..8c4c5a11 --- /dev/null +++ b/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)"} + } +} diff --git a/scripts/validate_panel_csv.py b/scripts/validate_panel_csv.py new file mode 100644 index 00000000..386e96e2 --- /dev/null +++ b/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()) diff --git a/tests/test_validate_panel_csv.py b/tests/test_validate_panel_csv.py new file mode 100644 index 00000000..ccc98044 --- /dev/null +++ b/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