Andy's insights

Opinions, thoughts, archivements

Wed, 12 Apr 2006

Singleton and Borg Design Pattern

The Singleton Design Pattern is quite popular.

As usual Python provides an easy and even better solution for most situations the Singleton seems appropriate. But first look at our options:

The classic Singleton pattern implemented in Python looks like this:

class Singleton(object):
    def __new__(cls, *p, **k):
        if not '_the_instance' in cls.__dict__:
            cls._the_instance = object.__new__(cls)
        return cls._the_instance

The Borg Design Pattern Borg Design Pattern shares just one state. I’ve found it at ASPN (Alex Martelli).

The main idea is to just share the state by replacing the default instance dictionary with a shared one:

class Borg(object):
    _state = {}
    def __new__(cls, *p, **k):
        self = object.__new__(cls, *p, **k)
        self.__dict__ = cls._state
        return self

The best solution I found is to just use the Python mechanics: A Module is a singleton, why not just use it?

class _SingleInstance(object):
    def __init__(self):
        self.foo = 23
SingleInstance = _SingleInstance
del _SingleInstance

A good discussion about those pattern (and some others) can be found here: http://www.aleax.it/Python/5ep.html

The Borg design pattern can be found at ASPN: http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/66531

posted at: 13:05 | path: /python | permanent link to this entry

Sat, 18 Mar 2006

Using PyBlosxom

Searching for an extendable but simple weblog system, I finally found PyBlosxom.

This one is really simple and easy to extend. As a prove of concept I wrote the mod_python hander to integrate the blog in Apache 2. There is an old code snippled for this purpose somewhere out there. But this one won't work with recend software.

#!/usr/bin/python
"""
Pyblosxom mod_python handler

Based on ideas from 
 * Wari Wahab <wari at wari.per.sg>
   http://roughingit.subtlehints.net/pyblosxom/2003/Jun/02/
 * pyblosxom.cgi 1.3.2
"""

from mod_python import apache
import sys
sys.path.append('/etc/pyblosxom')
from config import py as cfg

if cfg.has_key("codebase"):
    sys.path.insert(0, cfg["codebase"])

from Pyblosxom.pyblosxom import PyBlosxom

__author__ = 'Andreas Stricker <rapidmax@gmx.net>'
__version__ = "1.3.2"
__copyright__ = "Copyright (c) 2006 Andreas Stricker"
__license__ = "Python"

def handler(req):
    req.content_type = 'text/html'
    req.add_common_vars()

    env = {}
    for k, v in req.subprocess_env.items():
        env[k] = v

    env['wsgi.input'] = req
    env['wsgi.errors'] = sys.stderr

    # setup url_scheme for static rendering
    if cfg.has_key('base_url'):
    	if cfg['base_url'].find('://') > 0:
            env['wsgi.url_scheme'] = cfg['base_url'][:cfg['base_url'].find("://")]
        else:
            env['wsgi.url_scheme'] = "http"
	    if cfg['base_url'].startswith('/'):
	    	cfg['base_url'] = "http://" + req.hostname + cfg['base_url']
    else:
        env['wsgi.url_scheme'] = "http"
        cfg['base_url'] = "/blog/blog.py"

    p = PyBlosxom(cfg, env)

    p.run()

    response = p.getResponse()

    # detect if we have to return "404 Not found" status
    # This case is handled as suggested in mod_python documenation
    if response.status.startswith('404'):
        req.status = apache.HTTP_NOT_FOUND

    # set headers
    for k, v in response.headers.items():
        if k.lower() == 'content-type':
            req.content_type = v
        req.headers_out[k] = v

    response.sendBody(req)

    return apache.OK

To make it work, you must register this script as handler for a specific directory:

DirectoryIndex blog.py
AddHandler mod_python .py
PythonHandler blog
#PythonDebug On

posted at: 14:58 | path: /python | permanent link to this entry

Mon, 06 Mar 2006

Building Finite State Machines (FSM) with Python

A problem that many times occured to me. I tried to implement a finite state machine, but there was no convenient framework to do this. Of corse there are powerful RAD Tools (e.g. for SDL), but I didn’t wanted this dependency. The framework should be simple, clean and easy to use but still powerfull enough to describe the problem. I know there is alway a switch statement FSM. This is good enough for simple state machines.

I ended up by implementing something for myself. The first try was a C++ module (see SIPSec ). two years later I played a bit with a Implementation in Python. This one worked, but didn’t satisfied me enough. It was still to complex.

Once again nearly a year later I had the Idea. I’ll provide the full documented source later. First I developp the ideas mixed with concrete code.

For a full state machine we need:

Further we should have (but not necessary need for a FSM by definition):

Let’s start at the end: We like to implement an example FSM looking like that:

  +-----------------+
  |       +-+e1     |
  V e1,e3 V | e2    |e1
 (A) ---> (B) ---> (C) <-+
  ^        ^        |e2  |
  |        |        V    |
  |        |       /x\   |
  |        |      /0_1\  |
  |        |     /0 1 2\ |
  |        |      | | |  |
  |        +------+ | +--+ 
  +-----------------+

The first thing we need are States. We can define those like this:

a = State('A') 
b = State('B')
c = State('C')

Fine, but how does a state look like? We have one methode process() that takes an event and decide if and which transition it has to choose. And then we have some methods needed to construct the state machine, adding transitions and operations. Have a look at the code


class State(object): ON_ENTER = 'on_enter’ ON_LEAVE = 'on_leave’ def init(self, name): self.name = name self.on_enter = [] # callbacks self.on_leave = [] # callbacks self.events = {} # events -> transition def repr(self): return self.name

def add_operation(self, when, operation): getattr(self, when).append(operation) return self def add_event(self, event, transition): self.events[event] = transition return self def enter(self): for op in self.on_enter: op(self) def leave(self): for op in self.on_leave: op(self) def process(self, event): if not self.events.has_key(event): return self # sink self.leave() return self.events[event].process(event)

For the building phase the add_event() method is important: It add a transition initiated by an event to the internal lookup-table (dict). If an event occurs it is submitted to the process() method. This function look up the event in the table and execute its transition process() method if an event exists or return itself without furter processing if not. The enter() and leave() functions should be clear.

We talked about a transition, but never saw one. May I introduce them? A transition is the thing between two states on time they changes. This is if the old state is left and before the new one comes. A Transition defines which state comes next. And it also provides a hook for operations to execute on this transition. Likewise to the state the transition consist of methods to build the machine and of methodes used at execution time. When we instantiate a transition we define the end state we enter.

class Transition(object):
    def __init__(self, to_state):
        self.to_state = to_state
        self.operations = []
    def add_operation(self, op):
        self.operations.append(op)
        return self
    def process(self, event):
        for op in self.operations:
            op(event)
        self.to_state.enter()
        return self.to_state

The process() method returns the state we end in. To connect two states, we do something like this:

b.add_event(e2, Transition(c).add_operation(Printer("Transition b -> c")))

This adds a transition to state B on event e thats execute the callable Printer() on transition.

Finally we are missing the events. Those aren’t nothing special. We could have choosen to differentiate on class or just choosen a single object (dictionary for example). But I think it’s best to have a single Event class and differentiate based on a ID given on construction time. The ID should be a string (but could also be an integer). We are still able to subclass the Event if we need not only to set atributes but also to give a behaviour to an event. So look at the implementation: It’s somewhat pythonic:

class Event(object):
    def __init__(self, event_id):
        self.event_id = event_id
    def __hash__(self):
        return hash(self.event_id)
    def __cmp__(self, other):
        if isinstance(other, Event):
            return cmp(self.event_id,  other.event_id)
        else:
            return cmp(self.event_id, other)
    def __repr__(self):
        return "<Event %s>" % `self.event_id`

Finally we define some events for our example:

e1 = Event('e1')
e2 = Event('e2')
e3 = Event('e3')

This is nearly all we need for our example FSM. But what we need an entry point, an interface to access the whole thing. I’ll call them FSM because it represent an FSM (even if it’s deadly simple!):

class FSM(object):
    def __init__(self, start):
        self.state = self.start = start
    def process(self, event):
        self.state = self.state.process(event)

If you look at our example you seed that there is still on missing brick: The conditional. I implemented this one as generic as possible: First it subclass our transition and second it ends up in transitions (there must be at least one). The Conditional has finite ends (I call them paths) and a desition to choose one of those paths. The paths are simple implemented by yet another lookup table (dict) and the decition is just a callable object (function, method, lambda or callable class). Once again we don’t have to subclass the Conditional class for every single task.


class Conditional(Transition): def init(self, decision, paths): Transition.init(self, None) self.decision = decision self.paths = paths

def process(self, event): for op in self.operations: op(event) return self.paths[self.decision(event)].process(event)

Simple, isn’t it?

Let’s do some action: Here is the code to implement the FSM as shown above:


class Printer(object): def init(self, txt): self.txt = txt def call(self, *args): print self.txt, str(args)

a = State('A’)
b = State('B’)
c = State('C’)
e1 = Event('e1’)
e2 = Event('e2’)
e3 = Event('e3’)

t = Transition(b).add_operation(Printer(“Transition a -> b”))
a.add_event(e1, t).add_event(e3, t)
b.add_event(e1, Transition(b).add_operation(Printer(“Transition b -> b”))\ .add_operation(Printer(“2nd Operation”)))
b.add_event(e2, Transition©.add_operation(Printer(“Transition b -> c”)))
c.add_event(e1, Transition(a).add_operation(Printer(“Transition c -> a”)))
c.add_event(e2, Conditional(lambda event: event.x, { 0 : Transition(b).add_operation(Printer(“Transition c -> b”)), 1 : Transition(a).add_operation(Printer(“Transition c -> a”)), 2 : Transition©.add_operation(Printer(“Transition c -> c”)),
}).add_operation(Printer(“Decision (x = ?) -> a,b,c”)))

a.add_operation('on_enter’, Printer(“Enter State A”))
a.add_operation('on_leave’, Printer(“Leave State A”))
b.add_operation('on_enter’, Printer(“Enter State B”))
b.add_operation('on_leave’, Printer(“Leave State B”))
c.add_operation('on_enter’, Printer(“Enter State C”))
c.add_operation('on_leave’, Printer(“Leave State C”))

fsm = FSM

That’s it. Next we jump a bit around just to see if it works:

fsm.process(e1)
fsm.process(e1)
fsm.process(e2)
fsm.process(e1)
fsm.process(e3)
fsm.process(e2)
e2.x = 2
fsm.process(e2)
e2.x = 0
fsm.process(e2)
fsm.process(e2)
fsm.process(e2)

The output should look like this:

Leave State A (A,)
Transition a -> b (<Event 'e1'>,)
Enter State B (B,)
Leave State B (B,)
Transition b -> b (<Event 'e1'>,)
2nd Operation (<Event 'e1'>,)
Enter State B (B,)
Leave State B (B,)
Transition b -> c (<Event 'e2'>,)
Enter State C (C,)
Leave State C (C,)
Transition c -> a (<Event 'e1'>,)
Enter State A (A,)
Leave State A (A,)
Transition a -> b (<Event 'e3'>,)
Enter State B (B,)              
Leave State B (B,)              
Transition b -> c (<Event 'e2'>,)
Enter State C (C,)              
Leave State C (C,)              
Decision (x = ?) -> a,b,c (<Event 'e2'>,)
Transition c -> c (<Event 'e2'>,)
Enter State C (C,)              
Leave State C (C,)              
Decision (x = ?) -> a,b,c (<Event 'e2'>,)
Transition c -> b (<Event 'e2'>,)
Enter State B (B,)              
Leave State B (B,)              
Transition b -> c (<Event 'e2'>,)
Enter State C (C,)              
Leave State C (C,)              
Decision (x = ?) -> a,b,c (<Event 'e2'>,)
Transition c -> a (<Event 'e2'>,)
Enter State A (A,)

posted at: 23:48 | path: /python | permanent link to this entry

Wed, 01 Mar 2006

PyBlosxom generic Wiki markup

It’s nice to format the blog entries in HTML. Of course HTML is somewhat complicated. I like Wiki markup: It is straightforward and simple.

To add Wiki markup support to PyBlosxom, there is a plugin in entryparsers subdirectory of the contrib package. Not bad, but I’m missing a code block option. Once again I was able to enhance the source code.

Apropos: This entry is formated with the enhanced genericwiki parser. Here the prove:


“”“
Generic wiki markup PreFormatter 2002-11-18, for pyblosxom
CHANGE wikibaseurl to point to your wiki, & wikinamepattern to yours
Bug reports, comments, presents, etc. to John Abbe at johnca@ourpla.net
ToDo: Lists; code; InterWiki links; other wikinamepatterns

You can configure this as your default preformatter by configuring it in your
L{config} file as follows::

py['parser’] = 'genericwiki’

or in your blosxom entries, place a C{#parser wiki} line after the title of
your blog::

My Little Blog Entry #parser genericwiki This is a text in '’'wiki’'’ format

This preformatter also supports WikiWirds, you need to point out where your
Wiki site is. This is configured with a new variable in your config.py file,
'genericwiki_baseurl’::

py['genericwiki_baseurl’] = 'http://www.google.com/search?q=’

The above example would expand 'WikiWord’ to
http://www.google.com/search?q=WikiWord

Permission is hereby granted, free of charge, to any person
obtaining a copy of this software and associated documentation
files (the “Software”), to deal in the Software without restriction,
including without limitation the rights to use, copy, modify,
merge, publish, distribute, sublicense, and/or sell copies of the
Software, and to permit persons to whom the Software is furnished
to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

Copyright 2004, 2005 John Abbe
“”“
author = 'John Abbe '
version = “$Id: genericwiki.py,v 1.1.2.1 2005/06/21 18:22:33 willhelm Exp $”
PREFORMATTER_ID = 'genericwiki’
import re

  1. caching compiled regular expressions

re_cache = None

def cb_preformat(args): “”“ Preformat callback chain looks for this.

param args: a dict with 'parser' string and a list 'story' type args: dict “”“ if args['parser’] == PREFORMATTER_ID: config = args['request’].getConfiguration() baseurl = config.get('genericwiki_baseurl’, None) return parse(’'.join(args['story’]), baseurl)

def _build_cache(): “”“ Compile regular expression pattern and store them in global cache “”“ global re_cache # build the cache first time called (speedup later calls!) re_cache = {}

# WikiName pattern used in your wiki wikinamepattern = r’\b(([A-Z]+[a-z]+){2,})\b’ # original mailurlpattern = r’mailto\:[\”\-\_\.\w]+\@[\-\_\.\w]+\w’ newsurlpattern = r’news\:(?:\w+\.){1,}\w+’ fileurlpattern = r’(?:http|https|file|ftp):[/-_.\w-]+[\/\w][?&+=%\w/-_.#]*’ # Turn '[xxx:address label]’ into labeled link re_cache['bracket_links’] = re.compile(r’\[(’ + fileurlpattern + '|’ + mailurlpattern + '|’ + newsurlpattern + ')\s+(.+?)\]’) # Convert naked URLs into links — skip ones with a “ before re_cache['nacked_url’] = re.compile(r’(?def _parse_markup(text, wikibaseurl): “”“ Parse wiki markup for text

param text: Text for conversion type text: string “”“ if not len(text.strip()): return “” # Turn '[xxx:address label]’ into labeled link text = re_cache['bracket_links’].sub(r’\2', text) # Convert naked URLs into links — skip ones with a “ before text = re_cache['nacked_url’].sub(r’\1', text) # Convert WikiNames into links if wikibaseurl: text = re_cache['wiki_names’].sub(’\1', text) # '’ for emphasis, '’' for strong, —— for a horizontal rule text = re_cache['strong’].sub(r”\1“, text) text = re_cache['emphasis’].sub(r”\1“, text) text = re_cache['ruler’].sub(”
“, text) # Convert two or more newlines into

text = re_cache['para’].sub(r’

\n

', text) return “

“ + text + “

def _parse_code(text): if text.strip().startswith(’#!python’): keywords = “(and|del|for|is|raise|assert|elif|from|lambda|return|break”\ “|else|global|not|try|class|except|if|or|while|continue”\ “|exec|import|pass|yield|def|finally|in|print)” text = re.sub(keywords, r’\1', text) return text

def parse(text, wikibaseurl): “”“ The main workhorse that convert wiki text into html markup

param text: Text for conversion type text: string @return: string “”“ global re_cache if not re_cache: _build_cache() # replace entities text = text.replace(’&’, '&’).replace(’<', '<').replace('>', '>’) # convert preformated text and wiki markup text fragments = [ x.split(’} }}’, 1) for x in text.split(’{ {{’) ] result = [] for fragment in fragments: if len(fragment) > 1: result.append(”<" + "pre" + ">“ + _parse_code(fragment0) + ““) result.append(_parse_markup(fragment1, wikibaseurl)) else: result.append(_parse_markup(fragment0, wikibaseurl)) return “”.join(result)

if name == 'main': text = “”“This is a test To test the wiki

[http://roughingit.subtlehints.net/pyblosxom?blah=duh#spam A link] news:roughingit.subtlehints.net/pyblosxom/ – no, ''I’' do '’'not’'’ have a news server. mailto:wari@example should go link to an email. WikiWiki is a wiki Keyword A text for preformated text: {{{ #!python import this print “Yes! don’t replace ''source’' '’'code’'’ here: http://don.t.do.this.ch/” }}} That’s it, folks! “”“ print parse(text, 'http://wiki.subtlehints.net/moin/’)

Note: You should remove the space in curly brackets.

posted at: 23:36 | path: /python | permanent link to this entry

Mon, 27 Feb 2006

Adding new messages with e-mail

This new python module allows to write new messages in your mail program and send it to the blog

Following Items are needed to successful blog with mail:

  1. Subject for the title
  2. Passphrase to authenticate
  3. Path with category and filename (no trailing slash!)
  4. A empty line (as with RFC822 header/body
  5. The blog body, with markup

This script will do the work. It not finished yet, there are some problems to solve.

#!/usr/bin/python
"""
Script to check a IMAP folder for specific blog mails to
upload to a weblog via xmlrpc
"""
import getpass
import xmlrpclib
import imaplib
import email.Parser

passphrase = "A good passphrase"
USERNAME = 'yourname'

class IMAPError(Exception): pass

class BlogEntry(object):
    def __init__(self, data=None):
        self.title = None
        self.body = None
        self.path = None
        self.valid = False
        if data:
            self.parse(data)

    def __repr__(self): return self.title

    def parse(self, data):
        parser = email.Parser.Parser()
        em = parser.parsestr(data)
        if not em.has_key('Subject'):
            return
        content = em.get_payload()
        if em.is_multipart():
            content = content[0]            # only use first part
        content = parser.parsestr(content)  # body is a pseudo RFC822 text

        if not content.get('Passphrase') or not content.get('Path'):
            return

        #if not em.get('Passphrase') == passphrase:
        #    return

        self.title = em.get('Subject')
        self.path = content.get('Path')
        if not self.path.startswith('/'):
            self.path = '/' + self.path
        if not self.path.endswith('/'):
            self.path += '/'
        self.body = content.get_payload()
        self.valid = True


def get_blog_mails(server, mailbox, username, password):
    m = imaplib.IMAP4(server)
    
    status, data = m.login(username, password)
    if not status.startswith('OK'):
        raise IMAPError, status
    
    status, data = m.select(mailbox)
    if not status.startswith('OK'):
        raise IMAPError, status

    status, data = m.search(None, 'ALL')
    if not status.startswith('OK'):
        raise IMAPError, status
    mail_id_list = data[0].split()

    result = []
    for mailid in mail_id_list:
        status, data = m.fetch(mailid, '(RFC822)')
        result.append(data[0][1])
        # TODO ast: mark mail as read

    m.logout()
    return result
class BloggerAPI(object):
    APPKEY = "070F5B78CF8DF3450B732A53546CCFC6636A4547"

    def __init__(self, url, username, password):
        self.server = xmlrpclib.Server(url)
        self.username = username
        self.password = password

    def getUsersBlogs(self):
        return self.server.blogger.getUsersBlogs(self.APPKEY, 
                                                 self.username, 
                                                 self.password)

    def newPost(self, blogid, content, publish=1):
        self.server.blogger.newPost(self.APPKEY, 
                                    blogid, 
                                    self.username, 
                                    self.password, 
                                    content, 
                                    publish)

    def getRecentPosts(self, blogid, numberOfPosts=5):
        return self.server.blogger.getRecentPosts(self.APPKEY, 
                                                  blogid, 
                                                  self.username, 
                                                  self.password, 
                                                  numberOfPosts)

    def getUserInfo(self):
        return self.server.blogger.getUserInfo(self.APPKEY, 
                                               self.username, 
                                               self.password)

if __name__ == '__main__':
    password_mail = getpass.getpass("Your mailbox password: ")
    password_blog = getpass.getpass("Your blog password: ")
    blogger = BloggerAPI('http://sirius/blog/blog.py/RPC', 
                         USERNAME, password_blog)
    for mail in get_blog_mails("sirius.hogwarts.local", 
                               "INBOX.projects.weblog", 
                               USERNAME, password_mail):
        blog = BlogEntry(mail)
        print blog.title, blog.path
        try:
            blogger.newPost(blog.path, blog.title + "\n" + blog.body)
        except xmlrpclib.Fault, e:
            print e

posted at: 01:53 | path: /python | permanent link to this entry