Skip to content

Instantly share code, notes, and snippets.

@jml
Created October 7, 2013 21:49
Show Gist options
  • Save jml/6875518 to your computer and use it in GitHub Desktop.
Save jml/6875518 to your computer and use it in GitHub Desktop.
Streaming parser
from ometa.grammar import OMeta
from ometa.interp import TrampolinedGrammarInterpreter
from parsley import makeGrammar
import sys
grammar = """
hex_digit = digit | 'A' | 'B' | 'C' | 'D' | 'E' | 'F'
byte = <hex_digit hex_digit>:b -> int(b, 16)
"""
hex = makeGrammar(grammar, {}, name='hex', unwrap=True)
def iter_grammar(grammar, rule, input_stream, name='Grammar', buffer_size=1):
tokens = [] # Should really be an explicit queue.
interp = TrampolinedGrammarInterpreter(
OMeta(grammar).parseGrammar(name), rule, callback=lambda *a: tokens.append(a))
data = input_stream.read(buffer_size)
while data:
interp.receive(data)
for token in tokens:
yield token
tokens[:] = []
data = input_stream.read()
interp.end()
for token in iter_grammar(grammar, 'byte', sys.stdin):
print token
A0B0FFCD010283
$ python streaming.py < eg.data
(160, ParseError(2, None))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment