Raspberry Piにミクさんボイスでタイムラインをしゃべらせた(OpenJTalk)

Tweet

Pocket

Raspberry Piでやる意味があまりないのだけれど。

前提

以下を実行できるようにしておく。

Raspberry Pi2でミクさんに喋らせてみた(OpenJTalk)
Raspberry Pi2でミクさんに喋らせてみた(Python + OpenJTalk)

また、Twitter API関連の情報を入手しておく。

Twitter APIの利用方法

以下、ソースコード

TwitterTalkクラスの、streamメソッドを使うとtwitter上からワードを検索してきてしゃべる。
userstreamメソッドを使うと、自分のタイムライン上が更新されるたびにしゃべる。

#coding: utf-8
import os
import subprocess
import random
import time
import re
from datetime import datetime

from tweepy.streaming import StreamListener, Stream
from tweepy.auth import OAuthHandler
from tweepy.api import API


class Talk(object):
    """
    OpenJTalkを使ってテキストからwavファイル作成と音声出力をする。
    """

    def create_wav(self, text):
        """
        テキストからwavファイルを作成する。
        """

        file_name = self.create_filename()
        file_path = "/tmp/{}.wav".format(file_name)

        open_jtalk = ["open_jtalk"]
        td = ["-td", "/usr/share/hts-voice/miku/tree-dur.inf"]
        tf = ["-tf", "/usr/share/hts-voice/miku/tree-lf0.inf"]
        tm = ["-tm", "/usr/share/hts-voice/miku/tree-mgc.inf"]
        md = ["-md", "/usr/share/hts-voice/miku/dur.pdf"]
        mf = ["-mf", "/usr/share/hts-voice/miku/lf0.pdf"]
        mm = ["-mm", "/usr/share/hts-voice/miku/mgc.pdf"]
        dm = ["-dm", "/usr/share/hts-voice/miku/mgc.win1"]
        dm += ["-dm", "/usr/share/hts-voice/miku/mgc.win2"]
        dm += ["-dm", "/usr/share/hts-voice/miku/mgc.win3"]
        df = ["-df", "/usr/share/hts-voice/miku/lf0.win1"]
        df += ["-df", "/usr/share/hts-voice/miku/lf0.win2"]
        df += ["-df", "/usr/share/hts-voice/miku/lf0.win3"]
        dl = ["-dl", "/usr/share/hts-voice/miku/lpf.win1"]
        ef = ["-ef", "/usr/share/hts-voice/miku/tree-gv-lf0.inf"]
        em = ["-em", "/usr/share/hts-voice/miku/tree-gv-mgc.inf"]
        cf = ["-cf", "/usr/share/hts-voice/miku/gv-lf0.pdf"]
        cm = ["-cm", "/usr/share/hts-voice/miku/gv-mgc.pdf"]
        k = ["-k", "/usr/share/hts-voice/miku/gv-switch.inf"]
        s = ["-s", "16000"]
        a = ["-a", "0.05"]
        u = ["-u", "0.0"]
        jm = ["-jm", "1.0"]
        jf = ["-jf", "1.0"]
        jl = ["-jl", "1.0"]
        x = ["-x", "/var/lib/mecab/dic/open-jtalk/naist-jdic"]
        ow = ["-ow", file_path]

        cmd = open_jtalk + td + tf + tm + md + mf + mm + dm + df
        cmd += dl + ef + em + cf + cm + k + s + a + u + jm
        cmd += jf + jl + x + ow

        c = subprocess.Popen(cmd, stdin=subprocess.PIPE)
        c.stdin.write(text.encode("utf-8"))
        c.stdin.close()
        c.wait()

        return file_path

    def play_wav(self, file_path):
        """
        wavファイルを再生する。
        """

        aplay = ['aplay']
        q = ["-q", file_path]

        cmd = aplay + q
        wr = subprocess.Popen(cmd)


    def create_filename(self, l=8):
        """
        wavファイルのファイル名を作成する。
        """

        source = "abcdefghijklmnopqrstuvwxyz"
        source += "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
        source += "0123456789"

        return "".join(random.choice(source) for _ in range(l))


class TwitterTalk():

    def __init__(self, consumer_key, consumer_secret, access_token, access_token_secret):

        self.auth = OAuthHandler(consumer_key, consumer_secret)
        self.auth.set_access_token(access_token, access_token_secret)


    def stream(self, word):
        """
        Twitter Streamingで特定のワードを取得する。
        """

        stream = Stream(self.auth, ExtendStreamListener(), secure=True)
        stream.filter(track=[word])

    def userstream(self):
        """
        自分のタイムラインのツイートを取得する。
        """

        stream = Stream(self.auth, ExtendStreamListener(), secure=True)
        stream.userstream()


class ExtendStreamListener(StreamListener, Talk):
    """
    Twitter Streaming用クラス。
    """

    def reject_url_and_hashtag(self, t):
        """
        テキストからURL、ハッシュタグを除く。
        """

        text = re.sub('https?://[\w/:%#\$&\?\(\)~\.=\+\-]+', '', t)
        text = re.sub('[#＃]([\w一-龠ぁ-んァ-ヴーａ-ｚ]+)', '', text)
        text = re.sub('
', '', text)
        return text

    def on_status(self, status):
        """
        Twitter Streamingで取得したツイートの処理。
        """

        text = self.reject_url_and_hashtag(status.text)

        username = status.user.name
        print(username, ":", text)

        text = username + text

        file_path = self.create_wav(text)
        self.play_wav(file_path)

        # 一定時間経ったら音声ファイルを削除
        time.sleep(20)
        os.remove(file_path)

if __name__ == '__main__':

    tw_talk = TwitterTalk("CUSTOMER KEY",
                        "CUSTOMER SECRET",
                        "ACCESS TOKEN",
                        "ACCESS TOKEN SECRET")
    tw_talk.userstream()