Setting the file. One moment. 22
Validate Data
Tech Debt
62
Recruiting Pipeline
71
Vendor Check
125
Zoom Meeting SDK Web
88
Vendor Review
181
Create An Asset
Video Sdk/web
8 KBscripts/utils/validators.py
Python·256 lines·8 KB
14
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)
22
23 def __bool__(self):
24 return self.valid
25
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)
42
43
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"
49
50 if not config_path.exists():
51 return None
52
53 with open(config_path) as f:
54 return yaml.safe_load(f)
55
56
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.
64
65 Args:
66 rows: List of row dictionaries
67 pipeline: Pipeline name (e.g., 'rnaseq', 'sarek')
68 config: Optional pre-loaded config dict
69
70 Returns:
71 ValidationResult with errors, warnings, and suggestions
72 """
73 errors = []
74 warnings = []
75 suggestions = []
76
77 # Load config if not provided
78 if config is None:
79 config = load_pipeline_config(pipeline)
80
81 if config is None:
82 errors.append(f"Unknown pipeline: {pipeline}")
83 return ValidationResult(valid=False, errors=errors)
84
85 columns = config.get("samplesheet", {}).get("columns", [])
86 required_cols = [c["name"] for c in columns if c.get("required", False)]
87
88 if not rows:
89 errors.append("Samplesheet is empty - no samples found")
90 return ValidationResult(valid=False, errors=errors)
91
92 # Validate each row
93 for i, row in enumerate(rows):
94 row_num = i + 2 # Account for header row
95
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)
99
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
105
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}'")
111
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}")
120
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 )
132
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")
140
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 )
152
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)
158
159 return ValidationResult(
160 valid=len(errors) == 0,
161 errors=errors,
162 warnings=warnings,
163 suggestions=suggestions
164 )
165
166
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")
179
180 if patient not in patients:
181 patients[patient] = {"tumor": 0, "normal": 0, "unknown": 0}
182
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
189
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 )
200
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 )
206
207
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)
220
221 if sample not in samples:
222 samples[sample] = []
223
224 samples[sample].append(replicate)
225
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 )
233
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 )
240
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 )
247
248
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)
252
253
254def validate_absolute_path(path: str) -> bool:
255 """Check if path is absolute."""
256 return os.path.isabs(path)