センサー値をリアルタイムグラフ表示してみたところ、人感センサー(ir)は手を近づけたりすると敏感に反応することが観察できました。この値を使ってパペロの首を動かしたらどうなるのか？という思いつきを試してみました。

PaPeRo iの人感センサーについて

　PaPeRo iの人感センサーは
(1) 赤外線センサ自体の温度素子
(2) 上下左右に配置された4つの素子の検出値
(3) センサーコンポーネント内で判定した位置情報“Center”(中央)、”Left”(左)、”Right”(右)、”Lost”(不在)
の6値を返します。今回は(2)の値の、上下の差、左右の差の値に単に係数を掛けて首の絶対位置(角度)として指定しました。

プログラムについて

　リアルタイムグラフ表示のプログラムに首の制御を追加します。これはグラフ表示のための状態遷移プログラムになっていて、首の動きとは同期させづらいので、首は簡単にタイマーで2秒に1回、最新のセンサー値で動かすことにします。このため、状態とは無関係にイベントごとにstpypaperoから呼び出されるメソッドevent_hook()に処理を記述します。

    def event_hook(self, name, msg):
        if name == '_timer':
            if msg.get('TimerID') == 'head':
                self.do_move()

    def do_move(self):
        if self.irlst is None:
            return
        s2x = 0.5
        s2y = 0.5
        ofx = 0
        ofy = -10
        x = s2x * (self.irlst[1] - self.irlst[3]) + ofx
        y = s2y * (self.irlst[4] - self.irlst[2]) + ofy
        logger.info((x, y))
        self.move_to(x, y, sfx="S50L")
        self.set_timer(2, 'head')

センサー値から首の指令値に変換するための係数(0.5, 0.5)とオフセット(0, -10)は、動作を見て適当に決めたものです。

動作確認

　オリジナル同様、このプログラムはPaPeRo i上でもラズパイでもPCでも、pythonが動作する環境であれば動かすことが出来ます。センサー値をリアルタイムグラフ表示が動作する状態から、末尾記載のファイルpaperograph.pyとstpypapero.pyを上書きして下さい。PaPeRo iのアドレスはpaperograph.pyに直書きしているので必要なら修正し、以下で起動して下さい。

> python3 paperograph.py

　座布団前面中央の人感センサーの周りに手をかざすと、かざした方に首が向いてくれました。手は使わず、顔をぐっと近づけるだけでも、だいたいこちらを向いてくれるような動きになりました。グラフ通り、ノイズ(?)により不安定な動きになりますが、そのランダムさがこの場合には良い方に働く様です。
　アプリの中で、単に首をランダムに動かしている場合には、この方法に置き換えると、そこそこ周りの人の動きを反映する動きになり、面白いかも知れません。ただし、2秒に1回の動作を連続で続けるのは単調な動作音がうるさく、もう少しメリハリを付けるような工夫が必要かも知れません。

ソース

(1) paperograph.py

# -*- coding: utf-8 -*-
# センサー値をリアルタイムグラフ表示する
# 0.1.0  2018/05/31 IRセンサー値で首を移動する試み
# 0.0.0  2018/05/25 初版
import os
from logging import (getLogger, debug, info, warn, error, critical,
                     DEBUG, INFO, WARN, ERROR, CRITICAL, basicConfig)

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

from stpypapero import StPyPaperoThread as StPyPapero


logger = getLogger(__name__)


class HtmlHandler(tornado.web.RequestHandler):
    def get(self):
        self.render('paperograph.html')


class JsHandler(tornado.web.RequestHandler):
    def get(self):
        self.render('paperograph.js')


class Js2Handler(tornado.web.RequestHandler):
    def get(self):
        self.render('smoothie.js')


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

    def initialize(self):
        """ initialize (called by tornado) """
        self.clients.append(self)

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

    def on_message(self, msg):
        logger.debug("WsHandler msg:{}".format(msg))

    def on_close(self):
        logger.debug("WsHandler")
        self.clients.remove(self)

    def send(self, msg, binary=False):
        logger.debug("WsHandler msg:{}".format(msg))
        super().write_message(msg, binary)

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

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


def start_web():
    app = tornado.web.Application([
        (r'/paperograph.html', HtmlHandler),
        (r'/paperograph.js', JsHandler),
        (r'/smoothie.js', Js2Handler),
        (r'/__ws', WsHandler)
        ],
        template_path=os.path.dirname(__file__),
    )
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()


class SensorApp(StPyPapero):
    def initialize(self):
        self.handler = self.in_init
        self.irlst = None
        # states are:
        # init, sensors, time_wait, lum_start, lum_get, lum_stop

    # 処理メソッド

    def trans_state(self, newhandler):
        # ログを抑制するためのオーバーライド
        self.handler = newhandler

    def do_move(self):
        if self.irlst is None:
            return
        s2x = 0.5
        s2y = 0.5
        ofx = 0
        ofy = -10
        x = s2x * (self.irlst[1] - self.irlst[3]) + ofx
        y = s2y * (self.irlst[4] - self.irlst[2]) + ofy
        logger.info((x, y))
        self.move_to(x, y, sfx="S50L")
        self.set_timer(2, 'head')

    # 状態遷移メソッド

    def go_lum_start(self):
        self.papero.send_start_lum_sensor()
        self.trans_state(self.in_lum_start)

    def go_lum_get(self):
        self.papero.send_get_lum_sensor_value()
        self.trans_state(self.in_lum_get)

    def go_lum_stop(self):
        self.papero.send_stop_lum_sensor()
        self.trans_state(self.in_lum_stop)

    def go_sensors(self):
        self.papero.send_get_sensor_value()
        self.trans_state(self.in_sensors)

    def go_time_wait(self):
        self.set_timer(0.1, 'tm0')
        self.trans_state(self.in_time_wait)

    # 状態ごとのイベント処理メソッド

    def in_init(self, name, msg):
        if name == 'Ready':
            # self.go_sensors()
            self.set_timer(2, 'head')
            self.go_lum_start()

    def in_lum_start(self, name, msg):
        logger.debug('event: {}'.format(name))
        if name == 'startLumSensorRes':
            self.go_lum_get()

    def in_lum_get(self, name, msg):
        logger.debug('event: {}'.format(name))
        if name == 'getLumSensorValueRes':
            lx = msg.get('Return')
            WsHandler.invoke_broadcast({'lx': lx})
            #self.go_lum_stop()
            self.go_sensors()

    def in_lum_stop(self, name, msg):
        logger.debug('event: {}'.format(name))
        if name == 'stopLumSensorRes':
            self.go_sensors()

    def in_sensors(self, name, msg):
        logger.debug('event: {}'.format(name))
        if name == 'getSensorValueRes':
            if 'ret' not in msg or 0 <= msg.get('ret'):
                temp = msg.get('TEM')
                rh = msg.get('HYG')
                ir = msg.get('IR')
                irlst = ir.split(',') if ir else []
                WsHandler.invoke_broadcast({'temp': temp, 'rh': rh, 'ir': irlst})
                self.irlst = [int(s) for j, s in enumerate(irlst) if j < 5 ]
            else:
                logger.info('error ret={}'.format(msg.get('ret')))
            self.go_time_wait()

    def in_time_wait(self, name, msg):
        logger.debug('event: {}'.format(name))
        if name == '_timer':
            if msg.get('TimerID') == 'tm0':
                self.go_lum_start()
                # self.go_sensors()

    # 状態にかかわらず実行するイベントフック
    def event_hook(self, name, msg):
        if name == '_timer':
            if msg.get('TimerID') == 'head':
                self.do_move()


def main():
    #url = 'ws://192.168.1.188:8088/papero'
    url = 'ws://192.168.4.1:8088/papero'
    papero = SensorApp(url=url)
    papero.start()
    #threading.Thread(target=start_web).start()
    start_web()


if __name__ == '__main__':
    basicConfig(format='%(asctime)-15s %(levelname)s %(module)s.%(funcName)s %(message)s')
    logger = getLogger()  # root logger
    # logger.setLevel(DEBUG)
    logger.setLevel(INFO)
    logger.info('start')
    main()

(2) stpypapero.py

# -*- coding: utf-8 -*-
# pypaperoラッパークラスstpypapero
# (c) 2017-2018 Sophia Planning Inc.
# LICENSE: MIT
#
# Ver   Date       Description
# 0.3.1 2018/05/31 event_hook()が機能しなかった不具合修正
# 0.3.0 2018/05/29 put_ex_event(), move_to()、event_hook()追加
# 0.2.0 2018/05/22 状態変数を廃止、状態処理メソッドself.handlerを書き換える方法に変更
# 0.1.1 2018/02/20 speech()にmouth_en, head_en追加
# 0.1.0 2018/02/09 名称変更paperoappbase->stpypapero、on_startup削除、
#                  メッセージのリスト(現状要素数常に1)ではなく要素を渡すよう変更、
#                  paperoの引数指定追加、処理関数の返値をTRUEでプログラム終了(return書かなければ継続)に反転
#                  タイマー機能追加
# 0.0.2 2018/02/02 基本クラスAppBaseと派生クラスThreadAppBaseに分離
# 0.0.1 2017/12/19 Initial version
"""
使い方:
(1) StPyPaperoを継承したクラスを作る
(2) in_xxxで各状態ごとの処理を記述
(3) initialize()でself.handlerに初期状態用のin_xxxを設定
(4) 状態遷移時にはself.handlerを別のin_xxxに書き換える

機能：
(1) 終話検知
  stpypapero.speech(text)
  ...
  def in_xxxx(self, name, msg):
      if name == '_endOfSpeech':
          ...
 イベント: ''
(2) タイマー
  stpypapero.set_timer(delay, timer_id)
  stpypapero.cancel_timer(timer_id)
  ...
  def in_xxxx(self, name, msg):
      if name == '_timer':
          tid = msg.get('TimerID')
          if tid == timer_id:
              ...
(3) 外部イベント(パペロスレッドにパペロ以外のイベントを送る)
  stpypapero.put_ex_event('__exEvent1', {'data': [1,2,3,4]})
  ...
  def in_xxxx(self, name, msg):
      if name == '__exEvent1':
          ...
"""
import sys
import threading
import random
import json
from logging import (getLogger, basicConfig, DEBUG, INFO, WARN, ERROR,)

import pypapero

logger = getLogger(__name__)


# mouth data for speech
MOUTH_DAT = [
    [
        "G3" + "G3" + "G3" + "G3" + "G3" + "G3" + "G3" + "G3" + "G3", "3",
        "NNN" + "G3" + "G3" + "G3" + "NNN", "3",
        "NNNN" + "G3" + "NNNN", "3",
        "N" + "G3" + "G3" + "G3" + "G3" + "G3" + "G3" + "G3" + "N", "3",
        "NN" + "G3" + "G3" + "G3" + "G3" + "G3" + "NN", "3",
    ]
]

# head data for speech (random select)
HEAD_DAT = [
    [ # 0
        ["A4T300L",  "R0T300L",  "R0T300L",  "R0T300L", "A0T300L", ],  # x
        ["A3T300L", "A-3T300L", "A3T300L", "A-6T300L", "A0T300L", ],   # y
    ], [ # 1
        ["A-4T300L",  "R0T300L",  "R0T300L",  "R0T300L", "A0T300L", ],  # x
        ["A-3T300L", "A3T300L", "A-3T300L", "A6T300L", "A0T300L", ],    # y
    ], [ # 2
        ["A-3T300L",  "R0T300L",  "R5T300L",  "R0T300L", "A0T300L", ],  # x
        ["A-3T300L", "A3T300L", "A-2T300L", "A4T300L", "A0T300L", ],    # y
    ]
]


class StPyPapero(object):
    # papero application class
    def __init__(self, simid='', simname='', url=None, polling_sec=0.2, papero=None):
        if url is None:
            simid, simname, url = pypapero.get_params_from_commandline(sys.argv)
        self._polling_sec = polling_sec
        self._eos_wait_res = False  # send_get_speech_staus and waiting response
        self._eos_polling = False
        self._eos_callback = None  # end of speech callback
        self._timer_dic = {}
        if papero is None:
            papero = pypapero.Papero(simid, simname, url)
        self.papero = papero
        self.handler = None
        self.initialize()
        if 0 < self.papero.errOccurred:
            logger.error(self.papero.errDetail)
        else:
            self._inner_event_handler('Ready')

    """virtual
    def initialize(self):
        self.handler = self.in_init
    """

    def trans_state(self, newhandler):
        # transit state
        logger.info('state: {} -> {}'.format(self.handler, newhandler))
        self.handler = newhandler

    def speech(self, text,
               language=None, speaker_id=None,
               pitch=None, speed=None, volume=None,
               pause=None, comma_pause=None, urgent=None, priority="normal",
               eos_callback=None, mouth_en=True, head_en=True):
        # speech with head and mouth
        logger.debug('start speech')
        self._eos_polling = True
        self._eos_callback = eos_callback
        if mouth_en:
            self.papero.send_turn_led_on('mouth', MOUTH_DAT[0], repeat=True)
        if head_en:
            hidx = random.randint(0, len(HEAD_DAT) - 1)
            self.papero.send_move_head(HEAD_DAT[hidx][1], HEAD_DAT[hidx][0])
        self.papero.send_start_speech(text, language=language, speaker_id=speaker_id,
                                      pitch=pitch, speed=speed, volume=volume,
                                      pause=pause, comma_pause=comma_pause,
                                      urgent=urgent, priority=priority)

    def recv_msg(self):
        msg = self.papero.papero_robot_message_recv(self._polling_sec)
        return self._inner_event_handler(msg)

    def _inner_event_handler(self, msg):
        if self.papero.errOccurred != 0:
            return True
        # msg = self.recv_msg()
        if msg is None:
            if self._eos_polling and not self._eos_wait_res:
                # poll end of speech
                self.papero.send_get_speech_status()
                self._eos_wait_res = True
            msg = self._timeout_hook()
        if msg is None:
            return False
        else:
            res = self._handle_msg(msg)
            return res

    def run(self):
        while True:
            if self.recv_msg():
                break
        self.papero.papero_cleanup()

    def _timeout_hook(self):
        return None

    """
    def in_init(self, name, msg):
        if name == 'Ready':
            self.trans_state(self.State.IDLE)

    def in_idle(self, name, msg):
        pass
    """

    def _handle_msg(self, msg):
        if msg is None:
            return False
        if isinstance(msg, str):
            name = msg
        elif isinstance(msg, list):
            if 1 == len(msg):
                msg = msg[0]
                name = msg.get('Name')
            else:
                name = 'MsgList'
        else:
            name = msg.get('Name')
        if name == "getSpeechStatusRes":
            self._eos_wait_res = False
            ret = msg.get("Return")
            if self._eos_polling and (ret == 0):
                logger.debug('end of speech')
                self._eos_polling = False
                self.papero.send_turn_led_off('mouth')
                if callable(self._eos_callback):
                    self._eos_callback()
                # change name
                name = "_endOfSpeech"
        if hasattr(self, 'event_hook'):
            self.event_hook(name, msg)
        # call function for this state
        if callable(self.handler):
            self.handler(name, msg)

    # callbackが別スレッドで動く事に注意
    def _set_timeout(self, delay, callback, *args, **kwargs):
        hdl = threading.Timer(delay, callback, args, kwargs)
        hdl.start()
        return hdl

    def _remove_timeout(self, hdl):
        hdl.cancel()

    def _send_timeout_msg(self, timer_id):
        msg = ('{{"Name": "RobotMessage", "Messages": [' +
               '{{"Name": "_timer", "TimerID": "{}"}}] }}').format(timer_id)
        self.papero.queFromCom.put(msg)

    def set_timer(self, delay, timer_id):
        hdl = self._set_timeout(delay, self._send_timeout_msg, timer_id)
        self._timer_dic[timer_id] = hdl

    def cancel_timer(self, timer_id):
        if timer_id in self._timer_dic:
            hdl = self._timer_dic[timer_id]
            self._remove_timeout(hdl)

    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.papero.queFromCom.put(s)

    @staticmethod
    def limit(x, lb, ub):
        if x < lb:
            return lb
        if x > ub:
            return ub
        return x

    def move_to(self, xy, y=None, pfx='A', sfx='S40L'):
        if y is None:
            x, y = xy
        else:
            x = xy
        x = self.limit(x, -128, 128)
        y = self.limit(y, -20, 55)
        hs = ['{}{}{}'.format(pfx, int(x), sfx), ]
        vs = ['{}{}{}'.format(pfx, int(y), sfx), ]
        self.papero.send_move_head(vs, hs)
        return x, y


class StPyPaperoThread(threading.Thread, StPyPapero):
    def __init__(self, simid='', simname='', url=None, polling_sec=0.2, papero=None):
        if url is None:
            simid, simname, url = pypapero.get_params_from_commandline(sys.argv)
        # threading.Thread.__init__()
        super(StPyPaperoThread, self).__init__()
        # PyPaperoAppBase.__init__()
        super(threading.Thread, self).__init__(simid=simid, simname=simname,
                                               url=url, polling_sec=polling_sec,
                                               papero=papero)

    def run(self):
        return super(threading.Thread, self).run()