Finite state automata

Question

This is a non-deterministic finite state automata (NFA) meant to be used with a Regex engine. To show some example usage, suppose you wanted to construct an NFA for the regex (ad|[0-9])*. It might look something like this:

ad_literal = NFA.from_string('ad')
number_set = NFA.from_set({str(x) for x in range(10)})
union = ad_literal.union(number_set)
final_nfa = union.kleene()
final_nfa.match('adbeadf')
>>> ad

I'm mainly hoping for pointers for making this more organized and readable. I'm pretty happy with how the public functions turned out, but it seems like all the helpers are a bit of a mess. I think maybe there's another class hiding somewhere, but since it's all a light wrapper around a list anyway, I don't see anywhere to draw the line.

import string


class DFA(object):
    def __init__(self, table, terminals):
        self._table = table
        self._terminals = terminals

    def match(self, pattern: str) -> str:
        match_string = ''
        state = 0
        last_match = ''
        for char in pattern:
            try:
                state = self._table[state][ord(char)]
                if state == -1:
                    return last_match
                if char in string.printable:
                    match_string += char
                if state in self._terminals:
                    last_match = match_string
            except KeyError:
                return last_match
        return last_match


class NFA(object):
    def __init__(self):
        self._initial_state = 0
        self._terminal_state = 0
        self._table = []
        self._dfa = None  # type: DFA

    def match(self, source: str) -> str:
        if not self._dfa:
            self._dfa = self.to_dfa()
        return self._dfa.match(source)

    def concat(self, other: 'NFA') -> 'NFA':
        new_nfa = NFA()
        new_nfa._table = [x.copy() for x in self._table]
        self._copy_table(other._table,
                         new_nfa._table,
                         lambda s: s + self._terminal_state)
        new_nfa._terminal_state = len(new_nfa._table)
        return new_nfa

    def kleene(self):
        new_table = [self._empty_row()]
        self._set_transition(new_table[0], '\0', {1, self._terminal_state + 1})

        self._copy_table(self._table, new_table, lambda s: s + 1)

        # Add null transition to new terminal state, or to beginning
        new_table.append(self._empty_row())
        self._set_transition(new_table[-1], '\0', {self._terminal_state + 2, 1})

        new_nfa = NFA()
        new_nfa._table = new_table
        new_nfa._terminal_state = self._terminal_state + 2
        return new_nfa

    def union(self, other: 'NFA'):
        new_terminal_state = self._terminal_state + other._terminal_state + 1
        new_table = [self._empty_row()]
        new_table[0][ord('\0')] = {1, self._terminal_state + 1}

        self._copy_table(self._table, new_table,
                         lambda s: new_terminal_state if s == self._terminal_state else s + 1)

        self._copy_table(other._table, new_table,
                         lambda s: (new_terminal_state if s == other._terminal_state
                                    else s + 1 + self._terminal_state))
        new_nfa = NFA()
        new_nfa._table = new_table
        new_nfa._terminal_state = new_terminal_state
        return new_nfa

    def to_dfa(self) -> DFA:
        characters = [chr(i) for i in range(128)]
        start_state = frozenset(self._epsilon_closure({self._initial_state}))
        dfa_states = self._collect_nfa_states(characters, start_state)

        nfa_to_dfa_state_map = {start_state: 0}
        for i, state in enumerate(dfa_states.difference({start_state})):
            nfa_to_dfa_state_map[state] = i + 1
        nfa_to_dfa_state_map[frozenset()] = -1

        # Just invert it
        dfa_to_nfa_state_map = {v: k for k, v in nfa_to_dfa_state_map.items()}

        dfa_table = [[-1 for _ in characters] for _ in dfa_states]
        for dfa_state in dfa_to_nfa_state_map:
            if dfa_state == -1:
                continue
            nfa_state = dfa_to_nfa_state_map[dfa_state]
            for char in characters:
                if char == '\0':
                    continue
                next_nfa_state = frozenset(self._epsilon_closure(self._next_states(nfa_state, char)))
                next_dfa_state = (nfa_to_dfa_state_map[next_nfa_state]
                                  if next_nfa_state in nfa_to_dfa_state_map
                                  else -1)
                dfa_table[dfa_state][ord(char)] = next_dfa_state

        terminal_nfa_states = {state for state in dfa_states if self._terminal_state in state}
        terminal_dfa_states = {nfa_to_dfa_state_map[state] for state in terminal_nfa_states}

        return DFA(dfa_table, terminal_dfa_states)

    def _collect_nfa_states(self, characters, start_state):
        dfa_states = {start_state}
        checked_dfa_states = set()
        while dfa_states:
            current_state = dfa_states.pop()
            new_states = set()
            for char in characters:
                next_state = frozenset(self._epsilon_closure(self._next_states(current_state, char)))
                checked_dfa_states.add(current_state)
                if next_state and next_state not in checked_dfa_states:
                    new_states.add(next_state)
            dfa_states.update(new_states)
        return checked_dfa_states

    def _next_states(self, states: {int}, char: str) -> {int}:
        result = set()

        for state in states:
            result.update(self._at(state, char))
        return result

    def _single_state_closure(self, state: int) -> {int}:
        return self._at(state, '\0')

    def _epsilon_closure(self, state: {int}) -> {int}:
        if not state:
            return set()
        to_check = state.copy()
        checked = set()
        closure = state.copy()
        iteration = set()
        while to_check:

            # Copy states to current iteration
            while to_check:
                iteration.add(to_check.pop())

            for state in iteration:
                next_states = self._single_state_closure(state)
                if state not in checked:
                    checked.add(state)
                    to_check.update(next_states)
                closure.update(next_states)

        return closure

    def _at(self, state: int, char: str):
        if state >= len(self._table):
            return set()
        return self._table[state][ord(char)]

    def _add_row(self, row_number):
        while row_number >= len(self._table):
            self._table.append(self._empty_row())

    def _add_transition(self, start_state: int, next_state: int, char: str) -> None:
        assert len(char) == 1
        if start_state >= len(self._table):
            self._add_row(start_state)
        self._table[start_state][ord(char)].add(next_state)

    @staticmethod
    def _empty_row():
        return [set() for _ in range(128)]

    @staticmethod
    def _set_transition(row: [set], character: str, states: {int}):
        row[ord(character)] = states

    @staticmethod
    def from_string(pattern: str) -> 'NFA':
        nfa = NFA()
        current_state = nfa._initial_state
        for char in pattern:
            nfa._add_transition(current_state, current_state + 1, char)
            current_state += 1
        nfa._terminal_state = len(pattern)
        return nfa

    @staticmethod
    def from_set(union: {str}) -> 'NFA':
        nfa = NFA()
        nfa._add_row(0)
        for char in union:
            nfa._table[0][ord(char)] = {1}
        nfa._add_transition(1, 2, '\0')
        nfa._terminal_state = 2
        return nfa

    @staticmethod
    def _copy_table(source, dest, state_function):
        for row in source:
            row_copy = []
            for state_set in row:
                row_copy.append({state_function(state) for state in state_set})
            dest.append(row_copy)

Below are some more usage examples. Here are some simple unit tests:

import unittest
from automata import NFA


class TestNfa(unittest.TestCase):

    def test_union(self):
        nfa = NFA.from_string('abc')
        nfa = nfa.union(NFA.from_string('def'))

        self.assertEqual(nfa.match('abc'), 'abc')
        self.assertEqual(nfa.match('def'), 'def')
        self.assertEqual(nfa.match('de'), '')

    def test_kleene(self):
        nfa = NFA.from_string('abc')
        nfa = nfa.kleene()

        self.assertEqual(nfa.match(''), '')
        self.assertEqual(nfa.match('abc'), 'abc')
        self.assertEqual(nfa.match('abcabc'), 'abcabc')
        self.assertEqual(nfa.match('abcDabc'), 'abc')

    def test_concat(self):
        nfa = NFA.from_string('ab')
        nfa = nfa.concat(NFA.from_string('cd'))

        self.assertEqual(nfa.match('abcd'), 'abcd')
        self.assertEqual(nfa.match('abcde'), 'abcd')
        self.assertEqual(nfa.match('abc'), '')

And here is a (non-runnable) excerpt from my Regex class which shows the usage in context:

def parse_basic_re(self):
    """
    <elementary-re> "*" | <elementary-re> "+" | <elementary-re>
    """
    nfa = self.parse_elementary_re()
    if not nfa:
        return None
    next_match = self._lexer.peek()
    if not next_match or next_match.token != Token.METACHAR:
        return nfa
    if next_match.lexeme == '*':
        self._lexer.eat_metachar('*')
        return nfa.kleene()
    if next_match.lexeme == '+':
        self._lexer.eat_metachar('+')
        return nfa.union(nfa.kleene())
    return nfa

Answer 1

Thanks for the tests, it makes the code more understandable.

(My opinion) I think the biggest problem with the construction and design of the code is the recreation of the NFA() object inside itself. I believe because of how you've done this - it's forced you to jump through certain hoops to get your code working. From reading about Thompson's Construction and reviewing examples of regexes broken into graphs, I believe the construction should be much simpler. I would challenge you to go and learn some graph and vertex examples and then come back to try recreating your NFA/DFA. I believe you should be able to improve your code quite a bit.

Onto parts of the code: you have a duplicate loop (a single loop should always be sufficient):

    while to_check:

        # Copy states to current iteration
        while to_check:

and you modify the contents of what it's checking inside the loop:

    while to_check:

        # Copy states to current iteration
        while to_check:
            iteration.add(to_check.pop())

this is also never good. Even though you're using the copy functionality, the fact you're using it multiple times implies you really don't understand what is happening inside your data constructors. Duplicating the contents and then removing pieces of the content, you then later compare against an empty set if state not in checked: (on the first run) when the following statement modifies (again) the variable providing the control for the loop. This is not good - there is a confusing amount of changing of state in inner and outer loops.

(adding this quickly)

    def _single_state_closure(self, state: int) -> {int}:
        return self._at(state, '\0')

to continue the previous code:

            for state in iteration:
                next_states = self._single_state_closure(state)
                if state not in checked:
                    checked.add(state)
                    to_check.update(next_states)
                closure.update(next_states)

Your call to the self._single_state_closure(state) is only ever performed at this one place in the code (not worthy of a separate function and is confusing if people have to "pop out" when reading your code). Also, your def _at(self, state: int, char: str) function too is only used twice - once with a hardcoded value (as mentioned above with the _single_state_closure), and once more in a loop found inside _next_states.

Whilst the action might be the same, I think it best if you unravel the _at function into their separate statements at each of those points and discard the _at function. It will be much easier to understand the code if you do that.

Finally, (I know I've skipped over a lot), your def kleene(self): and def union(self, other: 'NFA'): functions are quite similar - obviously not for the hardcoded values inside the lambdas (those you should definitely extract to a separate function), but the creation of the table state, the copying of the table, modifying the terminal state - they are (kleen and union) functionally both the same.
This implies that both can share a standard creation function or a modification function inside them (or even both if you want to split it up and try to adhere to the Single Responsibility Principal).

I hope this isn't too confusing, and I hope I understood your code as well as you do. As mentioned, please have a go at graphs in python and then try writing this code all again from scratch.

Good luck!

Answer 2

Just reviewing DFA.

1. There are no docstrings. What does an object belonging to this class represent? What arguments do I pass to the constructor? What does the match method do?

2. In Python 3, all classes inherit from object so there is no need to specify this explicitly, unless you need to code to be portable to Python 2.7 (but the use of type annotations means that this can't be the case).

3. The initial state has to be the number 0. This seems arbitrary. Why not allow the caller to supply any Python object for the initial state?

4. The name table is vague: what kind of data is in this table? I would use a name like transition.

5. table needs to be a mapping from states to mappings from Unicode code points to states. Why not use characters instead of their Unicode code points? This would avoid the need for the call to ord, and would also make the tables easier to understand since the keys would be strings like 'a' and 'b' instead of mysterious numbers like 97 and 98.

6. The name pattern for the argument to the match method is misleading: normally this name is used for the pattern that needs to be matched, not the string that gets matched against the pattern. See for example re.match.

7. The match method returns the printable characters in the longest initial substring of the argument that matches, or the empty string if no initial substring matches. This means that if match returns the empty string, it's not clear whether the longest match consisted entirely of non-printing characters, or whether there was no match at all. Compare with re.match, which returns the distinguished value None when there is no match.

8. I don't understand the purpose of only returning the printable characters in the longest initial matching substring. This seems arbitrary — surely removing non-printing characters from the match is something that can be left to the caller (in the rare case that the caller needs this feature).

9. There are two mechanisms for the table to specify that a character is not matched — it can omit a transition for the Unicode code point for that character, or it can specify a transition to the special state -1. Having two mechanisms for the same feature is a bad idea because there could be a bug in one of the mechanisms that would be easy to miss because you only exercised the other mechanism. It is best to have only one way to do something.