AquesTalkをPythonから（音声合成）
AquesTalkは発声用のコーパスを持たない非常に軽量な音声合成ライブラリです．営利，非営利かわらず無償で使えるのでいろいろなところで使ってみたいですね．AquesTalkの2つのDLLと以下のスクリプトを一緒に放り込むだけですぐに使える手軽なところも魅力的です．
AquesTalk - テキスト音声合成ミドルウェア
AquesTalkは、組み込み用途向けのテキスト情報を音声波形に変換出力するライブラリです。システムに負担をかけずに簡単に組み込めることを目指して開発しました。 AquesTalkを使えば、様々なアプリケーションやサービスに、任意の音声メッセージを動的に生成できます。
・超軽量
他に類を見ない少ないリソースと処理量で高品質な音声を生成します。
・フリー（Win版のみ）
営利、非営利にかかわらず無償で使用でき、製品に組み込んで販売することも可能です
・高い移植性
ANSI準拠のCプログラムで記述されているので各種プラットフォームに容易に移植できます
PythonからWin32APIを使ってWindowを作成したい人にも参考にもなるかも．（移植性を考えるとwxWidgetsとかTkを使った方が良いかもしれないが配布時に重くなりそうなのでパス）
- AquesTalk.py
# -*- coding:sjis -*- import os, sys import ctypes from array import array from ctypes import windll from MessageEvents import MessageEvents class AquesWrapper(object): """ AquesTalk音声合成ライブラリの薄いラッパ Python2.4以下の場合はctypesライブラリが必要(2.5は標準ライブラリ) Download：http://python.net/crew/theller/ctypes/ """ def __init__(self): self.talk_dll = windll.LoadLibrary("AquesTalkDa.dll") self.synt_dll = windll.LoadLibrary("AquesTalk.dll") # 同期発声 self.aques_playsync = self.talk_dll.AquesTalkDa_PlaySync self.aques_playsync.argtypes = [ctypes.c_char_p, ctypes.c_int] self.aques_playsync.restype = ctypes.c_int # 非同期発声 self.aques_play = self.talk_dll.AquesTalkDa_Play self.aques_play.argtypes = [ctypes.c_uint, ctypes.c_char_p, ctypes.c_int, ctypes.c_uint, ctypes.c_ulong, ctypes.c_ulong] self.aques_play.restype = ctypes.c_int # 音声合成エンジン生成 self.aques_create = self.talk_dll.AquesTalkDa_Create self.aques_create.argtypes = [] self.aques_create.restype = ctypes.c_uint # 音声合成エンジン解放 self.aques_release = self.talk_dll.AquesTalkDa_Release self.aques_release.argtypes = [ctypes.c_uint] self.aques_release.restype = None # 発声ストップ self.aques_stop = self.talk_dll.AquesTalkDa_Stop self.aques_stop.argtypes = [ctypes.c_uint] self.aques_stop.restype = None # 発声チェック self.aques_isplay = self.talk_dll.AquesTalkDa_IsPlay self.aques_isplay.argtypes = [ctypes.c_uint] self.aques_isplay.restype = ctypes.c_int # 音声データ生成 self.aques_synthe = self.synt_dll.AquesTalk_Synthe self.aques_synthe.argtypes = [ctypes.c_char_p, ctypes.c_int, ctypes.c_void_p] self.aques_synthe.restype = ctypes.c_void_p # 音声データ解放 self.aques_free = self.synt_dll.AquesTalk_FreeWave self.aques_free.argtypes = [ctypes.c_void_p] self.aques_free.restype = None class AquesTalk(AquesWrapper): """ 非同期再生のコールバックを使いやすくしたラッパ Python for Windows extensionsが必要 Webサイト：http://starship.python.net/crew/mhammond/win32/Downloads.html Download ：https://sourceforge.net/projects/pywin32/ """ # 非同期再生を同時に利用する際に識別できる最大数 MAX_SYN_PLAY = 32 # 次に非同期再生する際に付与する発声ID（0-MAX_SYNPLAY） count = 0 # 現在再生中の発声ID play_list = [] def __init__(self): AquesWrapper.__init__(self) # 音声合成エンジンの生成 self.handle = self.Create() # このインスタンスで再生中の発声ID self.playing = None def PlaySync(self, text, speed): """ 発声が終了するまで戻らない同期タイプの音声合成 text: 発声記号列 speed: 50-300で指定（標準は100） """ return self.aques_playsync(text, speed) def Create(self): """ 非同期再生用の音声合成エンジンのインスタンスを生成する """ return self.aques_create() def Release(self): """ 音声合成エンジンのインスタンスを解放する """ return self.aques_release(self.handle) def Stop(self): """ このインスタンスの非同期発声を停止する """ return self.aques_stop(self.handle) def IsPlayNumber(self, number): """ 指定した番号の音声をどれかのインスタンスで再生中かどうか number: PlayASyncで再生したときの返り値を指定 """ if number in AquesTalk.play_list: return True return False def IsPlay(self): """ このインスタンスが持つ音声合成エンジンが再生中かどうか """ return self.aques_isplay(self.handle) def PlayASync(self, text, speed, callback): """ 発生が終了する前に戻る非同期タイプの音声合成 複数同時に発声させたい場合は複数のインスタンスを用いる callback: 発声終了時に呼ばれるコールバック関数 （引数はhwnd, msg, wp, lp） """ ret_code = AquesTalk.count AquesTalk.play_list.append(AquesTalk.count) self.playing = AquesTalk.count AquesTalk.count += 1 if AquesTalk.count >= AquesTalk.MAX_SYN_PLAY: AquesTalk.count = 0 def finish_wrapper(hwnd, msg, wparam, lparam): AquesTalk.play_list.remove(msg) self.playing = None return callback(hwnd, msg, wparam, lparam) msg = MessageEvents.set_callback(ret_code, finish_wrapper) hwnd = MessageEvents.get_hwnd() self.aques_play(self.handle, text, speed, hwnd, msg, 0) return ret_code def __synthe(self, text, speed): """ 音声記号列から音声波形を生成する """ err_code = ctypes.c_int() ret = self.aques_synthe(text, speed, ctypes.byref(err_code)) if ret == None: raise Exception, err_code.value return ret, err_code def SyntheData(self, text, speed): """ 音声データをarray（リスト形式）で返す """ data, size = self.__synthe(text, speed) byte_ary = array('B') for i in range(size.value): byte = ctypes.c_ubyte.from_address(data+i).value byte_ary.append(byte) self.aques_free(data) return byte_ary def SyntheFile(self, text, speed, filename): """ 音声データをファイルに書き込む """ byte_ary = self.SyntheData(text, speed) fp = open(filename, "wb") byte_ary.tofile(fp) fp.close() if __name__ == '__main__': def talk_test(): import time at = AquesTalk() at2 = AquesTalk() def finish(*args): print args at.PlayASync("こんにちわ", 100, finish) time.sleep(0.1) at2.PlayASync("おはようございます", 100, finish) time.sleep(0.5) at2.Stop() # 「こんにちわ」と「おはよう」を同時に発声 # 「おはようございます」を途中でストップ def synthe_test(): at = AquesTalk() at.SyntheFile("こんばんは", 100, "test.wav") at.Release() #synthe_test() talk_test()
AquesTalkのplay関数は非同期に発声を行ってその終了時に指定したウィンドウにメッセージを投げることでコールバックを実現していますが，できれば直接コールバック関数を指定して使いたいのでAquesTalk専用の見えないWindowを一つ作っています．ほかの言語から使う場合も同様の仕組みが必要ですが，.netについては以下のサイトに解説とソースがあります．
- MessageEvents.py
# -*- coding: sjis -*- import os, sys import time import threading import win32api import win32con import win32gui class Window(object): def __init__(self, lock): className = "win32gui_window_class" windowTitle = "win32gui Sample Window" style = win32con.WS_OVERLAPPEDWINDOW hIcon = win32gui.LoadIcon(0, win32con.IDI_APPLICATION) self.lock = lock # Initialize instance win32gui.InitCommonControls() self.hinst = 0 # win32api.GetModuleHandle(None) # Register class wc = win32gui.WNDCLASS() wc.style = win32con.CS_HREDRAW | win32con.CS_VREDRAW wc.lpfnWndProc = self.WndProc wc.hCursor = win32gui.LoadCursor( 0, win32con.IDC_ARROW ) wc.hbrBackground = win32con.COLOR_WINDOW + 1 wc.hIcon = hIcon wc.lpszClassName = className wc.cbWndExtra = 0 classAtom = win32gui.RegisterClass(wc) self.message_map = {} # Create window style = win32con.WS_OVERLAPPEDWINDOW self.hwnd = win32gui.CreateWindow(className, windowTitle, style, win32con.CW_USEDEFAULT, win32con.CW_USEDEFAULT, win32con.CW_USEDEFAULT, win32con.CW_USEDEFAULT, 0, 0, self.hinst, None) # Hook WndProc self.oldWndProc = win32gui.SetWindowLong(self.hwnd, win32con.GWL_WNDPROC, self.WndProc) def ShowWindow(self): win32gui.UpdateWindow(self.hwnd) win32gui.ShowWindow(self.hwnd, win32con.SW_SHOW) def WndProc(self, hwnd, message, wparam, lparam): self.lock.acquire() try: if message == win32con.WM_CLOSE: return win32gui.PostQuitMessage(0) elif message in self.message_map: return self.message_map[message]( hwnd, message - win32con.WM_USER,wparam, lparam) finally: self.lock.release() return win32gui.DefWindowProc(hwnd, message, wparam, lparam) def PumpMessages(self): return win32gui.PumpMessages() def PumpWaitingMessages(self): return win32gui.PumpWaitingMessages() class WindowThread(threading.Thread): def __init__(self, lock): threading.Thread.__init__(self) self.hwnd = 0 self.lock = lock def run(self): self.win = Window(self.lock) self.hwnd = self.win.hwnd self.win.PumpMessages() class MessageEvents(object): # メッセージを処理するためのウィンドウを作成 lock = threading.Lock() win = WindowThread(lock) win.setDaemon(True) # 親スレッド終了時に一緒に終了 win.start() # ウィンドウができるまで待ち while win.hwnd == 0: time.sleep(0.1) message_map = {} @staticmethod def set_callback(msg_id, func): MessageEvents.lock.acquire() MessageEvents.message_map[msg_id] = func MessageEvents.win.win.message_map[win32con.WM_USER + msg_id] = func MessageEvents.lock.release() return win32con.WM_USER + msg_id @staticmethod def get_hwnd(): return MessageEvents.win.hwnd
スレッドの知識が浅いので変なこと書いてるかも．．
