jpayne@68: # cython: embedsignature=True jpayne@68: # jpayne@68: # Code to read, write and edit VCF files jpayne@68: # jpayne@68: # VCF lines are encoded as a dictionary with these keys (note: all lowercase): jpayne@68: # 'chrom': string jpayne@68: # 'pos': integer jpayne@68: # 'id': string jpayne@68: # 'ref': string jpayne@68: # 'alt': list of strings jpayne@68: # 'qual': integer jpayne@68: # 'filter': None (missing value), or list of keys (strings); empty list parsed as ["PASS"] jpayne@68: # 'info': dictionary of values (see below) jpayne@68: # 'format': list of keys (strings) jpayne@68: # sample keys: dictionary of values (see below) jpayne@68: # jpayne@68: # The sample keys are accessible through vcf.getsamples() jpayne@68: # jpayne@68: # A dictionary of values contains value keys (defined in ##INFO or jpayne@68: # ##FORMAT lines) which map to a list, containing integers, floats, jpayne@68: # strings, or characters. Missing values are replaced by a particular jpayne@68: # value, often -1 or . jpayne@68: # jpayne@68: # Genotypes are not stored as a string, but as a list of 1 or 3 jpayne@68: # elements (for haploid and diploid samples), the first (and last) the jpayne@68: # integer representing an allele, and the second the separation jpayne@68: # character. Note that there is just one genotype per sample, but for jpayne@68: # consistency the single element is stored in a list. jpayne@68: # jpayne@68: # Header lines other than ##INFO, ##FORMAT and ##FILTER are stored as jpayne@68: # (key, value) pairs and are accessible through getheader() jpayne@68: # jpayne@68: # The VCF class can be instantiated with a 'regions' variable jpayne@68: # consisting of tuples (chrom,start,end) encoding 0-based half-open jpayne@68: # segments. Only variants with a position inside the segment will be jpayne@68: # parsed. A regions parser is available under parse_regions. jpayne@68: # jpayne@68: # When instantiated, a reference can be passed to the VCF class. This jpayne@68: # may be any class that supports a fetch(chrom, start, end) method. jpayne@68: # jpayne@68: # NOTE: the position that is returned to Python is 0-based, NOT jpayne@68: # 1-based as in the VCF file. jpayne@68: # NOTE: There is also preliminary VCF functionality in the VariantFile class. jpayne@68: # jpayne@68: # TODO: jpayne@68: # only v4.0 writing is complete; alleles are not converted to v3.3 format jpayne@68: # jpayne@68: jpayne@68: from collections import namedtuple, defaultdict jpayne@68: from operator import itemgetter jpayne@68: import sys, re, copy, bisect jpayne@68: jpayne@68: from libc.stdlib cimport atoi jpayne@68: from libc.stdint cimport int8_t, int16_t, int32_t, int64_t jpayne@68: from libc.stdint cimport uint8_t, uint16_t, uint32_t, uint64_t jpayne@68: jpayne@68: cimport pysam.libctabix as libctabix jpayne@68: cimport pysam.libctabixproxies as libctabixproxies jpayne@68: jpayne@68: from pysam.libcutils cimport force_str jpayne@68: jpayne@68: import pysam jpayne@68: jpayne@68: gtsRegEx = re.compile("[|/\\\\]") jpayne@68: alleleRegEx = re.compile('^[ACGTN]+$') jpayne@68: jpayne@68: # Utility function. Uses 0-based coordinates jpayne@68: def get_sequence(chrom, start, end, fa): jpayne@68: # obtain sequence from .fa file, without truncation jpayne@68: if end<=start: return "" jpayne@68: if not fa: return "N"*(end-start) jpayne@68: if start<0: return "N"*(-start) + get_sequence(chrom, 0, end, fa).upper() jpayne@68: sequence = fa.fetch(chrom, start, end).upper() jpayne@68: if len(sequence) < end-start: sequence += "N"*(end-start-len(sequence)) jpayne@68: return sequence jpayne@68: jpayne@68: # Utility function. Parses a region string jpayne@68: def parse_regions( string ): jpayne@68: result = [] jpayne@68: for r in string.split(','): jpayne@68: elts = r.split(':') jpayne@68: chrom, start, end = elts[0], 0, 3000000000 jpayne@68: if len(elts)==1: pass jpayne@68: elif len(elts)==2: jpayne@68: if len(elts[1])>0: jpayne@68: ielts = elts[1].split('-') jpayne@68: if len(ielts) != 2: ValueError("Don't understand region string '%s'" % r) jpayne@68: try: start, end = int(ielts[0])-1, int(ielts[1]) jpayne@68: except: raise ValueError("Don't understand region string '%s'" % r) jpayne@68: else: jpayne@68: raise ValueError("Don't understand region string '%s'" % r) jpayne@68: result.append( (chrom,start,end) ) jpayne@68: return result jpayne@68: jpayne@68: jpayne@68: FORMAT = namedtuple('FORMAT','id numbertype number type description missingvalue') jpayne@68: jpayne@68: ########################################################################################################### jpayne@68: # jpayne@68: # New class jpayne@68: # jpayne@68: ########################################################################################################### jpayne@68: jpayne@68: cdef class VCFRecord(libctabixproxies.TupleProxy): jpayne@68: '''vcf record. jpayne@68: jpayne@68: initialized from data and vcf meta jpayne@68: ''' jpayne@68: jpayne@68: cdef vcf jpayne@68: cdef char * contig jpayne@68: cdef uint32_t pos jpayne@68: jpayne@68: def __init__(self, vcf): jpayne@68: self.vcf = vcf jpayne@68: self.encoding = vcf.encoding jpayne@68: jpayne@68: # if len(data) != len(self.vcf._samples): jpayne@68: # self.vcf.error(str(data), jpayne@68: # self.BAD_NUMBER_OF_COLUMNS, jpayne@68: # "expected %s for %s samples (%s), got %s" % \ jpayne@68: # (len(self.vcf._samples), jpayne@68: # len(self.vcf._samples), jpayne@68: # self.vcf._samples, jpayne@68: # len(data))) jpayne@68: jpayne@68: def __cinit__(self, vcf): jpayne@68: # start indexed access at genotypes jpayne@68: self.offset = 9 jpayne@68: jpayne@68: self.vcf = vcf jpayne@68: self.encoding = vcf.encoding jpayne@68: jpayne@68: def error(self, line, error, opt=None): jpayne@68: '''raise error.''' jpayne@68: # pass to vcf file for error handling jpayne@68: return self.vcf.error(line, error, opt) jpayne@68: jpayne@68: cdef update(self, char * buffer, size_t nbytes): jpayne@68: '''update internal data. jpayne@68: jpayne@68: nbytes does not include the terminal '\0'. jpayne@68: ''' jpayne@68: libctabixproxies.TupleProxy.update(self, buffer, nbytes) jpayne@68: jpayne@68: self.contig = self.fields[0] jpayne@68: # vcf counts from 1 - correct here jpayne@68: self.pos = atoi(self.fields[1]) - 1 jpayne@68: jpayne@68: def __len__(self): jpayne@68: return max(0, self.nfields - 9) jpayne@68: jpayne@68: property contig: jpayne@68: def __get__(self): return self.contig jpayne@68: jpayne@68: property pos: jpayne@68: def __get__(self): return self.pos jpayne@68: jpayne@68: property id: jpayne@68: def __get__(self): return self.fields[2] jpayne@68: jpayne@68: property ref: jpayne@68: def __get__(self): jpayne@68: return self.fields[3] jpayne@68: jpayne@68: property alt: jpayne@68: def __get__(self): jpayne@68: # convert v3.3 to v4.0 alleles below jpayne@68: alt = self.fields[4] jpayne@68: if alt == ".": alt = [] jpayne@68: else: alt = alt.upper().split(',') jpayne@68: return alt jpayne@68: jpayne@68: property qual: jpayne@68: def __get__(self): jpayne@68: qual = self.fields[5] jpayne@68: if qual == b".": qual = -1 jpayne@68: else: jpayne@68: try: qual = float(qual) jpayne@68: except: self.vcf.error(str(self),self.QUAL_NOT_NUMERICAL) jpayne@68: return qual jpayne@68: jpayne@68: property filter: jpayne@68: def __get__(self): jpayne@68: f = self.fields[6] jpayne@68: # postpone checking that filters exist. Encode missing filter or no filtering as empty list jpayne@68: if f == b"." or f == b"PASS" or f == b"0": return [] jpayne@68: else: return f.split(';') jpayne@68: jpayne@68: property info: jpayne@68: def __get__(self): jpayne@68: col = self.fields[7] jpayne@68: # dictionary of keys, and list of values jpayne@68: info = {} jpayne@68: if col != b".": jpayne@68: for blurp in col.split(';'): jpayne@68: elts = blurp.split('=') jpayne@68: if len(elts) == 1: v = None jpayne@68: elif len(elts) == 2: v = elts[1] jpayne@68: else: self.vcf.error(str(self),self.ERROR_INFO_STRING) jpayne@68: info[elts[0]] = self.vcf.parse_formatdata(elts[0], v, self.vcf._info, str(self.vcf)) jpayne@68: return info jpayne@68: jpayne@68: property format: jpayne@68: def __get__(self): jpayne@68: return self.fields[8].split(':') jpayne@68: jpayne@68: property samples: jpayne@68: def __get__(self): jpayne@68: return self.vcf._samples jpayne@68: jpayne@68: def __getitem__(self, key): jpayne@68: jpayne@68: # parse sample columns jpayne@68: values = self.fields[self.vcf._sample2column[key]].split(':') jpayne@68: alt = self.alt jpayne@68: format = self.format jpayne@68: jpayne@68: if len(values) > len(format): jpayne@68: self.vcf.error(str(self.line),self.BAD_NUMBER_OF_VALUES,"(found %s values in element %s; expected %s)" %\ jpayne@68: (len(values),key,len(format))) jpayne@68: jpayne@68: result = {} jpayne@68: for idx in range(len(format)): jpayne@68: expected = self.vcf.get_expected(format[idx], self.vcf._format, alt) jpayne@68: if idx < len(values): value = values[idx] jpayne@68: else: jpayne@68: if expected == -1: value = "." jpayne@68: else: value = ",".join(["."]*expected) jpayne@68: jpayne@68: result[format[idx]] = self.vcf.parse_formatdata(format[idx], value, self.vcf._format, str(self.data)) jpayne@68: if expected != -1 and len(result[format[idx]]) != expected: jpayne@68: self.vcf.error(str(self.data),self.BAD_NUMBER_OF_PARAMETERS, jpayne@68: "id=%s, expected %s parameters, got %s" % (format[idx],expected,result[format[idx]])) jpayne@68: if len(result[format[idx]] ) < expected: result[format[idx]] += [result[format[idx]][-1]]*(expected-len(result[format[idx]])) jpayne@68: result[format[idx]] = result[format[idx]][:expected] jpayne@68: jpayne@68: return result jpayne@68: jpayne@68: jpayne@68: cdef class asVCFRecord(libctabix.Parser): jpayne@68: '''converts a :term:`tabix row` into a VCF record.''' jpayne@68: cdef vcffile jpayne@68: def __init__(self, vcffile): jpayne@68: self.vcffile = vcffile jpayne@68: jpayne@68: cdef parse(self, char * buffer, int len): jpayne@68: cdef VCFRecord r jpayne@68: r = VCFRecord(self.vcffile) jpayne@68: r.copy(buffer, len) jpayne@68: return r jpayne@68: jpayne@68: class VCF(object): jpayne@68: jpayne@68: # types jpayne@68: NT_UNKNOWN = 0 jpayne@68: NT_NUMBER = 1 jpayne@68: NT_ALLELES = 2 jpayne@68: NT_NR_ALLELES = 3 jpayne@68: NT_GENOTYPES = 4 jpayne@68: NT_PHASED_GENOTYPES = 5 jpayne@68: jpayne@68: _errors = { 0:"UNKNOWN_FORMAT_STRING:Unknown file format identifier", jpayne@68: 1:"BADLY_FORMATTED_FORMAT_STRING:Formatting error in the format string", jpayne@68: 2:"BADLY_FORMATTED_HEADING:Did not find 9 required headings (CHROM, POS, ..., FORMAT) %s", jpayne@68: 3:"BAD_NUMBER_OF_COLUMNS:Wrong number of columns found (%s)", jpayne@68: 4:"POS_NOT_NUMERICAL:Position column is not numerical", jpayne@68: 5:"UNKNOWN_CHAR_IN_REF:Unknown character in reference field", jpayne@68: 6:"V33_BAD_REF:Reference should be single-character in v3.3 VCF", jpayne@68: 7:"V33_BAD_ALLELE:Cannot interpret allele for v3.3 VCF", jpayne@68: 8:"POS_NOT_POSITIVE:Position field must be >0", jpayne@68: 9:"QUAL_NOT_NUMERICAL:Quality field must be numerical, or '.'", jpayne@68: 10:"ERROR_INFO_STRING:Error while parsing info field", jpayne@68: 11:"ERROR_UNKNOWN_KEY:Unknown key (%s) found in formatted field (info; format; or filter)", jpayne@68: 12:"ERROR_FORMAT_NOT_NUMERICAL:Expected integer or float in formatted field; got %s", jpayne@68: 13:"ERROR_FORMAT_NOT_CHAR:Eexpected character in formatted field; got string", jpayne@68: 14:"FILTER_NOT_DEFINED:Identifier (%s) in filter found which was not defined in header", jpayne@68: 15:"FORMAT_NOT_DEFINED:Identifier (%s) in format found which was not defined in header", jpayne@68: 16:"BAD_NUMBER_OF_VALUES:Found too many of values in sample column (%s)", jpayne@68: 17:"BAD_NUMBER_OF_PARAMETERS:Found unexpected number of parameters (%s)", jpayne@68: 18:"BAD_GENOTYPE:Cannot parse genotype (%s)", jpayne@68: 19:"V40_BAD_ALLELE:Bad allele found for v4.0 VCF (%s)", jpayne@68: 20:"MISSING_REF:Reference allele missing", jpayne@68: 21:"V33_UNMATCHED_DELETION:Deleted sequence does not match reference (%s)", jpayne@68: 22:"V40_MISSING_ANGLE_BRACKETS:Format definition is not deliminted by angular brackets", jpayne@68: 23:"FORMAT_MISSING_QUOTES:Description field in format definition is not surrounded by quotes", jpayne@68: 24:"V40_FORMAT_MUST_HAVE_NAMED_FIELDS:Fields in v4.0 VCF format definition must have named fields", jpayne@68: 25:"HEADING_NOT_SEPARATED_BY_TABS:Heading line appears separated by spaces, not tabs", jpayne@68: 26:"WRONG_REF:Wrong reference %s", jpayne@68: 27:"ERROR_TRAILING_DATA:Numerical field ('%s') has semicolon-separated trailing data", jpayne@68: 28:"BAD_CHR_TAG:Error calculating chr tag for %s", jpayne@68: 29:"ZERO_LENGTH_ALLELE:Found zero-length allele", jpayne@68: 30:"MISSING_INDEL_ALLELE_REF_BASE:Indel alleles must begin with single reference base", jpayne@68: 31:"ZERO_FOR_NON_FLAG_FIELD: number set to 0, but type is not 'FLAG'", jpayne@68: 32:"ERROR_FORMAT_NOT_INTEGER:Expected integer in formatted field; got %s", jpayne@68: 33:"ERROR_FLAG_HAS_VALUE:Flag fields should not have a value", jpayne@68: } jpayne@68: jpayne@68: # tag-value pairs; tags are not unique; does not include fileformat, INFO, FILTER or FORMAT fields jpayne@68: _header = [] jpayne@68: jpayne@68: # version number; 33=v3.3; 40=v4.0 jpayne@68: _version = 40 jpayne@68: jpayne@68: # info, filter and format data jpayne@68: _info = {} jpayne@68: _filter = {} jpayne@68: _format = {} jpayne@68: jpayne@68: # header; and required columns jpayne@68: _required = ["CHROM","POS","ID","REF","ALT","QUAL","FILTER","INFO","FORMAT"] jpayne@68: _samples = [] jpayne@68: jpayne@68: # control behaviour jpayne@68: _ignored_errors = set([11,31]) # ERROR_UNKNOWN_KEY, ERROR_ZERO_FOR_NON_FLAG_FIELD jpayne@68: _warn_errors = set([]) jpayne@68: _leftalign = False jpayne@68: jpayne@68: # reference sequence jpayne@68: _reference = None jpayne@68: jpayne@68: # regions to include; None includes everything jpayne@68: _regions = None jpayne@68: jpayne@68: # statefull stuff jpayne@68: _lineno = -1 jpayne@68: _line = None jpayne@68: _lines = None jpayne@68: jpayne@68: def __init__(self, _copy=None, reference=None, regions=None, jpayne@68: lines=None, leftalign=False): jpayne@68: # make error identifiers accessible by name jpayne@68: for id in self._errors.keys(): jpayne@68: self.__dict__[self._errors[id].split(':')[0]] = id jpayne@68: if _copy != None: jpayne@68: self._leftalign = _copy._leftalign jpayne@68: self._header = _copy._header[:] jpayne@68: self._version = _copy._version jpayne@68: self._info = copy.deepcopy(_copy._info) jpayne@68: self._filter = copy.deepcopy(_copy._filter) jpayne@68: self._format = copy.deepcopy(_copy._format) jpayne@68: self._samples = _copy._samples[:] jpayne@68: self._sample2column = copy.deepcopy(_copy._sample2column) jpayne@68: self._ignored_errors = copy.deepcopy(_copy._ignored_errors) jpayne@68: self._warn_errors = copy.deepcopy(_copy._warn_errors) jpayne@68: self._reference = _copy._reference jpayne@68: self._regions = _copy._regions jpayne@68: if reference: self._reference = reference jpayne@68: if regions: self._regions = regions jpayne@68: if leftalign: self._leftalign = leftalign jpayne@68: self._lines = lines jpayne@68: self.encoding = "ascii" jpayne@68: self.tabixfile = None jpayne@68: jpayne@68: def error(self,line,error,opt=None): jpayne@68: if error in self._ignored_errors: return jpayne@68: errorlabel, errorstring = self._errors[error].split(':') jpayne@68: if opt: errorstring = errorstring % opt jpayne@68: errwarn = ["Error","Warning"][error in self._warn_errors] jpayne@68: errorstring += " in line %s: '%s'\n%s %s: %s\n" % (self._lineno,line,errwarn,errorlabel,errorstring) jpayne@68: if error in self._warn_errors: return jpayne@68: raise ValueError(errorstring) jpayne@68: jpayne@68: def parse_format(self,line,format,filter=False): jpayne@68: if self._version == 40: jpayne@68: if not format.startswith('<'): jpayne@68: self.error(line,self.V40_MISSING_ANGLE_BRACKETS) jpayne@68: format = "<"+format jpayne@68: if not format.endswith('>'): jpayne@68: self.error(line,self.V40_MISSING_ANGLE_BRACKETS) jpayne@68: format += ">" jpayne@68: format = format[1:-1] jpayne@68: data = {'id':None,'number':None,'type':None,'descr':None} jpayne@68: idx = 0 jpayne@68: while len(format.strip())>0: jpayne@68: elts = format.strip().split(',') jpayne@68: first, rest = elts[0], ','.join(elts[1:]) jpayne@68: if first.find('=') == -1 or (first.find('"')>=0 and first.find('=') > first.find('"')): jpayne@68: if self._version == 40: self.error(line,self.V40_FORMAT_MUST_HAVE_NAMED_FIELDS) jpayne@68: if idx == 4: self.error(line,self.BADLY_FORMATTED_FORMAT_STRING) jpayne@68: first = ["ID=","Number=","Type=","Description="][idx] + first jpayne@68: if first.startswith('ID='): data['id'] = first.split('=')[1] jpayne@68: elif first.startswith('Number='): data['number'] = first.split('=')[1] jpayne@68: elif first.startswith('Type='): data['type'] = first.split('=')[1] jpayne@68: elif first.startswith('Description='): jpayne@68: elts = format.split('"') jpayne@68: if len(elts)<3: jpayne@68: self.error(line,self.FORMAT_MISSING_QUOTES) jpayne@68: elts = first.split('=') + [rest] jpayne@68: data['descr'] = elts[1] jpayne@68: rest = '"'.join(elts[2:]) jpayne@68: if rest.startswith(','): rest = rest[1:] jpayne@68: else: jpayne@68: self.error(line,self.BADLY_FORMATTED_FORMAT_STRING) jpayne@68: format = rest jpayne@68: idx += 1 jpayne@68: if filter and idx==1: idx=3 # skip number and type fields for FILTER format strings jpayne@68: if not data['id']: self.error(line,self.BADLY_FORMATTED_FORMAT_STRING) jpayne@68: if 'descr' not in data: jpayne@68: # missing description jpayne@68: self.error(line,self.BADLY_FORMATTED_FORMAT_STRING) jpayne@68: data['descr'] = "" jpayne@68: if not data['type'] and not data['number']: jpayne@68: # fine, ##filter format jpayne@68: return FORMAT(data['id'],self.NT_NUMBER,0,"Flag",data['descr'],'.') jpayne@68: if not data['type'] in ["Integer","Float","Character","String","Flag"]: jpayne@68: self.error(line,self.BADLY_FORMATTED_FORMAT_STRING) jpayne@68: # I would like a missing-value field, but it isn't there jpayne@68: if data['type'] in ['Integer','Float']: data['missing'] = None # Do NOT use arbitrary int/float as missing value jpayne@68: else: data['missing'] = '.' jpayne@68: if not data['number']: self.error(line,self.BADLY_FORMATTED_FORMAT_STRING) jpayne@68: try: jpayne@68: n = int(data['number']) jpayne@68: t = self.NT_NUMBER jpayne@68: except ValueError: jpayne@68: n = -1 jpayne@68: if data['number'] == '.': t = self.NT_UNKNOWN jpayne@68: elif data['number'] == '#alleles': t = self.NT_ALLELES jpayne@68: elif data['number'] == '#nonref_alleles': t = self.NT_NR_ALLELES jpayne@68: elif data['number'] == '#genotypes': t = self.NT_GENOTYPES jpayne@68: elif data['number'] == '#phased_genotypes': t = self.NT_PHASED_GENOTYPES jpayne@68: elif data['number'] == '#phased_genotypes': t = self.NT_PHASED_GENOTYPES jpayne@68: # abbreviations added in VCF version v4.1 jpayne@68: elif data['number'] == 'A': t = self.NT_ALLELES jpayne@68: elif data['number'] == 'G': t = self.NT_GENOTYPES jpayne@68: else: jpayne@68: self.error(line,self.BADLY_FORMATTED_FORMAT_STRING) jpayne@68: # if number is 0 - type must be Flag jpayne@68: if n == 0 and data['type'] != 'Flag': jpayne@68: self.error( line, self.ZERO_FOR_NON_FLAG_FIELD) jpayne@68: # force type 'Flag' if no number jpayne@68: data['type'] = 'Flag' jpayne@68: jpayne@68: return FORMAT(data['id'],t,n,data['type'],data['descr'],data['missing']) jpayne@68: jpayne@68: def format_format( self, fmt, filter=False ): jpayne@68: values = [('ID',fmt.id)] jpayne@68: if fmt.number != None and not filter: jpayne@68: if fmt.numbertype == self.NT_UNKNOWN: nmb = "." jpayne@68: elif fmt.numbertype == self.NT_NUMBER: nmb = str(fmt.number) jpayne@68: elif fmt.numbertype == self.NT_ALLELES: nmb = "#alleles" jpayne@68: elif fmt.numbertype == self.NT_NR_ALLELES: nmb = "#nonref_alleles" jpayne@68: elif fmt.numbertype == self.NT_GENOTYPES: nmb = "#genotypes" jpayne@68: elif fmt.numbertype == self.NT_PHASED_GENOTYPES: nmb = "#phased_genotypes" jpayne@68: else: jpayne@68: raise ValueError("Unknown number type encountered: %s" % fmt.numbertype) jpayne@68: values.append( ('Number',nmb) ) jpayne@68: values.append( ('Type', fmt.type) ) jpayne@68: values.append( ('Description', '"' + fmt.description + '"') ) jpayne@68: if self._version == 33: jpayne@68: format = ",".join([v for k,v in values]) jpayne@68: else: jpayne@68: format = "<" + (",".join( ["%s=%s" % (k,v) for (k,v) in values] )) + ">" jpayne@68: return format jpayne@68: jpayne@68: def get_expected(self, format, formatdict, alt): jpayne@68: fmt = formatdict[format] jpayne@68: if fmt.numbertype == self.NT_UNKNOWN: return -1 jpayne@68: if fmt.numbertype == self.NT_NUMBER: return fmt.number jpayne@68: if fmt.numbertype == self.NT_ALLELES: return len(alt)+1 jpayne@68: if fmt.numbertype == self.NT_NR_ALLELES: return len(alt) jpayne@68: if fmt.numbertype == self.NT_GENOTYPES: return ((len(alt)+1)*(len(alt)+2)) // 2 jpayne@68: if fmt.numbertype == self.NT_PHASED_GENOTYPES: return (len(alt)+1)*(len(alt)+1) jpayne@68: return 0 jpayne@68: jpayne@68: jpayne@68: def _add_definition(self, formatdict, key, data, line ): jpayne@68: if key in formatdict: return jpayne@68: self.error(line,self.ERROR_UNKNOWN_KEY,key) jpayne@68: if data == None: jpayne@68: formatdict[key] = FORMAT(key,self.NT_NUMBER,0,"Flag","(Undefined tag)",".") jpayne@68: return jpayne@68: if data == []: data = [""] # unsure what type -- say string jpayne@68: if type(data[0]) == type(0.0): jpayne@68: formatdict[key] = FORMAT(key,self.NT_UNKNOWN,-1,"Float","(Undefined tag)",None) jpayne@68: return jpayne@68: if type(data[0]) == type(0): jpayne@68: formatdict[key] = FORMAT(key,self.NT_UNKNOWN,-1,"Integer","(Undefined tag)",None) jpayne@68: return jpayne@68: formatdict[key] = FORMAT(key,self.NT_UNKNOWN,-1,"String","(Undefined tag)",".") jpayne@68: jpayne@68: jpayne@68: # todo: trim trailing missing values jpayne@68: def format_formatdata( self, data, format, key=True, value=True, separator=":" ): jpayne@68: output, sdata = [], [] jpayne@68: if type(data) == type([]): # for FORMAT field, make data with dummy values jpayne@68: d = {} jpayne@68: for k in data: d[k] = [] jpayne@68: data = d jpayne@68: # convert missing values; and silently add definitions if required jpayne@68: for k in data: jpayne@68: self._add_definition( format, k, data[k], "(output)" ) jpayne@68: for idx,v in enumerate(data[k]): jpayne@68: if v == format[k].missingvalue: data[k][idx] = "." jpayne@68: # make sure GT comes first; and ensure fixed ordering; also convert GT data back to string jpayne@68: for k in data: jpayne@68: if k != 'GT': sdata.append( (k,data[k]) ) jpayne@68: sdata.sort() jpayne@68: if 'GT' in data: jpayne@68: sdata = [('GT',map(self.convertGTback,data['GT']))] + sdata jpayne@68: for k,v in sdata: jpayne@68: if v == []: v = None jpayne@68: if key and value: jpayne@68: if v != None: output.append( k+"="+','.join(map(str,v)) ) jpayne@68: else: output.append( k ) jpayne@68: elif key: output.append(k) jpayne@68: elif value: jpayne@68: if v != None: output.append( ','.join(map(str,v)) ) jpayne@68: else: output.append( "." ) # should not happen jpayne@68: # snip off trailing missing data jpayne@68: while len(output) > 1: jpayne@68: last = output[-1].replace(',','').replace('.','') jpayne@68: if len(last)>0: break jpayne@68: output = output[:-1] jpayne@68: return separator.join(output) jpayne@68: jpayne@68: jpayne@68: def enter_default_format(self): jpayne@68: for f in [FORMAT('GT',self.NT_NUMBER,1,'String','Genotype','.'), jpayne@68: FORMAT('DP',self.NT_NUMBER,1,'Integer','Read depth at this position for this sample',-1), jpayne@68: FORMAT('FT',self.NT_NUMBER,1,'String','Sample Genotype Filter','.'), jpayne@68: FORMAT('GL',self.NT_UNKNOWN,-1,'Float','Genotype likelihoods','.'), jpayne@68: FORMAT('GLE',self.NT_UNKNOWN,-1,'Float','Genotype likelihoods','.'), jpayne@68: FORMAT('GQ',self.NT_NUMBER,1,'Integer','Genotype Quality',-1), jpayne@68: FORMAT('PL',self.NT_GENOTYPES,-1,'Integer','Phred-scaled genotype likelihoods', '.'), jpayne@68: FORMAT('GP',self.NT_GENOTYPES,-1,'Float','Genotype posterior probabilities','.'), jpayne@68: FORMAT('GQ',self.NT_GENOTYPES,-1,'Integer','Conditional genotype quality','.'), jpayne@68: FORMAT('HQ',self.NT_UNKNOWN,-1,'Integer','Haplotype Quality',-1), # unknown number, since may be haploid jpayne@68: FORMAT('PS',self.NT_UNKNOWN,-1,'Integer','Phase set','.'), jpayne@68: FORMAT('PQ',self.NT_NUMBER,1,'Integer','Phasing quality',-1), jpayne@68: FORMAT('EC',self.NT_ALLELES,1,'Integer','Expected alternate allel counts',-1), jpayne@68: FORMAT('MQ',self.NT_NUMBER,1,'Integer','RMS mapping quality',-1), jpayne@68: ]: jpayne@68: if f.id not in self._format: jpayne@68: self._format[f.id] = f jpayne@68: jpayne@68: def parse_header(self, line): jpayne@68: jpayne@68: assert line.startswith('##') jpayne@68: elts = line[2:].split('=') jpayne@68: key = elts[0].strip() jpayne@68: value = '='.join(elts[1:]).strip() jpayne@68: if key == "fileformat": jpayne@68: if value == "VCFv3.3": jpayne@68: self._version = 33 jpayne@68: elif value == "VCFv4.0": jpayne@68: self._version = 40 jpayne@68: elif value == "VCFv4.1": jpayne@68: # AH - for testing jpayne@68: self._version = 40 jpayne@68: elif value == "VCFv4.2": jpayne@68: # AH - for testing jpayne@68: self._version = 40 jpayne@68: else: jpayne@68: self.error(line,self.UNKNOWN_FORMAT_STRING) jpayne@68: elif key == "INFO": jpayne@68: f = self.parse_format(line, value) jpayne@68: self._info[ f.id ] = f jpayne@68: elif key == "FILTER": jpayne@68: f = self.parse_format(line, value, filter=True) jpayne@68: self._filter[ f.id ] = f jpayne@68: elif key == "FORMAT": jpayne@68: f = self.parse_format(line, value) jpayne@68: self._format[ f.id ] = f jpayne@68: else: jpayne@68: # keep other keys in the header field jpayne@68: self._header.append( (key,value) ) jpayne@68: jpayne@68: jpayne@68: def write_header( self, stream ): jpayne@68: stream.write("##fileformat=VCFv%s.%s\n" % (self._version // 10, self._version % 10)) jpayne@68: for key,value in self._header: stream.write("##%s=%s\n" % (key,value)) jpayne@68: for var,label in [(self._info,"INFO"),(self._filter,"FILTER"),(self._format,"FORMAT")]: jpayne@68: for f in var.itervalues(): stream.write("##%s=%s\n" % (label,self.format_format(f,filter=(label=="FILTER")))) jpayne@68: jpayne@68: jpayne@68: def parse_heading( self, line ): jpayne@68: assert line.startswith('#') jpayne@68: assert not line.startswith('##') jpayne@68: headings = line[1:].split('\t') jpayne@68: # test for 8, as FORMAT field might be missing jpayne@68: if len(headings)==1 and len(line[1:].split()) >= 8: jpayne@68: self.error(line,self.HEADING_NOT_SEPARATED_BY_TABS) jpayne@68: headings = line[1:].split() jpayne@68: jpayne@68: for i,s in enumerate(self._required): jpayne@68: jpayne@68: if len(headings)<=i or headings[i] != s: jpayne@68: jpayne@68: if len(headings) <= i: jpayne@68: err = "(%sth entry not found)" % (i+1) jpayne@68: else: jpayne@68: err = "(found %s, expected %s)" % (headings[i],s) jpayne@68: jpayne@68: #self.error(line,self.BADLY_FORMATTED_HEADING,err) jpayne@68: # allow FORMAT column to be absent jpayne@68: if len(headings) == 8: jpayne@68: headings.append("FORMAT") jpayne@68: else: jpayne@68: self.error(line,self.BADLY_FORMATTED_HEADING,err) jpayne@68: jpayne@68: self._samples = headings[9:] jpayne@68: self._sample2column = dict( [(y,x+9) for x,y in enumerate( self._samples ) ] ) jpayne@68: jpayne@68: def write_heading( self, stream ): jpayne@68: stream.write("#" + "\t".join(self._required + self._samples) + "\n") jpayne@68: jpayne@68: def convertGT(self, GTstring): jpayne@68: if GTstring == ".": return ["."] jpayne@68: try: jpayne@68: gts = gtsRegEx.split(GTstring) jpayne@68: if len(gts) == 1: return [int(gts[0])] jpayne@68: if len(gts) != 2: raise ValueError() jpayne@68: if gts[0] == "." and gts[1] == ".": return [gts[0],GTstring[len(gts[0]):-len(gts[1])],gts[1]] jpayne@68: return [int(gts[0]),GTstring[len(gts[0]):-len(gts[1])],int(gts[1])] jpayne@68: except ValueError: jpayne@68: self.error(self._line,self.BAD_GENOTYPE,GTstring) jpayne@68: return [".","|","."] jpayne@68: jpayne@68: def convertGTback(self, GTdata): jpayne@68: return ''.join(map(str,GTdata)) jpayne@68: jpayne@68: def parse_formatdata( self, key, value, formatdict, line ): jpayne@68: # To do: check that the right number of values is present jpayne@68: f = formatdict.get(key,None) jpayne@68: if f == None: jpayne@68: self._add_definition(formatdict, key, value, line ) jpayne@68: f = formatdict[key] jpayne@68: if f.type == "Flag": jpayne@68: if value is not None: self.error(line,self.ERROR_FLAG_HAS_VALUE) jpayne@68: return [] jpayne@68: values = value.split(',') jpayne@68: # deal with trailing data in some early VCF files jpayne@68: if f.type in ["Float","Integer"] and len(values)>0 and values[-1].find(';') > -1: jpayne@68: self.error(line,self.ERROR_TRAILING_DATA,values[-1]) jpayne@68: values[-1] = values[-1].split(';')[0] jpayne@68: if f.type == "Integer": jpayne@68: for idx,v in enumerate(values): jpayne@68: try: jpayne@68: if v == ".": values[idx] = f.missingvalue jpayne@68: else: values[idx] = int(v) jpayne@68: except: jpayne@68: self.error(line,self.ERROR_FORMAT_NOT_INTEGER,"%s=%s" % (key, str(values))) jpayne@68: return [0] * len(values) jpayne@68: return values jpayne@68: elif f.type == "String": jpayne@68: self._line = line jpayne@68: if f.id == "GT": values = list(map( self.convertGT, values )) jpayne@68: return values jpayne@68: elif f.type == "Character": jpayne@68: for v in values: jpayne@68: if len(v) != 1: self.error(line,self.ERROR_FORMAT_NOT_CHAR) jpayne@68: return values jpayne@68: elif f.type == "Float": jpayne@68: for idx,v in enumerate(values): jpayne@68: if v == ".": values[idx] = f.missingvalue jpayne@68: try: return list(map(float,values)) jpayne@68: except: jpayne@68: self.error(line,self.ERROR_FORMAT_NOT_NUMERICAL,"%s=%s" % (key, str(values))) jpayne@68: return [0.0] * len(values) jpayne@68: else: jpayne@68: # can't happen jpayne@68: self.error(line,self.ERROR_INFO_STRING) jpayne@68: jpayne@68: def inregion(self, chrom, pos): jpayne@68: if not self._regions: return True jpayne@68: for r in self._regions: jpayne@68: if r[0] == chrom and r[1] <= pos < r[2]: return True jpayne@68: return False jpayne@68: jpayne@68: def parse_data( self, line, lineparse=False ): jpayne@68: cols = line.split('\t') jpayne@68: if len(cols) != len(self._samples)+9: jpayne@68: # gracefully deal with absent FORMAT column jpayne@68: # and those missing samples jpayne@68: if len(cols) == 8: jpayne@68: cols.append("") jpayne@68: else: jpayne@68: self.error(line, jpayne@68: self.BAD_NUMBER_OF_COLUMNS, jpayne@68: "expected %s for %s samples (%s), got %s" % (len(self._samples)+9, len(self._samples), self._samples, len(cols))) jpayne@68: jpayne@68: chrom = cols[0] jpayne@68: jpayne@68: # get 0-based position jpayne@68: try: pos = int(cols[1])-1 jpayne@68: except: self.error(line,self.POS_NOT_NUMERICAL) jpayne@68: if pos < 0: self.error(line,self.POS_NOT_POSITIVE) jpayne@68: jpayne@68: # implement filtering jpayne@68: if not self.inregion(chrom,pos): return None jpayne@68: jpayne@68: # end of first-pass parse for sortedVCF jpayne@68: if lineparse: return chrom, pos, line jpayne@68: jpayne@68: id = cols[2] jpayne@68: jpayne@68: ref = cols[3].upper() jpayne@68: if ref == ".": jpayne@68: self.error(line,self.MISSING_REF) jpayne@68: if self._version == 33: ref = get_sequence(chrom,pos,pos+1,self._reference) jpayne@68: else: ref = "" jpayne@68: else: jpayne@68: for c in ref: jpayne@68: if c not in "ACGTN": self.error(line,self.UNKNOWN_CHAR_IN_REF) jpayne@68: if "N" in ref: ref = get_sequence(chrom,pos,pos+len(ref),self._reference) jpayne@68: jpayne@68: # make sure reference is sane jpayne@68: if self._reference: jpayne@68: left = max(0,pos-100) jpayne@68: faref_leftflank = get_sequence(chrom,left,pos+len(ref),self._reference) jpayne@68: faref = faref_leftflank[pos-left:] jpayne@68: if faref != ref: self.error(line,self.WRONG_REF,"(reference is %s, VCF says %s)" % (faref,ref)) jpayne@68: ref = faref jpayne@68: jpayne@68: # convert v3.3 to v4.0 alleles below jpayne@68: if cols[4] == ".": alt = [] jpayne@68: else: alt = cols[4].upper().split(',') jpayne@68: jpayne@68: if cols[5] == ".": qual = -1 jpayne@68: else: jpayne@68: try: qual = float(cols[5]) jpayne@68: except: self.error(line,self.QUAL_NOT_NUMERICAL) jpayne@68: jpayne@68: # postpone checking that filters exist. Encode missing filter or no filtering as empty list jpayne@68: if cols[6] == "." or cols[6] == "PASS" or cols[6] == "0": filter = [] jpayne@68: else: filter = cols[6].split(';') jpayne@68: jpayne@68: # dictionary of keys, and list of values jpayne@68: info = {} jpayne@68: if cols[7] != ".": jpayne@68: for blurp in cols[7].split(';'): jpayne@68: elts = blurp.split('=') jpayne@68: if len(elts) == 1: v = None jpayne@68: elif len(elts) == 2: v = elts[1] jpayne@68: else: self.error(line,self.ERROR_INFO_STRING) jpayne@68: info[elts[0]] = self.parse_formatdata(elts[0], jpayne@68: v, jpayne@68: self._info, jpayne@68: line) jpayne@68: jpayne@68: # Gracefully deal with absent FORMAT column jpayne@68: if cols[8] == "": format = [] jpayne@68: else: format = cols[8].split(':') jpayne@68: jpayne@68: # check: all filters are defined jpayne@68: for f in filter: jpayne@68: if f not in self._filter: self.error(line,self.FILTER_NOT_DEFINED, f) jpayne@68: jpayne@68: # check: format fields are defined jpayne@68: if self._format: jpayne@68: for f in format: jpayne@68: if f not in self._format: self.error(line,self.FORMAT_NOT_DEFINED, f) jpayne@68: jpayne@68: # convert v3.3 alleles jpayne@68: if self._version == 33: jpayne@68: if len(ref) != 1: self.error(line,self.V33_BAD_REF) jpayne@68: newalts = [] jpayne@68: have_deletions = False jpayne@68: for a in alt: jpayne@68: if len(a) == 1: a = a + ref[1:] # SNP; add trailing reference jpayne@68: elif a.startswith('I'): a = ref[0] + a[1:] + ref[1:] # insertion just beyond pos; add first and trailing reference jpayne@68: elif a.startswith('D'): # allow D and D jpayne@68: have_deletions = True jpayne@68: try: jpayne@68: l = int(a[1:]) # throws ValueError if sequence jpayne@68: if len(ref) < l: # add to reference if necessary jpayne@68: addns = get_sequence(chrom,pos+len(ref),pos+l,self._reference) jpayne@68: ref += addns jpayne@68: for i,na in enumerate(newalts): newalts[i] = na+addns jpayne@68: a = ref[l:] # new deletion, deleting pos...pos+l jpayne@68: except ValueError: jpayne@68: s = a[1:] jpayne@68: if len(ref) < len(s): # add Ns to reference if necessary jpayne@68: addns = get_sequence(chrom,pos+len(ref),pos+len(s),self._reference) jpayne@68: if not s.endswith(addns) and addns != 'N'*len(addns): jpayne@68: self.error(line,self.V33_UNMATCHED_DELETION, jpayne@68: "(deletion is %s, reference is %s)" % (a,get_sequence(chrom,pos,pos+len(s),self._reference))) jpayne@68: ref += addns jpayne@68: for i,na in enumerate(newalts): newalts[i] = na+addns jpayne@68: a = ref[len(s):] # new deletion, deleting from pos jpayne@68: else: jpayne@68: self.error(line,self.V33_BAD_ALLELE) jpayne@68: newalts.append(a) jpayne@68: alt = newalts jpayne@68: # deletion alleles exist, add dummy 1st reference allele, and account for leading base jpayne@68: if have_deletions: jpayne@68: if pos == 0: jpayne@68: # Petr Danacek's: we can't have a leading nucleotide at (1-based) position 1 jpayne@68: addn = get_sequence(chrom,pos+len(ref),pos+len(ref)+1,self._reference) jpayne@68: ref += addn jpayne@68: alt = [allele+addn for allele in alt] jpayne@68: else: jpayne@68: addn = get_sequence(chrom,pos-1,pos,self._reference) jpayne@68: ref = addn + ref jpayne@68: alt = [addn + allele for allele in alt] jpayne@68: pos -= 1 jpayne@68: else: jpayne@68: # format v4.0 -- just check for nucleotides jpayne@68: for allele in alt: jpayne@68: if not alleleRegEx.match(allele): jpayne@68: self.error(line,self.V40_BAD_ALLELE,allele) jpayne@68: jpayne@68: # check for leading nucleotide in indel calls jpayne@68: for allele in alt: jpayne@68: if len(allele) != len(ref): jpayne@68: if len(allele) == 0: self.error(line,self.ZERO_LENGTH_ALLELE) jpayne@68: if ref[0].upper() != allele[0].upper() and "N" not in (ref[0]+allele[0]).upper(): jpayne@68: self.error(line,self.MISSING_INDEL_ALLELE_REF_BASE) jpayne@68: jpayne@68: # trim trailing bases in alleles jpayne@68: # AH: not certain why trimming this needs to be added jpayne@68: # disabled now for unit testing jpayne@68: # if alt: jpayne@68: # for i in range(1,min(len(ref),min(map(len,alt)))): jpayne@68: # if len(set(allele[-1].upper() for allele in alt)) > 1 or ref[-1].upper() != alt[0][-1].upper(): jpayne@68: # break jpayne@68: # ref, alt = ref[:-1], [allele[:-1] for allele in alt] jpayne@68: jpayne@68: # left-align alleles, if a reference is available jpayne@68: if self._leftalign and self._reference: jpayne@68: while left < pos: jpayne@68: movable = True jpayne@68: for allele in alt: jpayne@68: if len(allele) > len(ref): jpayne@68: longest, shortest = allele, ref jpayne@68: else: jpayne@68: longest, shortest = ref, allele jpayne@68: if len(longest) == len(shortest) or longest[:len(shortest)].upper() != shortest.upper(): jpayne@68: movable = False jpayne@68: if longest[-1].upper() != longest[len(shortest)-1].upper(): jpayne@68: movable = False jpayne@68: if not movable: jpayne@68: break jpayne@68: ref = ref[:-1] jpayne@68: alt = [allele[:-1] for allele in alt] jpayne@68: if min([len(allele) for allele in alt]) == 0 or len(ref) == 0: jpayne@68: ref = faref_leftflank[pos-left-1] + ref jpayne@68: alt = [faref_leftflank[pos-left-1] + allele for allele in alt] jpayne@68: pos -= 1 jpayne@68: jpayne@68: # parse sample columns jpayne@68: samples = [] jpayne@68: for sample in cols[9:]: jpayne@68: dict = {} jpayne@68: values = sample.split(':') jpayne@68: if len(values) > len(format): jpayne@68: self.error(line,self.BAD_NUMBER_OF_VALUES,"(found %s values in element %s; expected %s)" % (len(values),sample,len(format))) jpayne@68: for idx in range(len(format)): jpayne@68: expected = self.get_expected(format[idx], self._format, alt) jpayne@68: if idx < len(values): value = values[idx] jpayne@68: else: jpayne@68: if expected == -1: value = "." jpayne@68: else: value = ",".join(["."]*expected) jpayne@68: jpayne@68: dict[format[idx]] = self.parse_formatdata(format[idx], jpayne@68: value, jpayne@68: self._format, jpayne@68: line) jpayne@68: if expected != -1 and len(dict[format[idx]]) != expected: jpayne@68: self.error(line,self.BAD_NUMBER_OF_PARAMETERS, jpayne@68: "id=%s, expected %s parameters, got %s" % (format[idx],expected,dict[format[idx]])) jpayne@68: if len(dict[format[idx]] ) < expected: dict[format[idx]] += [dict[format[idx]][-1]]*(expected-len(dict[format[idx]])) jpayne@68: dict[format[idx]] = dict[format[idx]][:expected] jpayne@68: samples.append( dict ) jpayne@68: jpayne@68: # done jpayne@68: d = {'chrom':chrom, jpayne@68: 'pos':pos, # return 0-based position jpayne@68: 'id':id, jpayne@68: 'ref':ref, jpayne@68: 'alt':alt, jpayne@68: 'qual':qual, jpayne@68: 'filter':filter, jpayne@68: 'info':info, jpayne@68: 'format':format} jpayne@68: for key,value in zip(self._samples,samples): jpayne@68: d[key] = value jpayne@68: jpayne@68: return d jpayne@68: jpayne@68: jpayne@68: def write_data(self, stream, data): jpayne@68: required = ['chrom','pos','id','ref','alt','qual','filter','info','format'] + self._samples jpayne@68: for k in required: jpayne@68: if k not in data: raise ValueError("Required key %s not found in data" % str(k)) jpayne@68: if data['alt'] == []: alt = "." jpayne@68: else: alt = ",".join(data['alt']) jpayne@68: if data['filter'] == None: filter = "." jpayne@68: elif data['filter'] == []: jpayne@68: if self._version == 33: filter = "0" jpayne@68: else: filter = "PASS" jpayne@68: else: filter = ';'.join(data['filter']) jpayne@68: if data['qual'] == -1: qual = "." jpayne@68: else: qual = str(data['qual']) jpayne@68: jpayne@68: output = [data['chrom'], jpayne@68: str(data['pos']+1), # change to 1-based position jpayne@68: data['id'], jpayne@68: data['ref'], jpayne@68: alt, jpayne@68: qual, jpayne@68: filter, jpayne@68: self.format_formatdata( jpayne@68: data['info'], self._info, separator=";"), jpayne@68: self.format_formatdata( jpayne@68: data['format'], self._format, value=False)] jpayne@68: jpayne@68: for s in self._samples: jpayne@68: output.append(self.format_formatdata( jpayne@68: data[s], self._format, key=False)) jpayne@68: jpayne@68: stream.write( "\t".join(output) + "\n" ) jpayne@68: jpayne@68: def _parse_header(self, stream): jpayne@68: self._lineno = 0 jpayne@68: for line in stream: jpayne@68: line = force_str(line, self.encoding) jpayne@68: self._lineno += 1 jpayne@68: if line.startswith('##'): jpayne@68: self.parse_header(line.strip()) jpayne@68: elif line.startswith('#'): jpayne@68: self.parse_heading(line.strip()) jpayne@68: self.enter_default_format() jpayne@68: else: jpayne@68: break jpayne@68: return line jpayne@68: jpayne@68: def _parse(self, line, stream): jpayne@68: # deal with files with header only jpayne@68: if line.startswith("##"): return jpayne@68: if len(line.strip()) > 0: jpayne@68: d = self.parse_data( line.strip() ) jpayne@68: if d: yield d jpayne@68: for line in stream: jpayne@68: self._lineno += 1 jpayne@68: if self._lines and self._lineno > self._lines: raise StopIteration jpayne@68: d = self.parse_data( line.strip() ) jpayne@68: if d: yield d jpayne@68: jpayne@68: ###################################################################################################### jpayne@68: # jpayne@68: # API follows jpayne@68: # jpayne@68: ###################################################################################################### jpayne@68: jpayne@68: def getsamples(self): jpayne@68: """ List of samples in VCF file """ jpayne@68: return self._samples jpayne@68: jpayne@68: def setsamples(self,samples): jpayne@68: """ List of samples in VCF file """ jpayne@68: self._samples = samples jpayne@68: jpayne@68: def getheader(self): jpayne@68: """ List of header key-value pairs (strings) """ jpayne@68: return self._header jpayne@68: jpayne@68: def setheader(self,header): jpayne@68: """ List of header key-value pairs (strings) """ jpayne@68: self._header = header jpayne@68: jpayne@68: def getinfo(self): jpayne@68: """ Dictionary of ##INFO tags, as VCF.FORMAT values """ jpayne@68: return self._info jpayne@68: jpayne@68: def setinfo(self,info): jpayne@68: """ Dictionary of ##INFO tags, as VCF.FORMAT values """ jpayne@68: self._info = info jpayne@68: jpayne@68: def getformat(self): jpayne@68: """ Dictionary of ##FORMAT tags, as VCF.FORMAT values """ jpayne@68: return self._format jpayne@68: jpayne@68: def setformat(self,format): jpayne@68: """ Dictionary of ##FORMAT tags, as VCF.FORMAT values """ jpayne@68: self._format = format jpayne@68: jpayne@68: def getfilter(self): jpayne@68: """ Dictionary of ##FILTER tags, as VCF.FORMAT values """ jpayne@68: return self._filter jpayne@68: jpayne@68: def setfilter(self,filter): jpayne@68: """ Dictionary of ##FILTER tags, as VCF.FORMAT values """ jpayne@68: self._filter = filter jpayne@68: jpayne@68: def setversion(self, version): jpayne@68: if version != 33 and version != 40: raise ValueError("Can only handle v3.3 and v4.0 VCF files") jpayne@68: self._version = version jpayne@68: jpayne@68: def setregions(self, regions): jpayne@68: self._regions = regions jpayne@68: jpayne@68: def setreference(self, ref): jpayne@68: """ Provide a reference sequence; a Python class supporting a fetch(chromosome, start, end) method, e.g. PySam.FastaFile """ jpayne@68: self._reference = ref jpayne@68: jpayne@68: def ignoreerror(self, errorstring): jpayne@68: try: self._ignored_errors.add(self.__dict__[errorstring]) jpayne@68: except KeyError: raise ValueError("Invalid error string: %s" % errorstring) jpayne@68: jpayne@68: def warnerror(self, errorstring): jpayne@68: try: self._warn_errors.add(self.__dict__[errorstring]) jpayne@68: except KeyError: raise ValueError("Invalid error string: %s" % errorstring) jpayne@68: jpayne@68: def parse(self, stream): jpayne@68: """ Parse a stream of VCF-formatted lines. Initializes class instance and return generator """ jpayne@68: last_line = self._parse_header(stream) jpayne@68: # now return a generator that does the actual work. In this way the pre-processing is done jpayne@68: # before the first piece of data is yielded jpayne@68: return self._parse(last_line, stream) jpayne@68: jpayne@68: def write(self, stream, datagenerator): jpayne@68: """ Writes a VCF file to a stream, using a data generator (or list) """ jpayne@68: self.write_header(stream) jpayne@68: self.write_heading(stream) jpayne@68: for data in datagenerator: self.write_data(stream,data) jpayne@68: jpayne@68: def writeheader(self, stream): jpayne@68: """ Writes a VCF header """ jpayne@68: self.write_header(stream) jpayne@68: self.write_heading(stream) jpayne@68: jpayne@68: def compare_calls(self, pos1, ref1, alt1, pos2, ref2, alt2): jpayne@68: """ Utility function: compares two calls for equality """ jpayne@68: # a variant should always be assigned to a unique position, one base before jpayne@68: # the leftmost position of the alignment gap. If this rule is implemented jpayne@68: # correctly, the two positions must be equal for the calls to be identical. jpayne@68: if pos1 != pos2: return False jpayne@68: # from both calls, trim rightmost bases when identical. Do this safely, i.e. jpayne@68: # only when the reference bases are not Ns jpayne@68: while len(ref1)>0 and len(alt1)>0 and ref1[-1] == alt1[-1]: jpayne@68: ref1 = ref1[:-1] jpayne@68: alt1 = alt1[:-1] jpayne@68: while len(ref2)>0 and len(alt2)>0 and ref2[-1] == alt2[-1]: jpayne@68: ref2 = ref2[:-1] jpayne@68: alt2 = alt2[:-1] jpayne@68: # now, the alternative alleles must be identical jpayne@68: return alt1 == alt2 jpayne@68: jpayne@68: ########################################################################################################### jpayne@68: ########################################################################################################### jpayne@68: ## API functions added by Andreas jpayne@68: ########################################################################################################### jpayne@68: jpayne@68: def connect(self, filename, encoding="ascii"): jpayne@68: '''connect to tabix file.''' jpayne@68: self.encoding=encoding jpayne@68: self.tabixfile = pysam.Tabixfile(filename, encoding=encoding) jpayne@68: self._parse_header(self.tabixfile.header) jpayne@68: jpayne@68: def __del__(self): jpayne@68: self.close() jpayne@68: self.tabixfile = None jpayne@68: jpayne@68: def close(self): jpayne@68: if self.tabixfile: jpayne@68: self.tabixfile.close() jpayne@68: self.tabixfile = None jpayne@68: jpayne@68: def fetch(self, jpayne@68: reference=None, jpayne@68: start=None, jpayne@68: end=None, jpayne@68: region=None ): jpayne@68: """ Parse a stream of VCF-formatted lines. jpayne@68: Initializes class instance and return generator """ jpayne@68: return self.tabixfile.fetch( jpayne@68: reference, jpayne@68: start, jpayne@68: end, jpayne@68: region, jpayne@68: parser = asVCFRecord(self)) jpayne@68: jpayne@68: def validate(self, record): jpayne@68: '''validate vcf record. jpayne@68: jpayne@68: returns a validated record. jpayne@68: ''' jpayne@68: jpayne@68: raise NotImplementedError("needs to be checked") jpayne@68: jpayne@68: chrom, pos = record.chrom, record.pos jpayne@68: jpayne@68: # check reference jpayne@68: ref = record.ref jpayne@68: if ref == ".": jpayne@68: self.error(str(record),self.MISSING_REF) jpayne@68: if self._version == 33: ref = get_sequence(chrom,pos,pos+1,self._reference) jpayne@68: else: ref = "" jpayne@68: else: jpayne@68: for c in ref: jpayne@68: if c not in "ACGTN": self.error(str(record),self.UNKNOWN_CHAR_IN_REF) jpayne@68: if "N" in ref: ref = get_sequence(chrom, jpayne@68: pos, jpayne@68: pos+len(ref), jpayne@68: self._reference) jpayne@68: jpayne@68: # make sure reference is sane jpayne@68: if self._reference: jpayne@68: left = max(0,self.pos-100) jpayne@68: faref_leftflank = get_sequence(chrom,left,self.pos+len(ref),self._reference) jpayne@68: faref = faref_leftflank[pos-left:] jpayne@68: if faref != ref: self.error(str(record),self.WRONG_REF,"(reference is %s, VCF says %s)" % (faref,ref)) jpayne@68: ref = faref jpayne@68: jpayne@68: # check: format fields are defined jpayne@68: for f in record.format: jpayne@68: if f not in self._format: self.error(str(record),self.FORMAT_NOT_DEFINED, f) jpayne@68: jpayne@68: # check: all filters are defined jpayne@68: for f in record.filter: jpayne@68: if f not in self._filter: self.error(str(record),self.FILTER_NOT_DEFINED, f) jpayne@68: jpayne@68: # convert v3.3 alleles jpayne@68: if self._version == 33: jpayne@68: if len(ref) != 1: self.error(str(record),self.V33_BAD_REF) jpayne@68: newalts = [] jpayne@68: have_deletions = False jpayne@68: for a in alt: jpayne@68: if len(a) == 1: a = a + ref[1:] # SNP; add trailing reference jpayne@68: elif a.startswith('I'): a = ref[0] + a[1:] + ref[1:] # insertion just beyond pos; add first and trailing reference jpayne@68: elif a.startswith('D'): # allow D and D jpayne@68: have_deletions = True jpayne@68: try: jpayne@68: l = int(a[1:]) # throws ValueError if sequence jpayne@68: if len(ref) < l: # add to reference if necessary jpayne@68: addns = get_sequence(chrom,pos+len(ref),pos+l,self._reference) jpayne@68: ref += addns jpayne@68: for i,na in enumerate(newalts): newalts[i] = na+addns jpayne@68: a = ref[l:] # new deletion, deleting pos...pos+l jpayne@68: except ValueError: jpayne@68: s = a[1:] jpayne@68: if len(ref) < len(s): # add Ns to reference if necessary jpayne@68: addns = get_sequence(chrom,pos+len(ref),pos+len(s),self._reference) jpayne@68: if not s.endswith(addns) and addns != 'N'*len(addns): jpayne@68: self.error(str(record),self.V33_UNMATCHED_DELETION, jpayne@68: "(deletion is %s, reference is %s)" % (a,get_sequence(chrom,pos,pos+len(s),self._reference))) jpayne@68: ref += addns jpayne@68: for i,na in enumerate(newalts): newalts[i] = na+addns jpayne@68: a = ref[len(s):] # new deletion, deleting from pos jpayne@68: else: jpayne@68: self.error(str(record),self.V33_BAD_ALLELE) jpayne@68: newalts.append(a) jpayne@68: alt = newalts jpayne@68: # deletion alleles exist, add dummy 1st reference allele, and account for leading base jpayne@68: if have_deletions: jpayne@68: if pos == 0: jpayne@68: # Petr Danacek's: we can't have a leading nucleotide at (1-based) position 1 jpayne@68: addn = get_sequence(chrom,pos+len(ref),pos+len(ref)+1,self._reference) jpayne@68: ref += addn jpayne@68: alt = [allele+addn for allele in alt] jpayne@68: else: jpayne@68: addn = get_sequence(chrom,pos-1,pos,self._reference) jpayne@68: ref = addn + ref jpayne@68: alt = [addn + allele for allele in alt] jpayne@68: pos -= 1 jpayne@68: else: jpayne@68: # format v4.0 -- just check for nucleotides jpayne@68: for allele in alt: jpayne@68: if not alleleRegEx.match(allele): jpayne@68: self.error(str(record),self.V40_BAD_ALLELE,allele) jpayne@68: jpayne@68: jpayne@68: # check for leading nucleotide in indel calls jpayne@68: for allele in alt: jpayne@68: if len(allele) != len(ref): jpayne@68: if len(allele) == 0: self.error(str(record),self.ZERO_LENGTH_ALLELE) jpayne@68: if ref[0].upper() != allele[0].upper() and "N" not in (ref[0]+allele[0]).upper(): jpayne@68: self.error(str(record),self.MISSING_INDEL_ALLELE_REF_BASE) jpayne@68: jpayne@68: # trim trailing bases in alleles jpayne@68: # AH: not certain why trimming this needs to be added jpayne@68: # disabled now for unit testing jpayne@68: # for i in range(1,min(len(ref),min(map(len,alt)))): jpayne@68: # if len(set(allele[-1].upper() for allele in alt)) > 1 or ref[-1].upper() != alt[0][-1].upper(): jpayne@68: # break jpayne@68: # ref, alt = ref[:-1], [allele[:-1] for allele in alt] jpayne@68: jpayne@68: # left-align alleles, if a reference is available jpayne@68: if self._leftalign and self._reference: jpayne@68: while left < pos: jpayne@68: movable = True jpayne@68: for allele in alt: jpayne@68: if len(allele) > len(ref): jpayne@68: longest, shortest = allele, ref jpayne@68: else: jpayne@68: longest, shortest = ref, allele jpayne@68: if len(longest) == len(shortest) or longest[:len(shortest)].upper() != shortest.upper(): jpayne@68: movable = False jpayne@68: if longest[-1].upper() != longest[len(shortest)-1].upper(): jpayne@68: movable = False jpayne@68: if not movable: jpayne@68: break jpayne@68: ref = ref[:-1] jpayne@68: alt = [allele[:-1] for allele in alt] jpayne@68: if min([len(allele) for allele in alt]) == 0 or len(ref) == 0: jpayne@68: ref = faref_leftflank[pos-left-1] + ref jpayne@68: alt = [faref_leftflank[pos-left-1] + allele for allele in alt] jpayne@68: pos -= 1 jpayne@68: jpayne@68: __all__ = [ jpayne@68: "VCF", "VCFRecord", ]