--- /dev/null
+# 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:
+
+ <some_info>Spam<some_info>
+ <some_info>More Spam<some_info>
+
+ 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
git://source.jalview.org / jabaws.git / blobdiff