1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84
|
#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""
Four operations calculator, asynchronous. Due to various buffering
problems you probably won't see what's the point unless you force
stdin to be noninteractive, e.g.
$ echo '3*4+6' | python3 ./async_calc.py
"""
import operator, os, asyncio, sys, codecs
from ptk.async_lexer import token, AsyncLexer, EOF
from ptk.async_parser import production, leftAssoc, AsyncLRParser, ParseError
@leftAssoc('+', '-')
@leftAssoc('*', '/')
class Parser(AsyncLRParser, AsyncLexer):
async def asyncNewSentence(self, result):
print('== Result:', result)
# Lexer
def ignore(self, char):
return char in [' ', '\t']
@token(r'[1-9][0-9]*')
def number(self, tok):
tok.value = int(tok.value)
# Parser
@production('E -> "-" E<value>', priority='*')
async def minus(self, value):
print('== Neg: - %d' % value)
return -value
@production('E -> "(" E<value> ")"')
async def paren(self, value):
return value
@production('E -> number<number>')
async def litteral(self, number):
return number
@production('E -> E<left> "+"<op> E<right>')
@production('E -> E<left> "-"<op> E<right>')
@production('E -> E<left> "*"<op> E<right>')
@production('E -> E<left> "/"<op> E<right>')
async def binaryop(self, left, op, right):
print('Binary operation: %s %s %s' % (left, op, right))
return {
'+': operator.add,
'-': operator.sub,
'*': operator.mul,
'/': operator.floordiv
}[op](left, right)
async def main():
reader = asyncio.StreamReader()
await asyncio.get_event_loop().connect_read_pipe(lambda: asyncio.StreamReaderProtocol(reader), sys.stdin)
decoder = codecs.getincrementaldecoder('utf_8')()
parser = Parser()
while True:
byte = await reader.read(1)
if not byte:
break
char = decoder.decode(byte)
if char:
if char == '\n':
char = EOF
else:
print('Input char: "%s"' % repr(char))
await parser.asyncFeed(char)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
|