bulls-and-cows-profiler.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import Queue
import argparse
import os
import random
import subprocess
import sys
import threading
import time

class Tester(threading.Thread):
    def __init__(self, testee, answer_queue, counter_queue):
        threading.Thread.__init__(self)
        self.testee = testee
        self.answer_queue = answer_queue
        self.counter_queue = counter_queue
    def run(self):
        while True:
            ans = self.answer_queue.get()
            counter = 1
            p = subprocess.Popen(self.testee,
                    stdin=subprocess.PIPE, stdout=subprocess.PIPE)
            while True:
                guess = p.stdout.readline().strip()
                bulls = sum(1 for i, j in zip(ans, guess) if i == j)
                cows = sum(1 for i, j in zip(ans, guess)if i != j and j in ans)
                score = str(bulls) + 'a' + str(cows) + 'b' + '\n'
                p.stdin.write(score)
                if bulls == 4:
                    self.answer_queue.task_done()
                    self.counter_queue.put(counter)
                    counter = 1
                    break
                counter += 1
            if self.answer_queue.empty():
                break

def get_args():
    args_parser = argparse.ArgumentParser(
            formatter_class=argparse.RawDescriptionHelpFormatter,
            description='Profile a bulls-and-cows guesser.',
            epilog='EXAMPLE:\n' \
                    "  {} -j 8 './bulls-and-cows-guesser.py'".format(os.path.basename(__file__))
                    )
    args_parser.add_argument('testee' , metavar='TESTEE',
            help='the guesser you want to evaluate')
    args_parser.add_argument('-j', '--jobs', metavar='N', type=int, default=1,
            help='allow N jobs at once')
    args_parser.add_argument('--no-shuffle', action='store_true',
            help='do not shuffle the test casees')
    return args_parser.parse_args()

def get_all_answers(shuffle):
    answer_candidates = [str(i) + str(j) + str(k) + str(l) \
                for i in xrange(0, 10) for j in xrange(0, 10) \
                for k in xrange(0, 10) for l in xrange(0, 10) \
                if len(set(str(i) + str(j) + str(k) + str(l))) == 4]

    if shuffle:
        random.shuffle(answer_candidates)

    return answer_candidates

def main():
    args = get_args()
    answer_candidates = get_all_answers(not args.no_shuffle)

    answer_queue = Queue.Queue()
    counter_queue = Queue.Queue()
    for ans in answer_candidates:
        answer_queue.put(ans)
    for i in xrange(args.jobs):
        t = Tester(args.testee, answer_queue, counter_queue)
        t.daemon = True
        t.start()

    answer_queue.join()
    while True:
        if counter_queue.qsize() == len(answer_candidates):
            break

    counter = []
    for i in xrange(len(answer_candidates)):
        counter.append(counter_queue.get())

    print 'avg: ', float(sum(counter)) / len(counter)
    d = {}
    for v in counter:
        if v not in d:
            d[v] = 1
            continue
        d[v] += 1
    print str(d)[1:len(str(d))-1]

if __name__ == '__main__':
    main()