This is an implementation of a general Purpose Parser/Matcher that can operate on any input stream of objects and will return all possible combinations.
In this case I do not value performance that much, but if you find something major, tell me.
This does by design no optimizations before parsing to be easily editable.
from __future__ import annotations
from typing import Iterator, TypeVar, Iterable, Generic, List
T = TypeVar('T')
class Rewindable(Iterator[T]):
"""
A Rewindable Wrapper around an iterator the allows for 'rewinding'
the iterator by reloading an old state
"""
def __init__(self, iterator: Iterable[T]):
self._iterator = iter(iterator)
self._index = 0
self._cache = []
def __next__(self) -> T:
self._index += 1
while self._index > len(self._cache):
self._cache.append(next(self._iterator))
return self._cache[self._index - 1]
def get_state(self) -> object:
"""Returns an object indicating the current position
that can be reloaded via `load_state`"""
return self._index
def load_state(self, state: object):
"""Reloads a state returned by `get_state`"""
assert isinstance(state, int)
self._index = state
IN = TypeVar('IN')
OUT = TypeVar('OUT')
class Parser(Generic[IN, OUT]):
"""
Base class for Parser which match pattern in an rewindable
input stream and return all possible matches
"""
def parse(self, in_stream: Rewindable[IN]) -> Iterator[Iterator[OUT]]:
"""
Matches the pattern in the Rewindable stream `in_stream`
Returns an Iterator over all possible matches (also iterator)
The implementation must insure that `in_stream` is reset
after StopIteration is thrown
"""
raise NotImplementedError
class DirectParser(Parser):
"""
A Parser that matches the value `expected` and yields `result` in response
"""
def __init__(self, expected: IN, result: OUT):
self.expected = expected
self.result = result
def __repr__(self):
return f"<{self.expected!r}>"
def parse(self, in_stream: Rewindable[IN]) -> Iterator[Iterator[OUT]]:
state = in_stream.get_state()
try:
n = next(in_stream)
except StopIteration:
return
if n == self.expected:
yield iter((self.result,))
in_stream.load_state(state)
class ORParser(Parser[IN, OUT]):
"""
A Parser that matches any Parser of `sub_parser` and returns all possible results of each one
"""
def __init__(self, sub_parser: List[Parser[IN, OUT]]):
self._sub_parser = list(sub_parser)
def __repr__(self):
return '(' + '|'.join(repr(sp) for sp in self._sub_parser) + ')'
def parse(self, in_stream: Rewindable[IN]) -> Iterator[Iterator[OUT]]:
for sp in self._sub_parser:
yield from sp.parse(in_stream)
class ANDParser(Parser[IN, OUT]):
"""
A Parser that matches the Parsers of `sub_parser` and returns all possible combinations
"""
def __init__(self, sub_parser: List[Parser[IN, OUT]]):
self._sub_parser = list(sub_parser)
def __repr__(self):
return '(' + ' '.join(repr(sp) for sp in self._sub_parser) + ')'
def parse(self, in_stream: Rewindable[IN]) -> Iterator[Iterator[OUT]]:
iterator_stack = [self._sub_parser[0].parse(in_stream)]
try:
result = [tuple(next(iterator_stack[0]))]
except StopIteration:
return
skip = False
while len(iterator_stack) > 0:
if not skip:
if len(iterator_stack) == len(self._sub_parser):
assert len(result) == len(self._sub_parser)
yield (v for t in result for v in t)
else:
iterator_stack.append(self._sub_parser[len(iterator_stack)].parse(in_stream))
result.append(())
try:
result[-1] = tuple(next(iterator_stack[-1]))
skip = False
except StopIteration:
skip = True
result.pop()
iterator_stack.pop()
class RepeatingNGParser(Parser[IN, OUT]):
"""
A Parser that matches `min_count` to `max_count` repetitions of Parser `parser`
and returns all possible combinations (Non Greedy, matches as few repetitions as possible first
`max_count` can be `None` to indicate that the repetition can be infinite
"""
def __init__(self, parser: Parser[IN, OUT], min_count: int, max_count: int = None):
self.parser = parser
self.min_count = min_count
self.max_count = max_count
def parse(self, in_stream: Rewindable[IN]) -> Iterator[Iterator[OUT]]:
count = self.min_count
if count == 0: # Special case Optional matching
yield iter(())
count = 1
while self.max_count is None or self.max_count > count:
iterator_stack = [self.parser.parse(in_stream)]
try:
result = [tuple(next(iterator_stack[0]))]
except StopIteration:
return
skip = False
any_result = False
while len(iterator_stack) > 0:
if not skip:
if len(iterator_stack) >= count:
assert len(result) == len(iterator_stack)
any_result = True
yield (v for t in result for v in t)
else:
iterator_stack.append(self.parser.parse(in_stream))
result.append(())
try:
result[-1] = tuple(next(iterator_stack[-1]))
skip = False
except StopIteration:
skip = True
result.pop()
iterator_stack.pop()
count += 1
if not any_result:
return
