File: async_calc.py

package info (click to toggle)
python-ptk 1.3.8%2Bdfsg-3
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 416 kB
  • sloc: python: 3,616; makefile: 200
file content (84 lines) | stat: -rw-r--r-- 2,237 bytes parent folder | download | duplicates (3)
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())