Source code for ferenda.fsmparser

# -*- coding: utf-8 -*-
from __future__ import (absolute_import, division,
                        print_function, unicode_literals)
from builtins import *

import builtins
from collections import deque
import logging
import inspect

from ferenda.errors import FSMStateError

[docs]class FSMParser(): """A configurable finite state machine (FSM) for parsing documents with nested structure. You provide a set of *recognizers*, a set of *constructors*, a *transition table* and a *stream* of document text chunks, and it returns a hierarchical document object structure. See :doc:`../fsmparser`. """ def __init__(self): self.debug = False self.transitions = None # set by set_transitions self.recognizers = None # set by set_recognizers() or set_transitions() self.reader = None # set by parse() # somewhat magic self.initial_state = None self.initial_constructor = None # pseudo-internal self._state_stack = [] self.log = logging.getLogger(__name__) def _debug(self, msg): """Prints a debug message, indented to show how far down in the nested structure we are""" if self.debug: stack = inspect.stack() calling_frame = [x[3] for x in stack][1] relative_depth = len(self._state_stack) # print("%s[%s(%r)] %s" % (". " * relative_depth, calling_frame, self._state_stack, msg)) state = "/".join(self._state_stack) builtins.print("%s/%s(): %s" % (state, calling_frame, msg))
[docs] def set_recognizers(self, *args): """Set the list of functions (or other callables) used in order to recognize symbols from the stream of text chunks. Recognizers are tried in the order specified here.""" self.recognizers = args
[docs] def remove_recognizer(self, recognizer): self.recognizers = tuple(x for x in self.recognizers if x != recognizer)
[docs] def set_transitions(self, transitions): """Set the transition table for the state matchine. :param transitions: The transition table, in the form of a mapping between two tuples. The first tuple should be the current state (or a list of possible current states) and a callable function that determines if a particular symbol is recognized ``(currentstate, recognizer)``. The second tuple should be a constructor function (or `False```) and the new state to transition into. """ self.transitions = {} for (before, after) in transitions.items(): (before_states, recognizer) = before if not callable(after): (constructor, after_state) = after assert (constructor == False) or callable( constructor), "Specified constructor %r not callable" % constructor assert callable(recognizer), "Specified recognizer %r not callable" % recognizer if (not isinstance(before_states, (list, tuple))): before_states = [before_states] for before_state in before_states: if callable(after): self._debug("%r,%s() -> %s()" % (before_state, recognizer.__name__, after.__name__)) elif callable(after[0]): self._debug("%r,%s() -> %s(), %r" % (before_state, recognizer.__name__, after[0].__name__, after[1])) else: self._debug("%r,%s() -> %r, %r" % (before_state, recognizer.__name__, after[0], after[1])) self.transitions[(before_state, recognizer)] = after
[docs] def parse(self, chunks): """Parse a document in the form of an iterable of suitable chunks -- often lines or elements. each chunk should be a string or a string-like obje ct. Some examples:: p = FSMParser() reader = TextReader("foo.txt") body = p.parse(reader.getiterator(reader.readparagraph),"body", make_body) body = p.parse(BeautifulSoup("foo.html").find_all("#main p"), "body", make_body) body = p.parse(ElementTree.parse("foo.xml").find(".//paragraph"), "body", make_body) :param chunks: The document to be parsed, as a list or any other iterable of text-like objects. :param initialstate: The initial state for the machine. The state must be present in the transition table. This could be any object, but strings are preferrable as they make error messages easier to understand. :param initialconstructor: A function that creates a document root object, and then fills it with child objects using .make_children() :type initialconstructor: callable :returns: A document object tree. """ self._debug("Starting parse") self.reader = Peekable(chunks) self._state_stack = [self.initial_state] return self.initial_constructor(self)
[docs] def analyze_symbol(self): """Internal function used by make_children()""" try: rawchunk = self.reader.peek() chunk = str(rawchunk) # chunk = repr(rawchunk) if len(chunk) > 90: # chunk = chunk[:25] + "[...]" + chunk[-10:] seg = (chunk[:25], chunk[-10:]) try: chunk = "%s [...] %s" % seg except UnicodeDecodeError: chunk = "%r [...] %r" % seg else: chunk = chunk except StopIteration: self._debug("We're done!") return None applicable_tmp = [x[1] for x in self.transitions.keys() if x[0] == self._state_stack[-1]] # Create correct sorting of applicable_recognizers applicable_recognizers = [] for recognizer in self.recognizers: if recognizer in applicable_tmp: applicable_recognizers.append(recognizer) applicable_display = ", ".join([x.__name__ for x in applicable_recognizers]) for recognizer in applicable_recognizers: if recognizer(self): self._debug("Tested '%s' against %s -> %s " % (chunk, applicable_display, recognizer.__name__)) # self._debug("%r -> %s" % (chunk, recognizer.__name__)) return recognizer raise FSMStateError( "No recognizer match for %s (tried %s)" % (chunk, applicable_display))
[docs] def transition(self, currentstate, symbol): """Internal function used by make_children()""" assert (currentstate, symbol) in self.transitions, "(%r, %r) should be in self.transitions" % ( currentstate, symbol) t = self.transitions[(currentstate, symbol)] if callable(t): return t(symbol, self._state_stack) else: return t
[docs] def make_child(self, constructor, childstate): """Internal function used by make_children(), which calls one of the constructors defined in the transition table.""" if not childstate: childstate = self._state_stack[-1] # self._debug("calling child constructor %s" % constructor.__name__) else: # self._debug("calling child constructor %s in state %r" % # (constructor.__name__, childstate)) pass self._state_stack.append(childstate) ret = constructor(self) self._state_stack.pop() # do something with this? return ret
[docs] def make_children(self, parent): """Creates child nodes for the current (parent) document node. :param parent: The parent document node, as any list-like object (preferrably a subclass of :py:class:`ferenda.elements.CompoundElement`) :returns: The same ``parent`` object. """ self._debug("Making children for %s" % parent.__class__.__name__) while True: # we'll break out of this when transition() # returns a constructor that is False symbol = self.analyze_symbol() if symbol is None: # no more symbols self._debug("We're done!") return parent (constructor, newstate) = self.transition(self._state_stack[-1], symbol) if constructor is False: self._debug("transition(%r,%s()) -> (False,%r)" % (self._state_stack[-1], symbol.__name__, newstate)) else: self._debug("transition(%r,%s()) -> (%s(),%r)" % (self._state_stack[-1], symbol.__name__, constructor.__name__, newstate)) # if transition() indicated that we should change state, # first find out whether the constructor will call # make_child, creating a new stack frame. This is # indicated by the callable having the 'newstate' # attribute (set by the @ferenda.decorators.newstate # decorator) if newstate and not hasattr(constructor, 'newstate'): self._debug("Changing top of state stack (%r->%r)" % (self._state_stack[-1], newstate)) self._state_stack[-1] = newstate if constructor: try: element = self.make_child(constructor, newstate) except StopIteration: self._debug("Couldn't make child -- seems we're done!") element = None return parent if element is not None: parent.append(element) else: # special weird hack - set the state we'll be # returning to by manipulating self._state_stack # FIXME: we have no regular test case for this path, # but integrationRFC excercises it if newstate: self._debug( "Changing the state we'll return to (self._state_stack[-2])") self._debug(" (from %r to %r)" % (self._state_stack[-2], newstate)) self._state_stack[-2] = newstate return parent
# inspired by recipe 19.18 in the python cookbook. A implementation detail # helper for FSMParser. class Peekable(object): def __init__(self, iterable): self._iterable = iter(iterable) self._cache = deque() def __iter__(self): return self def _fillcache(self, cachesize=1): while len(self._cache) < cachesize: self._cache.append(next(self._iterable)) def __next__(self): self._fillcache() result = self._cache.popleft() return result # useful alias next = __next__ def peek(self, chunkno=1): self._fillcache(chunkno) result = self._cache[chunkno-1] return result