Skills › Knowledge Work Plugins › nextflow-development › detect_data_type.py
detect_data_type.py
300 lines · bio-research/skills/nextflow-development/scripts/detect_data_type.py at 2574e29
1#!/usr/bin/env python3
2"""
3Auto-detect appropriate nf-core pipeline from data directory.
5Analyzes filenames, directory structure, and file content hints to suggest
6the most appropriate pipeline for the data.
8Usage:
9 python detect_data_type.py /path/to/data
10 python detect_data_type.py /path/to/data --json
11"""
13import argparse
14import json
15import os
16import sys
17from pathlib import Path
18from typing import Dict, List, Tuple
20import yaml
23def load_all_pipeline_configs() -> Dict[str, Dict]:
24 """Load all pipeline configurations."""
25 config_dir = Path(__file__).parent / "config" / "pipelines"
26 configs = {}
28 for config_file in config_dir.glob("*.yaml"):
29 if config_file.stem.startswith("_"):
30 continue
31 with open(config_file) as f:
32 configs[config_file.stem] = yaml.safe_load(f)
34 return configs
37def scan_directory(directory: str) -> Dict:
38 """Scan directory and collect file information."""
39 info = {
40 'fastq_count': 0,
41 'bam_count': 0,
42 'cram_count': 0,
43 'filenames': [],
44 'directories': [],
45 'total_size_gb': 0,
46 }
48 directory = os.path.abspath(directory)
50 for root, dirs, files in os.walk(directory):
51 # Collect directory names
52 rel_root = os.path.relpath(root, directory)
53 if rel_root != '.':
54 info['directories'].append(rel_root.lower())
56 for filename in files:
57 filename_lower = filename.lower()
59 # Count file types
60 if any(filename_lower.endswith(ext) for ext in ['.fastq.gz', '.fq.gz', '.fastq', '.fq']):
61 info['fastq_count'] += 1
62 elif filename_lower.endswith('.bam'):
63 info['bam_count'] += 1
64 elif filename_lower.endswith('.cram'):
65 info['cram_count'] += 1
67 # Collect filenames for pattern matching
68 info['filenames'].append(filename_lower)
70 # Sum file sizes
71 try:
72 size = os.path.getsize(os.path.join(root, filename))
73 info['total_size_gb'] += size / (1024**3)
74 except Exception:
75 pass
77 return info
80def calculate_pipeline_scores(scan_info: Dict, configs: Dict) -> Dict[str, Dict]:
81 """Calculate confidence scores for each pipeline."""
82 scores = {}
84 for pipeline_name, config in configs.items():
85 score = 0
86 matches = []
88 # Check detection hints
89 hints = config.get('detection_hints', {})
91 # Filename hints
92 filename_hints = hints.get('filename', [])
93 for hint in filename_hints:
94 hint_lower = hint.lower()
95 for filename in scan_info['filenames']:
96 if hint_lower in filename:
97 score += 10
98 matches.append(f"Filename contains '{hint}'")
99 break
101 # Directory hints
102 directory_hints = hints.get('directory', [])
103 for hint in directory_hints:
104 hint_lower = hint.lower()
105 for dirname in scan_info['directories']:
106 if hint_lower in dirname:
107 score += 15
108 matches.append(f"Directory contains '{hint}'")
109 break
111 # Check data type compatibility
112 data_types = config.get('data_types', [])
113 input_types = config.get('samplesheet', {}).get('input_types', ['fastq'])
115 # Prefer pipelines that support the available file types
116 if 'fastq' in input_types and scan_info['fastq_count'] > 0:
117 score += 5
118 if 'bam' in input_types and scan_info['bam_count'] > 0:
119 score += 5
120 if 'cram' in input_types and scan_info['cram_count'] > 0:
121 score += 5
123 # Pipeline-specific boosts
124 if pipeline_name == 'sarek':
125 # Check for tumor/normal indicators
126 tumor_indicators = ['tumor', 'tumour', 'cancer', 'met', 'primary']
127 normal_indicators = ['normal', 'germline', 'blood', 'control']
129 has_tumor = any(ind in ' '.join(scan_info['filenames']) for ind in tumor_indicators)
130 has_normal = any(ind in ' '.join(scan_info['filenames']) for ind in normal_indicators)
132 if has_tumor or has_normal:
133 score += 20
134 if has_tumor:
135 matches.append("Found tumor sample indicators")
136 if has_normal:
137 matches.append("Found normal sample indicators")
139 # DNA-related hints
140 dna_hints = ['wgs', 'wes', 'exome', 'dna', 'variant', 'snp', 'indel']
141 for hint in dna_hints:
142 if hint in ' '.join(scan_info['filenames'] + scan_info['directories']):
143 score += 10
144 matches.append(f"Found DNA/variant indicator: '{hint}'")
145 break
147 elif pipeline_name == 'rnaseq':
148 # RNA-related hints
149 rna_hints = ['rna', 'rnaseq', 'mrna', 'expression', 'transcript', 'counts']
150 for hint in rna_hints:
151 if hint in ' '.join(scan_info['filenames'] + scan_info['directories']):
152 score += 15
153 matches.append(f"Found RNA indicator: '{hint}'")
154 break
156 elif pipeline_name == 'atacseq':
157 # ATAC-related hints
158 atac_hints = ['atac', 'atacseq', 'chromatin', 'accessibility', 'peak', 'macs']
159 for hint in atac_hints:
160 if hint in ' '.join(scan_info['filenames'] + scan_info['directories']):
161 score += 20
162 matches.append(f"Found ATAC-seq indicator: '{hint}'")
163 break
165 scores[pipeline_name] = {
166 'score': score,
167 'matches': matches,
168 'description': config.get('description', ''),
169 'version': config.get('version', 'unknown'),
170 }
172 return scores
175def detect_pipeline(directory: str) -> Tuple[str, Dict]:
176 """
177 Detect the most appropriate pipeline for the data.
179 Args:
180 directory: Path to data directory
182 Returns:
183 Tuple of (recommended_pipeline, all_scores)
184 """
185 if not os.path.isdir(directory):
186 raise ValueError(f"Not a directory: {directory}")
188 configs = load_all_pipeline_configs()
189 scan_info = scan_directory(directory)
191 # Check if any sequencing files found
192 total_files = scan_info['fastq_count'] + scan_info['bam_count'] + scan_info['cram_count']
193 if total_files == 0:
194 raise ValueError(f"No sequencing files (FASTQ/BAM/CRAM) found in {directory}")
196 scores = calculate_pipeline_scores(scan_info, configs)
198 # Find highest scoring pipeline
199 best_pipeline = max(scores.keys(), key=lambda k: scores[k]['score'])
201 return best_pipeline, scores
204def print_results(
205 directory: str,
206 recommended: str,
207 scores: Dict,
208 scan_info: Dict,
209 output_json: bool = False
210):
211 """Print detection results."""
212 if output_json:
213 result = {
214 'recommended': recommended,
215 'scores': scores,
216 'scan_info': {
217 'fastq_count': scan_info['fastq_count'],
218 'bam_count': scan_info['bam_count'],
219 'cram_count': scan_info['cram_count'],
220 'total_size_gb': round(scan_info['total_size_gb'], 2),
221 }
222 }
223 print(json.dumps(result, indent=2))
224 return
226 print("\n" + "=" * 50)
227 print(" nf-core Pipeline Detection")
228 print("=" * 50)
229 print(f"\nDirectory: {directory}")
230 print(f"Files found: {scan_info['fastq_count']} FASTQ, "
231 f"{scan_info['bam_count']} BAM, {scan_info['cram_count']} CRAM")
232 print(f"Total size: {scan_info['total_size_gb']:.1f} GB")
234 print("\n--- Pipeline Scores ---")
235 sorted_pipelines = sorted(scores.keys(), key=lambda k: scores[k]['score'], reverse=True)
237 for pipeline in sorted_pipelines:
238 info = scores[pipeline]
239 indicator = "→" if pipeline == recommended else " "
240 print(f"\n{indicator} {pipeline} (score: {info['score']})")
241 print(f" {info['description']}")
242 if info['matches']:
243 print(f" Matches: {', '.join(info['matches'][:3])}")
245 print(f"\n{'=' * 50}")
246 print(f"\n\033[92mRecommended: {recommended}\033[0m")
247 print(f"Version: {scores[recommended]['version']}")
249 # Print suggested next steps
250 print(f"\n--- Next Steps ---")
251 print(f"1. Run environment check:")
252 print(f" python scripts/check_environment.py")
253 print(f"\n2. Run test profile:")
254 config = load_all_pipeline_configs().get(recommended, {})
255 test_cmd = config.get('test_profile', {}).get('command', '')
256 if test_cmd:
257 print(f" {test_cmd}")
258 print(f"\n3. Generate samplesheet:")
259 print(f" python scripts/generate_samplesheet.py {directory} {recommended}")
262def main():
263 parser = argparse.ArgumentParser(
264 description='Detect appropriate nf-core pipeline for data',
265 formatter_class=argparse.RawDescriptionHelpFormatter,
266 epilog="""
267Examples:
268 %(prog)s ./data
269 %(prog)s ./fastqs --json
270 """
271 )
273 parser.add_argument('directory', help='Directory containing sequencing data')
274 parser.add_argument('--json', action='store_true', help='Output as JSON')
276 args = parser.parse_args()
278 try:
279 scan_info = scan_directory(args.directory)
280 recommended, scores = detect_pipeline(args.directory)
281 print_results(args.directory, recommended, scores, scan_info, args.json)
282 sys.exit(0)
284 except ValueError as e:
285 if args.json:
286 print(json.dumps({'error': str(e)}))
287 else:
288 print(f"Error: {e}")
289 sys.exit(1)
291 except Exception as e:
292 if args.json:
293 print(json.dumps({'error': str(e)}))
294 else:
295 print(f"Error: {e}")
296 sys.exit(1)
299if __name__ == '__main__':
300 main()