1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
# ----------------------------------------------------------------------------
# Copyright (c) 2016-2023, QIIME 2 development team.
#
# Distributed under the terms of the Modified BSD License.
#
# The full license is in the file LICENSE, distributed with this software.
# ----------------------------------------------------------------------------
import os
import subprocess
from q2_types.feature_data import AlignedDNAFASTAFormat
from q2_types.tree import NewickFormat
def run_command(cmd, output_fp, verbose=True, env=None):
if verbose:
print("Running external command line application. This may print "
"messages to stdout and/or stderr.")
print("The command being run is below. This command cannot "
"be manually re-run as it will depend on temporary files that "
"no longer exist.")
print("\nCommand:", end=' ')
print(" ".join(cmd), end='\n\n')
with open(output_fp, 'w') as output_f:
subprocess.run(cmd, stdout=output_f, check=True, env=env)
def fasttree(alignment: AlignedDNAFASTAFormat,
n_threads: int = 1) -> NewickFormat:
result = NewickFormat()
aligned_fp = str(alignment)
tree_fp = str(result)
env = None
if n_threads == 1:
cmd = ['FastTree']
else:
env = os.environ.copy()
if n_threads == 0:
env.pop('OMP_NUM_THREADS', 0)
else:
env.update({'OMP_NUM_THREADS': str(n_threads)})
cmd = ['FastTreeMP']
cmd.extend(['-quote', '-nt', aligned_fp])
run_command(cmd, tree_fp, env=env)
return result
|
