2013 samples

Author: eliben

2012/rd_infix_precedence.py

@@ -0,0 +1,156 @@
+#-----------------------------------------------
+# Precedence climbing expression parser.
+#
+# Eli Bendersky ([email protected])
+# License: this code is in the public domain
+# Last modified: July 2012
+#-----------------------------------------------
+from collections import namedtuple
+import re
+
+
+Tok = namedtuple('Tok', 'name value')
+
+
+class Tokenizer(object):
+    """ Simple tokenizer object. The cur_token attribute holds the current
+        token (Tok). Call get_next_token() to advance to the
+        next token. cur_token is None before the first token is
+        taken and after the source ends.
+    """
+    TOKPATTERN = re.compile("\s*(?:(\d+)|(.))")
+
+    def __init__(self, source):
+        self._tokgen = self._gen_tokens(source)
+        self.cur_token = None
+
+    def get_next_token(self):
+        """ Advance to the next token, and return it.
+        """
+        try:
+            self.cur_token = self._tokgen.next()
+        except StopIteration:
+            self.cur_token = None
+        return self.cur_token
+
+    def _gen_tokens(self, source):
+        for number, operator in self.TOKPATTERN.findall(source):
+            if number:
+                yield Tok('NUMBER', number)
+            elif operator == '(':
+                yield Tok('LEFTPAREN', '(')
+            elif operator == ')':
+                yield Tok('RIGHTPAREN', ')')
+            else:
+                yield Tok('BINOP', operator)
+
+    def __repr__(self):
+        return 'Tokenizer(cur_token=%s)' % str(self.cur_token)
+
+
+
+# For each operator, a (precedence, associativity) pair.
+OpInfo = namedtuple('OpInfo', 'prec assoc')
+
+OPINFO_MAP = {
+    '+':    OpInfo(1, 'LEFT'),
+    '-':    OpInfo(1, 'LEFT'),
+    '*':    OpInfo(2, 'LEFT'),
+    '/':    OpInfo(2, 'LEFT'),
+    '^':    OpInfo(3, 'RIGHT'),
+}
+
+
+def parse_error(msg):
+    raise RuntimeError(msg)
+
+
+from eblib.tracer import TraceCalls
+
+@TraceCalls(show_ret=True)
+def compute_atom(tokenizer):
+    tok = tokenizer.cur_token
+    if tok.name == 'LEFTPAREN':
+        tokenizer.get_next_token()
+        val = compute_expr(tokenizer, 1)
+        if tokenizer.cur_token.name != 'RIGHTPAREN':
+            parse_error('unmatched "("')
+        tokenizer.get_next_token()
+        return val
+    elif tok is None:
+            parse_error('source ended unexpectedly')
+    elif tok.name == 'BINOP':
+        parse_error('expected an atom, not an operator "%s"' % tok.value)
+    else:
+        assert tok.name == 'NUMBER'
+        tokenizer.get_next_token()
+        return int(tok.value)
+
+
+@TraceCalls(show_ret=True)
+def compute_expr(tokenizer, min_prec):
+    atom_lhs = compute_atom(tokenizer)
+
+    while True:
+        cur = tokenizer.cur_token
+        if (cur is None or cur.name != 'BINOP'
+                        or OPINFO_MAP[cur.value].prec < min_prec):
+            break
+
+        # Inside this loop the current token is a binary operator
+        assert cur.name == 'BINOP'
+
+        # Get the operator's precedence and associativity, and compute a
+        # minimal precedence for the recursive call
+        op = cur.value
+        prec, assoc = OPINFO_MAP[op]
+        next_min_prec = prec + 1 if assoc == 'LEFT' else prec
+
+        # Consume the current token and prepare the next one for the
+        # recursive call
+        tokenizer.get_next_token()
+        atom_rhs = compute_expr(tokenizer, next_min_prec)
+
+        # Update lhs with the new value
+        atom_lhs = compute_op(op, atom_lhs, atom_rhs)
+
+    return atom_lhs
+
+
+def compute_op(op, lhs, rhs):
+    lhs = int(lhs); rhs = int(rhs)
+    if op == '+':   return lhs + rhs
+    elif op == '-': return lhs - rhs
+    elif op == '*': return lhs * rhs
+    elif op == '/': return lhs / rhs
+    elif op == '^': return lhs ** rhs
+    else:
+        parse_error('unknown operator "%s"' % op)
+
+
+def test():
+    def compute(s):
+        t = Tokenizer(s)
+        t.get_next_token()
+        return compute_expr(t, 1)
+
+    assert compute('1 + 2 * 3') == 7
+    assert compute('7 - 9 * (2 - 3)') == 16
+    assert compute('2 * 3 * 4') == 24
+    assert compute('2 ^ 3 ^ 4') == 2 ** (3 ** 4)
+    assert compute('(2 ^ 3) ^ 4') == 4096
+    assert compute('5') == 5
+    assert compute('4 + 2') == 6
+    assert compute('9 - 8 - 7') == -6
+    assert compute('9 - (8 - 7)') == 8
+    assert compute('(9 - 8) - 7') == -6
+    assert compute('2 + 3 ^ 2 * 3 + 4') == 33
+
+
+if __name__ == '__main__':
+    #test()
+
+    t = Tokenizer('2 + 3^2*3 + 4')
+    t.get_next_token()
+    print compute_expr(t, min_prec=1)
+

2013/twisted_irc_server.py

@@ -0,0 +1,44 @@
+#-------------------------------------------------------------------------------
+# twisted_irc_server.py
+#
+# A simple IRC server that flattens out what's happening behind the scenes of
+# 'twistd words'. Useful mainly to test clients on localhost.
+#
+# Eli Bendersky ([email protected])
+# Last updated: 2013.01.27
+# This code is in the public domain
+#-------------------------------------------------------------------------------
+import sys
+
+from twisted.cred import checkers, portal
+from twisted.internet import reactor
+from twisted.internet.endpoints import TCP4ServerEndpoint
+from twisted.python import log
+from twisted.words import service
+
+
+ROOM = 'room'
+USERS = dict(
+    user1='pass1',
+    user2='pass2',
+    user3='pass3',
+    user4='pass4')
+
+
+if __name__ == '__main__':
+    log.startLogging(sys.stdout)
+
+    # Initialize the Cred authentication system used by the IRC server.
+    realm = service.InMemoryWordsRealm('testrealm')
+    realm.addGroup(service.Group(ROOM))
+    user_db = checkers.InMemoryUsernamePasswordDatabaseDontUse(**USERS)
+    portal = portal.Portal(realm, [user_db])
+
+    # IRC server factory.
+    ircfactory = service.IRCFactory(realm, portal)
+
+    # Connect a server to the TCP port 6667 endpoint and start listening.
+    endpoint = TCP4ServerEndpoint(reactor, 6667)
+    endpoint.listen(ircfactory)
+    reactor.run()
+

2013/twisted_irc_testbot.py

@@ -0,0 +1,101 @@
+#-------------------------------------------------------------------------------
+# twisted_irc_testbot.py
+#
+# A sample IRC bot based on the example in Twisted's docs.
+#
+# Eli Bendersky ([email protected])
+# Last updated: 2013.01.27
+# This code is in the public domain
+#-------------------------------------------------------------------------------
+import argparse
+import sys
+
+from twisted.internet import reactor, protocol
+from twisted.python import log
+from twisted.words.protocols import irc
+
+
+class TestBot(irc.IRCClient):
+    def __init__(self, channel, nickname, password):
+        self.channel = channel
+        self.nickname = nickname
+        self.password = password
+
+    def connectionMade(self):
+        irc.IRCClient.connectionMade(self)
+        log.msg("[connected]")
+
+    def connectionLost(self, reason):
+        irc.IRCClient.connectionLost(self, reason)
+        log.msg("[disconnected]")
+
+    def signedOn(self):
+        """Called after sucessfully signing on to the server."""
+        self.join(self.channel)
+
+    def joined(self, channel):
+        """Called when I finish joining a channel.
+
+        channel has the starting character intact.
+        """
+        log.msg("[I have joined %s]" % channel)
+        self.msg(channel, "user1: bonbon")
+
+    def privmsg(self, user, channel, msg):
+        """Called when I have a message from a user to me or a channel."""
+        user = user.split('!', 1)[0]
+        log.msg("<%s> %s" % (user, msg))
+
+        # Check to see if they're sending me a private message
+        if channel == self.nickname:
+            self.msg(user, 'Thanks for the private message')
+        else:
+            # Otherwise check to see if it is a message directed at me
+            if msg.startswith(self.nickname + ":"):
+                msg = "%s: I am a bot" % user
+                self.msg(channel, msg)
+                log.msg("<%s> %s" % (self.nickname, msg))
+
+    def lineReceived(self, line):
+        """Low level LineReceiver callback, used for debugging..."""
+        log.msg('>> %s' % line)
+        # Twisted's classes are old-style, so no super(), oh my...
+        irc.IRCClient.lineReceived(self, line)
+
+
+class TestBotFactory(protocol.ClientFactory):
+    def __init__(self, channel, nickname, password):
+        self.channel = channel
+        self.nickname = nickname
+        self.password = password
+
+    def buildProtocol(self, addr):
+        return TestBot(self.channel, self.nickname, self.password)
+
+    def clientConnectionLost(self, connector, reason):
+        reactor.stop()
+        connector.connect()
+
+    def clientConnectionFailed(self, connector, reason):
+        reactor.stop()
+
+
+if __name__ == '__main__':
+    argparser = argparse.ArgumentParser()
+    argparser.add_argument('--server', help='the server to connect to',
+                           default='localhost')
+    argparser.add_argument('--port', help='TCP port',
+                           default=6667)
+    argparser.add_argument('--channel', help='channel/room name to join',
+                           default='room')
+    argparser.add_argument('--nickname', default='user1')
+    argparser.add_argument('--password', default='pass1')
+    args = argparser.parse_args()
+
+    log.startLogging(sys.stdout)
+
+    # Create client, connect to server and run
+    f = TestBotFactory(args.channel, args.nickname, args.password)
+    reactor.connectTCP(args.server, args.port, f)
+    reactor.run()
+