1
2
3
4
5
6
7
8
9
10
11
12
13
14 __version__ = "$Revision: 1.17 $"
15
16 import os
17 import sys
18 import tempfile
19
20 from Bio import SeqIO
21
22
24 """Helper function to build a command line string (PRIVATE).
25
26 >>> os.environ["WISE_KBYTE"]="300000"
27 >>> if os.isatty(sys.stderr.fileno()):
28 ... c = _build_align_cmdline(["dnal"], ("seq1.fna", "seq2.fna"),
29 ... "/tmp/output", kbyte=100000)
30 ... assert c == 'dnal -kbyte 100000 seq1.fna seq2.fna > /tmp/output', c
31 ... c = _build_align_cmdline(["psw"], ("seq1.faa", "seq2.faa"),
32 ... "/tmp/output_aa")
33 ... assert c == 'psw -kbyte 300000 seq1.faa seq2.faa > /tmp/output_aa', c
34 ... else:
35 ... c = _build_align_cmdline(["dnal"], ("seq1.fna", "seq2.fna"),
36 ... "/tmp/output", kbyte=100000)
37 ... assert c == 'dnal -kbyte 100000 -quiet seq1.fna seq2.fna > /tmp/output', c
38 ... c = _build_align_cmdline(["psw"], ("seq1.faa", "seq2.faa"),
39 ... "/tmp/output_aa")
40 ... assert c == 'psw -kbyte 300000 -quiet seq1.faa seq2.faa > /tmp/output_aa', c
41
42 """
43 cmdline = cmdline[:]
44
45
46
47 if kbyte is None:
48 try:
49 cmdline.extend(("-kbyte", os.environ["WISE_KBYTE"]))
50 except KeyError:
51 pass
52 else:
53 cmdline.extend(("-kbyte", str(kbyte)))
54
55 if not os.isatty(sys.stderr.fileno()):
56 cmdline.append("-quiet")
57
58 cmdline.extend(pair)
59 cmdline.extend((">", output_filename))
60 if quiet:
61 cmdline.extend(("2>", "/dev/null"))
62 cmdline_str = ' '.join(cmdline)
63
64 return cmdline_str
65
66
67 -def align(cmdline, pair, kbyte=None, force_type=None, dry_run=False, quiet=False, debug=False):
68 """
69 Returns a filehandle
70 """
71 assert len(pair) == 2
72
73 output_file = tempfile.NamedTemporaryFile(mode='r')
74 input_files = tempfile.NamedTemporaryFile(mode="w"), tempfile.NamedTemporaryFile(mode="w")
75
76 if dry_run:
77 print _build_align_cmdline(cmdline,
78 pair,
79 output_file.name,
80 kbyte,
81 force_type,
82 quiet)
83 return
84
85 for filename, input_file in zip(pair, input_files):
86
87
88
89 records = SeqIO.parse(open(filename), 'fasta')
90 SeqIO.write(records, input_file, 'fasta')
91 input_file.flush()
92
93 input_file_names = [input_file.name for input_file in input_files]
94
95 cmdline_str = _build_align_cmdline(cmdline,
96 input_file_names,
97 output_file.name,
98 kbyte,
99 force_type,
100 quiet)
101
102 if debug:
103 print >>sys.stderr, cmdline_str
104
105 status = os.system(cmdline_str) >> 8
106
107 if status > 1:
108 if kbyte != 0:
109 print >>sys.stderr, "INFO trying again with the linear model"
110 return align(cmdline, pair, 0, force_type, dry_run, quiet, debug)
111 else:
112 raise OSError("%s returned %s" % (" ".join(cmdline), status))
113
114 return output_file
115
116
118 """
119 Generate pairs list for all-against-all alignments
120
121 >>> all_pairs(range(4))
122 [(0, 1), (0, 2), (0, 3), (1, 2), (1, 3), (2, 3)]
123 """
124 pairs = []
125
126 singles = list(singles)
127 while singles:
128 suitor = singles.pop(0)
129 pairs.extend([(suitor, single) for single in singles])
130
131 return pairs
132
133
136
137
138 -def _test(*args, **keywds):
139 import doctest
140 doctest.testmod(sys.modules[__name__], *args, **keywds)
141
142 if __name__ == "__main__":
143 if __debug__:
144 _test()
145 main()
146