音声強化版PaPeRo i の音声認識アプリをリモートホストで動かす

 音声強化版PaPeRo i で音声認識を行うの最後でpapebotパッケージがないPCでpapebotを利用するアプリを動かす方法を紹介しましたが、これは単なるダミーで、音声認識はPCのキー入力で代用し、papebotのおしゃべりAPIなどは実際には機能しないものでした。本体の音声認識機能を利用するアプリはリモートホストでも動かせたのに較べ、不便です。そこで、音声強化版PaPeRo i のTinker Board S でpapebot機能を提供するサーバを動かし、リモートホストから通信によって音声認識機能を含むpapebotの一部機能を利用できる様にしてみました。
※記事公開時「papebotの全機能を利用できる様に」としていましたが誤りでした。全APIを呼ぶことはできますがauto_talk()などpapebotがpypaperoを制御する機能についてはにせpypaperoを制御することになり、正常に動作しません。(2020/2/25)

にせpypapero

 papebotには初期化時にpypaperoのインスタンスを渡す必要がありますが、今回実際にパペロを制御するアプリはリモートホスト上で動かしたいので、papebotには実際にはパペロを制御する動作を行わないにせのインスタンスを渡すことにします。このためにpypapero.Paperoクラスを継承し通信でパペロを制御する部分を何もしない様にオーバーライドしたにせpypaperoクラスを作りました。

nullpypapero.py:

import json
import queue
import time
from logging import getLogger, debug, info, warn, error, critical

import pypapero

logger = getLogger(__name__)


nullversion = "0.0.0"
nullversion_info = (0, 0, 0)

class Papero(pypapero.Papero):
    def __init__(self, simulator_id, robot_name, arg_ws_server_addr):
        """
        @param simulator_id:シミュレータID
        @param robot_name:ロボット名
        @param arg_ws_server_addr:WebSocket接続先(""ならデフォルトの接続先)
        """
        ws_server_addr = "wss://smilerobo.com:8000/papero"  # デフォルトの接続先
        if arg_ws_server_addr != "":
            ws_server_addr = arg_ws_server_addr
        logger.debug("simulator_id=" + str(simulator_id))
        logger.debug("robot_name=" + str(robot_name))
        logger.debug("ws_server_addr=" + str(ws_server_addr))
        self.simulatorID = simulator_id
        self.robotName = robot_name
        self.robotID = 0
        self.messageID = 0
        self.errOccurred = 0
        self.errDetail = ""
        self.scriptMayFinish = False
        # WebSocket関連
        self.wsAvail = False
        # self.ws = PaperoClient(ws_server_addr, protocols=['http-only'])
        self.ws = pypapero.PaperoClient(ws_server_addr, protocols=None)
        self.ws.papero = self
        ###self.ws.connect()
        # スレッド間通信用キュー
        self.queFromCom = queue.Queue()
        # Ver.1.01 発話コマンド個数管理
        self.remain_speech_count = 0
        self.papero_init()

    def put_dummy_ready(self):
        dic = {
            'Name': 'Ready',
            "RobotID": "dummyrobotid#",
        }
        s = json.dumps(dic)
        self.queFromCom.put(s)

    def put_ex_event(self, name, msg=None):
        if msg is None:
            msg = {}
        if name is not None:
            msg['Name'] = name
        dic = {
            'Name': 'RobotMessage',
            "Messages": [msg, ],
        }
        #s = json.dumps(dic)
        #self.queFromCom.put(s)
        # dictをそのまま載せるように改造
        self.queFromCom.put(dic)

    def papero_init(self):
        """
        パペロ初期化
        """
        #while not self.wsAvail:
        #    time.sleep(0.1)
        time.sleep(0.1)
        #self.send_select_sim_robot()
        self.put_dummy_ready() ###
        while True:
            message = self.papero_recv(None)
            logger.error('message: {}'.format(message))
            if message is not None:
                msg_dic_rcv = json.loads(message)
                if msg_dic_rcv["Name"] == "Ready":
                    self.robotID = msg_dic_rcv["RobotID"]
                    break
                elif msg_dic_rcv["Name"] == "Error":
                    logger.error("------Received error (papero_init()). Detail : " + msg_dic_rcv["Detail"])
                    self.papero_send(None)
                    self.errOccurred = 1
                    self.errDetail = "Inithalize failed"
                    break
            elif self.errOccurred != 0:
                logger.error("------Error occurred(papero_init()). Detail : " + self.errDetail)
                self.papero_send(None)
                self.errDetail = "Inithalize failed"
                break

    def papero_send(self, msg_dic_snd):
        """
        伝文送信
        @param msg_dic_snd:送信するJSON文字列(Noneを設定すると通信終了となる)
        """
        # (pypapero4papebot) kill papero control function
        logger.debug(msg_dic_snd)
        return

    def papero_robot_message_recv(self, t):
        """
        ロボット伝文受信
        @param t:タイムアウト
        @return 受信伝文の配列、t秒以内に受信できなかった場合はNone
        """
        robot_message = self.papero_recv(t)
        messages = None
        if robot_message is not None:
            if isinstance(robot_message, dict):  # dictをそのまま載せられるように改造
                msg_dic_rcv = robot_message
            else:
                msg_dic_rcv = json.loads(robot_message)
            if msg_dic_rcv["Name"] == "RobotMessage":
                messages = msg_dic_rcv["Messages"]
                # Ver.1.01 発話コマンド個数管理
                if messages[0]["Name"] == "startSpeechRes":
                    self.remain_speech_count -= 1
            elif msg_dic_rcv["Name"] == "RobotEnd":
                self.errOccurred = 3
                self.errDetail = "ScriptEnd"
                if not self.scriptMayFinish:
                    self.send_script_abort()
        return messages

オリジナルではメッセージは辞書をjsonに変換していましたがメッセージに関数(コールバック)を載せる必要が生じたため、辞書をそのまま載せる様に改造しています。
なお最初はpapebotが動作せず、色々試したところ本来パペロから送られてくるReadyメッセージを作って自分に送ったところ正常に動作する様になりました。papebotは渡されたpypaperoインスタンスを改造する様ですがソースは非公開なのでこの様な手探りになります。またpapebotのバージョンアップによりこれらのコードが動作しなくなる可能性があります。

papebot API実行専用スレッド

 今回リモートホストからの通信によってpapebotの全APIを呼べるようにしたいのですが、papebotのAPIがスレッドセーフでなかった場合に発生すると予想される難しい不具合を回避するため、papebotのAPIは専用のスレッドだけから呼ぶ事にしました。これは、

(1) papebotと同等の公開メソッドを提供するクラスを作成
(2) そのクラスの公開メソッドはpypaperoの内部キュー(queue.Queue)にメッセージを入れる動作とする
(3) papebot API専用のスレッドはpypaperoの内部のキューの受信イベントドリブンで動作し、メッセージに従いpapebot APIを呼ぶ

とし、papebotの機能を利用する際にはこのクラスインスタンスの公開メソッドを呼ぶことにするという方法で行いました。queue.Queueへデータを入れるput()はスレッドセーフなので、これで目的が達成できます。上記(1)のpapebotと同等の公開メソッドを持ち、(3)の受信メッセージに従いpapebot APIを呼ぶ、という機能は今回作成する複数のクラスの共通機能になるので、(1)(3)を実装する基本クラスと(2)を実装する派生クラスに分離してこのクラスを作成しました。なおこのクラスには音声認識などイベントが発生した時にあらかじめ設定したコールバックを呼ぶ機能も実装しておきます。

基本クラス=papebotcom.py(抜粋):

class PapebotCom(object):

    # 要求を送信するメソッド
    def _send_msg(self, name, dic=None):
        raise NotImplementedError

    # 要求受付けメソッド

    def init(self, papero, *, config_file=None, config_data=None):
        # エンジン初期設定
        dic = {}
        if config_file is not None:
            dic[KEY_CONFIG_FILE] = config_file
        if config_data is not None:
            dic[KEY_CONFIG_DATA] = config_data
        self._send_msg(MID_INIT, dic)

    def set_auto_operation(self, params=None, ear_led=None, head_move=None,
                           mouth_led=None, cheek_led=None):
        # 自動動作設定
        # オリジナルはparamsのみ、個別に指定できる様に拡張している
        dic = params
        if dic is None:
            dic = {}
            if ear_led is not None:
                dic['EarLED'] = ear_led
            if head_move is not None:
                dic['HeadMove'] = head_move
            if mouth_led is not None:
                dic['MouthLED'] = mouth_led
            if cheek_led is not None:
                dic['CheekLED'] = cheek_led
        logger.info('dic={} ear={} head={} mouth={} cheek={}'.format(
            dic, ear_led, head_move, mouth_led, cheek_led
        ))
        self._send_msg(MID_INIT, dic)

    def start_listening(self, chat_talk=True, callback=None):
        # 動作開始
        if chat_talk is None:
            chat_talk = True
        dic = {KEY_CHAT_TALK: chat_talk, KEY_CALLBACK: callback}
        self._send_msg(MID_START_LISTENING, dic)

    def stop_listening(self, callback=None):
        # 動作停止
        self._send_msg(MID_STOP_LISTENIN

    ...


    def _proc_init(self, config_file=None, config_data=None):
        # 初期化
        logger.info('config_file={} config_data={}'.format(
            config_file, config_data))
        self.mypapebot.init(self.papero, config_file=config_file,
                            config_data=config_data)

    def _proc_set_auto_operation(self, params):
        # 自動動作設定
        self.mypapebot.set_auto_operation(params)

    def _proc_start_listening(self, chat_talk=True, cmd=None, callback=None):
        # 動作開始
        if chat_talk is None:
            chat_talk = True
        cbtype = 'b'
        if self.mypapebot_has_callback:
            def cb(res):
                self.do_proc_callback(cmd, callback, cbtype, res)
            logger.info('chat_talk={}'.format(chat_talk))
            self.mypapebot.start_listening(chat_talk=chat_talk, callback=cb)
        else:
            res = self.mypapebot.start_listening(chat_talk=chat_talk)
            logger.info('chat_talk={} res={}'.format(chat_talk, res))
            self.do_proc_callback(cmd, callback, cbtype, res)
            return res

    def _proc_stop_listening(self, cmd=None, callback=None):
        # 動作停止
        cbtype = 'b'
        if self.mypapebot_has_callback:
            def cb(res):
                self.do_proc_callback(cmd, callback, cbtype, res)
            self.mypapebot.stop_listening(callback=cb)
            logger.info('end.')
        else:
            res = self.mypapebot.stop_listening()
            logger.info('res={}'.format(res))
            self.do_proc_callback(cmd, callback, cbtype, res)
            return res

    ...                                            

    def handle_cmd(self, name, callback, msg):
        # 受信コマンドに従い各_proc_xxxメソッドを呼び出す
        if name == MID_INIT:
            config_file = msg.get(KEY_CONFIG_FILE)
            config_data = msg.get(KEY_CONFIG_DATA)
            #logger.info('init: config_file={} config_data={}'.format(config_file, config_data))
            self._proc_init(config_file=config_file, config_data=config_data)
        elif name == MID_SET_AUTO_OPERATION:
            params = msg.get(KEY_PARAMS)
            #logger.info('set_auto_operation: params={}'.format(params))
            self._proc_set_auto_operation(params=params)
        elif name == MID_START_LISTENING:
            chat_talk = msg.get(KEY_CHAT_TALK)
            res = self._proc_start_listening(chat_talk=chat_talk,
                                            cmd=name, callback=callback)
            #logger.info('start_listening: chat_talk={} res={}'.format(chat_talk, res))
        elif name == MID_STOP_LISTENING:
            res = self._proc_stop_listening(cmd=name, callback=callback)
            #logger.info('stop_listening: res={}'.format(res))
        ...

派生クラス=papebotcallbacker.py(抜粋):

class PapebotCallbacker(papebotcom.PapebotCom):

    def run(self):
        # message loop
        self._msg_loop()

    def start(self):
        self.th = threading.Thread(target=self.run)
        self.th.start()

    def _msg_loop(self):
        while True:
            #msgs = self.papero.papero_robot_message_recv(1.0)
            msgs = self.papero.papero_robot_message_recv(None)
            res = self._handle_msg(msgs)
            if res:
                break

    def _handle_msg(self, msgs):
        if msgs is None:
            return False
        msg = msgs[0]
        name = msg.get(KEY_NAME)
        callback = msg.get(KEY_CALLBACK)
        logger.debug('event: {}.'.format(name))
        is_event = False
        if self.handle_cmd(name, callback, msg):
            pass
        elif name == MID_RECOG_START:
            # papero.send_turn_led_on("ear", ["W3W3", "10"], repeat=True)
            logger.info('papebotRecogStart.')
            is_event = True
        elif name == MID_RECOGNIZED:
            # papero.send_turn_led_off("ear")
            txt = msg.get(KEY_SENTENCE)
            logger.info('papebotRecognized: sentence={}.'.format(txt))
            is_event = True
        ...
        if is_event and callable(self.event_callback):
            try:
                self.event_callback(msg)
            except Exception as e:
                logger.error('event callback exception: {}'.format(e))

    def _send_msg(self, name, dic=None):
        # 公開メソッドのデータ送信。paperoのキューにputする
        self.papero.put_ex_event(name, dic)

papebotサーバ

 リモートホストへ通信でpapebot機能を提供するサーバはtornadoを利用したWebSocketサーバとしました。WebSocket経由で受け取ったメッセージに従い前述のPapebotCallbackerのAPIを呼び出します。

papebotserver.py(抜粋):

class WsHandler(tornado.websocket.WebSocketHandler):
    clients = []
    single_mode = True

    def takeover(self):
        # close others
        for cli in WsHandler.clients:
            if cli != self:
                cli.close()
        WsHandler.clients.clear()
        WsHandler.clients.append(self)

    def initialize(self):
        """ initialize (called by tornado) """
        if self.single_mode:
            self.takeover()
        else:
            self.clients.append(self)

    def open(self, *args, **kwargs):
        logger.debug("WsHandler args={} kwargs={}".format(args, kwargs))

    def on_close(self):
        logger.debug("BaseWsHandler")
        if self in WsHandler.clients:
            WsHandler.clients.remove(self)
        if 0 == len(WsHandler.clients) and self.papecb is not None:
            logger.info('do safe_kill()')
            self.papecb.safe_kill(timeout=5.0)
            self.papecb = None
            self.svpapebot = None

    def init_papecb(self):
        papero = nullpypapero.Papero('', '', '')

        def eventcb(msg):
            # papebotのイベントは全wsクライアントに送る
            if msg is None:
                logger.error('None.')
                return False
            # msg = msgs[0]
            name = msg.get(KEY_NAME)
            logger.info('event: {}.'.format(name))
            WsHandler.invoke_broadcast_dic(msg)

        def ext_proc_callback(obj, cmd, callback, cbtype, res):
            dic = {KEY_NAME: cmd + 'Res'}
            if cbtype == 'b':
                # bool
                dic['bool'] = res
            elif cbtype == 'i':
                dic['int'] = res
            elif cbtype == 't':
                dic['text'] = res
            else:
                logger.error('unknown callback_type: {}'.format(cbtype))
            self.send_dic(dic)
        #ext_proc_callbackp = ext_proc_callback
        ext_proc_callbackp = None

        papecb = papebotcb.PapebotCallbacker(papero=papero,
                                             event_callback=eventcb,
                                             ext_proc_callback=ext_proc_callbackp)
        # start papebotcb thread
        papecb.start()
        self.papecb = papecb
        self.svpapebot = ServerPapebot(mypapebot=papecb, send_dic=self.send_dic)

        #self.handle_cmd = self.papecb.handle_cmd
        self.handle_cmd = self.svpapebot.handle_cmd

    def on_message(self, dat):
        logger.debug("WsHandler msg={}".format(dat))
        msg = json.loads(dat)
        name = msg.get(KEY_NAME)
        logger.debug("name={} dic={}".format(name, msg))
        if name == MID_INIT:
            self.init_papecb()
        if self.handle_cmd(name, None, msg):
            pass
        else:
            logger.error('not supported. name={}'.format(name))

    def send(self, msg, binary=False):
        logger.debug("BaseWsHandler self:{} msg:{}".format(self, msg))
        try:
            super().write_message(msg, binary)
        except Exception as e:
            logger.error(e)

    def send_dic(self, dic):
        try:
            msg = json.dumps(dic)
            #super().write_message(msg, False)
            tornado.ioloop.IOLoop.current().add_callback(self.send, msg)
        except Exception as e:
            logger.error(e)

    @classmethod
    def broadcast(cls, msg, binary=False):
        for con in cls.clients:
            con.send(msg, binary)

    @classmethod
    def broadcast_dic(cls, dic):
        try:
            msg = json.dumps(dic)
            for con in cls.clients:
                con.send(msg, binary=False)
        except Exception as e:
            logger.error(e)

    @classmethod
    def invoke_broadcast(cls, msg, binary=False):
        tornado.ioloop.IOLoop.current().add_callback(cls.broadcast, msg, binary)

    @classmethod
    def invoke_broadcast_dic(cls, dic):
        tornado.ioloop.IOLoop.current().add_callback(cls.broadcast_dic, dic)


class ServerPapebot(papebotcom.PapebotCom):
    def __init__(self, mypapebot, send_dic):
        super().__init__(papero=None, mypapebot=mypapebot,
                         mypapebot_has_callback=True)
        self.send_dic = send_dic

    def do_proc_callback(self, cmd, callback, cbtype, res):
        dic = {KEY_NAME: cmd + 'Res'}
        if cbtype == 'b':
            # bool
            dic['bool'] = res
        elif cbtype == 'i':
            dic['int'] = res
        elif cbtype == 't':
            dic['text'] = res
        else:
            logger.error('unknown callback_type: {}'.format(cbtype))
        self.send_dic(dic)

papebotクライアント

 リモートホスト上で動作しpapebotサーバと通信するクライアントもtonadoを利用しました。これもPepebotComの派生クラスで、papebotと同様の公開メソッドを持ち、WebSocketにメッセージを送る様に動作します。

papebotclient.py(抜粋):

class PapebotClient(papebotcom.PapebotCom):
    def start(self):
        self._fu = tornado.websocket.websocket_connect(url=self._url,
                                                      callback=self._on_open)
    def _on_open(self, fu):
        logger.info('')
        self._conn = self._fu.result()
        # self._unsafe_send_dic({KEY_NAME: MID_INIT})
        self.init(None)
        if callable(self._open_callback):
            self._open_callback()
        else:
            self.init(None, config_file=self._config_file, config_data=self._config_data)
            if ((self._ear_led is not None) or (self._head_move is not None) or
                    (self._mouth_led is not None) or (self._cheek_led is not None)):
                self.set_auto_operation(ear_led=self._ear_led, head_move=self._head_move,
                                        mouth_led=self._mouth_led, cheek_led=self._cheek_led)
            self.set_triggerword_onoff(self._triggerword_onoff)
            self.start_listening(chat_talk=self._chat_talk)
        self._conn.read_message(callback=self._on_message)
        logger.info('end.')

    def _on_message(self, fu):
        #logger.info('')
        if callable(self._event_callback):
            try:
                dat = fu.result()
                if dat is None:
                    # closed
                    tornado.ioloop.IOLoop.instance().stop()
                    logger.error('closed.')
                    return
                dic = json.loads(dat)
                self._event_callback(self, dic)
            except Exception as e:
                logger.error(e)
        else:
            logger.error('event callback not set.')
        self._conn.read_message(callback=self._on_message)
        #logger.info('end.')

    def _unsafe_send_dic(self, dic):
        if self._conn is None:
            return False
        msg = json.dumps(dic)
        res = self._conn.write_message(msg, binary=False)
        return res

    def _unsafe_send_msg(self, name, dic=None):
        if dic is None:
            dic = {}
        dic[KEY_NAME] = name
        res = self._unsafe_send_dic(dic)
        return res

    def _send_msg(self, name, dic=None):
        tornado.ioloop.IOLoop.current().add_callback(self._unsafe_send_msg, name, dic)

テストコード

 リモートホストで動かすテストコードです。ここではpypaperoのラッパークラスstpypapero.pyを使用しています。音声認識した文字列が8文字以下ならオウム返しし、8文字以上の場合、おしゃべり機能による返答を発話します。音声発話などイベントの処理はローカル(Tinker Board S 上)で動かした場合と同じなのですが、do_chatbot_operation()の様に返値があるAPIはリモートではレスポンスメッセージが返ってくるという動作になるため、どうしてもリモートで動かす場合とローカルで動かす場合で条件分けする必要が生じます(この例はリモート専用です)。

papebotcli_test1.py:

from logging import (getLogger, Formatter, debug, info, warn, error, critical,
                     DEBUG, INFO, WARN, ERROR, CRITICAL, basicConfig)

import tornado.ioloop
import tornado.web
import tornado.websocket

import papebotclient

from stpypapero import StPyPaperoThread

from papebotclient import (
    MID_RECOGNIZED,
    KEY_SENTENCE,
)

MID_READY = 'Ready'
MID_END_OF_SPEECH = '_endOfSpeech'

KEY_STATUS = 'Status'
KEY_TEXT = 'Text'

logger = getLogger(__name__)


class App1(StPyPaperoThread):
    def initialize(self):
        self.handler = self.in_init
        self.papebot = None

    def set_papebot(self, papebot):
        self.papebot = papebot

    def speech(self, text):
        if self.papebot is not None:
            self.papebot.voice_mute()
        super().speech(text=text)

    def do_chat(self, text):
        if self.papebot is not None:
            self.papebot.do_chatbot_operation(sentence=text, talk=False)

    def in_init(self, name, msg):
        if name == MID_READY:
            self.speech('デモアプリ')
            self.trans_state(self.in_speech)

    def in_idle(self, name, msg):
        logger.debug('event: {}'.format(name))
        if name == MID_RECOGNIZED:
            txt = msg.get(KEY_SENTENCE)
            if txt is not None and 0 < len(txt):
                if 8 < len(txt):
                    self.do_chat(txt)
                    self.trans_state(self.in_chat)
                else:
                    self.speech('{}って何のこと?'.format(txt))
                    self.trans_state(self.in_speech)

    def in_chat(self, name, msg):
        if name == MID_DO_CHATBOT_OPERATION_RES:
            txt = msg.get('text')
            if (txt is not None) and (0 < len(txt)):
                self.speech(txt)
                self.trans_state(self.in_speech)
            else:
                logger.error('bad message: {}'.format(msg))
                self.trans_state(self.in_idle)

    def in_speech(self, name, msg):
        if name == MID_END_OF_SPEECH:
            if self.papebot is not None:
                self.papebot.voice_unmute()
            self.trans_state(self.in_idle)


def main():
    logger.info('start.')

    ipnet = '192.168.5'
    boturl = 'ws://{}.100:8867/papebot'.format(ipnet)
    papeurl = 'ws://{}.1:8088/papero'.format(ipnet)
    papero = App1(simid='', simname='', url=papeurl)

    def eventcb(rpapebot, msg):
        if msg is None:
            logger.error('None.')
            return False
        name = msg.get('Name')
        if name is not None:
            papero.put_ex_event(name, msg)

    try:
        # tw = True   # 「パペロ」必要
        tw = False  # 「パペロ」不要
        rmt = papebotclient.PapebotClient(boturl, event_callback=eventcb, triggerword_onoff=tw, chat_talk=False)
        papero.set_papebot(rmt)
        papero.start()
        rmt.start()
        tornado.ioloop.IOLoop.current().start()

    except KeyboardInterrupt:
        logger.info('KeyboardInterrupt!')
    except Exception as e:
        logger.error(e)
        logger.info('end.')


if __name__ == "__main__":
    loglevel = INFO
    logfmt = '%(asctime)s %(levelname)s %(thread)d.%(name)s.%(funcName)s %(message)s'
    basicConfig(level=loglevel, format=logfmt)

    main()

動作確認

(1) サーバ側ではpapebotserver.pyを実行します。

python3 papebotserver.py

(2) Windows PCでpapebotcli_test1.pyをPyCharmから、またはコマンドプロンプトから実行します。

python3 papebotcli_test1.py

ソースはこちらからダウンロードできます。