VIm command to open a Python module

pyhelp

#!/usr/bin/env python

# Show help for Python

__author__ = "Miki Tebeka <miki.tebeka@gmail.com>"
__license__ = "BSD"

def main(argv=None):
    import sys
    from optparse import OptionParser
    import webbrowser

    argv = argv or sys.argv

    parser = OptionParser("%prog [MODULE[.FUNCITON]]")
    opts, args = parser.parse_args(argv[1:])

    if len(args) not in (0, 1):
        parser.error("wrong number of arguments") # Will exit

    root = "http://docs.python.org"

    if not args:
        url = "%s/%s" % (root, "modindex.html")
    elif "." in args[0]:
        module, function = args[0].split(".")
        url = "%s/library/%s.html#%s.%s" % (root, module, module, function)
    else:
        url = "%s/library/%s.html" % (root, args[0])

    webbrowser.open(url)

if __name__ == "__main__":
    main()

Two gnupg Wrappers

encrypt

#!/bin/bash

# Encrypt input using gnupg and shred it
# THIS DELETES THE ORIGINAL FILE

err=0
for file in $@
do
    if [ $file != ${file/.gpg/} ]; then
        echo "$file already encrypted"
        continue
    fi
    if [ -d $file ]; then
        fname=$file.tar.bz2
        tar -cjf $fname $file
        find $file -type f -exec shred -u {} \;
        rm -r $file
    elif [ -f $file ]; then
        fname=$file
    else
        echo "ERROR: Unknown file type $file"
        err=1
        continue
    fi

    gpg -r miki -e $fname
    if [ $? -ne 0 ]; then
        echo "ERROR: can't encrypt"
        err=1
        continue
    fi

    shred -u $fname
    if [ $? -ne 0 ]; then
        echo "ERROR: can't shred"
        err=1
        continue
    fi
done

exit $err

decrypt

#!/bin/bash

# Decrypt a file encrypted with gpg


# Check command line
if [ $# -lt 1 ]; then
 echo "usage: `basename $0` FILE[s]"
 exit 1
fi

err=0
for file in $@
do
 if [ ! -f $file ]; then
  echo "$file: Not found"
  err=1
  continue
 fi

 # Output to current directory
 bfile=`basename $file`
    ext=`echo $file | sed -e 's/.*\.\([!.]*\)/\1/'`
    if [ "$ext" != "gpg" ] && [ "$ext" != "pgp" ]; then
  echo "Cowardly refusing to decrypt a file without gpg/pgp extension"
        err=1
        continue
 fi

 outfile=${bfile%.$ext}

    echo $file '->' $outfile
 gpg -d -o $outfile $file
 if [ $? -ne 0 ]; then
        echo "error decrypting $file"
        err=1
 fi
done

exit $err

What's My IP?

#!/usr/bin/env python
'''Find local machine IP (cross platform)'''

from subprocess import Popen, PIPE
from sys import platform
import re

COMMANDS = {
    "darwin" : "/sbin/ifconfig",
    "linux" : "/sbin/ifconfig",
    "linux2" : "/sbin/ifconfig",
    "win32" : "ipconfig",
}

def my_ip():
    command = COMMANDS.get(platform, "")
    assert command, "don't know how to get IP for current platform"

    pipe = Popen([command], stdout=PIPE)
    pipe.wait()
    output = pipe.stdout.read()

    for ip in re.findall("(\d+\.\d+\.\d+\.\d+)", output):
        if ip.startswith("127.0.0"):
            continue
        return ip

if __name__ == "__main__":
    print my_ip()

Little Genetics

#!/usr/bin/env python
'''Playing with genetic algorithms, see 
http://en.wikipedia.org/wiki/Genetic_programming.

The main idea that the "chromosome" represents variables in our algorithm and we
have a fitness function to check how good is it. For each generation we keep the
best and then mutate and crossover some of them.

Since the best chromosomes move from generation to generation, we cache the
fitness function results.

I'm pretty sure I got the basis for this from somewhere on the net, just don't
remeber where :)
'''

from itertools import starmap
from random import random, randint, choice
from sys import stdout

MUTATE_PROBABILITY = 0.1

def mutate_gene(n, range):
    if random() > MUTATE_PROBABILITY:
        return n

    while 1:
        # Make sure we mutated something
        new = randint(range[0], range[1])
        if new != n:
            return new

def mutate(chromosome, ranges):
    def mutate(gene, range):
        return mutate_gene(gene, range)

    while 1:
        new = tuple(starmap(mutate, zip(chromosome, ranges)))
        if new != chromosome:
            return new

def crossover(chromosome1, chromosome2):
    return tuple(map(choice, zip(chromosome1, chromosome2)))

def make_chromosome(ranges):
    return tuple(starmap(randint, ranges))

def breed(population, size, ranges):
    new = population[:]
    while len(new) < size:
        new.append(crossover(choice(population), choice(population)))
        new.append(mutate(choice(population), ranges))

    return new[:size]

def evaluate(fitness, chromosome, data, cache):
    if chromosome not in cache:
        cache[chromosome] = fitness(chromosome, data)

    return cache[chromosome]

def update_score_cache(population, fitness, data, cache):
    for chromosome in population:
        if chromosome in cache:
            continue
        cache[chromosome] = fitness(chromosome, data)

def find_solution(fitness, data, ranges, popsize, nruns, verbose=0):
    score_cache = {}
    population = [make_chromosome(ranges) for i in range(popsize)]
    for generation in xrange(nruns):
        update_score_cache(population, fitness, data, score_cache)
        population.sort(key=score_cache.get, reverse=1)
        if verbose:
            best = population[0]
            err = score_cache[best]
            print "%s: a=%s, b=%s, err=%s" % (generation, best[0], best[1], err)

        base = population[:popsize/4]
        population = breed(base, popsize, ranges)

    population.sort(key=score_cache.get, reverse=1)
    return population[0], score_cache[population[0]]

def test(show_graph=1):
    '''Try to find a linear equation a*x + b that is closest to log(x)'''
    from math import log
    xs = range(100)
    data = map(lambda i: log(i+1) * 100, xs)
    def fitness(chromosome, data):
        '''Calculate average error'''
        a, b = chromosome
        def f(x):
            return a * x + b
        values = map(f, xs)
        diffs = map(lambda i: abs(values[i] - data[i]), xs)

        # We want minimal error so return 1/error
        return 1 / (sum(diffs) / len(diffs))

    # Show a nice plot
    (a, b), err = find_solution(fitness, data, ((0, 100), (0, 100)), 10, 100, 1)
    print "best: a=%s, b=%s (error=%s)" % (a, b, err)

    data2 = map(lambda x: a * x + b, range(100))
    if not show_graph:
        return

    import pylab
    l1, l2 = pylab.plot(xs, data, xs, data2)
    pylab.legend((l1, l2), ("log(x+1)", "%s * x + %s" % (a, b)))
    pylab.show()

if __name__ == "__main__":
    test()

A Twitter Trends / Google News meshup

This tried to automatically explain why things are trending on Twitter.

The Python Server (trends.py)

#!/usr/bin/env python
'''A twitter trends/google news mesh

loader_thread get news trends every 1min and set _TRENDS_HTML
The web server serves _TRENDS_HTML with is refresed via JavaScript every 30sec
'''

from urllib import urlopen, urlencode
import feedparser
from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
from os.path import dirname, join, splitext
import json
from threading import Thread
from time import sleep

def ascii_clean(text):
    return text.encode("ascii", "ignore") # Pure ACII

def trend_news(trend):
    query = {
        "q" : ascii_clean(trend),
        "output" : "rss"
    }
    url = "http://news.google.com/news?" + urlencode(query)
    return feedparser.parse(url).entries

def current_trends():
    url = "http://search.twitter.com/trends.json"
    return json.load(urlopen(url))["trends"]

def news_html(news):
    html = '<li><a href="%(link)s">%(title)s</a></li>'
    chunks = map(lambda e: html % e, news)
    return "\n".join(["<ul>"] + chunks + ["</ul>"])

def trend_html(trend):
   thtml = '<a href="%(url)s">%(name)s' % trend
   nhtml = news_html(trend_news(trend["name"]))
   return '<tr><td>%s</td><td>%s</td><tr>' % (thtml, nhtml)

def table_html(trends):
    return ("<table>" + 
            "<tr><th>Trend</th><th>Related News</th></tr>" +
            "".join(map(trend_html, trends)) +
            "</table>"
           )

_TRENDS_HTML = ""
def loader_thread():
    global _TRENDS_HTML

    while 1:
        _TRENDS_HTML = ascii_clean(table_html(current_trends()))
        sleep(60)

def run_loader_thread():
    t = Thread(target=loader_thread)
    t.daemon = 1
    t.start()

def trends_html():
    while not _TRENDS_HTML:
        sleep(0.1)

    return _TRENDS_HTML

def index_html():
    return open(join(dirname(__file__), "index.html")).read()

class RequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/":
            self.wfile.write(index_html())
        elif self.path.startswith("/trends"):
            self.wfile.write(trends_html())
        elif splitext(self.path)[1] in (".js", ".css"):
            self.wfile.write(open(".%s" % self.path).read())
        else:
            self.send_error(404, "Not Found")

if __name__ == "__main__":
    run_loader_thread()
    server = HTTPServer(("", 8888), RequestHandler)
    server.serve_forever()

The Auto-Refreshing Web Client (index.html)

<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html>
    <head>
        <title>Trends</title>
        <link rel="stylesheet" href="trends.css" />
        <style>
            body {
                font-family: Serif, Helvetica;
            }

            table {
                width: 100%;
                border: 2px solid gray;
            }

            td {
                vertical-align: top;
                border-bottom: 1px solid black;
            }

            h1 {
                text-align: center;
                font-variant: small-caps;
            }

            a {
                text-decoration: none;
                color: black;
            }

            a:hover {
                background: silver;
            }

            ul {
                margin: 0px;
            }
        </style>
    </head>
    <body>
        <h1>Trends</h1>
        <div id="trends">
            Loading ...
        </div>
    </body>
    <script src="jquery.js"></script>
    <script>
        function load() {
            $('#trends').load('/trends');
        }
        $(document).ready(function() {
            load();
            setInterval(load, 30 * 1000);
        });
    </script>
</html>

Show "digits only" version of phone number

#!/bin/bash
# Convert phone numbers to numeric (1-800-T-MOBILE -> 1-800-8-662453)

if [ $# -ne 1 ]; then
    echo "usage: $(basename $0) PHONE-NUMBER"
    exit 1
fi

echo $1 | \
    tr '[:lower:]' '[:upper:]' | \
    tr ABCDEFGHIJKLMNOPQRSTUVWXYZ 22233344455566677778889999

JSON Pretiffier

#!/usr/bin/env python
'''JSON prettifier'''

def shift(array):
   try:
       return array.pop(0)
   except IndexError:
       return None

def fopen(name, mode, default):
   if name in (None, "-"):
       return default
   return open(name, mode)

def main(argv=None):
   import sys
   from optparse import OptionParser

   argv = argv or sys.argv

   indent = 4
   parser = OptionParser("%prog [INFILE [OUTFILE]]")
   parser.add_option("-i", "--indent", help="indent size (%s)" % indent,
                     dest="indent", default=indent, type="int")
   opts, args = parser.parse_args(argv[1:])
   if len(args) not in (0, 1, 2):
       parser.error("wrong number of arguments") # Will exit

   try:
       info = fopen(shift(args), "r", sys.stdin)
       outfo = fopen(shift(args), "w", sys.stdout)
   except IOError, e:
       raise SystemExit("error: %s" % e)

   import json
   json.dump(json.load(info), outfo, indent=opts.indent)
      
if __name__ == "__main__":
   main()

Update (Aug 14, 2009): Just found out about python -mjson.tool, oh well ...

strftime for JavaScript

/* strftime for JavaScript 

   Field description (taken from http://tinyurl.com/65s2qw)

    %a  Locale’s abbreviated weekday name.   
    %A  Locale’s full weekday name.   
    %b  Locale’s abbreviated month name.   
    %B  Locale’s full month name.   
    %c  Locale’s appropriate date and time representation.   
    %d  Day of the month as a decimal number [01,31].   
    %H  Hour (24-hour clock) as a decimal number [00,23].   
    %I  Hour (12-hour clock) as a decimal number [01,12].   
    %j  Day of the year as a decimal number [001,366].   
    %m  Month as a decimal number [01,12].   
    %M  Minute as a decimal number [00,59].   
    %p  Locale’s equivalent of either AM or PM.
    %S  Second as a decimal number [00,61].
    %U  Week number of the year (Sunday as the first day of the week) as a
        decimal number [00,53]. All days in a new year preceding the first
        Sunday are considered to be in week 0.
    %w  Weekday as a decimal number [0(Sunday),6].   
    %W  Week number of the year (Monday as the first day of the week) as a
        decimal number [00,53]. All days in a new year preceding the first
        Monday are considered to be in week 0.
    %x  Locale’s appropriate date representation.   
    %X  Locale’s appropriate time representation.   
    %y  Year without century as a decimal number [00,99].   
    %Y  Year with century as a decimal number.   
    %Z  Time zone name (no characters if no time zone exists).   
    %%  A literal '%' character.
*/

var days = [ 
    'Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday',
    'Saturday' 
];

var months = [
    'January', 'February', 'March', 'April', 'May', 'June', 'July',
    'August', 'September', 'October', 'November', 'December'
];

function shortname(name) {
    return name.substr(0, 3);
}

function zeropad(n, size) {
    n = '' + n; /* Make sure it's a string */
    size = size || 2;
    while (n.length < size) {
        n = '0' + n;
    }
    return n;
}

function twelve(n) {
    return (n <= 12) ? n : 24 - n;
}

function strftime(format, date) {
    date = date || new Date();
    var fields = {
        a: shortname(days[date.getDay()]),
        A: days[date.getDay()],
        b: shortname(months[date.getMonth()]),
        B: months[date.getMonth()],
        c: date.toString(),
        d: zeropad(date.getDate()),
        H: zeropad(date.getHours()),
        I: zeropad(twelve(date.getHours())),
        /* FIXME: j: */
        m: zeropad(date.getMonth() + 1),
        M: zeropad(date.getMinutes()),
        p: (date.getHours() >= 12) ? 'PM' : 'AM',
        S: zeropad(date.getSeconds()),
        w: zeropad(date.getDay() + 1),
        /* FIXME: W: */
        x: date.toLocaleDateString(),
        X: date.toLocaleTimeString(),
        y: ('' + date.getFullYear()).substr(2, 4),
        Y: '' + date.getFullYear(),
        /* FIXME: Z: */
        '%' : '%'
    };

    var result = '', i = 0;
    while (i < format.length) {
        if (format[i] === '%') {
            result = result + fields[format[i + 1]];
            ++i;
        }
        else {
            result = result + format[i];
        }
        ++i;
    }
    return result;
}

Quick search using xapian "omega"

A little hack to search current directory using Xapian's omega.

#!/bin/bash

# Minimal search using Xpians Omega (http://xapian.org/docs/omega/overview.html)
# Use "omsearch -i" to index current directoy, then use "omsearch QUERY" to
# search


index() {
    mkdir .omega
    cat >> .omega/omega.conf << EOF
    database_dir $PWD/.omega
    template_dir /var/lib/xapian-omega/templates
    log_dir /tmp
EOF
    omindex --db .omega/default .
}

search() {
    omega=/usr/lib/cgi-bin/omega/omega
    config=$PWD/.omega/omega.conf
    results=.omega/results.html
    OMEGA_CONFIG_FILE=$config $omega "P=$1" "HITSPERPAGE=1000" > $results
    egrep -o '<B><A HREF=".*"' $results | cut -d\" -f2 | \
        xargs -i printf ".%s\n" {}

}

USAGE="usage: $(basename $0) [-i | QUERY]"

index=no
while getopts "ih" opt
do
    case $opt in
        h | H ) echo $USAGE; exit;;
        i ) index=yes;;
        * ) echo "error: unknown option" >&2; exit 1;;
    esac
done

shift $(($OPTIND - 1))

if [ "$index" == "yes" ]; then
    if [ $# -gt 0 ]; then
        echo "error: -i don't take any parameters"
        exit 1
    fi
    index
    exit
fi

if [ $# -ne 1 ]; then
    echo $USAGE
    exit 1
fi

if [ ! -d .omega ]; then
    echo "error: can't find .omega, run with -i to index first"
    exit 1
fi
search $1

Avoiding "peek" with "itertools.chain"

Sometimes you need to filter out some header information. However the only clear anchor you have is the start of real data - you need a "peek" function.

It's easy to avoid writing the "peek" logic, just use itertools.chain.

#!/usr/bin/env python
'''Avoiding the need for "peek" with itertools.chain'''

from itertools import chain

def is_first_data(line):
    return line.startswith("Name:")

def skip_header(data):
    data = iter(data)
    for line in data:
        if is_first_data(line):
            return chain([line], data)

    # FIXME: We might want to raise something here
    return []


if __name__ == "__main__":
    data = [
        "this is the header",
        "it might change every time",
        "and you'll never find a good regexp for it",
        "The first line of data is easy to know",
        "Name: Duffy",
        "Type: Duck",
        "Anger: 10",
        "",
        "Name: Bugs",
        "Type: Bunny",
        "Anger: 0",
    ]

    data = skip_header(data)
    for line in data:
        print line

Subversion IRC Bot

#!/usr/bin/env python
'''Push changes in subversion repository to IRC channel'''

from twisted.words.protocols import irc
from twisted.internet.protocol import ReconnectingClientFactory
import re
from subprocess import Popen, PIPE
from xml.etree.cElementTree import parse as xmlparse
from cStringIO import StringIO

class IRCClient(irc.IRCClient):
    nickname = "svnbot"
    realname = "Subversion Bot"
    channel = "#dev"

    instance = None # Running instance

    def signedOn(self):
        IRCClient.instance = self
        self.join(self.channel)

    def svn(self, revision, author, comment):
        comment = (comment[:57] + "...") if  len(comment) > 60 else comment
        message = "SVN revision %s by %s: %s" % (revision, author, comment)
        self.say(self.channel, message)


class SVNPoller:
    def __init__(self, root, user, password):
        self.pre = ["svn", "--xml", "--username", user, "--password", password]
        self.root = root
        self.last_revision = self.get_last_revision()

    def check(self):
        if not IRCClient.instance:
            return

        try:
            last_revision = self.get_last_revision()
            if (not last_revision) or (last_revision == self.last_revision):
                return

            for rev in range(self.last_revision + 1, last_revision + 1):
                author, comment = self.revision_info(rev)
                IRCClient.instance.svn(rev, author, comment)
            self.last_revision = last_revision
        except Exception, e:
            print "ERROR: %s" % e

    def svn(self, *cmd):
        pipe = Popen(self.pre +  list(cmd) + [self.root], stdout=PIPE)
        try:
            data = pipe.communicate()[0]
        except IOError:
            data = ""
        return xmlparse(StringIO(data))

    def get_last_revision(self):
        tree = self.svn("info")
        revision = tree.find("//commit").get("revision")
        return int(revision)

    def revision_info(self, revision):
        tree = self.svn("log", "-r", str(revision))
        author = tree.find("//author").text
        comment = tree.find("//msg").text

        return author, comment

if __name__ == "__main__":
    from twisted.internet import reactor
    from twisted.internet.task import LoopingCall

    factory = ReconnectingClientFactory()
    factory.protocol = IRCClient
    reactor.connectTCP("irc.mycompany.com", 6667, factory)

    poller = SVNPoller("http://svn.mycompany.com", "bugs", "carrot")
    task = LoopingCall(poller.check)
    task.start(1)

    reactor.run()

Solving Euler Question 24

I'm learning Clojure by solving project Euler. Here is a python version of one of the solutions.

#!/usr/bin/env python
''' Solving http://projecteuler.net/index.php?section=problems&id=24

Note that Python's 2.6 itertools.permutations return the permutation in order so
we can just write:
    from itertools import islice, permutations
    print islice(permutations(range(10)), 999999, None).next()

And it'll work much faster :)
'''

from itertools import islice, ifilter

def is_last_permutation(n):
    return n == sorted(n, reverse=1)

def next_head(n):
    '''Find next number to be 'head'. 

    It is smallest number if the tail that is bigger than the head.
    In the case of (2 4 3 1) it will pick 3 to get the next permutation
    of (3 1 2 4)
    '''
    return sorted(filter(lambda i: i > n[0], n[1:]))[0]

def remove(element, items):
    return filter(lambda i: i != element, items)

def next_permutation(n):
    if is_last_permutation(n):
        return None

    sub = next_permutation(n[1:])
    if sub:
        return [n[0]] + sub

    head = next_head(n)
    return [head] + sorted([n[0]] + remove(head, n[1:]))

def nth(it, n):
    '''Return the n'th element of an iterator'''
    return islice(it, n, None).next()

def iterate(func, n):
    '''iterate(func, n) -> n, func(n), func(func(n)) ...'''
    while 1:
        yield n
        n = func(n)

def permutations(n):
    return ifilter(None, iterate(next_permutation, n))

if __name__ == "__main__":
    n = range(10)
    m = 1000000
    print "Calculateing the %d permutation of %s" % (m, n)
    print nth(permutations(n), m-1)