# Copyright 1999 by Jeffrey Chang. All rights reserved. # This code is part of the Biopython distribution and governed by its # license. Please see the LICENSE file that should have been included # as part of this package. """Code to support writing parsers. Classes: AbstractParser Base class for parsers. AbstractConsumer Base class of all Consumers. TaggingConsumer Consumer that tags output with its event. For debugging SGMLStrippingConsumer Consumer that strips SGML tags from output. EventGenerator Generate Biopython Events from Martel XML output (note that Martel is now DEPRECATED) Functions: safe_readline Read a line from a handle, with check for EOF. safe_peekline Peek at next line, with check for EOF. read_and_call Read a line from a handle and pass it to a method. read_and_call_while Read many lines, as long as a condition is met. read_and_call_until Read many lines, until a condition is met. attempt_read_and_call Like read_and_call, but forgiving of errors. is_blank_line Test whether a line is blank. """ import sys import traceback from types import * from Bio import File # XML from python 2.0 try: from xml.sax import handler xml_support = 1 except ImportError: sys.stderr.write("Warning: Could not import SAX for dealing with XML.\n" + "This causes problems with some ParserSupport modules\n") xml_support = 0 class AbstractParser: """Base class for other parsers. """ def parse(self, handle): raise NotImplementedError("Please implement in a derived class") def parse_str(self, string): return self.parse(File.StringHandle(string)) def parse_file(self, filename): h = open(filename) try: retval = self.parse(h) finally: h.close() return retval class AbstractConsumer: """Base class for other Consumers. Derive Consumers from this class and implement appropriate methods for each event that you want to receive. """ def _unhandled_section(self): pass def _unhandled(self, data): pass def __getattr__(self, attr): if attr[:6] == 'start_' or attr[:4] == 'end_': method = self._unhandled_section else: method = self._unhandled return method class TaggingConsumer(AbstractConsumer): """A Consumer that tags the data stream with the event and prints it to a handle. Useful for debugging. """ def __init__(self, handle=None, colwidth=15, maxwidth=80): """TaggingConsumer(handle=sys.stdout, colwidth=15, maxwidth=80)""" # I can't assign sys.stdout to handle in the argument list. # If I do that, handle will be assigned the value of sys.stdout # the first time this function is called. This will fail if # the user has assigned sys.stdout to some other file, which may # be closed or invalid at a later time. if handle is None: handle = sys.stdout self._handle = handle self._colwidth = colwidth self._maxwidth = maxwidth def unhandled_section(self): self._print_name('unhandled_section') def unhandled(self, data): self._print_name('unhandled', data) def _print_name(self, name, data=None): if data is None: # Write the name of a section. self._handle.write("%s %s\n" % ("*"*self._colwidth, name)) else: # Write the tag and line. self._handle.write("%-*s: %s\n" % ( self._colwidth, name[:self._colwidth], data[:self._maxwidth-self._colwidth-2].rstrip())) def __getattr__(self, attr): if attr[:6] == 'start_' or attr[:4] == 'end_': method = lambda a=attr, s=self: s._print_name(a) else: method = lambda x, a=attr, s=self: s._print_name(a, x) return method class SGMLStrippingConsumer: """A consumer that strips off SGML tags. This is meant to be used as a decorator for other consumers. """ def __init__(self, consumer): if type(consumer) is not InstanceType: raise ValueError("consumer should be an instance") self._consumer = consumer self._prev_attr = None self._stripper = File.SGMLStripper() def _apply_clean_data(self, data): clean = self._stripper.strip(data) self._prev_attr(clean) def __getattr__(self, name): if name in ['_prev_attr', '_stripper']: return getattr(self, name) attr = getattr(self._consumer, name) # If this is not a method, then return it as is. if type(attr) is not MethodType: return attr # If it's a section method, then return it. if name[:6] == 'start_' or name[:4] == 'end_': return attr # Otherwise, it's an info event, and return my method. self._prev_attr = attr return self._apply_clean_data # onle use the Event Generator if XML handling is okay if xml_support: class EventGenerator(handler.ContentHandler): """Handler to generate events associated with a Martel parsed file. This acts like a normal SAX handler, and accepts XML generated by Martel during parsing. These events are then converted into 'Biopython events', which can then be caught by a standard biopython consumer. Note that Martel is now DEPRECATED. """ def __init__(self, consumer, interest_tags, callback_finalizer = None, exempt_tags = []): """Initialize to begin catching and firing off events. Arguments: o consumer - The consumer that we'll send Biopython events to. o interest_tags - A listing of all the tags we are interested in. o callback_finalizer - A function to deal with the collected information before passing it on to the consumer. By default the collected information is a list of all of the lines read for a particular tag -- if there are multiple tags in a row like: Spam More Spam In this case the list of information would be: ['Spam', 'More Spam'] This list of lines will be passed to the callback finalizer if it is present. Otherwise the consumer will be called with the list of content information. o exempt_tags - A listing of particular tags that are exempt from being processed by the callback_finalizer. This allows you to use a finalizer to deal with most tags, but leave those you don't want touched. """ self._consumer = consumer self.interest_tags = interest_tags self._finalizer = callback_finalizer self._exempt_tags = exempt_tags # a dictionary of content for each tag of interest # the information for each tag is held as a list of the lines. # This allows us to collect information from multiple tags # in a row, and return it all at once. self.info = {} for tag in self.interest_tags: self.info[tag] = [] # the previous tag we were collecting information for. # We set a delay in sending info to the consumer so that we can # collect a bunch of tags in a row and append all of the info # together. self._previous_tag = '' # the current character information for a tag self._cur_content = [] # whether we should be collecting information self._collect_characters = 0 def startElement(self, name, attrs): """Determine if we should collect characters from this tag. """ if name in self.interest_tags: self._collect_characters = 1 def characters(self, content): """Extract the information if we are interested in it. """ if self._collect_characters: self._cur_content.append(content) def endElement(self, name): """Send the information to the consumer. Once we've got the end element we've collected up all of the character information we need, and we need to send this on to the consumer to do something with it. We have a delay of one tag on doing this, so that we can collect all of the info from multiple calls to the same element at once. """ # only deal with the tag if it is something we are # interested in and potentially have information for if self._collect_characters: # add all of the information collected inside this tag self.info[name].append("".join(self._cur_content)) # reset our information and flags self._cur_content = [] self._collect_characters = 0 # if we are at a new tag, pass on the info from the last tag if self._previous_tag and self._previous_tag != name: self._make_callback(self._previous_tag) # set this tag as the next to be passed self._previous_tag = name def _make_callback(self, name): """Call the callback function with the info with the given name. """ # strip off whitespace and call the consumer callback_function = getattr(self._consumer, name) # --- pass back the information # if there is a finalizer, use that if self._finalizer is not None and name not in self._exempt_tags: info_to_pass = self._finalizer(self.info[name]) # otherwise pass back the entire list of information else: info_to_pass = self.info[name] callback_function(info_to_pass) # reset the information for the tag self.info[name] = [] def endDocument(self): """Make sure all of our information has been passed. This just flushes out any stored tags that need to be passed. """ if self._previous_tag: self._make_callback(self._previous_tag) def read_and_call(uhandle, method, **keywds): """read_and_call(uhandle, method[, start][, end][, contains][, blank][, has_re]) Read a line from uhandle, check it, and pass it to the method. Raises a ValueError if the line does not pass the checks. start, end, contains, blank, and has_re specify optional conditions that the line must pass. start and end specifies what the line must begin or end with (not counting EOL characters). contains specifies a substring that must be found in the line. If blank is a true value, then the line must be blank. has_re should be a regular expression object with a pattern that the line must match somewhere. """ line = safe_readline(uhandle) errmsg = _fails_conditions(*(line,), **keywds) if errmsg is not None: raise ValueError(errmsg) method(line) def read_and_call_while(uhandle, method, **keywds): """read_and_call_while(uhandle, method[, start][, end][, contains][, blank][, has_re]) -> number of lines Read a line from uhandle and pass it to the method as long as some condition is true. Returns the number of lines that were read. See the docstring for read_and_call for a description of the parameters. """ nlines = 0 while 1: line = safe_readline(uhandle) # If I've failed the condition, then stop reading the line. if _fails_conditions(*(line,), **keywds): uhandle.saveline(line) break method(line) nlines = nlines + 1 return nlines def read_and_call_until(uhandle, method, **keywds): """read_and_call_until(uhandle, method, start=None, end=None, contains=None, blank=None) -> number of lines Read a line from uhandle and pass it to the method until some condition is true. Returns the number of lines that were read. See the docstring for read_and_call for a description of the parameters. """ nlines = 0 while 1: line = safe_readline(uhandle) # If I've met the condition, then stop reading the line. if not _fails_conditions(*(line,), **keywds): uhandle.saveline(line) break method(line) nlines = nlines + 1 return nlines def attempt_read_and_call(uhandle, method, **keywds): """attempt_read_and_call(uhandle, method, **keywds) -> boolean Similar to read_and_call, but returns a boolean specifying whether the line has passed the checks. Does not raise exceptions. See docs for read_and_call for a description of the function arguments. """ line = safe_readline(uhandle) passed = not _fails_conditions(*(line,), **keywds) if passed: method(line) else: uhandle.saveline(line) return passed def _fails_conditions(line, start=None, end=None, contains=None, blank=None, has_re=None): if start is not None: if line[:len(start)] != start: return "Line does not start with '%s':\n%s" % (start, line) if end is not None: if line.rstrip()[-len(end):] != end: return "Line does not end with '%s':\n%s" % (end, line) if contains is not None: if line.find(contains) == -1: return "Line does not contain '%s':\n%s" % (contains, line) if blank is not None: if blank: if not is_blank_line(line): return "Expected blank line, but got:\n%s" % line else: if is_blank_line(line): return "Expected non-blank line, but got a blank one" if has_re is not None: if has_re.search(line) is None: return "Line does not match regex '%s':\n%s" % ( has_re.pattern, line) return None def is_blank_line(line, allow_spaces=0): """is_blank_line(line, allow_spaces=0) -> boolean Return whether a line is blank. allow_spaces specifies whether to allow whitespaces in a blank line. A true value signifies that a line containing whitespaces as well as end-of-line characters should be considered blank. """ if not line: return 1 if allow_spaces: return line.rstrip() == '' return line[0] == '\n' or line[0] == '\r' def safe_readline(handle): """safe_readline(handle) -> line Read a line from an UndoHandle and return it. If there are no more lines to read, I will raise a ValueError. """ line = handle.readline() if not line: raise ValueError("Unexpected end of stream.") return line def safe_peekline(handle): """safe_peekline(handle) -> line Peek at the next line in an UndoHandle and return it. If there are no more lines to peek, I will raise a ValueError. """ line = handle.peekline() if not line: raise ValueError("Unexpected end of stream.") return line