作業中のメモ

よく「計算機」を使って作業をする.知らなかったことを中心にまとめるつもり.

音声認識による赤外線機器の操作 その 6【単語解析編】

どうも,筆者です.

前回は,以下の 3 つを実装した.

  • Julius の起動
  • Julius からの認識結果の取得
  • WebSocket を用いて認識結果を送信

前回までの記事は以下にある.

workspacememory.hatenablog.com

製作状況

製作状況を以下に示す.パーサーの部分は製作が完了している.今回は,単語解析の部分を実装する.

f:id:mathematicsphysical:20180603192707p:plain
製作状況

今回の対象

今回対象とする単語解析の部分の構成を以下に示す.

f:id:mathematicsphysical:20180603192721p:plain
単語解析

流れとして,受信した単語を単語解析関数に入力し,戻り値として,解析結果,赤外線コマンド,音声ファイルのデータを返すものとする.これらの結果から,音声を流すかどうか,赤外線コマンドを送信するかどうかを決める.

単語の認識方法

さて,問題となる単語の認識方法であるが,ここでは,受信した単語に規定の単語が含まれている場合,その単語が発音されたものとする.具体例を以下に示す.

f:id:mathematicsphysical:20180603192752p:plain
単語認識用データ

ここで,必須項目は,解析をするにあたり,必ず含まれている必要がある単語となる.必要項目は,列挙されているもののうち,ひとつでも含めばよいものである.空リストになっているものは,解析対象に含まれないことを意味する.それぞれの単語が見つかった場合の具体的な処理方法も別途データとして用意しておく必要がある.今回は,以下のようなデータを用意した.

f:id:mathematicsphysical:20180603192746p:plain
解析時に利用するデータ

起動時に上記のデータを読み込み,処理に利用する.それぞれの項目の働きを以下に示す.

  • machine:機器の状態を管理するために利用する(Boolean 型).
  • json:読み込む JSON ファイル
  • cmd:JSON ファイルに記述されているコマンドのうち,実行するコマンド
  • voice:コマンド実行時に再生する音声ファイル
  • chkState:現在の状態から次の内部状態を決めるパラメータ

「machine」と「chkState」は関連しており,例えば,テレビがついているのに「テレビつけて」と発声した場合,ほとんどのテレビはオン/オフが同一ボタンであるため,テレビの電源がオフになってしまう.これを防ぐために,現状のテレビの状態を監視しておく必要がある.その際に利用するのが「machine」の部分となる.

また,誤認識防止のため,過去 1 回前までの単語のキーを記憶しておくようにしている.ただ,これに伴い,電気を明るくしたり,(ここにはないが)テレビの音量を上げる場合に必要となる「連続動作」ができなくなる.連続動作の対象となるものは,今回認識した単語を記憶しないようにしたいため,「chkState」を利用する.「chkState」は他にも,テレビがオンのとき「テレビをオンにして」という命令を棄却するのにも利用している.

内部状態を決めるパラメータの定義

現状,「chkState」では,1 Byte のデータを用いて,いくつかの状態を表せるようになっている.その内訳を以下に示す.

f:id:mathematicsphysical:20180603192727p:plainf:id:mathematicsphysical:20180603192733p:plain
内部状態の内訳

これらのうち,予約済みのものを以下に示す.

f:id:mathematicsphysical:20180603192740p:plain
予約済みの内部状態パラメータ

単語ファイルの更新が追いついていないが,最終的には,「Ok,Google」と話しかけたら認識開始,「ありがとう(仮)」と話しかけたら認識終了としようと考えている.

単語解析コード

上記の設定を実装したものを以下に示す.ここでは,「configuration.py」として保存する.また,実行には,「jsonData」のディレクトリに対象とするデータが追加されている必要がある.

~/juliusKit $ touch configuration.py
~/juliusKit $ vim configuration.py # エディタは自分の使いやすいものを利用する
#!/usr/bin/python3
# -*- coding: utf-8 -*-

# configuration.py

import json
from VR_ConstClass import CONST_CLASS

class voiceConfigClass():
    def __init__(self):
        # 定数定義(public,結果解析で利用するため,外部から参照可能にしている)
        self.juliusReturnState = {
            True:      int('0x00', 16),
            False:     int('0x01', 16),
            'weather': int('0x02', 16)
        }

        # コンフィグ情報のマスク処理用パラメータ
        self.__maskDict = {'upper': int('0xF0', 16), 'lower': int('0x0F', 16)}
        # 機器の状態(True: 起動状態,False: 停止状態)
        self.__state = {'julius': False}
        # Julius の処理対象
        self.__JuliusConfig = {
            'invalid':     int('0x01', 16), # 無効値
            'wakeUp':      int('0xFF', 16), # 命令解析開始
            'sleep':       int('0xF8', 16), # 命令解析終了
            'weather':     int('0xF0', 16), # 天気予報読み上げ
            'init':        int('0xF1', 16), # 状態変数の初期化
            'active':      int('0x10', 16), # 実行中
            'stopped':     int('0x80', 16), # 停止中
            'update':      int('0x01', 16), # 状態を更新する
            'delPrevWord': int('0x02', 16)  # 直前の単語を記憶しない
        }
        # 語彙情報(定数)
        self.__confParamFile = 'confParam.json' # 図に示した「解析時に利用するデータ」が格納されている JSON ファイル
        self.__vocabInfo = None
        # 語彙リスト(図に示した「単語認識用データ」)
        # key: 単語ラベル
        # val: 語彙候補(mandatory: 必須項目,required: 必要項目)
        self.__vocabListData = {
            'lightOn': {
                'mandatory': [str('電気')],
                'required': [str('つけて'), str('オン')]
            },
            'lightOff': {
                'mandatory': [str('電気')],
                'required':  [str('オフ'), str('切って')]
            },
            'lightUp': {
                'mandatory': [str('明るくして')],
                'required':  []
            },
            'lightDown': {
                'mandatory': [str('暗くして')],
                'required':  []
            },
            'nightLight': {
                'mandatory': [str('こだまにして')],
                'required':  []
            },
            'TVOn': {
                'mandatory': [str('テレビ')],
                'required':  [str('つけて'), str('オン')]
            },
            'TVOff': {
                'mandatory': [str('テレビ')],
                'required':  [str('オフ'), str('切って')]
            }
        }
        # 作業用変数
        self.__jsonData = {}
        self.__prevWordIdx = ''
        self.__initConfigData()

    # 状態変数の初期化
    def __initStat(self, invalidList):
        keyList = [key for key, _ in self.__state.items()]
        for key in list(set(keyList) - set(invalidList)):
            self.__state[key] = False

    # 初期化処理関数
    def __initConfigData(self):
        # config データの取得
        jsonFileList = []
        machineTypeList = []
        # json file から config 情報を読み込む
        with open(CONST_CLASS.JSON_DIR + self.__confParamFile, 'r') as fin:
            self.__vocabInfo = json.load(fin)

        for key in self.__vocabInfo.keys():
            useMachine = self.__vocabInfo[key]['machine'].lower()
            self.__vocabInfo[key]['machine'] = useMachine
            jsonFileList.append(self.__vocabInfo[key]['json'])
            machineTypeList.append(useMachine)

        # 状態変数のキーの追加
        for key in list(set(machineTypeList)):
            self.__state[key] = False

        # json file の読み込み(list(set(---)) で重複を削除)
        for targetJsonFile in list(set(jsonFileList)):
            try:
                with open(CONST_CLASS.JSON_DIR + targetJsonFile, 'r') as fin:
                    self.__jsonData[targetJsonFile] = json.load(fin)
            except:
                # 該当する json file が存在しない場合,処理せず読み飛ばす
                pass

    # julius の状態の設定
    def setJuliusState(self, status):
        retVal = self.__state['julius']
        self.__state['julius'] = status
        return retVal

    # 状態の出力
    def getStatus(self):
        return [(key, self.__state[key]) for key in sorted(list(self.__state.keys()))]

    # 合致する単語を探す
    def __findMatchWord(self, wordData):
        retWordIdx = None

        # 単語の一覧から合致するパターンを抽出
        for key, wordDictList in self.__vocabListData.items():
            mandatoryData = wordDictList['mandatory']
            requiredData = wordDictList['required']
            # 必須項目の単語が含まれているか調査
            isExistMandatory = True
            for targetWord in mandatoryData:
                if targetWord not in wordData:
                    isExistMandatory = False
                    break
            # 必要項目の単語が含まれているか調査
            if len(requiredData) == 0:
                isExistRequired = True
            else:
                isExistRequired = False
                for targetWord in requiredData:
                    if targetWord in wordData:
                        isExistRequired = True

            if isExistMandatory and isExistRequired:
                retWordIdx = key
                break
        return retWordIdx

    # コマンドの実行可能判定
    def chkCmdExection(self, wordData):
        local_tmpVal = False
        returnState = self.juliusReturnState[local_tmpVal]
        retCmd = None
        wordIdx = self.__findMatchWord(wordData)
        retAudio = ''

        # 単語が候補に存在するかつ,前回と同じ単語でない
        if wordIdx is not None and wordIdx != self.__prevWordIdx:
            useMachine = self.__vocabInfo[wordIdx]['machine']
            useState = int(self.__vocabInfo[wordIdx]['chkState'], 16)
            useJsonFile = self.__vocabInfo[wordIdx]['json']
            useCmd = self.__vocabInfo[wordIdx]['cmd']
            tmpAudio = self.__vocabInfo[wordIdx]['voice']
            useAudio = {False: '', True: tmpAudio, 'weather': tmpAudio}

            try:
                retCmd = self.__jsonData[useJsonFile][useCmd]
            except:
                # コマンドが存在しない場合
                retCmd = None
            # 命令解析開始メッセージの場合
            if useState == self.__JuliusConfig['wakeUp']:
                self.setJuliusState(True)
                local_tmpVal = True
                self.__prevWordIdx = ''
            elif self.__state['julius']:
                # 命令解析終了メッセージの場合
                if useState == self.__JuliusConfig['sleep']:
                    self.setJuliusState(False)
                    local_tmpVal = True
                    self.__prevWordIdx = ''
                # Julius の初期化要求の場合
                elif useState == self.__JuliusConfig['init']:
                    # 状態を初期化
                    self.__initStat(['julius'])
                    local_tmpVal = True
                    self.__prevWordIdx = ''
                # 天気予報読み上げ要求の場合
                elif useState == self.__JuliusConfig['weather']:
                    local_tmpVal = 'weather'
                    self.__prevWordIdx = ''
                # 無効値でない場合
                elif useState != self.__JuliusConfig['invalid']:
                    # 入力コマンドの動作条件を確認し,コマンドを実行するか決める
                    upperInfo = (useState & self.__maskDict['upper'])
                    lowerInfo = (useState & self.__maskDict['lower'])

                    if upperInfo == self.__JuliusConfig['active']:
                        # [期待値]対象機器が動作中
                        local_tmpVal = self.__state[useMachine]
                    elif upperInfo == self.__JuliusConfig['stopped']:
                        # [期待値]対象機器が停止中
                        local_tmpVal = not self.__state[useMachine]
                    else:
                        # [期待値]なし で初期化
                        local_tmpVal = True

                    # コマンドを実行する場合
                    if local_tmpVal:
                        # 現在の入力単語を記録
                        self.__prevWordIdx = wordIdx
                        # 下位 4 ビットを確認し処理する
                        if lowerInfo == self.__JuliusConfig['update']:
                            # 状態を反転
                            self.__state[useMachine] = not self.__state[useMachine]
                        elif lowerInfo == self.__JuliusConfig['delPrevWord']:
                            # 記録した単語を削除
                            self.__prevWordIdx = ''
            retAudio = useAudio[local_tmpVal]
            returnState = self.juliusReturnState[local_tmpVal]
        return returnState, retCmd, retAudio

自分が利用している「confParam.json」ファイルを以下に示す.

{
    "lightOn": {
        "cmd": "all_light",
        "machine": "light",
        "json": "lightData.json",
        "voice": "lightOn.wav",
        "chkState": "0x81"
    },
    "lightOff": {
        "cmd": "light_off",
        "machine": "light",
        "json": "lightData.json",
        "voice": "lightOff.wav",
        "chkState": "0x11"
    },
    "lightUp": {
        "cmd": "up",
        "machine": "light",
        "json": "lightData.json",
        "voice": "lightUp.wav",
        "chkState": "0x12"
    },
    "lightDown": {
        "cmd": "down",
        "machine": "light",
        "json": "lightData.json",
        "voice": "lightDown.wav",
        "chkState": "0x12"
    },
    "nightLight": {
        "cmd": "night_light",
        "machine": "light",
        "json": "lightData.json",
        "voice": "nightLight.wav",
        "chkState": "0x11"
    },
    "TVOn": {
        "cmd": "power",
        "machine": "tv",
        "json": "TVData.json",
        "voice": "TVOn.wav",
        "chkState": "0x81"
    },
    "TVOff": {
        "cmd": "power",
        "machine": "tv",
        "json": "TVData.json",
        "voice": "TVOff.wav",
        "chkState": "0x11"
    }
}

長くなってしまったため,main 関数部分と解析結果部分に関しては,次回にまわす.main 関数部分が軸となって動作するため,これまでの関数を利用してスレッドを立ち上げつつ動作する.ここまでのディレクトリ構成を以下に示す.

~/juliusKit
   |--dictationKit_v4.3.1
       |--word.dic
       |--word.jconf
   |--grammarKit
   |   |--controller
   |        |--compile.sh
   |        |--mkdfa.pl
   |        |--mkfa
   |        |--utf8_controller.grammar
   |        |--utf8_controller.voca
   |--outYomi.sh
   |--word.yomi
   |--VR_ConstClass.py
   |--parseJuliusData.py
   |--webSocketClient.py
   |--configuration.py    # 追加部分
   |--jsonData            # 追加部分
   |   |--TVData.json
   |   |--confParam.json
   |   |--lightData.json
   |--wavFile             # 追加部分
         |--TVOff.wav
         |--TVOn.wav
         |--lightDown.wav
         |--lightOff.wav
         |--lightOn.wav
         |--lightUp.wav
         |--nightLight.wav