import xml.etree.ElementTree import re class DataNode(object): """ A storage class that allows setting arbitrary attributes """ class Convert(object): """ Conversion to a certain type """ def __init__(self, type): self.type = type def __call__(self, data): try: value = self.type( data ) except ValueError, e: raise RuntimeError, ( "%s cannot be converted to %s: %s" % ( data, self.type, e ) ) return value to_int = Convert( type = int ) to_float = Convert( type = float ) noconversion = lambda data: data def Direct(processing, elem, node): """ Direct attachment to node """ processing.populate( elem = elem, node = node ) class Child(object): """ Attach a child node """ def __init__(self, tag, attribute): self.tag = tag self.attribute = attribute def __call__(self, processing, elem, node): elem_child = elem.find( self.tag ) if elem_child is None: raise RuntimeError, "Missing %s tag" % self.tag node_child = DataNode() setattr( node, self.attribute, node_child ) processing.populate( elem = elem_child, node = node_child ) class Submerged(object): """ Climb the tree, but keep current node """ def __init__(self, tag): self.tag = tag def __call__(self, processing, elem, node): elem_child = elem.find( self.tag ) if elem_child is None: raise RuntimeError, "Missing %s tag" % self.tag processing.populate( elem = elem_child, node = node ) class Raised(object): """ Climb node hierarchy, but keep current element """ def __init__(self, attribute): self.attribute = attribute def __call__(self, processing, elem, node): node_child = DataNode() setattr( node, self.attribute, node_child ) processing.populate( elem = elem, node = node_child ) class Multiple(object): """ Attach an array of children """ def __init__(self, attribute, tag): self.attribute = attribute self.tag = tag def __call__(self, processing, elem, node): elem_child = elem.find( self.tag ) if elem_child is None: raise RuntimeError, "Missing %s tag" % self.tag array = [] setattr( node, self.attribute, array ) for elem_grandchild in elem_child: node_child = DataNode() processing.populate( elem = elem_grandchild, node = node_child ) array.append( node_child ) class MultipleEntry(object): """ Attach nodes as an array """ def __init__(self, attribute, tag): self.attribute = attribute self.tag = tag def __call__(self, processing, elem, node): array = [] setattr( node, self.attribute, array ) for elem_child in elem.findall( self.tag ): node_child = DataNode() processing.populate( elem = elem_child, node = node_child ) array.append( node_child ) class NodeProcessing(object): """ Processing data XML node """ def __init__(self, instructions, attachment): self.instructions = instructions self.attachment = attachment def process(self, elem, node): self.attachment( processing = self, elem = elem, node = node ) def populate(self, elem, node): for instruction in self.instructions: instruction.process( elem = elem, node = node ) class TextProcessing(object): """ Converts text to some type """ def __init__(self, attribute, tag, conversion, default = None): self.attribute = attribute self.tag = tag self.conversion = conversion self.default = default def process(self, elem, node): child = elem.find( self.tag ) if child is None: if self.default is None: raise RuntimeError, "Missing %s tag" % self.tag else: value = self.default else: value = child.text setattr( node, self.attribute, self.conversion( value ) ) class AttributeProcessing(object): """ Converts an attribute to some type """ def __init__(self, attribute, keyword, conversion, default = None): self.attribute = attribute self.keyword = keyword self.conversion = conversion self.default = default def process(self, elem, node): if self.keyword not in elem.attrib: if self.default is None: raise RuntimeError, "Missing %s attribute" % self.keyword else: value = self.default else: value = elem.attrib[ self.keyword ] setattr( node, self.attribute, self.conversion( value ) ) class DataProcessing(object): """ Converts the text attribute to some type """ def __init__(self, attribute, conversion): self.attribute = attribute self.conversion = conversion def process(self, elem, node): setattr( node, self.attribute, self.conversion( elem.text ) ) BLAST_BUILDER = NodeProcessing( instructions = [ TextProcessing( attribute = "program", tag = "BlastOutput_program", conversion = noconversion ), TextProcessing( attribute = "version", tag = "BlastOutput_version", conversion = noconversion ), TextProcessing( attribute = "reference", tag = "BlastOutput_reference", conversion = noconversion ), TextProcessing( attribute = "db", tag = "BlastOutput_db", conversion = noconversion ), TextProcessing( attribute = "query_id", tag = "BlastOutput_query-ID", conversion = noconversion ), TextProcessing( attribute = "query_def", tag = "BlastOutput_query-def", conversion = noconversion ), TextProcessing( attribute = "query_len", tag = "BlastOutput_query-len", conversion = to_int ), NodeProcessing( instructions = [ NodeProcessing( instructions = [ TextProcessing( attribute = "matrix", tag = "Parameters_matrix", conversion = noconversion ), TextProcessing( attribute = "expect", tag = "Parameters_expect", conversion = to_float ), TextProcessing( attribute = "gap_open", tag = "Parameters_gap-open", conversion = to_float ), TextProcessing( attribute = "gap_extend", tag = "Parameters_gap-extend", conversion = to_float ), TextProcessing( attribute = "filter", tag = "Parameters_filter", conversion = noconversion ), ], attachment = Submerged( tag = "Parameters" ) ) ], attachment = Child( attribute = "parameters", tag = "BlastOutput_param" ) ), NodeProcessing( instructions = [ TextProcessing( attribute = "num", tag = "Iteration_iter-num", conversion = to_int ), TextProcessing( attribute = "query_id", tag = "Iteration_query-ID", conversion = noconversion, default = "" ), TextProcessing( attribute = "query_def", tag = "Iteration_query-def", conversion = noconversion, default = "" ), TextProcessing( attribute = "query_len", tag = "Iteration_query-len", conversion = to_int, default = 0 ), NodeProcessing( instructions = [ TextProcessing( attribute = "num", tag = "Hit_num", conversion = to_int ), TextProcessing( attribute = "identifier", tag = "Hit_id", conversion = noconversion ), TextProcessing( attribute = "annotation", tag = "Hit_def", conversion = noconversion ), TextProcessing( attribute = "accession", tag = "Hit_accession", conversion = noconversion ), TextProcessing( attribute = "length", tag = "Hit_len", conversion = to_int ), NodeProcessing( instructions = [ TextProcessing( attribute = "num", tag = "Hsp_num", conversion = to_int ), TextProcessing( attribute = "bit_score", tag = "Hsp_bit-score", conversion = to_float ), TextProcessing( attribute = "score", tag = "Hsp_score", conversion = to_float ), TextProcessing( attribute = "evalue", tag = "Hsp_evalue", conversion = to_float ), NodeProcessing( instructions = [ TextProcessing( attribute = "start", tag = "Hsp_query-from", conversion = to_int ), TextProcessing( attribute = "end", tag = "Hsp_query-to", conversion = to_int ), TextProcessing( attribute = "frame", tag = "Hsp_query-frame", conversion = to_int, default = 0 ), TextProcessing( attribute = "seq", tag = "Hsp_qseq", conversion = noconversion ), ], attachment = Raised( attribute = "query" ) ), NodeProcessing( instructions = [ TextProcessing( attribute = "start", tag = "Hsp_hit-from", conversion = to_int ), TextProcessing( attribute = "end", tag = "Hsp_hit-to", conversion = to_int ), TextProcessing( attribute = "frame", tag = "Hsp_hit-frame", conversion = to_int, default = 0 ), TextProcessing( attribute = "seq", tag = "Hsp_hseq", conversion = noconversion ), ], attachment = Raised( attribute = "hit" ) ), TextProcessing( attribute = "identity", tag = "Hsp_identity", conversion = to_int ), TextProcessing( attribute = "positive", tag = "Hsp_positive", conversion = to_int ), TextProcessing( attribute = "gaps", tag = "Hsp_gaps", conversion = to_int, default = 0 ), TextProcessing( attribute = "length", tag = "Hsp_align-len", conversion = to_int ), TextProcessing( attribute = "midline", tag = "Hsp_midline", conversion = noconversion ), ], attachment = Multiple( attribute = "hsps", tag = "Hit_hsps" ) ) ], attachment = Multiple( attribute = "hits", tag = "Iteration_hits" ) ), NodeProcessing( instructions = [ NodeProcessing( instructions = [ TextProcessing( attribute = "db_num", tag = "Statistics_db-num", conversion = to_int ), TextProcessing( attribute = "db_len", tag = "Statistics_db-len", conversion = to_int ), TextProcessing( attribute = "hsp_len", tag = "Statistics_hsp-len", conversion = to_int ), TextProcessing( attribute = "eff_space", tag = "Statistics_eff-space", conversion = to_float ), TextProcessing( attribute = "kappa", tag = "Statistics_kappa", conversion = to_float ), TextProcessing( attribute = "lambdav", tag = "Statistics_lambda", conversion = to_float ), TextProcessing( attribute = "entropy", tag = "Statistics_entropy", conversion = to_float ), ], attachment = Submerged( tag = "Statistics" ) ) ], attachment = Child( attribute = "statistics", tag = "Iteration_stat" ) ), ], attachment = Multiple( attribute = "iterations", tag = "BlastOutput_iterations" ) ) ], attachment = Direct, ) class Blast(object): """ Homology search by BLAST, XML output """ REGEX = re.compile( r"([A-Za-z0-9]{4})[_\-]([A-Za-z0-9]?)$" ) def __init__(self, root): self.root = root def restrict(self, max_count): if self.root.iterations: self.root.iterations[-1].hits = self.root.iterations[-1].hits[:max_count] def hits(self): if not self.root.iterations: raise StopIteration from iotbx import bioinformatics for h in self.root.iterations[-1].hits: match = self.REGEX.search( h.accession ) if match: ( pdb, chain ) = match.groups() else: ( pdb, chain ) = ( "", "" ) alignment = bioinformatics.clustal_alignment( names = [ "target", "%s_%s" % ( pdb, chain ) ], alignments = [ h.hsps[0].query.seq, h.hsps[0].hit.seq ], program = "NCBI-BLAST" ) yield bioinformatics.homology_search_hit( identifier = pdb, chain = chain, annotation = h.annotation, alignment = alignment ) ebi_ns = lambda s: xml.etree.ElementTree.QName( "http://www.ebi.ac.uk/schema", s ).text EBI_WU_BLAST_BUILDER = NodeProcessing( instructions = [ NodeProcessing( instructions = [ NodeProcessing( instructions = [ AttributeProcessing( attribute = "name", keyword = "name", conversion = noconversion ), AttributeProcessing( attribute = "version", keyword = "version", conversion = noconversion ), AttributeProcessing( attribute = "citation", keyword = "citation", conversion = noconversion, default = "" ), ], attachment = Child( tag = ebi_ns( "program" ), attribute = "program" ) ), NodeProcessing( instructions = [ AttributeProcessing( attribute = "command", keyword = "command", conversion = noconversion ), ], attachment = Submerged( tag = ebi_ns( "commandLine" ) ) ), NodeProcessing( instructions = [ NodeProcessing( instructions = [ AttributeProcessing( attribute = "number", keyword = "number", conversion = to_int ), AttributeProcessing( attribute = "name", keyword = "name", conversion = noconversion ), AttributeProcessing( attribute = "type", keyword = "type", conversion = noconversion ), AttributeProcessing( attribute = "length", keyword = "length", conversion = to_int ), ], attachment = Multiple( tag = ebi_ns( "sequences" ), attribute = "sequences" ) ), NodeProcessing( instructions = [ AttributeProcessing( attribute = "number", keyword = "number", conversion = to_int ), AttributeProcessing( attribute = "name", keyword = "name", conversion = noconversion ), AttributeProcessing( attribute = "type", keyword = "type", conversion = noconversion ), AttributeProcessing( attribute = "created", keyword = "created", conversion = noconversion, default = "" ), ], attachment = Multiple( tag = ebi_ns( "databases" ), attribute = "databases" ) ), TextProcessing( attribute = "scores", tag = ebi_ns( "scores" ), conversion = to_float, default = 0.0 ), TextProcessing( attribute = "alignments", tag = ebi_ns( "alignments" ), conversion = to_int, default = 0 ), TextProcessing( attribute = "matrix", tag = ebi_ns( "matrix" ), conversion = noconversion, default = "" ), TextProcessing( attribute = "expectation_upper", tag = ebi_ns( "expectationUpper" ), conversion = to_float, default = 0.0 ), TextProcessing( attribute = "statistics", tag = ebi_ns( "statistics" ), conversion = noconversion, default = "" ), TextProcessing( attribute = "filter", tag = ebi_ns( "filter" ), conversion = noconversion, default = "" ), ], attachment = Child( tag = ebi_ns( "parameters" ), attribute = "parameters" ) ), NodeProcessing( instructions = [ AttributeProcessing( attribute = "start", keyword = "start", conversion = noconversion ), AttributeProcessing( attribute = "end", keyword = "end", conversion = noconversion ), AttributeProcessing( attribute = "search", keyword = "search", conversion = noconversion ), ], attachment = Child( tag = ebi_ns( "timeInfo" ), attribute = "time" ) ), ], attachment = Submerged( tag = ebi_ns( "Header" ) ) ), NodeProcessing( instructions = [ NodeProcessing( instructions = [ AttributeProcessing( attribute = "number", keyword = "number", conversion = to_int ), AttributeProcessing( attribute = "database", keyword = "database", conversion = noconversion ), AttributeProcessing( attribute = "identifier", keyword = "id", conversion = noconversion ), AttributeProcessing( attribute = "length", keyword = "length", conversion = to_int ), AttributeProcessing( attribute = "description", keyword = "description", conversion = noconversion ), NodeProcessing( instructions = [ AttributeProcessing( attribute = "number", keyword = "number", conversion = to_int ), TextProcessing( attribute = "score", tag = ebi_ns( "score" ), conversion = to_float ), TextProcessing( attribute = "bits", tag = ebi_ns( "bits" ), conversion = to_float ), TextProcessing( attribute = "expectation", tag = ebi_ns( "expectation" ), conversion = to_float ), TextProcessing( attribute = "probability", tag = ebi_ns( "probability" ), conversion = to_float, default = 0.0 ), TextProcessing( attribute = "identity", tag = ebi_ns( "identity" ), conversion = to_float ), TextProcessing( attribute = "positives", tag = ebi_ns( "positives" ), conversion = to_float ), NodeProcessing( instructions = [ AttributeProcessing( attribute = "start", keyword = "start", conversion = to_int ), AttributeProcessing( attribute = "end", keyword = "end", conversion = to_int ), DataProcessing( attribute = "seq", conversion = noconversion ), ], attachment = Child( tag = ebi_ns( "querySeq" ), attribute = "query" ) ), NodeProcessing( instructions = [ AttributeProcessing( attribute = "start", keyword = "start", conversion = to_int ), AttributeProcessing( attribute = "end", keyword = "end", conversion = to_int ), DataProcessing( attribute = "seq", conversion = noconversion ), ], attachment = Child( tag = ebi_ns( "matchSeq" ), attribute = "match" ) ), TextProcessing( attribute = "pattern", tag = ebi_ns( "pattern" ), conversion = noconversion ), ], attachment = Multiple( tag = ebi_ns( "alignments" ), attribute = "alignments" ) ) ], attachment = Multiple( tag = ebi_ns( "hits" ), attribute = "hits" ) ), ], attachment = Submerged( tag = ebi_ns( "SequenceSimilaritySearchResult" ) ) ), ], attachment = Direct ) class EBIWUBlast(object): """ Homology search by WU-BLAST at EBI """ def __init__(self, root): self.root = root def restrict(self, max_count): self.root.hits = self.root.hits[:max_count] def hits(self): from iotbx import bioinformatics for h in self.root.hits: pdb = h.identifier[:4] chain = h.identifier[5:] alignment = bioinformatics.clustal_alignment( names = [ "target", "%s_%s" % ( pdb, chain ) ], alignments = [ h.alignments[0].query.seq, h.alignments[0].match.seq ], program = "WU-BLAST" ) yield bioinformatics.homology_search_hit( identifier = pdb, chain = chain, annotation = h.description, alignment = alignment ) def parse_tree(tree, builder, data_type): root = DataNode() builder.process( elem = tree, node = root ) return data_type( root = root ) def parse(data, builder, data_type): tree = xml.etree.ElementTree.fromstring( data ) return parse_tree( tree = tree, builder = builder, data_type = data_type ) # Legacy function def get_ncbi_blast_parser(): import functools return functools.partial( parse, builder = BLAST_BUILDER, data_type = Blast ) def get_ebi_blast_parser(): import functools return functools.partial( parse, builder = EBI_WU_BLAST_BUILDER, data_type = EBIWUBlast ) # CATH XML parsing CATH_ANNOTATION_BUILDER = NodeProcessing( instructions = [ NodeProcessing( instructions = [ AttributeProcessing( attribute = "pdb_code", keyword = "pdb_code", conversion = noconversion ), AttributeProcessing( attribute = "chain_code", keyword = "chain_code", conversion = noconversion ), AttributeProcessing( attribute = "domain_code", keyword = "domain_code", conversion = noconversion ), AttributeProcessing( attribute = "domain_count", keyword = "cath_domain_count", conversion = to_int ), NodeProcessing( instructions = [ AttributeProcessing( attribute = "domain_id", keyword = "domain_id", conversion = noconversion ), AttributeProcessing( attribute = "pdb_id", keyword = "pdb_id", conversion = noconversion ), AttributeProcessing( attribute = "chain_code", keyword = "chain_code", conversion = noconversion ), AttributeProcessing( attribute = "domain_code", keyword = "domain_code", conversion = noconversion ), AttributeProcessing( attribute = "dom_length", keyword = "dom_length", conversion = to_int ), AttributeProcessing( attribute = "dom_atom_sequence", keyword = "dom_atom_sequence", conversion = noconversion ), AttributeProcessing( attribute = "dom_combs_sequence", keyword = "dom_combs_sequence", conversion = noconversion ), NodeProcessing( instructions = [ AttributeProcessing( attribute = "class_code", keyword = "class_code", conversion = noconversion ), AttributeProcessing( attribute = "arch_code", keyword = "arch_code", conversion = noconversion ), AttributeProcessing( attribute = "top_code", keyword = "top_code", conversion = noconversion ), AttributeProcessing( attribute = "homol_code", keyword = "homol_code", conversion = noconversion ), AttributeProcessing( attribute = "s35_code", keyword = "s35_code", conversion = noconversion ), AttributeProcessing( attribute = "s60_code", keyword = "s60_code", conversion = noconversion ), AttributeProcessing( attribute = "s95_code", keyword = "s95_code", conversion = noconversion ), AttributeProcessing( attribute = "s100_code", keyword = "s100_code", conversion = noconversion ), AttributeProcessing( attribute = "s100_count", keyword = "s100_count", conversion = noconversion ), DataProcessing( attribute = "uni_code", conversion = noconversion ), ], attachment = Child( tag = "cath_code", attribute = "cath_code" ) ), NodeProcessing( instructions = [ AttributeProcessing( attribute = "seg_num", keyword = "seg_num", conversion = to_int ), AttributeProcessing( attribute = "pdb_start", keyword = "pdb_start", conversion = noconversion ), AttributeProcessing( attribute = "pdb_stop", keyword = "pdb_stop", conversion = noconversion ), AttributeProcessing( attribute = "slength", keyword = "slength", conversion = to_int ), AttributeProcessing( attribute = "sseqs", keyword = "sseqs", conversion = noconversion ), ], attachment = Multiple( tag = "segments", attribute = "segments" ) ), ], attachment = MultipleEntry( tag = "cath_domain", attribute = "domains" ) ) ], attachment = Child( tag = "cath_pdb_query", attribute = "query" ) ), ], attachment = Direct, ) class CathAnnotation(object): """ Cath query results """ def __init__(self, root): self.root = root def empty(self): return self.root.query.domain_count == 0 def parse_cath_xml(data): tree = xml.etree.ElementTree.fromstring( data ) if tree.find( "error" ) is not None: return None return parse_tree( tree = tree, builder = CATH_ANNOTATION_BUILDER, data_type = CathAnnotation ) def get_cath_parser(): return parse_cath_xml