Setting the file. One moment.
Detect Data Type · Nextflow Development · anthropics/knowledge-work-plugins · Skills Docs
Back to the top of the page 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
Next
Script Generate Samplesheet
scripts/ detect_data_type.py
Python · 300 lines · 10 KB
import
sys
17 from pathlib import Path
18 from typing import Dict, List, Tuple
19
20 import yaml
21
22
23 def load_all_pipeline_configs () -> Dict[ str , Dict]:
24 """Load all pipeline configurations."""
25 config_dir = Path( __file__ ).parent / "config" / "pipelines"
26 configs = {}
27
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)
33
34 return configs
35
36
37 def 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 }
47
48 directory = os.path.abspath(directory)
49
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())
55
56 for filename in files:
57 filename_lower = filename.lower()
58
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
66
67 # Collect filenames for pattern matching
68 info[ 'filenames' ].append(filename_lower)
69
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
76
77 return info
78
79
80 def calculate_pipeline_scores (scan_info: Dict, configs: Dict) -> Dict[ str , Dict]:
81 """Calculate confidence scores for each pipeline."""
82 scores = {}
83
84 for pipeline_name, config in configs.items():
85 score = 0
86 matches = []
87
88 # Check detection hints
89 hints = config.get( 'detection_hints' , {})
90
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
100
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
110
111 # Check data type compatibility
112 data_types = config.get( 'data_types' , [])
113 input_types = config.get( 'samplesheet' , {}).get( 'input_types' , [ 'fastq' ])
114
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
122
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' ]
128
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)
131
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" )
138
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
146
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
155
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
164
165 scores[pipeline_name] = {
166 'score' : score,
167 'matches' : matches,
168 'description' : config.get( 'description' , '' ),
169 'version' : config.get( 'version' , 'unknown' ),
170 }
171
172 return scores
173
174
175 def detect_pipeline (directory: str ) -> Tuple[ str , Dict]:
176 """
177 Detect the most appropriate pipeline for the data.
178
179 Args:
180 directory: Path to data directory
181
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 } " )
187
188 configs = load_all_pipeline_configs()
189 scan_info = scan_directory(directory)
190
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 } " )
195
196 scores = calculate_pipeline_scores(scan_info, configs)
197
198 # Find highest scoring pipeline
199 best_pipeline = max (scores.keys(), key =lambda k: scores[k][ 'score' ])
200
201 return best_pipeline, scores
202
203
204 def 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
225
226 print ( " \n " + "=" * 50 )
227 print ( " nf-core Pipeline Detection" )
228 print ( "=" * 50 )
229 print ( f " \n Directory: { 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" )
233
234 print ( " \n --- Pipeline Scores ---" )
235 sorted_pipelines = sorted (scores.keys(), key =lambda k: scores[k][ 'score' ], reverse = True )
236
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 ]) } " )
244
245 print ( f " \n{ '=' * 50 } " )
246 print ( f " \n\033 [92mRecommended: { recommended }\033 [0m" )
247 print ( f "Version: { scores[recommended][ 'version' ] } " )
248
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 " \n 2. 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 " \n 3. Generate samplesheet:" )
259 print ( f " python scripts/generate_samplesheet.py { directory } { recommended } " )
260
261
262 def main ():
263 parser = argparse.ArgumentParser(
264 description = 'Detect appropriate nf-core pipeline for data' ,
265 formatter_class = argparse.RawDescriptionHelpFormatter,
266 epilog = """
267 Examples:
268 %(prog)s ./data
269 %(prog)s ./fastqs --json
270 """
271 )
272
273 parser.add_argument( 'directory' , help = 'Directory containing sequencing data' )
274 parser.add_argument( '--json' , action = 'store_true' , help = 'Output as JSON' )
275
276 args = parser.parse_args()
277
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 )
283
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 )
290
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 )
297
298
299 if __name__ == '__main__' :
300 main()