Newer
Older
from svreader import SVRecord, SVReader, SVWriter
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
from pysam import VariantFile
HOM_REF = (0, 0)
HET_VAR = (0, 1)
HOM_VAR = (1, 1)
class AnnotatedRecord(object):
"""
A lightweight object to annotated the final records
"""
def __init__(self, record):
"""
A pysam VariantRecord wrapper
"""
self.__record = record
if "SVTYPE" in record.info.keys():
self._sv_type = record.info["SVTYPE"]
else:
self._sv_type = record.alts[0][1:-1]
@property
def record(self):
return self.__record
@property
def id(self):
return self.record.id
@id.setter
def id(self, id):
self.__record.id = id
@property
def pos(self):
return self.record.pos
@property
def chrom(self):
return self.record.chrom
@property
def stop(self):
return self.record.stop
@property
def svtype(self):
return self._sv_type
@property
def filter(self):
return self.record.filter
@property
def samples(self):
return self.record.samples
class AnnotatedRecordSample(object):
"""
A lightweight object to VariantRecordSample
"""
def __init__(self, variantsample):
self.variantsample = variantsample
class VCFReader(SVReader):
super(VCFReader, self).__init__(file_name,
generic_name,
reference_handle)
self.vcf_reader = VariantFile(file_name)
self.filename = file_name
self.sv_to_report =sv_to_report
self.vcf_reader = VariantFile(file_name)
def __iter__(self):
return self
def __next__(self):
while True:
raw_record = next(self.vcf_reader)
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
return record
def getHeader(self):
return self.vcf_reader.header
def addInfo(self, name, number, type, description):
self.vcf_reader.header.info.add(id=name,
number=number,
type=type,
description=description)
def addFilter(self, name, description):
self.vcf_reader.header.filters.add(id=name,
number=None,
type=None,
description=description)
def getOrderedSamples(self):
samples = self.vcf_reader.header.samples
sample_names = [sample.rsplit('.')[0] for sample in samples]
return sample_names
def numSamples(self):
return len(self.vcf_reader.header.samples)
class VCFWriter(SVWriter):
def __init__(self, file_name, template_reader, index=True):
super(VCFWriter, self).__init__(file_name,
template_reader.tool_name,
template_reader)
def _open(self):
self.vcf_writer = VariantFile(self.filename, 'w',
header=self.template_reader.getHeader())
self._isopen = True
def _write(self, record):
self.vcf_writer.write(record.record)
def _close(self):
if self._isopen:
self.vcf_writer.close()
else: # nothing was written
self._dumpemptyvcf()