Skills › Knowledge Work Plugins › nextflow-development › validators.py
validators.py
256 lines · bio-research/skills/nextflow-development/scripts/utils/validators.py at 2574e29
1"""
2Samplesheet validation utilities.
4Validates samplesheet rows against pipeline configuration before writing,
5catching errors early with helpful messages.
6"""
8import os
9from dataclasses import dataclass, field
10from pathlib import Path
11from typing import Dict, List, Optional
12import yaml
15@dataclass
16class ValidationResult:
17 """Result of samplesheet validation."""
18 valid: bool
19 errors: List[str] = field(default_factory=list)
20 warnings: List[str] = field(default_factory=list)
21 suggestions: List[str] = field(default_factory=list)
23 def __bool__(self):
24 return self.valid
26 def summary(self) -> str:
27 """Generate human-readable summary."""
28 lines = []
29 if self.errors:
30 lines.append("Errors:")
31 for e in self.errors:
32 lines.append(f" - {e}")
33 if self.warnings:
34 lines.append("Warnings:")
35 for w in self.warnings:
36 lines.append(f" - {w}")
37 if self.suggestions:
38 lines.append("Suggestions:")
39 for s in self.suggestions:
40 lines.append(f" - {s}")
41 return "\n".join(lines)
44def load_pipeline_config(pipeline: str) -> Optional[Dict]:
45 """Load pipeline configuration from YAML file."""
46 # Find config directory relative to this file
47 script_dir = Path(__file__).parent.parent.parent
48 config_path = script_dir / "config" / "pipelines" / f"{pipeline}.yaml"
50 if not config_path.exists():
51 return None
53 with open(config_path) as f:
54 return yaml.safe_load(f)
57def validate_samplesheet(
58 rows: List[Dict],
59 pipeline: str,
60 config: Optional[Dict] = None
61) -> ValidationResult:
62 """
63 Validate samplesheet rows against pipeline requirements.
65 Args:
66 rows: List of row dictionaries
67 pipeline: Pipeline name (e.g., 'rnaseq', 'sarek')
68 config: Optional pre-loaded config dict
70 Returns:
71 ValidationResult with errors, warnings, and suggestions
72 """
73 errors = []
74 warnings = []
75 suggestions = []
77 # Load config if not provided
78 if config is None:
79 config = load_pipeline_config(pipeline)
81 if config is None:
82 errors.append(f"Unknown pipeline: {pipeline}")
83 return ValidationResult(valid=False, errors=errors)
85 columns = config.get("samplesheet", {}).get("columns", [])
86 required_cols = [c["name"] for c in columns if c.get("required", False)]
88 if not rows:
89 errors.append("Samplesheet is empty - no samples found")
90 return ValidationResult(valid=False, errors=errors)
92 # Validate each row
93 for i, row in enumerate(rows):
94 row_num = i + 2 # Account for header row
96 # Check required columns
97 for col_name in required_cols:
98 col_config = next((c for c in columns if c["name"] == col_name), None)
100 # Skip columns with conditions that don't apply
101 if col_config and "condition" in col_config:
102 # Simple condition check - skip for now
103 # Full implementation would evaluate conditions
104 pass
106 if col_name not in row or row[col_name] is None or row[col_name] == "":
107 # Check if there's a default
108 if col_config and "default" in col_config:
109 continue
110 errors.append(f"Row {row_num}: Missing required column '{col_name}'")
112 # Validate path columns exist
113 for col_name in ["fastq_1", "fastq_2", "bam", "bai"]:
114 if col_name in row and row[col_name]:
115 path = row[col_name]
116 if not os.path.exists(path):
117 errors.append(f"Row {row_num}: File not found: {path}")
118 elif not os.path.isfile(path):
119 errors.append(f"Row {row_num}: Not a file: {path}")
121 # Validate enum values
122 for col_config in columns:
123 col_name = col_config["name"]
124 if col_name in row and row[col_name] and "allowed" in col_config:
125 value = row[col_name]
126 allowed = col_config["allowed"]
127 if value not in allowed:
128 errors.append(
129 f"Row {row_num}: Invalid value '{value}' for '{col_name}'. "
130 f"Allowed: {allowed}"
131 )
133 # Check R1/R2 pairing consistency
134 r1 = row.get("fastq_1", "")
135 r2 = row.get("fastq_2", "")
136 if r1 and not r2:
137 warnings.append(f"Row {row_num}: Single-end data (no R2 file)")
138 elif r2 and not r1:
139 errors.append(f"Row {row_num}: R2 present but R1 missing")
141 # Check for duplicate samples
142 sample_col = "sample" if "sample" in rows[0] else "patient"
143 if sample_col in rows[0]:
144 samples = [r.get(sample_col, "") for r in rows]
145 duplicates = [s for s in set(samples) if samples.count(s) > 1]
146 if duplicates:
147 warnings.append(f"Duplicate sample names: {duplicates}")
148 suggestions.append(
149 "Duplicates may be intentional (multi-lane sequencing). "
150 "Verify sample grouping is correct."
151 )
153 # Pipeline-specific validation
154 if pipeline == "sarek":
155 _validate_sarek_specific(rows, errors, warnings, suggestions)
156 elif pipeline == "atacseq":
157 _validate_atacseq_specific(rows, errors, warnings, suggestions)
159 return ValidationResult(
160 valid=len(errors) == 0,
161 errors=errors,
162 warnings=warnings,
163 suggestions=suggestions
164 )
167def _validate_sarek_specific(
168 rows: List[Dict],
169 errors: List[str],
170 warnings: List[str],
171 suggestions: List[str]
172):
173 """Sarek-specific validation for tumor/normal pairing."""
174 # Group by patient
175 patients = {}
176 for row in rows:
177 patient = row.get("patient", "")
178 status = row.get("status")
180 if patient not in patients:
181 patients[patient] = {"tumor": 0, "normal": 0, "unknown": 0}
183 if status == 1:
184 patients[patient]["tumor"] += 1
185 elif status == 0:
186 patients[patient]["normal"] += 1
187 else:
188 patients[patient]["unknown"] += 1
190 # Check pairing
191 for patient, counts in patients.items():
192 if counts["tumor"] > 0 and counts["normal"] == 0:
193 warnings.append(
194 f"Patient '{patient}': Tumor sample(s) without matched normal. "
195 "Somatic calling works best with paired tumor-normal."
196 )
197 suggestions.append(
198 f"For patient '{patient}': Add a normal sample or use tumor-only mode."
199 )
201 if counts["unknown"] > 0:
202 warnings.append(
203 f"Patient '{patient}': {counts['unknown']} sample(s) with unknown status. "
204 "Set status column to 0 (normal) or 1 (tumor)."
205 )
208def _validate_atacseq_specific(
209 rows: List[Dict],
210 errors: List[str],
211 warnings: List[str],
212 suggestions: List[str]
213):
214 """ATAC-seq specific validation for replicates."""
215 # Group by sample (condition)
216 samples = {}
217 for row in rows:
218 sample = row.get("sample", "")
219 replicate = row.get("replicate", 1)
221 if sample not in samples:
222 samples[sample] = []
224 samples[sample].append(replicate)
226 # Check replicates
227 for sample, reps in samples.items():
228 if len(reps) < 2:
229 warnings.append(
230 f"Sample '{sample}': Only {len(reps)} replicate(s). "
231 "Consensus peaks require 2+ replicates."
232 )
234 # Check for duplicate replicate numbers
235 if len(reps) != len(set(reps)):
236 errors.append(
237 f"Sample '{sample}': Duplicate replicate numbers detected. "
238 "Each replicate must have a unique number."
239 )
241 # Check all samples have R2 (ATAC-seq requires paired-end)
242 for i, row in enumerate(rows):
243 if not row.get("fastq_2"):
244 errors.append(
245 f"Row {i+2}: ATAC-seq requires paired-end data. R2 file missing."
246 )
249def validate_file_exists(path: str) -> bool:
250 """Check if file exists and is accessible."""
251 return os.path.isfile(path) and os.access(path, os.R_OK)
254def validate_absolute_path(path: str) -> bool:
255 """Check if path is absolute."""
256 return os.path.isabs(path)