Last active
December 21, 2015 12:00
-
-
Save mgronhol/e428c728d82c6014a883 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3.4 | |
| import sys | |
| import shlex | |
| class ParserError(Exception): | |
| pass | |
| _example = """ | |
| step steppi1 | |
| begin | |
| requirements | |
| begin | |
| input1: false; | |
| input2: true; | |
| output1: false; | |
| end | |
| set valve1 open; | |
| set valve2 close; | |
| move axis1 position1; | |
| end | |
| """ | |
| def consume( token, tokens ): | |
| if tokens[0] != token: | |
| raise ParserError("Invalid token '%s', should be '%s'." % (tokens[0], token) ) | |
| tokens = tokens[1:] | |
| return tokens | |
| def parse_block( tokens ): | |
| tokens = consume( "begin", tokens ) | |
| bucket = [] | |
| counter = 1 | |
| for token in tokens: | |
| if token == "begin": | |
| counter += 1 | |
| elif token == "end": | |
| counter -= 1 | |
| if counter == 0: | |
| break | |
| bucket.append( token ) | |
| if counter > 0: | |
| raise ValueError | |
| tokens = tokens[len(bucket) + 1:] | |
| return tokens, bucket | |
| def parse_identifier( ctx, tokens ): | |
| idn = tokens[0] | |
| if len(idn) < 1 or idn[0].isdigit(): | |
| raise ValueError | |
| tokens = tokens[1:] | |
| return tokens, idn | |
| def parse_boolean( tokens ): | |
| value = None | |
| if tokens[0] == "false": | |
| value = False | |
| elif tokens[0] == "true": | |
| value = True | |
| # for more readability | |
| elif tokens[0] == "close": | |
| value = False | |
| elif tokens[0] == "open": | |
| value = True | |
| else: | |
| raise ParserError( "Invalid value for boolean: '%s'." % tokens[0] ) | |
| tokens = tokens[1:] | |
| return tokens, value | |
| def parse_single_requirement( ctx, tokens ): | |
| tokens, name = parse_identifier( ctx, tokens ) | |
| tokens = consume( ":", tokens ) | |
| tokens, value = parse_boolean( tokens ) | |
| tokens = consume( ";", tokens ) | |
| if (name not in ctx["inputs"]) and (name not in ctx["outputs"]): | |
| raise ParserError( "Input/Output '%s' not defined." % name ) | |
| return tokens, {name: value} | |
| def parse_requirements( ctx, tokens ): | |
| tokens = consume( "requirements", tokens ) | |
| tokens, body = parse_block( tokens ) | |
| reqs = [] | |
| while len(body) > 0: | |
| body, req = parse_single_requirement( ctx, body ) | |
| reqs.append( req ) | |
| return tokens, reqs | |
| def parse_statement( ctx, tokens ): | |
| stmt = None | |
| if tokens[0] == "set": | |
| tokens = consume( "set", tokens ) | |
| tokens, name = parse_identifier( ctx, tokens ) | |
| tokens, value = parse_boolean( tokens ) | |
| tokens = consume( ";", tokens ) | |
| stmt = ("set", name, value) | |
| if (name not in ctx["inputs"]) and (name not in ctx["outputs"]): | |
| raise ParserError( "Input/Output '%s' not defined." % name ) | |
| if name not in ctx["writable inputs"]: | |
| raise ParserError( "Writing to '%s' is forbidden." % name ) | |
| elif tokens[0] == "move": | |
| tokens = consume( "move", tokens ) | |
| tokens, name = parse_identifier( ctx, tokens ) | |
| tokens, value = parse_identifier( ctx, tokens ) | |
| tokens = consume( ";", tokens ) | |
| stmt = ("move", name, value) | |
| if name not in ctx["positions"]: | |
| raise ParserError( "Axis '%s' not defined." % name ) | |
| else: | |
| if value not in ctx["positions"][name]: | |
| raise ParserError( "Position '%s' not defined for axis '%s'." % (value, name) ) | |
| else: | |
| raise ParserError( "Unknown operation '%s'." % tokens[0] ) | |
| return tokens, stmt | |
| def parse_step( ctx, tokens ): | |
| tokens = consume( "step", tokens ) | |
| tokens, step_name = parse_identifier( ctx, tokens ) | |
| tokens, body = parse_block( tokens ) | |
| body, requirements = parse_requirements( ctx, body ) | |
| statements = [] | |
| while len(body) > 0: | |
| body, stmt = parse_statement( ctx, body ) | |
| statements.append( stmt ) | |
| return tokens, {"name": step_name, "requirements": requirements, "statements": statements} | |
| def parse_steps( ctx, tokens ): | |
| steps = [] | |
| while len( tokens ) >0: | |
| tokens, step = parse_step( ctx, tokens ) | |
| steps.append( step ) | |
| return tokens, steps | |
| def parse( ctx, txt ): | |
| lexer = shlex.shlex( txt ) | |
| tokens = [token for token in lexer] | |
| tokens, steps = parse_steps( ctx, tokens ) | |
| return steps | |
| context = { | |
| 'inputs': ["input1", "input2", "valve1", "valve2"], | |
| 'writable inputs': ["valve1", "valve2"], | |
| 'outputs': ["output1"], | |
| 'positions': {"axis1": ["position1"]} | |
| } | |
| import pprint | |
| pprint.pprint( parse(context, _example) ) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment