Skills › Scientific Agent Skills › bulk-rnaseq › validate_samplesheet.py
validate_samplesheet.py
249 lines · skills/bulk-rnaseq/scripts/validate_samplesheet.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate a bulk RNA-seq samplesheet (and optional design metadata).
4Catches the cheap-to-miss, expensive-to-debug problems before you spend compute:
5missing/duplicate FASTQs, inconsistent paired/single-end rows, invalid
6strandedness, and (with --metadata) too few replicates or batch fully confounded
7with condition.
9Checks an nf-core/rnaseq-style samplesheet:
11 sample,fastq_1,fastq_2,strandedness
13Exit code 0 if no errors (warnings allowed), 1 if any error is found.
15Examples
16--------
17python validate_samplesheet.py --samplesheet samplesheet.csv
18python validate_samplesheet.py --samplesheet samplesheet.csv \
19 --metadata metadata.csv --condition-col condition
20python validate_samplesheet.py --samplesheet sheet.csv --no-check-files # skip local file existence
22Requires: pandas.
23"""
24from __future__ import annotations
26import argparse
27import sys
28from pathlib import Path
30import pandas as pd
32from _tabular import validate_header
34VALID_STRANDEDNESS = {"auto", "forward", "reverse", "unstranded"}
35REMOTE_PREFIXES = ("http://", "https://", "ftp://", "s3://", "gs://", "az://")
38class Report:
39 """Collects errors (fatal) and warnings (advisory)."""
41 def __init__(self) -> None:
42 self.errors: list[str] = []
43 self.warnings: list[str] = []
45 def error(self, msg: str) -> None:
46 self.errors.append(msg)
48 def warn(self, msg: str) -> None:
49 self.warnings.append(msg)
51 def summarize(self) -> int:
52 for w in self.warnings:
53 print(f" WARN {w}")
54 for e in self.errors:
55 print(f" ERROR {e}")
56 print()
57 if self.errors:
58 print(f"FAILED: {len(self.errors)} error(s), {len(self.warnings)} warning(s)")
59 return 1
60 print(f"PASSED: 0 errors, {len(self.warnings)} warning(s)")
61 return 0
64def _is_remote(path: str) -> bool:
65 return str(path).startswith(REMOTE_PREFIXES)
68def validate_samplesheet(path: Path, check_files: bool, rep: Report,
69 nfcore: bool = False) -> pd.DataFrame | None:
70 if not path.is_file():
71 rep.error(f"samplesheet not found: {path}")
72 return None
74 try:
75 validate_header(path)
76 df = pd.read_csv(path, dtype=str, keep_default_na=False)
77 except ValueError as exc:
78 rep.error(f"invalid samplesheet CSV: {exc}")
79 return None
80 df.columns = [c.strip() for c in df.columns]
81 for col in df.columns:
82 df[col] = df[col].str.strip()
84 if "sample" not in df.columns or "fastq_1" not in df.columns:
85 rep.error("samplesheet must have at least 'sample' and 'fastq_1' columns")
86 return None
87 has_r2 = "fastq_2" in df.columns
88 has_strand = "strandedness" in df.columns
89 if df.empty:
90 rep.error("samplesheet has no sample rows")
91 if nfcore and list(df.columns[:4]) != ["sample", "fastq_1", "fastq_2", "strandedness"]:
92 rep.error("nf-core requires the first four columns: sample,fastq_1,fastq_2,strandedness")
93 if not has_strand:
94 rep.warn("no 'strandedness' column; nf-core/rnaseq requires one (use 'auto')")
96 seen_files: dict[str, int] = {}
97 for i, row in df.iterrows():
98 ln = i + 2 # +1 for header, +1 for 1-based
99 sample, r1 = row["sample"], row["fastq_1"]
100 r2 = row["fastq_2"] if has_r2 else ""
102 if not sample:
103 rep.error(f"row {ln}: empty 'sample'")
104 if not r1:
105 rep.error(f"row {ln}: empty 'fastq_1'")
106 if r1 and r2 and r1 == r2:
107 rep.error(f"row {ln} ({sample}): fastq_1 and fastq_2 are the same file")
108 for fp in (r1, r2):
109 if fp:
110 identity = fp if _is_remote(fp) else str(Path(fp).resolve())
111 seen_files[identity] = seen_files.get(identity, 0) + 1
112 if nfcore and not fp.endswith((".fastq.gz", ".fq.gz")):
113 rep.error(f"row {ln}: nf-core FASTQ must end in .fastq.gz or .fq.gz: {fp}")
115 if has_strand:
116 s = row["strandedness"].lower()
117 if s and s not in VALID_STRANDEDNESS:
118 rep.error(f"row {ln} ({sample}): strandedness '{s}' not in {sorted(VALID_STRANDEDNESS)}")
119 if nfcore and row["strandedness"] not in VALID_STRANDEDNESS:
120 rep.error(f"row {ln}: nf-core strandedness must be non-empty and lowercase")
122 if check_files:
123 for label, fp in (("fastq_1", r1), ("fastq_2", r2)):
124 if not fp:
125 continue
126 if _is_remote(fp):
127 rep.warn(f"row {ln} ({sample}): {label} is remote; existence not checked")
128 elif not Path(fp).is_file():
129 rep.error(f"row {ln} ({sample}): {label} not found: {fp}")
130 elif not fp.endswith((".fastq.gz", ".fq.gz", ".fastq", ".fq")):
131 rep.warn(f"row {ln} ({sample}): {label} has an unusual extension: {fp}")
133 for fp, n in seen_files.items():
134 if n > 1:
135 rep.error(f"FASTQ appears {n} times (each library file must be unique across both mates): {fp}")
137 if has_strand:
138 for sample, grp in df.groupby("sample"):
139 if grp["strandedness"].str.lower().nunique() > 1:
140 rep.error(f"sample '{sample}' has inconsistent strandedness across lanes")
141 normalized = df["sample"].str.replace(" ", "_", regex=False)
142 if df.assign(normalized=normalized).groupby("normalized")["sample"].nunique().gt(1).any():
143 rep.error("sample IDs collide after nf-core replaces spaces with underscores")
145 # Per-sample paired/single-end consistency (same sample over lanes is allowed in nf-core).
146 if has_r2:
147 for sample, grp in df.groupby("sample"):
148 layouts = {"paired" if r2 else "single" for r2 in grp["fastq_2"]}
149 if len(layouts) > 1:
150 rep.error(f"sample '{sample}' mixes paired-end and single-end rows")
151 dup_samples = df["sample"][df["sample"].duplicated()].unique()
152 if len(dup_samples):
153 rep.warn(f"samples appear on multiple rows (will be lane-merged): {list(dup_samples)}")
155 print(f"samplesheet: {df['sample'].nunique()} unique sample(s) across {len(df)} row(s)")
156 return df
159def validate_metadata(meta_path: Path, sheet: pd.DataFrame | None,
160 condition_col: str, min_rep: int, rep: Report) -> None:
161 if not meta_path.is_file():
162 rep.error(f"metadata not found: {meta_path}")
163 return
165 try:
166 validate_header(meta_path)
167 meta = pd.read_csv(meta_path, dtype=str, keep_default_na=False, index_col=0)
168 except ValueError as exc:
169 rep.error(f"invalid metadata CSV: {exc}")
170 return
171 meta.index = meta.index.astype(str).str.strip()
172 meta.columns = meta.columns.str.strip()
173 for col in meta.columns:
174 meta[col] = meta[col].str.strip()
175 if meta.index.has_duplicates or (meta.index == "").any():
176 rep.error("metadata sample IDs must be non-empty and unique; repeated rows are not biological replicates")
177 return
179 if condition_col not in meta.columns:
180 rep.error(f"metadata has no '{condition_col}' column (columns: {list(meta.columns)})")
181 return
183 # Cross-check sample IDs against the samplesheet.
184 if sheet is not None:
185 sheet_samples = set(sheet["sample"].unique())
186 meta_samples = set(meta.index)
187 missing = sheet_samples - meta_samples
188 extra = meta_samples - sheet_samples
189 if missing:
190 rep.error(f"samples in samplesheet but missing from metadata: {sorted(missing)}")
191 if extra:
192 rep.warn(f"samples in metadata but not in samplesheet: {sorted(extra)}")
193 meta = meta.loc[meta.index.isin(sheet_samples)]
195 # Replication per condition group.
196 groups = meta[condition_col].replace("", pd.NA).dropna()
197 if groups.empty:
198 rep.error(f"'{condition_col}' is empty for all samples")
199 return
200 if len(groups) != len(meta) or groups.eq("CHANGE_ME").any():
201 rep.error(f"'{condition_col}' has missing or CHANGE_ME values")
202 if groups.nunique() < 2:
203 rep.error(f"'{condition_col}' needs at least two groups for this comparison workflow")
204 sizes = groups.value_counts()
205 print(f"design: '{condition_col}' groups -> {sizes.to_dict()}")
206 for level, n in sizes.items():
207 if n < 2:
208 rep.error(f"group '{level}' has {n} replicate(s); this workflow requires >=2 biological replicates per group")
209 elif n < min_rep:
210 rep.warn(f"group '{level}' has {n} replicate(s); >={min_rep} recommended for reliable DE")
212 # Light confounding check: batch fully nested within condition.
213 if "batch" in meta.columns and meta["batch"].replace("", pd.NA).notna().any():
214 ct = pd.crosstab(meta[condition_col], meta["batch"])
215 each_batch_single_condition = (ct > 0).sum(axis=0).eq(1).all()
216 if ct.shape[0] > 1 and ct.shape[1] > 1 and each_batch_single_condition:
217 rep.warn(
218 "'batch' looks fully confounded with "
219 f"'{condition_col}' (each batch holds a single condition); "
220 "the batch effect cannot be separated from the biology. See design-and-qc.md."
221 )
224def main() -> None:
225 p = argparse.ArgumentParser(
226 description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
227 )
228 p.add_argument("--samplesheet", type=Path, required=True, help="CSV samplesheet to validate.")
229 p.add_argument("--metadata", type=Path, help="Optional design metadata CSV (index=sample).")
230 p.add_argument("--condition-col", default="condition", help="Condition column in metadata (default: condition).")
231 p.add_argument("--min-replicates", type=int, default=3, dest="min_rep",
232 help="Replicates per group to recommend (default: 3).")
233 p.add_argument("--no-check-files", action="store_true", help="Skip local FASTQ existence checks.")
234 p.add_argument("--nfcore", action="store_true", help="Also enforce nf-core/rnaseq header, gzip extension and strandedness rules.")
235 args = p.parse_args()
237 rep = Report()
238 print(f"Validating {args.samplesheet} ...")
239 sheet = validate_samplesheet(args.samplesheet, not args.no_check_files, rep, args.nfcore)
240 if args.metadata:
241 print(f"Validating {args.metadata} ...")
242 validate_metadata(args.metadata, sheet, args.condition_col, args.min_rep, rep)
244 print()
245 sys.exit(rep.summarize())
248if __name__ == "__main__":
249 main()