【Pepper】ロボットと人の会話履歴をロギングしてローカルに保存する【Choregraphe/Python】

タイトルの通りです。

はじめに/対象読者
いいからソースコードだ！
コードの内容
Pepperのローカルに保存したファイルはどうやって手元のPCに持ってくるの？
まとめ

はじめに/対象読者

タイトル通りなので繰り返してもアレなんですが、Pepperの発話と、人がPepperに話しかけた発言の履歴をログとしてローカルファイルに保存する方法を紹介します。平たく言うと、Choregrapheの「ダイアログ」ウィンドウに表示される内容をアプリ内プログラムで拾ってテキストに保存しよう、という話です。

本記事の想定対象読者は下記の通りです。

Choregrapheの基本的な使い方は分かってる方(ボックス検索して貼りつけ)
Pythonコードを読める方

いいからソースコードだ！

雰囲気をお伝えするために結論から。下記のコードはChoregrapheで使う想定のPythonプログラムです。

from datetime import datetime
import time

class MyClass(GeneratedClass):
    def __init__(self):
        GeneratedClass.__init__(self)
        #アプリ起動時の時刻を使ってファイル名を生成
        self.filename = datetime.now().strftime("dialog_%Y_%m%d_%H%M_%S.txt")
                                     
    def onLoad(self):
        
        #保存先を指定してファイルオープン、フルパスは下記
        #/home/nao/.local/share/<アプリID>/(self.filename)
        #(Windowsでシミュレータを使ってる場合のパス例, 下記のAppDataフォルダは隠しフォルダな点に注意)
        #(C:\Users\<ユーザ名>\AppData\Roaming\.lastUploadedChoregrapheBehavior\(self.filename))

        appId = self.packageUid()
        self.dataPath = qi.path.userWritableDataPath(appId, self.filename)
        self.writer = open(self.dataPath, "w+")
        
        #qi Frameworkをベースにしたモジュールを取得(新しいフレームワーク:こっちが推奨されてる)
        self.mem = ALProxy("ALMemory").session().service("ALMemory")

        #subscriber Id
        self.signalHuman = None
        self.signalRobot = None
        self.subHumanId = 0
        self.subRobotId = 0
        
        self.isRunning = False
      
        
    def onUnload(self):
        if self.subHumanId and self.signalHuman:
            self.signalHuman.disconnect(self.subHumanId)
            self.signalHuman = None
            self.subHumanId = 0
            
        if self.subRobotId and self.signalRobot:
            self.signalRobot.disconnect(self.subRobotId)
            self.signalRobot = None
            self.subRobotId = 0
            
        if self.writer != None:
            self.writer.close()
            self.writer = None
    
        self.isRunning = False   
    
    def onInput_onStart(self): 
        if self.isRunning:
            return
            
        self.isRunning = True

        if not self.writer:
            self.writer = open(self.dataPath, "w+")
        
        #人の会話
        subscriberHuman = self.mem.subscriber("Dialog/LastInput")
        self.signalHuman = subscriberHuman.signal
        self.subHumanId = self.signalHuman.connect(self.onHumanSpeechDetected)
        
        #Pepperからの発話
        subscriberRobot = self.mem.subscriber("ALTextToSpeech/CurrentSentence")
        self.signalRobot = subscriberRobot.signal
        self.subRobotId = self.signalRobot.connect(self.onRobotSpeechDetected)

    
    def onInput_onStop(self):
        self.onUnload()
        self.onStopped()

    def onHumanSpeechDetected(self, val):
        self.writeSentenceToFile(val, "Human")
        
    def onRobotSpeechDetected(self, val):
        self.writeSentenceToFile(val, "Robot")
            
    def writeSentenceToFile(self, val, whoTalk):
        if self.writer and val:
            content = "{0}({1}):{2}\n".format(
                whoTalk,
                datetime.now().strftime("%H:%M:%S"),
                val
                )
            self.writer.write(content)
            self.logger.info(content)
            self.writer.flush()

上記のPythonコードは次のようにして使えます。

Choregrapheでボックスライブラリから「Python Script」ボックスを選んで配置
ボックスをダブルクリックしてスクリプト編集画面を開く
既存のコードを全て消し、代わりに上のスクリプトをコピペ
アプリ起動時に、作ったボックスがスタートするように線をつなぐ

StartだけでなくStopにも対応させたつもりですが、あまりマジメにデバッグしてないのでご注意下さい。

コードの内容

上記のコードで使っているテクニックを簡単に紹介しておきます。

1. 書き込み可能なローカルファイルのパスを取得する

qi.path.userWritableDataPathという関数を使うことで、Pepperのローカルストレージ内に保存するデータへの適切なパスが取得できます。

appId = self.packageUid()
self.dataPath = qi.path.userWritableDataPath(appId, self.filename)
self.writer = open(self.dataPath, "w+")

全体のソースのコメントにも書いてますが、このようにするとself.dataPathとして「"/home/nao/.local/(アプリID)/(ファイル名)"」という形でパスが取得できます。この方法で取得したパスに対してであれば、ファイルの読み書きが安全に行えます。アプリに関するファイルパスを取得する似たような機能としてはALFrameManagerとかself.behaviorAbsolutePath()といったものもありますが、これらはアプリ自体がインストールされている場所のパスを取得するものです。気分としては

qi.path.userWritableDataPath関数で拾えるパス: Windowsのマイドキュメント
ALFrameManager / self.behaviorAbsolutePathで拾えるパス: WindowsのProgram Files

くらいの違いだと認識すると良さそうに思います。ちなみにqi.pathを使う方法は「Pepper プログラミング」で紹介されているテクニックの一つです。

オンラインドキュメンテーションの一つであるqi Frameworkのドキュメントでも同じ内容の紹介がありますが、Python版だと内容が薄すぎるので、真面目に使うならC++版を確認する必要があるでしょう。

2. イベント監視を行うためのALMemoryモジュールを取得する(新方式)

先ほど紹介したコードでは通常のALProxyからモジュールを取得する方法にヒネリが加えられています。

#qi Frameworkをベースにしたモジュールを取得(新しいフレームワーク:こっちが推奨されてる)
self.mem = ALProxy("ALMemory").session().service("ALMemory")

対比として紹介しますが、従来のコードは通常こう書かれています。

self.mem = ALProxy("ALMemory")

この違いについては「【和訳記事】NAOqiのプログラムをqi Frameworkに移行しよう！」で紹介しているのですが、ALProxyで取ってきた機能をそのまま使うのは古いスタイルで非推奨とされているため、上例では新しいスタイルのコードを使うように変更を加えています。結果として見た目が少し奇妙になってますが、まあテンプレだと思ってください。

3. 実際にイベントを監視する

Pepperの発話や人が話しかけた言葉を取得する処理は対応するイベントの監視処理が必要であり、そのイベント監視を開始するのが次のコードです。実はこれについても「【和訳記事】NAOqiのプログラムをqi Frameworkに移行しよう！」で非常に近い話題を紹介しています。

#人の会話
subscriberHuman = self.mem.subscriber("Dialog/LastInput")
self.signalHuman = subscriberHuman.signal
self.subHumanId = self.signalHuman.connect(self.onHumanSpeechDetected)

Javascript版のAPIでタブレットとPepperの連携プログラムを書いた人であれば、上のコードはそんなに目新しいものではないと思います。手順はテンプレート的です。

ALMemoryのsubscriber関数に、監視したいメモリのキーを指定(上の場合だと"Dialog/LastInput")
1で取得した結果からsignalプロパティを取得
2で取得したsignalのconnect関数を呼び出し、connect関数の引数にコールバック関数を与える
3の戻り値としてIDを取得できるので、イベントの監視を終了する際にはそのIDを指定してsignalのdisconnect関数を呼び出す(onUnloadで行っている処理)

3で指定するコールバック関数には「引数がひとつだけあり、戻り値が無い」という関数を指定します。今回の場合はログファイルに追記を行う関数をコールバック関数として登録しています。

4. 監視するイベントのキーについて

これはコードに書かれているままの話題ですが、本記事のロギング処理を実現するために一番大切な部分なので一応きちんと書いておきます。上のコードでは監視対象となるイベントのキーとして次の二つを使っています。

"Dialog/LastInput": 人がPepperに喋りかけてくると発生するイベント。中身は人が喋った言葉(=音声認識の結果)
"ALTextToSpeech/CurrentSentence": Pepperが人へ喋りかけるときに発生するイベント。中身はPepperが喋った言葉

こういうイベントを探し当てるにはロボットからのあらゆるイベントを監視すればいいんですが、これは本記事の内容から逸れるので割愛します¹。

それともう一つ。細かい注意になりますが、ALTextToSpeech/CurrentSentenceイベントは喋り始めた時だけでなく喋り終わったときにも発生する仕様になっています。こういう仕様になってる理由は定かではありませんが、喋り終わってイベントが発生した場合にはデータの中身として空文字が渡されます。

今回のロギングプログラムでは「喋り始めた所はログ取りたいけど、終わったタイミングは別に興味ないや」という発想で作ってあり、if文で空文字列のデータが来たケースを弾くようにしています。

def writeSentenceToFile(self, val, whoTalk):
    #val(Pepperが喋った言葉)が空だとif文を通らないのでログが書き込まれない
    if self.writer and val:
        #実際の処理

Pepperのローカルに保存したファイルはどうやって手元のPCに持ってくるの？

本記事ではファイルを"/home/nao/.local/(アプリID)/(ファイル名)"というパスに保存できますよー、という話はしましたが、プログラムを実行し終わった際にPepperからPCへファイルを持ってくる手段の話はしてきませんでした。これはググれば出てくる話題ですが、とのさまラボさんのQiita記事等に載っている通り、Choregrapheの機能としてファイルダウンロード機能があるので使いましょう。Pepperは所詮Linuxだと割り切れている人はSSHで通信してファイルをダウンロードすることもできます。

まとめ

今回は以上です。実は本記事の実装には一つ改善点があって、人からPepperに話しかけた際の認識精度(50%とか)が取得できてませんが、これについては課題という事で残しておきたいと思います。

Pepperに登録されたイベント形式のメモリデータを全て監視する ↩

Bakulog

獏の夢日記的な何か。