Ask Babby

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# "How is babby formed?" --kavya (http://www.youtube.com/watch?v=w_RaPOOVX1Y)
#
# ask_babby.py - Speak a question in natural language, get back an answer spoken
# in natural language. Uses Google's speech recognition, Google's search, and
# Yahoo Answers. This will only work on a Mac.
#
# Other people's code strung together by: Tony ([email protected])
import sys
import subprocess
from subprocess import PIPE
import StringIO
import pyaudio
import wave
import random
import time
import threading
import audio
import urllib
import urllib2
import simplejson
import re
import HTMLParser
from nltk import sent_tokenize

# If you send FLAC audio data to this URL, you get a transcription of the
# audio as a response. This probably works better than any downloadable speech
# to text software out there for short utterances like questions.
GOOGLE_SR_URL = 'http://www.google.com/speech-api/v1/recognize?lang=en-us&client=chromium'
SAMPLE_RATE = 16000

# The official and unofficial Google search API url.
GOOGLE_API_URL = 'http://ajax.googleapis.com/ajax/services/search/web?v=1.0&%s'
GOOGLE_SEARCH_URL = 'http://www.google.com/search?%s'

# Some stuff used to deal with HTML.
HTML_PARSER = HTMLParser.HTMLParser()
ANSWER_PAT = re.compile(r'<div class="content">(.*?)</div>')
URL_PAT = re.compile(r'http://[^ ]*')
TAG_PAT = re.compile(r'<[^>]*?>')
YA_URL_PAT = re.compile(r'href="(http://answers.yahoo.com/question/index\?qid=[^"]*)"')

# Pass the time by muttering these phrases
THINKING_PHRASES = ["Wait a second", "I'm thinking", "I think I know this...",
    'Let me see...']

# Records for the given number of seconds and returns the wav recording
# I have no idea what these default settings actually mean. Check out
# the pyaudio documentation for more information. This code is lifted
# from one of their demo scripts.
def sample_wav(secs, format=pyaudio.paInt16, channels=1, rate=SAMPLE_RATE,
    frames_per_buffer=1024):

    p = pyaudio.PyAudio()
    stream = p.open(format=format, channels=channels, rate=rate,
                    input=True, frames_per_buffer=frames_per_buffer)

    all = []
    for i in range(0, rate / frames_per_buffer * secs):
        data = stream.read(frames_per_buffer)
        all.append(data)

    stream.close()
    p.terminate()

    data = ''.join(all)
    obj = StringIO.StringIO()
    wf = wave.open(obj, 'wb')
    wf.setnchannels(channels)
    wf.setsampwidth(p.get_sample_size(format))
    wf.setframerate(rate)
    wf.writeframes(data)
    wf.close()
    return obj.getvalue()

# Takes some wav data as input, and converts it to FLAC format. I could not
# figure out how to do this in Python, so it just class the flac command line
# utility by opening a subprocess.
def wav_to_flac(wav_data, rate=SAMPLE_RATE):
    cmd = ['flac', '--silent', '--sample-rate=%s' % rate, '-']
    proc = subprocess.Popen(cmd, stdin=PIPE, stdout=PIPE, stderr=PIPE)
    flac_data = proc.communicate(wav_data)[0]
    return flac_data

# Takes some flac data as input and sends it to Google's speech recognition
# API. I'm not sure if this is actually allowed, so this part of the code
# could break at any time. In fact, it's probably not allowed at all since I
# have to lie about the user agent.
def flac_to_text(flac_data, rate=SAMPLE_RATE):
    req = urllib2.Request(GOOGLE_SR_URL, data=flac_data)
    req.add_header('User-Agent', 'Mozilla/5.0')
    content_type = 'audio/x-flac; rate=%s' % rate
    req.add_header('Content-Type', content_type)
    res = urllib2.urlopen(req)
    result = simplejson.loads(res.read())
    # Google returns a JSON object that has a list of hypotheses about what
    # the transcription should be. I just take the first one. This code will
    # break if something goes wrong in opening the URL etc.
    return result['hypotheses'][0]['utterance']

# Takes some text as input and does text -> speech using the Mac OS X "say"
# command. This won't work on non-Mac systems.
def say(s, voice='Alex', rate=250):
    cmd = ['say', '-v', voice, '[[rate %s]] %s' % (rate,s)]
    subprocess.call(cmd)

# Cleans up some text.
def clean(text):
    text = unicode(text,errors='ignore')
    text = HTML_PARSER.unescape(TAG_PAT.sub('', text))
    text = URL_PAT.sub(' ', text)
    return text.replace('\n', ' ').replace('\r', ' ')

# Given a Yahoo Answers page, looks for the "Best Answer" and returns the
# text. If there isn't one there then... this will break.
def get_best_answer(data):
    start = data.index('Best Answer')
    got = ANSWER_PAT.findall(data[start:].replace('\n', ' '))[0]
    return clean(got)

# Searches Yahoo Answers using the official Google search API. For some reason,
# the results aren't as good as just scraping the results off of the regular
# Google search engine result page. Returns the text of the Best Answer on the
# answers page, or fails otherwise. Or breaks. Or something bad happens, I
# really don't know.
def get_yahoo_answer_official(q):
    query = urllib.urlencode({'q' : 'site:answers.yahoo.com ' + q})
    search_results = urllib.urlopen(GOOGLE_API_URL % query)
    json = simplejson.loads(search_results.read())
    results = json['responseData']['results']
    url = urllib.unquote(results[0]['url'])
    data = urllib.urlopen(url).read()
    return get_best_answer(data)

# Searches Yahoo Answers using the "unofficial" Google search API, also known
# as setting the user agent to be a desktop browser, and then scraping the
# results page.
def get_yahoo_answer_unofficial(q):
    query = urllib.urlencode({'q': 'site:answers.yahoo.com ' + q})
    req = urllib2.Request(url=GOOGLE_SEARCH_URL % query)
    req.add_header('User-Agent', 'Mozilla/5.0 (Macintosh; U; Intel Mac OS X 10.5; en-US; rv:1.9.0.6) Gecko/2009011912 Firefox/3.0.6')
    req.add_header('Accept', 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8')
    req.add_header('Accept-Language', 'en-us,en;q=0.5')
    out = urllib2.urlopen(req)
    data = out.read()
    data = data.replace("\n", ' ')
    url = YA_URL_PAT.findall(data)[0]
    data = urllib.urlopen(url).read()
    return get_best_answer(data)

# A class for fetching the best Yahoo answer in the background, given an
# wav audio recording of a question.
class AnswerFetcher(threading.Thread):
    def __init__(self, wav_data):
        threading.Thread.__init__(self)
        self.wav_data = wav_data
        self.answer = None
    def run(self):
        try:
            flac_data = wav_to_flac(self.wav_data)
            question = flac_to_text(flac_data)
            self.answer = get_yahoo_answer_unofficial(question)
        # Whatever, just catch the error.
        except Exception, e:
            print >>sys.stderr, e
            self.answer = "Sorry. I'm too stupid to answer."

# Prints text out one character at a time to make it look oh so cool. The
# font is hard coded to be white on pink, because that is the ultimate font
# choice and should never be changed.
class SlowPrinter(threading.Thread):
    def __init__(self, text, pause=0.05):
        threading.Thread.__init__(self)
        self.text = text
        self.pause = pause
        self.kill = False
    def run(self):
        # This means print white text on a pink background, bold
        sys.stdout.write('\x1b[1m\x1b[45m\x1b[37m')
        for c in self.text:
            if self.kill: break
            sys.stdout.write(c)
            sys.stdout.flush()
            time.sleep(self.pause)
        sys.stdout.write('\x1b[0m')
        sys.stdout.write('\n')
        sys.stdout.flush()

# Breaks a text into sentences, and "says" the sentences one by one.
def say_sent_by_sent(text, voice='Alex', rate=150):
    sents = sent_tokenize(text)
    for sent in sents:
        try:
            sp = SlowPrinter(sent)
            sp.start()
            say(sent, voice=voice, rate=rate)
            sp.join()
        except KeyboardInterrupt, e:
            sp.kill = True
            break

# The main routine for Babby. Records a question and says the answer.
def interface(secs):
    wav_data = sample_wav(secs)
    fetcher = AnswerFetcher(wav_data)
    fetcher.start()
    while fetcher.answer is None:
        say(random.choice(THINKING_PHRASES), rate=130)
        time.sleep(1)
    fetcher.join()
    answer = fetcher.answer
    say_sent_by_sent(answer, rate=180)

# Keeps asking for questions and answering them. Press any key and then
# enter to record a question. Type "exit" to exit, or just Control-C.
if __name__ == '__main__':
    while True:
        x = raw_input('ASK BABBY>')
        if x == 'exit': break
        dots = SlowPrinter('.....', pause=1)
        dots.start()
        interface(5)
        dots.join()