rev |
line source |
me@261
|
1 import sys |
bnagaev@357
|
2 import re |
me@261
|
3 |
me@315
|
4 import util |
bnagaev@418
|
5 import fileio |
me@260
|
6 |
dendik@382
|
7 # import this very module as means of having all related classes in one place |
dendik@382
|
8 import base |
dendik@382
|
9 |
me@306
|
10 default_gaps = set((".", "-", "~")) |
me@306
|
11 """Set of characters to recoginze as gaps when parsing alignment.""" |
me@306
|
12 |
me@328
|
13 class Monomer(object): |
me@328
|
14 """Monomer object.""" |
me@260
|
15 |
me@328
|
16 type = None |
me@328
|
17 """Either of 'dna', 'rna', 'protein'.""" |
me@260
|
18 |
dendik@382
|
19 types = base |
dendik@382
|
20 """Mapping of related types. SHOULD be redefined in subclasses.""" |
dendik@382
|
21 |
me@260
|
22 by_code1 = {} |
me@328
|
23 """A mapping from 1-letter code to Monomer subclass.""" |
me@328
|
24 |
me@260
|
25 by_code3 = {} |
me@328
|
26 """A mapping from 3-letter code to Monomer subclass.""" |
me@328
|
27 |
me@260
|
28 by_name = {} |
me@328
|
29 """A mapping from full monomer name to Monomer subclass.""" |
me@260
|
30 |
me@260
|
31 @classmethod |
me@328
|
32 def _subclass(cls, name='', code1='', code3='', is_modified=False): |
me@328
|
33 """Create new subclass of Monomer for given monomer type.""" |
me@328
|
34 class TheMonomer(cls): |
me@328
|
35 pass |
me@328
|
36 name = name.strip().capitalize() |
me@328
|
37 code1 = code1.upper() |
me@328
|
38 code3 = code3.upper() |
bnagaev@357
|
39 TheMonomer.__name__ = re.sub(r"[^\w]", "_", name) |
me@328
|
40 TheMonomer.name = name |
me@328
|
41 TheMonomer.code1 = code1 |
me@328
|
42 TheMonomer.code3 = code3 |
me@328
|
43 TheMonomer.is_modified = is_modified |
me@328
|
44 if not is_modified: |
me@328
|
45 cls.by_code1[code1] = TheMonomer |
me@328
|
46 cls.by_code3[code3] = TheMonomer |
me@328
|
47 cls.by_name[name] = TheMonomer |
me@328
|
48 # We duplicate distinguished long names into Monomer itself, so that we |
me@328
|
49 # can use Monomer.from_code3 to create the relevant type of monomer. |
me@328
|
50 Monomer.by_code3[code3] = TheMonomer |
me@328
|
51 Monomer.by_name[name] = TheMonomer |
me@260
|
52 |
me@328
|
53 @classmethod |
me@353
|
54 def _initialize(cls, codes=None): |
me@328
|
55 """Create all relevant subclasses of Monomer.""" |
bnagaev@378
|
56 for code1, is_modified, code3, name in codes: |
bnagaev@378
|
57 cls._subclass(name, code1, code3, is_modified) |
me@260
|
58 |
me@260
|
59 @classmethod |
me@260
|
60 def from_code1(cls, code1): |
me@328
|
61 """Create new monomer from 1-letter code.""" |
me@328
|
62 return cls.by_code1[code1.upper()]() |
me@260
|
63 |
me@260
|
64 @classmethod |
me@260
|
65 def from_code3(cls, code3): |
me@328
|
66 """Create new monomer from 3-letter code.""" |
me@328
|
67 return cls.by_code3[code3.upper()]() |
me@260
|
68 |
me@260
|
69 @classmethod |
me@260
|
70 def from_name(cls, name): |
me@328
|
71 """Create new monomer from full name.""" |
me@328
|
72 return cls.by_name[name.strip().capitalize()]() |
me@260
|
73 |
me@329
|
74 def __repr__(self): |
me@329
|
75 return '<Monomer %s>' % self.code3 |
me@329
|
76 |
me@329
|
77 def __str__(self): |
me@329
|
78 """Returns one-letter code""" |
me@329
|
79 return self.code1 |
me@329
|
80 |
me@260
|
81 def __eq__(self, other): |
me@328
|
82 """Monomers within same monomer type are compared by code1.""" |
me@328
|
83 assert self.type == other.type |
me@328
|
84 return self.code1 == other.code1 |
bnagaev@239
|
85 |
bnagaev@239
|
86 class Sequence(list): |
me@274
|
87 """Sequence of Monomers. |
bnagaev@243
|
88 |
me@274
|
89 This behaves like list of monomer objects. In addition to standard list |
me@274
|
90 behaviour, Sequence has the following attributes: |
me@270
|
91 |
me@274
|
92 * name -- str with the name of the sequence |
me@274
|
93 * description -- str with description of the sequence |
me@274
|
94 * source -- str denoting source of the sequence |
me@266
|
95 |
me@274
|
96 Any of them may be empty (i.e. hold empty string) |
me@274
|
97 """ |
me@270
|
98 |
dendik@382
|
99 types = base |
dendik@382
|
100 """Mapping of related types. SHOULD be redefined in subclasses.""" |
me@270
|
101 |
me@275
|
102 name = '' |
me@275
|
103 description = '' |
me@275
|
104 source = '' |
me@275
|
105 |
me@347
|
106 @classmethod |
me@347
|
107 def from_monomers(cls, monomers=[], name=None, description=None, source=None): |
me@347
|
108 """Create sequence from a list of monomer objecst.""" |
bnagaev@378
|
109 result = cls(monomers) |
me@275
|
110 if name: |
me@347
|
111 result.name = name |
me@275
|
112 if description: |
me@347
|
113 result.description = description |
me@275
|
114 if source: |
me@347
|
115 result.source = source |
me@347
|
116 return result |
me@347
|
117 |
me@347
|
118 @classmethod |
me@347
|
119 def from_string(cls, string, name='', description='', source=''): |
me@347
|
120 """Create sequences from string of one-letter codes.""" |
dendik@382
|
121 monomer = cls.types.Monomer.from_code1 |
me@347
|
122 monomers = [monomer(letter) for letter in string] |
me@347
|
123 return cls.from_monomers(monomers, name, description, source) |
me@270
|
124 |
me@329
|
125 def __repr__(self): |
me@329
|
126 return '<Sequence %s>' % str(self) |
me@329
|
127 |
me@262
|
128 def __str__(self): |
me@329
|
129 """Returns sequence of one-letter codes.""" |
me@275
|
130 return ''.join(monomer.code1 for monomer in self) |
me@270
|
131 |
me@316
|
132 def __hash__(self): |
me@316
|
133 """Hash sequence by identity.""" |
me@316
|
134 return id(self) |
me@316
|
135 |
me@295
|
136 class Alignment(object): |
me@295
|
137 """Alignment. It is a list of Columns.""" |
bnagaev@249
|
138 |
dendik@382
|
139 types = base |
dendik@382
|
140 """Mapping of related types. SHOULD be redefined in subclasses.""" |
me@288
|
141 |
me@289
|
142 sequences = None |
me@289
|
143 """Ordered list of sequences in alignment. Read, but DO NOT FIDDLE!""" |
bnagaev@249
|
144 |
me@287
|
145 def __init__(self): |
me@287
|
146 """Initialize empty alignment.""" |
me@287
|
147 self.sequences = [] |
me@295
|
148 self.columns = [] |
me@282
|
149 |
me@362
|
150 # Alignment grow & IO methods |
me@299
|
151 # ============================== |
me@299
|
152 |
me@294
|
153 def append_sequence(self, sequence): |
me@365
|
154 """Add sequence to alignment. Return self. |
me@294
|
155 |
me@294
|
156 If sequence is too short, pad it with gaps on the right. |
me@294
|
157 """ |
me@294
|
158 self.sequences.append(sequence) |
dendik@388
|
159 self._pad_to_width(len(sequence)) |
dendik@388
|
160 for column, monomer in zip(self.columns, sequence): |
dendik@388
|
161 column[sequence] = monomer |
me@365
|
162 return self |
me@294
|
163 |
me@364
|
164 def append_row_from_string(self, string, |
me@364
|
165 name='', description='', source='', gaps=default_gaps): |
me@364
|
166 """Add row from a string of one-letter codes and gaps. Return self.""" |
dendik@382
|
167 Sequence = self.types.Sequence |
me@349
|
168 without_gaps = util.remove_each(string, gaps) |
me@321
|
169 sequence = Sequence.from_string(without_gaps, name, description, source) |
dendik@388
|
170 self._pad_to_width(len(string)) |
dendik@388
|
171 non_gap_columns = [column |
dendik@390
|
172 for column, char in zip(self.columns, string) |
dendik@388
|
173 if char not in gaps |
dendik@388
|
174 ] |
dendik@388
|
175 for monomer, column in zip(sequence, non_gap_columns): |
dendik@388
|
176 column[sequence] = monomer |
me@287
|
177 self.sequences.append(sequence) |
me@364
|
178 return self |
me@287
|
179 |
dendik@388
|
180 def _pad_to_width(self, n): |
dendik@388
|
181 """Pad alignment with empty columns on the right to width n.""" |
dendik@388
|
182 for i in range(len(self.columns), n): |
me@302
|
183 self.columns.append(Column()) |
me@302
|
184 |
me@362
|
185 def append_file(self, file, format='fasta', gaps=default_gaps): |
me@365
|
186 """Append sequences from file to alignment. Return self. |
me@299
|
187 |
me@362
|
188 If sequences in file have gaps (detected as characters belonging to |
me@362
|
189 `gaps` set), treat them accordingly. |
me@362
|
190 """ |
bnagaev@418
|
191 sequences = [] |
bnagaev@418
|
192 if format == 'fasta': |
bnagaev@418
|
193 sequences = fileio.FastaIo(file).get_all_strings() |
bnagaev@418
|
194 elif format == 'msf': |
bnagaev@418
|
195 sequences = fileio.MsfIo(file).get_all_strings() |
bnagaev@418
|
196 else: |
bnagaev@418
|
197 raise Exception("We don't support other formats yet") |
bnagaev@418
|
198 for (name, description, body) in sequences: |
bnagaev@378
|
199 self.append_row_from_string(body, name, description, file.name, gaps) |
me@287
|
200 return self |
bnagaev@249
|
201 |
bnagaev@418
|
202 def to_file(self, file, format='fasta', gap='-'): |
me@292
|
203 """Write alignment in FASTA file as sequences with gaps.""" |
me@367
|
204 assert format == "fasta", "We don't support other formats yet" |
me@292
|
205 def char(monomer): |
me@292
|
206 if monomer: |
me@292
|
207 return monomer.code1 |
bnagaev@418
|
208 return gap |
bnagaev@418
|
209 if format == 'fasta': |
bnagaev@418
|
210 io = fileio.FastaIo(file) |
bnagaev@418
|
211 elif format == 'msf': |
bnagaev@418
|
212 io = fileio.MsfIo(file) |
bnagaev@418
|
213 else: |
bnagaev@418
|
214 raise Exception("We don't support other formats yet") |
me@292
|
215 for row in self.rows_as_lists(): |
me@292
|
216 seq = row.sequence |
me@292
|
217 line = "".join(map(char, row)) |
bnagaev@418
|
218 io.save_string(line, seq.name, seq.description) |
me@292
|
219 |
me@299
|
220 # Data access methods for alignment |
me@299
|
221 # ================================= |
me@299
|
222 |
me@299
|
223 def rows(self): |
me@299
|
224 """Return list of rows (temporary objects) in alignment. |
me@299
|
225 |
me@299
|
226 Each row is a dictionary of { column : monomer }. |
me@363
|
227 |
me@299
|
228 For gap positions there is no key for the column in row. |
me@299
|
229 |
me@299
|
230 Each row has attribute `sequence` pointing to the sequence the row is |
me@299
|
231 describing. |
me@299
|
232 |
me@299
|
233 Modifications of row have no effect on the alignment. |
me@299
|
234 """ |
me@299
|
235 # For now, the function returns a list rather than iterator. |
me@299
|
236 # It is yet to see, whether memory performance here becomes critical, |
me@299
|
237 # or is random access useful. |
me@299
|
238 rows = [] |
me@299
|
239 for sequence in self.sequences: |
me@299
|
240 row = util.UserDict() |
me@299
|
241 row.sequence = sequence |
me@299
|
242 for column in self.columns: |
me@299
|
243 if sequence in column: |
me@299
|
244 row[column] = column[sequence] |
me@299
|
245 rows.append(row) |
me@299
|
246 return rows |
me@299
|
247 |
me@299
|
248 def rows_as_lists(self): |
me@299
|
249 """Return list of rows (temporary objects) in alignment. |
me@299
|
250 |
me@299
|
251 Each row here is a list of either monomer or None (for gaps). |
me@299
|
252 |
me@299
|
253 Each row has attribute `sequence` pointing to the sequence of row. |
me@299
|
254 |
me@299
|
255 Modifications of row have no effect on the alignment. |
me@299
|
256 """ |
me@299
|
257 rows = [] |
me@299
|
258 for sequence in self.sequences: |
me@299
|
259 row = util.UserList() |
me@299
|
260 row.sequence = sequence |
me@299
|
261 for column in self.columns: |
me@299
|
262 row.append(column.get(sequence)) |
me@299
|
263 rows.append(row) |
me@299
|
264 return rows |
me@299
|
265 |
me@299
|
266 def columns_as_lists(self): |
me@299
|
267 """Return list of columns (temorary objects) in alignment. |
me@299
|
268 |
me@299
|
269 Each column here is a list of either monomer or None (for gaps). |
me@299
|
270 |
me@299
|
271 Items of column are sorted in the same way as alignment.sequences. |
me@299
|
272 |
me@299
|
273 Modifications of column have no effect on the alignment. |
me@299
|
274 """ |
me@299
|
275 columns = [] |
me@299
|
276 for column in self.columns: |
me@299
|
277 col = [] |
me@299
|
278 for sequence in self.sequences: |
me@299
|
279 col.append(column.get(sequence)) |
me@299
|
280 columns.append(col) |
me@299
|
281 return columns |
me@299
|
282 |
me@368
|
283 # Alignment / Block editing methods |
me@368
|
284 # ================================= |
me@368
|
285 |
me@368
|
286 def _flush_row(self, row, whence='left'): |
me@368
|
287 """Helper for `flush`: flush to one side all monomers in one row.""" |
me@368
|
288 row = filter(None, row) |
me@368
|
289 padding = [None] * len(self.columns) |
me@368
|
290 if whence == 'left': |
me@368
|
291 return row + padding |
me@368
|
292 if whence == 'right': |
me@368
|
293 return padding + row |
me@368
|
294 if whence == 'center': |
me@368
|
295 pad_len = (len(self.columns) - len(row)) // 2 |
me@368
|
296 # vvv fix padding for case when length is odd: better have more |
me@368
|
297 pad_len += len(self.columns) - 2 * pad_len |
me@368
|
298 padding = [None] * pad_len |
me@368
|
299 return padding + row + padding |
me@368
|
300 assert True, "whence must be either 'left' or 'right' or 'center'" |
me@368
|
301 |
me@368
|
302 def flush(self, whence='left'): |
me@368
|
303 """Remove all gaps from alignment and flush results to one side. |
me@368
|
304 |
me@368
|
305 `whence` must be one of 'left', 'right' or 'center' |
me@368
|
306 """ |
me@368
|
307 for row in self.rows_as_lists(): |
me@368
|
308 sequence = row.sequence |
me@368
|
309 row = self._flush_row(row, whence) |
me@368
|
310 for monomer, column in zip(row, self.columns): |
me@368
|
311 if monomer: |
me@368
|
312 column[sequence] = monomer |
me@368
|
313 elif sequence in column: |
me@368
|
314 del column[sequence] |
me@368
|
315 |
me@369
|
316 def remove_gap_columns(self): |
me@369
|
317 """Remove all empty columns.""" |
me@369
|
318 for n, column in reversed(enumerate(self.columns)): |
me@369
|
319 if column == {}: |
me@369
|
320 self.columns[n:n+1] = [] |
me@369
|
321 |
me@371
|
322 def _wipe(self): |
me@371
|
323 """Make all positions gaps (but keep sequences intact).""" |
me@371
|
324 for column in self.columns: |
dendik@412
|
325 for sequence in list(column): |
me@371
|
326 del column[sequence] |
me@371
|
327 |
me@372
|
328 def _merge(self, dst, new, merge): |
me@373
|
329 """Replace contents of `dst` with those of `new`. |
me@372
|
330 |
me@372
|
331 Replace contents of elements using function `merge(dst_el, new_le)`. |
me@372
|
332 """ |
bnagaev@384
|
333 for el, new_el in zip(dst, new): |
bnagaev@384
|
334 merge(el, new_el) |
me@372
|
335 dst[len(dst):] = new[len(dst):] |
me@372
|
336 del dst[len(new):] |
me@371
|
337 |
me@373
|
338 def _replace_sequence_contents(self, new, copy_descriptions): |
me@373
|
339 """Replace contents of sequences with those of `new` alignment.""" |
me@371
|
340 # XXX: we manually copy sequence contents here |
me@372
|
341 # XXX: we only copy, overlapping parts and link to the rest |
me@372
|
342 def merge_monomers(dst, new): |
me@372
|
343 dst.__class__ = new.__class__ |
me@372
|
344 def merge_sequences(dst, new): |
me@373
|
345 if copy_descriptions: |
me@373
|
346 vars(dst).update(vars(new)) |
me@372
|
347 self._merge(dst, new, merge_monomers) |
me@372
|
348 self._merge(self.sequences, new.sequences, merge_sequences) |
me@371
|
349 |
me@371
|
350 def _replace_column_contents(self, new): |
me@373
|
351 """Replace column contents with those of `new` alignment. |
me@371
|
352 |
me@373
|
353 Synonym: copy gap patterns from `new` to `self`. |
me@372
|
354 |
me@373
|
355 `self.sequences` and `new.sequences` should have the same contents. |
me@371
|
356 """ |
me@371
|
357 self._wipe() |
me@371
|
358 not_gap = lambda (a,b): a != None |
me@371
|
359 for sequence, new_row in zip(self.sequences, new.rows_as_lists()): |
me@371
|
360 assert len(sequence) == len(new_row.sequence) |
dendik@388
|
361 non_gap_columns = [column |
dendik@388
|
362 for column, monomer in zip(self.columns, new_row) |
dendik@388
|
363 if monomer |
dendik@388
|
364 ] |
dendik@388
|
365 for monomer, column in zip(sequence, non_gap_columns): |
dendik@388
|
366 column[sequence] = monomer |
me@371
|
367 |
me@373
|
368 def _replace_contents(self, new, copy_descriptions, copy_contents): |
me@371
|
369 """Replace alignment contents with those of other alignment.""" |
me@373
|
370 if copy_contents: |
me@373
|
371 self._replace_sequence_contents(new, copy_descriptions) |
bnagaev@378
|
372 self._replace_column_contents(new) |
me@371
|
373 |
me@373
|
374 def process(self, function, copy_descriptions=True, copy_contents=True): |
me@371
|
375 """Apply function to the alignment (or block); inject results back. |
me@371
|
376 |
me@373
|
377 - `function(block)` must return block with same line order. |
me@373
|
378 - if `copy_descriptions` is False, ignore new sequence names. |
me@373
|
379 - if `copy_contents` is False, don't copy sequence contents too. |
dendik@380
|
380 |
dendik@380
|
381 `function` (object) may have attributes `copy_descriptions` and |
dendik@380
|
382 `copy_contents`, which override the same named arguments. |
me@371
|
383 """ |
me@371
|
384 new = function(self) |
dendik@380
|
385 if hasattr(function, 'copy_descriptions'): |
dendik@380
|
386 copy_descriptions = function.copy_descriptions |
dendik@380
|
387 if hasattr(function, 'copy_contents'): |
dendik@380
|
388 copy_contents = function.copy_contents |
me@373
|
389 self._replace_contents(new, copy_descriptions, copy_contents) |
me@371
|
390 |
me@300
|
391 class Column(dict): |
me@300
|
392 """Column of alignment. |
me@300
|
393 |
me@300
|
394 Column is a dict of { sequence : monomer }. |
me@300
|
395 |
me@300
|
396 For sequences that have gaps in current row, given key is not present in |
me@300
|
397 the column. |
me@300
|
398 """ |
me@325
|
399 |
dendik@382
|
400 types = base |
dendik@382
|
401 """Mapping of related types. SHOULD be redefined in subclasses.""" |
dendik@382
|
402 |
me@325
|
403 def __hash__(self): |
me@325
|
404 """Return hash by identity.""" |
me@325
|
405 return id(self) |
me@300
|
406 |
me@317
|
407 class Block(Alignment): |
me@307
|
408 """Block of alignment. |
me@301
|
409 |
dendik@415
|
410 Block is an intersection of several rows & columns. (The collections of |
dendik@415
|
411 rows and columns are represented as ordered lists, to retain display order |
dendik@415
|
412 of Alignment or add ability to tweak it). Most of blocks look like |
dendik@415
|
413 rectangular part of alignment if you shuffle alignment rows the right way. |
me@261
|
414 """ |
me@270
|
415 |
me@307
|
416 alignment = None |
me@307
|
417 """Alignment the block belongs to.""" |
me@270
|
418 |
me@307
|
419 sequences = () |
me@307
|
420 """List of sequences in block.""" |
me@307
|
421 |
me@307
|
422 columns = () |
me@307
|
423 """List of columns in block.""" |
me@307
|
424 |
me@317
|
425 @classmethod |
me@317
|
426 def from_alignment(cls, alignment, sequences=None, columns=None): |
me@307
|
427 """Build new block from alignment. |
me@307
|
428 |
me@307
|
429 If sequences are not given, the block uses all sequences in alignment. |
me@307
|
430 |
me@307
|
431 If columns are not given, the block uses all columns in alignment. |
me@307
|
432 |
me@307
|
433 In both cases we use exactly the list used in alignment, thus, if new |
me@307
|
434 sequences or columns are added to alignment, the block tracks this too. |
me@261
|
435 """ |
me@307
|
436 if sequences is None: |
me@307
|
437 sequences = alignment.sequences |
me@318
|
438 if columns is None: |
me@307
|
439 columns = alignment.columns |
me@320
|
440 block = cls() |
me@320
|
441 block.alignment = alignment |
me@320
|
442 block.sequences = sequences |
me@320
|
443 block.columns = columns |
me@320
|
444 return block |
me@270
|
445 |
me@260
|
446 # vim: set ts=4 sts=4 sw=4 et: |