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
scripts/generate_samplesheet.py
Python·455 lines·15 KB
"""
17
18import argparse
19import os
20import sys
21from pathlib import Path
22from typing import Dict, List, Optional, Tuple
23
24import yaml
25
26# Add parent directory to path for utils import
27sys.path.insert(0, str(Path(__file__).parent))
28
29from utils.file_discovery import discover_files, detect_input_type, find_index_file
30from utils.sample_inference import (
31 extract_sample_info,
32 infer_tumor_normal_status,
33 match_read_pairs,
34 extract_replicate_number
35)
36from utils.validators import validate_samplesheet, ValidationResult
37
38
39def load_pipeline_config(pipeline: str) -> Dict:
40 """Load pipeline configuration from YAML."""
41 config_dir = Path(__file__).parent / "config" / "pipelines"
42 config_file = config_dir / f"{pipeline}.yaml"
43
44 if not config_file.exists():
45 available = [f.stem for f in config_dir.glob("*.yaml") if not f.stem.startswith("_")]
46 raise ValueError(f"Unknown pipeline '{pipeline}'. Available: {', '.join(available)}")
47
48 with open(config_file) as f:
49 return yaml.safe_load(f)
50
51
52def generate_samplesheet(
53 input_dir: str,
54 pipeline: str,
55 output_file: Optional[str] = None,
56 input_type: str = "auto",
57 single_end: bool = False,
58 interactive: bool = True
59) -> Tuple[Optional[str], ValidationResult]:
60 """
61 Generate samplesheet for specified pipeline.
62
63 Args:
64 input_dir: Directory containing sequencing files
65 pipeline: Pipeline name (rnaseq, sarek, atacseq)
66 output_file: Output CSV path (default: samplesheet_{pipeline}.csv)
67 input_type: File type (auto, fastq, bam, cram)
68 single_end: Suppress pairing warnings for single-end data
69 interactive: Prompt for missing info
70
71 Returns:
72 Tuple of (output_path, validation_result)
73 """
74 config = load_pipeline_config(pipeline)
75 samplesheet_config = config.get("samplesheet", {})
76 supported_types = samplesheet_config.get("input_types", ["fastq"])
77
78 # Determine input type
79 if input_type == "auto":
80 input_type = detect_input_type(input_dir)
81 print(f"Auto-detected input type: {input_type.upper()}")
82
83 if input_type not in supported_types:
84 return None, ValidationResult(
85 valid=False,
86 errors=[f"Pipeline '{pipeline}' does not support {input_type.upper()} input. "
87 f"Supported: {supported_types}"]
88 )
89
90 # Discover files
91 try:
92 files = discover_files(input_dir, input_type)
93 except ValueError as e:
94 return None, ValidationResult(valid=False, errors=[str(e)])
95
96 if not files:
97 return None, ValidationResult(
98 valid=False,
99 errors=[f"No {input_type.upper()} files found in {input_dir}"],
100 suggestions=[
101 "Check directory path is correct",
102 "Verify file extensions (.fastq.gz, .fq.gz, .bam, .cram)",
103 f"Run: ls {input_dir}"
104 ]
105 )
106
107 print(f"Found {len(files)} {input_type.upper()} files")
108
109 # Process based on input type
110 if input_type == "fastq":
111 rows = _process_fastq_files(files, config, single_end)
112 else:
113 rows = _process_alignment_files(files, config, input_type)
114
115 if not rows:
116 return None, ValidationResult(
117 valid=False,
118 errors=["Could not generate any samplesheet rows from files"]
119 )
120
121 print(f"Generated {len(rows)} samplesheet rows")
122
123 # Pipeline-specific processing
124 if pipeline == "sarek":
125 rows = _process_sarek_samples(rows, interactive)
126 elif pipeline == "atacseq":
127 rows = _process_atacseq_samples(rows)
128
129 # Validate before writing
130 validation = validate_samplesheet(rows, pipeline, config)
131
132 if not validation.valid:
133 print("\nValidation errors:")
134 for error in validation.errors:
135 print(f" - {error}")
136
137 if interactive:
138 response = input("\nProceed anyway? [y/N]: ").strip().lower()
139 if response != 'y':
140 return None, validation
141 elif validation.warnings:
142 print("\nWarnings:")
143 for warning in validation.warnings:
144 print(f" - {warning}")
145
146 # Determine output path
147 output_path = output_file or f"samplesheet_{pipeline}.csv"
148
149 # Write samplesheet
150 _write_samplesheet(rows, config, output_path)
151
152 print(f"\nGenerated: {output_path}")
153 print(f" Pipeline: {pipeline} v{config.get('version', 'unknown')}")
154 print(f" Samples: {len(set(r.get('sample', r.get('patient', '')) for r in rows))}")
155 print(f" Rows: {len(rows)}")
156
157 # Preview
158 _print_preview(rows, config)
159
160 return output_path, validation
161
162
163def _process_fastq_files(files, config: Dict, single_end: bool) -> List[Dict]:
164 """Process FASTQ files into samplesheet rows."""
165 pairs = match_read_pairs(files)
166
167 if not pairs:
168 return []
169
170 # Check for unpaired files
171 unpaired = [k for k, v in pairs.items() if v.get('r1') and not v.get('r2')]
172 if unpaired and not single_end:
173 print(f"\nNote: {len(unpaired)} samples appear to be single-end (no R2)")
174
175 rows = []
176 columns = config.get("samplesheet", {}).get("columns", [])
177
178 for sample_key, pair_info in sorted(pairs.items()):
179 if not pair_info.get('r1'):
180 continue # Skip entries with only R2
181
182 info = pair_info.get('info', {})
183
184 row = {
185 'sample': info.get('sample', sample_key),
186 'fastq_1': str(Path(pair_info['r1']).absolute()),
187 'fastq_2': str(Path(pair_info['r2']).absolute()) if pair_info.get('r2') else '',
188 }
189
190 # Add additional info from filename
191 if 'patient' in [c['name'] for c in columns]:
192 row['patient'] = info.get('patient', info.get('sample', sample_key))
193
194 if 'lane' in [c['name'] for c in columns]:
195 row['lane'] = info.get('lane', 'L001')
196
197 # Apply defaults from config
198 for col in columns:
199 if col['name'] not in row and 'default' in col:
200 row[col['name']] = col['default']
201
202 rows.append(row)
203
204 return rows
205
206
207def _process_alignment_files(files, config: Dict, input_type: str) -> List[Dict]:
208 """Process BAM/CRAM files into samplesheet rows."""
209 rows = []
210 columns = config.get("samplesheet", {}).get("columns", [])
211
212 for file_info in files:
213 # Find index file
214 index_path = find_index_file(file_info.path)
215
216 info = extract_sample_info(file_info.path)
217
218 row = {
219 'sample': info.get('sample', file_info.stem),
220 'bam': str(Path(file_info.path).absolute()),
221 'bai': str(Path(index_path).absolute()) if index_path else '',
222 }
223
224 # Add patient for sarek
225 if 'patient' in [c['name'] for c in columns]:
226 row['patient'] = info.get('patient', info.get('sample', file_info.stem))
227
228 # Apply defaults
229 for col in columns:
230 if col['name'] not in row and 'default' in col:
231 row[col['name']] = col['default']
232
233 # Warn if no index found
234 if not index_path:
235 print(f" Warning: No index found for {file_info.name}")
236
237 rows.append(row)
238
239 return rows
240
241
242def _process_sarek_samples(rows: List[Dict], interactive: bool) -> List[Dict]:
243 """Process sarek samples: infer and confirm tumor/normal status."""
244 # Auto-infer status from sample names
245 for row in rows:
246 sample_name = row.get('sample', '')
247 inferred = infer_tumor_normal_status(sample_name)
248 if inferred is not None:
249 row['status'] = inferred
250
251 # Report inference results
252 inferred_tumor = [r for r in rows if r.get('status') == 1]
253 inferred_normal = [r for r in rows if r.get('status') == 0]
254 unknown = [r for r in rows if 'status' not in r]
255
256 if inferred_tumor or inferred_normal:
257 print(f"\nTumor/normal inference:")
258 print(f" Tumor samples: {len(inferred_tumor)}")
259 print(f" Normal samples: {len(inferred_normal)}")
260
261 # Handle unknown samples
262 if unknown and interactive:
263 print(f"\n{len(unknown)} sample(s) with unknown status:")
264 for r in unknown:
265 print(f" - {r.get('sample')}")
266
267 print("\nSpecify status for each (0=normal, 1=tumor, Enter=skip):")
268 for r in unknown:
269 response = input(f" {r.get('sample')} [0/1/Enter]: ").strip()
270 if response in ['0', '1']:
271 r['status'] = int(response)
272 else:
273 r['status'] = 0 # Default to normal
274 print(f" Defaulting to normal (0)")
275 elif unknown:
276 # Non-interactive: default to normal
277 for r in unknown:
278 r['status'] = 0
279
280 return rows
281
282
283def _process_atacseq_samples(rows: List[Dict]) -> List[Dict]:
284 """Process ATAC-seq samples: ensure replicate numbers."""
285 # Group by sample name
286 sample_counts = {}
287 for row in rows:
288 sample = row.get('sample', '')
289 if sample not in sample_counts:
290 sample_counts[sample] = 0
291 sample_counts[sample] += 1
292
293 # Assign replicate numbers if not present
294 sample_rep = {}
295 for row in rows:
296 sample = row.get('sample', '')
297
298 if 'replicate' not in row or not row['replicate']:
299 # Try to extract from filename
300 extracted = extract_replicate_number(row.get('fastq_1', ''))
301 if extracted:
302 row['replicate'] = extracted
303 else:
304 # Auto-assign sequential
305 if sample not in sample_rep:
306 sample_rep[sample] = 0
307 sample_rep[sample] += 1
308 row['replicate'] = sample_rep[sample]
309
310 return rows
311
312
313def _write_samplesheet(rows: List[Dict], config: Dict, output_path: str):
314 """Write samplesheet to CSV file."""
315 columns = config.get("samplesheet", {}).get("columns", [])
316 column_names = [c['name'] for c in columns]
317
318 # Filter to columns that have data
319 active_columns = [c for c in column_names if any(c in row and row[c] for row in rows)]
320
321 # Ensure fastq_1/fastq_2 or bam/bai are included
322 for required in ['fastq_1', 'bam']:
323 if required in column_names and required not in active_columns:
324 if any(required in row for row in rows):
325 active_columns.append(required)
326
327 # Maintain original column order
328 active_columns = [c for c in column_names if c in active_columns]
329
330 with open(output_path, 'w') as f:
331 f.write(','.join(active_columns) + '\n')
332 for row in rows:
333 values = [str(row.get(col, '')) for col in active_columns]
334 f.write(','.join(values) + '\n')
335
336
337def _print_preview(rows: List[Dict], config: Dict):
338 """Print preview of generated samplesheet."""
339 columns = config.get("samplesheet", {}).get("columns", [])
340 column_names = [c['name'] for c in columns]
341 active_columns = [c for c in column_names if any(c in row for row in rows)]
342
343 print(f"\nPreview (first 3 rows):")
344 print(','.join(active_columns))
345 for row in rows[:3]:
346 values = [str(row.get(col, ''))[:40] for col in active_columns] # Truncate long paths
347 print(','.join(values))
348 if len(rows) > 3:
349 print(f"... ({len(rows) - 3} more rows)")
350
351
352def validate_existing_samplesheet(csv_path: str, pipeline: str) -> ValidationResult:
353 """Validate an existing samplesheet file."""
354 import csv
355
356 if not os.path.exists(csv_path):
357 return ValidationResult(valid=False, errors=[f"File not found: {csv_path}"])
358
359 try:
360 with open(csv_path, 'r') as f:
361 reader = csv.DictReader(f)
362 rows = list(reader)
363 except Exception as e:
364 return ValidationResult(valid=False, errors=[f"Failed to read CSV: {e}"])
365
366 if not rows:
367 return ValidationResult(valid=False, errors=["Samplesheet is empty"])
368
369 config = load_pipeline_config(pipeline)
370 return validate_samplesheet(rows, pipeline, config)
371
372
373def main():
374 parser = argparse.ArgumentParser(
375 description='Generate nf-core samplesheet from data directory',
376 formatter_class=argparse.RawDescriptionHelpFormatter,
377 epilog="""
378Examples:
379 # Generate samplesheet for RNA-seq
380 %(prog)s ./fastqs rnaseq -o samples.csv
381
382 # Generate samplesheet for sarek from BAM files
383 %(prog)s ./bams sarek --input-type bam
384
385 # Validate existing samplesheet
386 %(prog)s --validate samplesheet.csv rnaseq
387
388Supported pipelines: rnaseq, sarek, atacseq
389 """
390 )
391
392 parser.add_argument('input', help='Directory with data files, or CSV path for --validate')
393 parser.add_argument('pipeline', help='Pipeline name (rnaseq, sarek, atacseq)')
394 parser.add_argument('-o', '--output', help='Output CSV filename')
395 parser.add_argument('--input-type', choices=['auto', 'fastq', 'bam', 'cram'],
396 default='auto', help='Input file type (default: auto-detect)')
397 parser.add_argument('--single-end', action='store_true',
398 help='Treat as single-end data (suppress pairing warnings)')
399 parser.add_argument('--validate', action='store_true',
400 help='Validate existing samplesheet instead of generating')
401 parser.add_argument('--no-interactive', action='store_true',
402 help='Non-interactive mode (use defaults)')
403
404 args = parser.parse_args()
405
406 try:
407 if args.validate:
408 # Validate existing samplesheet
409 result = validate_existing_samplesheet(args.input, args.pipeline)
410 if result.valid:
411 print(f"✓ Samplesheet is valid for {args.pipeline}")
412 if result.warnings:
413 print("\nWarnings:")
414 for w in result.warnings:
415 print(f" - {w}")
416 sys.exit(0)
417 else:
418 print(f"✗ Samplesheet validation failed")
419 print(result.summary())
420 sys.exit(1)
421 else:
422 # Generate new samplesheet
423 if not os.path.isdir(args.input):
424 print(f"Error: Not a directory: {args.input}")
425 sys.exit(1)
426
427 output_path, result = generate_samplesheet(
428 args.input,
429 args.pipeline,
430 args.output,
431 args.input_type,
432 args.single_end,
433 interactive=not args.no_interactive
434 )
435
436 if output_path is None:
437 print("\nFailed to generate samplesheet.")
438 if result.suggestions:
439 print("\nSuggestions:")
440 for s in result.suggestions:
441 print(f" - {s}")
442 sys.exit(1)
443
444 sys.exit(0)
445
446 except ValueError as e:
447 print(f"Error: {e}")
448 sys.exit(1)
449 except KeyboardInterrupt:
450 print("\nAborted.")
451 sys.exit(1)
452
453
454if __name__ == '__main__':
455 main()