Setting the file. One moment.
Check Environment · 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
Number 2.7
Position 7 of 21
Type Python
Size 14 KB
Lines 452 scripts/ check_environment.py
Python · 452 lines · 14 KB
subprocess
17 import sys
18 from dataclasses import dataclass, field, asdict
19 from typing import List, Optional
20
21
22 @dataclass
23 class CheckResult :
24 """Result of a single environment check."""
25 name: str
26 passed: bool
27 message: str
28 details: Optional[ str ] = None
29 fix: Optional[ str ] = None
30
31
32 @dataclass
33 class EnvironmentReport :
34 """Complete environment validation report."""
35 ready: bool
36 checks: List[CheckResult] = field( default_factory = list )
37 recommendations: List[ str ] = field( default_factory = list )
38
39 def to_dict (self):
40 return {
41 "ready" : self .ready,
42 "checks" : [asdict(c) for c in self .checks],
43 "recommendations" : self .recommendations
44 }
45
46
47 def check_docker () -> CheckResult:
48 """Check Docker availability, daemon status, and permissions."""
49 if not shutil.which( "docker" ):
50 return CheckResult(
51 name = "Docker" ,
52 passed = False ,
53 message = "Docker not found in PATH" ,
54 fix = "Install Docker: https://docs.docker.com/get-docker/"
55 )
56
57 try :
58 result = subprocess.run(
59 [ "docker" , "info" ],
60 capture_output = True ,
61 text = True ,
62 timeout = 15
63 )
64
65 if result.returncode != 0 :
66 stderr_lower = result.stderr.lower()
67 if "permission denied" in stderr_lower:
68 return CheckResult(
69 name = "Docker" ,
70 passed = False ,
71 message = "Docker permission denied" ,
72 details = "Cannot connect to Docker daemon" ,
73 fix = "sudo usermod -aG docker $USER && newgrp docker"
74 )
75 elif "cannot connect" in stderr_lower or "is the docker daemon running" in stderr_lower:
76 return CheckResult(
77 name = "Docker" ,
78 passed = False ,
79 message = "Docker daemon not running" ,
80 details = result.stderr[: 200 ] if result.stderr else None ,
81 fix = "sudo systemctl start docker"
82 )
83 else :
84 return CheckResult(
85 name = "Docker" ,
86 passed = False ,
87 message = "Docker error" ,
88 details = result.stderr[: 200 ] if result.stderr else None ,
89 fix = "Check Docker installation and daemon status"
90 )
91
92 return CheckResult(
93 name = "Docker" ,
94 passed = True ,
95 message = "Docker is available and running"
96 )
97
98 except subprocess.TimeoutExpired:
99 return CheckResult(
100 name = "Docker" ,
101 passed = False ,
102 message = "Docker command timed out" ,
103 fix = "Check Docker daemon status: sudo systemctl status docker"
104 )
105 except Exception as e:
106 return CheckResult(
107 name = "Docker" ,
108 passed = False ,
109 message = f "Docker check failed: { str (e) } "
110 )
111
112
113 def check_nextflow () -> CheckResult:
114 """Check Nextflow installation and version (requires >= 23.04)."""
115 if not shutil.which( "nextflow" ):
116 return CheckResult(
117 name = "Nextflow" ,
118 passed = False ,
119 message = "Nextflow not found in PATH" ,
120 fix = "curl -s https://get.nextflow.io | bash && mv nextflow ~/bin/ && export PATH=$HOME/bin:$PATH"
121 )
122
123 try :
124 result = subprocess.run(
125 [ "nextflow" , "-version" ],
126 capture_output = True ,
127 text = True ,
128 timeout = 30
129 )
130
131 output = result.stdout + result.stderr
132 version_line = output.strip().split( ' \n ' )[ 0 ] if output else ""
133
134 import re
135 match = re.search( r ' (\d + ) \. (\d + ) \. (\d + ) ' , version_line)
136
137 if match:
138 major, minor, patch = int (match.group( 1 )), int (match.group( 2 )), int (match.group( 3 ))
139 version_str = f " { major } . { minor } . { patch } "
140
141 # Require version >= 23.04
142 if major > 23 or (major == 23 and minor >= 4 ):
143 return CheckResult(
144 name = "Nextflow" ,
145 passed = True ,
146 message = f "Nextflow { version_str } installed" ,
147 details = version_line
148 )
149 else :
150 return CheckResult(
151 name = "Nextflow" ,
152 passed = False ,
153 message = f "Nextflow { version_str } is outdated (requires >= 23.04)" ,
154 details = version_line,
155 fix = "nextflow self-update"
156 )
157
158 return CheckResult(
159 name = "Nextflow" ,
160 passed = True ,
161 message = "Nextflow installed (version unknown)" ,
162 details = version_line
163 )
164
165 except subprocess.TimeoutExpired:
166 return CheckResult(
167 name = "Nextflow" ,
168 passed = False ,
169 message = "Nextflow command timed out" ,
170 fix = "Check Nextflow installation"
171 )
172 except Exception as e:
173 return CheckResult(
174 name = "Nextflow" ,
175 passed = False ,
176 message = f "Nextflow check failed: { str (e) } "
177 )
178
179
180 def check_java () -> CheckResult:
181 """Check Java version (requires >= 11)."""
182 if not shutil.which( "java" ):
183 return CheckResult(
184 name = "Java" ,
185 passed = False ,
186 message = "Java not found in PATH" ,
187 fix = "Install Java 11+: sudo apt install openjdk-11-jdk"
188 )
189
190 try :
191 result = subprocess.run(
192 [ "java" , "-version" ],
193 capture_output = True ,
194 text = True ,
195 timeout = 10
196 )
197
198 # Java version is typically in stderr
199 output = result.stderr or result.stdout
200 import re
201 match = re.search( r 'version " (\d + ) ' , output)
202
203 if match:
204 version = int (match.group( 1 ))
205 version_line = output.strip().split( ' \n ' )[ 0 ]
206
207 if version >= 11 :
208 return CheckResult(
209 name = "Java" ,
210 passed = True ,
211 message = f "Java { version } installed" ,
212 details = version_line
213 )
214 else :
215 return CheckResult(
216 name = "Java" ,
217 passed = False ,
218 message = f "Java { version } is too old (requires >= 11)" ,
219 details = version_line,
220 fix = "Install Java 11+: sudo apt install openjdk-11-jdk"
221 )
222
223 return CheckResult(
224 name = "Java" ,
225 passed = True ,
226 message = "Java installed" ,
227 details = output.strip().split( ' \n ' )[ 0 ] if output else None
228 )
229
230 except Exception as e:
231 return CheckResult(
232 name = "Java" ,
233 passed = False ,
234 message = f "Java check failed: { str (e) } "
235 )
236
237
238 def check_resources () -> CheckResult:
239 """Check system resources (CPU, memory, disk)."""
240 try :
241 # CPU cores
242 cpu_count = os.cpu_count() or 1
243
244 # Memory
245 mem_gb = 0
246 try :
247 # Linux: read from /proc/meminfo
248 with open ( '/proc/meminfo' , 'r' ) as f:
249 for line in f:
250 if line.startswith( 'MemTotal:' ):
251 mem_kb = int (line.split()[ 1 ])
252 mem_gb = mem_kb / ( 1024 * 1024 )
253 break
254 except ( FileNotFoundError , PermissionError ):
255 # macOS: use sysctl
256 try :
257 result = subprocess.run(
258 [ 'sysctl' , '-n' , 'hw.memsize' ],
259 capture_output = True , text = True , timeout = 5
260 )
261 if result.returncode == 0 :
262 mem_gb = int (result.stdout.strip()) / ( 1024 ** 3 )
263 except Exception :
264 pass
265
266 # Disk space (current directory)
267 disk_gb = 0
268 try :
269 statvfs = os.statvfs( '.' )
270 disk_gb = (statvfs.f_frsize * statvfs.f_bavail) / ( 1024 ** 3 )
271 except Exception :
272 pass
273
274 details = f "CPUs: { cpu_count } , Memory: { mem_gb :.1f} GB, Disk: { disk_gb :.1f} GB available"
275
276 # Check minimums
277 warnings = []
278 if cpu_count < 4 :
279 warnings.append( f "Low CPU count ( { cpu_count } ). Consider --max_cpus { cpu_count } " )
280 if 0 < mem_gb < 8 :
281 warnings.append( f "Low memory ( { mem_gb :.1f} GB). Use --max_memory ' { int (mem_gb) } GB'" )
282 if 0 < disk_gb < 50 :
283 warnings.append( f "Low disk space ( { disk_gb :.1f} GB). Pipelines need ~100GB for human data" )
284
285 if warnings:
286 return CheckResult(
287 name = "Resources" ,
288 passed = True ,
289 message = "Resources available (with warnings)" ,
290 details = details,
291 fix = "; " .join(warnings)
292 )
293
294 return CheckResult(
295 name = "Resources" ,
296 passed = True ,
297 message = "Sufficient resources available" ,
298 details = details
299 )
300
301 except Exception as e:
302 return CheckResult(
303 name = "Resources" ,
304 passed = True , # Don't fail on resource check errors
305 message = f "Could not fully check resources: { str (e) } "
306 )
307
308
309 def check_network () -> CheckResult:
310 """Check network connectivity to Docker Hub and nf-core."""
311 try :
312 import urllib.request
313
314 # User-Agent header to avoid 403 from sites that block default Python agent
315 headers = { 'User-Agent' : 'nf-core-helper/1.0' }
316
317 # Try Docker Hub
318 try :
319 req = urllib.request.Request( "https://hub.docker.com" , headers = headers)
320 urllib.request.urlopen(req, timeout = 10 )
321 docker_hub_ok = True
322 except Exception :
323 docker_hub_ok = False
324
325 # Try nf-core (for pipeline downloads)
326 try :
327 req = urllib.request.Request( "https://nf-co.re" , headers = headers)
328 urllib.request.urlopen(req, timeout = 10 )
329 nfcore_ok = True
330 except Exception :
331 nfcore_ok = False
332
333 if docker_hub_ok and nfcore_ok:
334 return CheckResult(
335 name = "Network" ,
336 passed = True ,
337 message = "Network connectivity OK (Docker Hub & nf-core reachable)"
338 )
339 elif docker_hub_ok:
340 return CheckResult(
341 name = "Network" ,
342 passed = True ,
343 message = "Docker Hub reachable (nf-core.re not reachable)" ,
344 details = "Pipeline downloads may still work via GitHub"
345 )
346 else :
347 return CheckResult(
348 name = "Network" ,
349 passed = False ,
350 message = "Cannot reach Docker Hub" ,
351 fix = "Check network connection. Containers require Docker Hub access."
352 )
353
354 except Exception as e:
355 return CheckResult(
356 name = "Network" ,
357 passed = False ,
358 message = f "Network check failed: { str (e) } " ,
359 fix = "Check network connection and proxy settings"
360 )
361
362
363 def run_all_checks () -> EnvironmentReport:
364 """Run all environment checks and return comprehensive report."""
365 checks = [
366 check_docker(),
367 check_nextflow(),
368 check_java(),
369 check_resources(),
370 check_network(),
371 ]
372
373 # Critical checks that must pass
374 critical_checks = [ "Docker" , "Nextflow" , "Java" ]
375 ready = all (c.passed for c in checks if c.name in critical_checks)
376
377 # Build recommendations
378 recommendations = []
379 for check in checks:
380 if not check.passed and check.fix:
381 recommendations.append( f " { check.name } : { check.fix } " )
382 elif check.passed and check.fix: # Warnings
383 recommendations.append( f " { check.name } (warning): { check.fix } " )
384
385 return EnvironmentReport(
386 ready = ready,
387 checks = checks,
388 recommendations = recommendations
389 )
390
391
392 def print_report (report: EnvironmentReport):
393 """Print human-readable report to stdout."""
394 print ( " \n " + "=" * 50 )
395 print ( " nf-core Environment Check" )
396 print ( "=" * 50 + " \n " )
397
398 for check in report.checks:
399 status = " \033 [92m[PASS] \033 [0m" if check.passed else " \033 [91m[FAIL] \033 [0m"
400 print ( f " { status } { check.name } : { check.message } " )
401
402 if check.details:
403 print ( f " { check.details } " )
404
405 if not check.passed and check.fix:
406 print ( f " \033 [93mFix: \033 [0m { check.fix } " )
407 elif check.passed and check.fix: # Warning
408 print ( f " \033 [93mWarning: \033 [0m { check.fix } " )
409
410 print ()
411 if report.ready:
412 print ( " \033 [92m✓ Environment is READY for nf-core pipelines. \033 [0m" )
413 else :
414 print ( " \033 [91m✗ Environment is NOT READY. Please address the issues above. \033 [0m" )
415
416 if report.recommendations:
417 print ( " \n --- Recommendations ---" )
418 for i, rec in enumerate (report.recommendations, 1 ):
419 print ( f " { i } . { rec } " )
420
421 print ()
422
423
424 def main ():
425 import argparse
426
427 parser = argparse.ArgumentParser(
428 description = "Check environment for nf-core pipeline execution" ,
429 formatter_class = argparse.RawDescriptionHelpFormatter,
430 epilog = """
431 Examples:
432 python check_environment.py # Human-readable output
433 python check_environment.py --json # JSON output for parsing
434 """
435 )
436 parser.add_argument( "--json" , action = "store_true" ,
437 help = "Output results as JSON" )
438
439 args = parser.parse_args()
440
441 report = run_all_checks()
442
443 if args.json:
444 print (json.dumps(report.to_dict(), indent = 2 ))
445 else :
446 print_report(report)
447
448 sys.exit( 0 if report.ready else 1 )
449
450
451 if __name__ == "__main__" :
452 main()