0

現在,Pythonのtensorflowでprdict.pyという未知の値を予測するためのart neural netのコードを書いています.
argmax の値が空というエラーが出ており,行列値が取得できません.

直接,値を入れたりとしているのですが(temp = [1, 1, 2]のように),違うような感じがします.下にコードをを記載させていただきますが、temp = [] という部分がおそらくエラーの原因ではないかと考えております.

また,参考に致しましたソースコードは以下のリンクのものです.(predict.py)

https://github.com/exelban/myo-armband-nn

エラー文は以下のようなものです

ValueError: attempt to get argmax of an empty sequence

よろしければ,ぜひご教示頂きたいです.何卒,よろしく,お願いいたします.

from collections import deque
#import collections
import myo
from threading import Lock
import time
import numpy as np
import tensorflow as tf
from include.model import model

x, y, output, global_step, y_pred_cls = model()

saver = tf.train.Saver()
_SAVE_PATH = "./data/tensorflow_sessions/myo_armband/"
sess = tf.Session()

try:
    print("Trying to restore last checkpoint ...")
    last_chk_path = tf.train.latest_checkpoint(checkpoint_dir=_SAVE_PATH)
    print(last_chk_path)
    saver.restore(sess, save_path=last_chk_path)
    print("Restored checkpoint from:", last_chk_path)
except:
    print("Failed to restore checkpoint. Initializing variables instead.")
    sess.run(tf.global_variables_initializer())

class ApiDeviceListener(myo.DeviceListener):

    def __init__(self, queue_size=8):
        #self.lock = threading.Lock()
        self.lock = Lock()
        #self.emg_data_queue = collections.deque(maxlen=queue_size)
        self.emg_data_queue = deque(maxlen=queue_size)
        self.orientation_data_queue = deque(maxlen=queue_size)
        self.acceleration_data_queue = deque(maxlen=queue_size)
        self.gyroscope_data_queue = deque(maxlen=queue_size)
        self.rssi_data_queue = deque(maxlen=100)
    def on_connect(self, device, timestamp, firmware_version, event):
        device.set_stream_emg(myo.StreamEmg.enabled)
        self.device = event.device  # sauvegarde de l'instance au myo arm
        event.device.unlock()  # demande de desappareiller
        event.device.lock()  # demande d'appareiller (génère des vibrations)
        event.device.stream_emg(True)  # lance l'acquisition des emg
        self.connected = True  # mise à jour du flag de connection du myo
        self.device_name = event.device_name  # on récupère le petit nom du myo
        # on récupère également le numéro du firmware (non exploité dans l'UI)
        self.myo_firmware = '.'.join(map(str, event.firmware_version[:-1]))
    def on_emg_data(self, device, timestamp, emg_data, event):
        with self.lock:
            self.emg_data_queue.append((timestamp, emg_data))
        with self.lock:
            self.emg_data_queue.append((event.timestamp,
                                        event.emg))
    def get_emg_data(self):
        with self.lock:
            return list(self.emg_data_queue)
        with self.lock:
            return list(self.emg_data_queue)
    
myo.init(bin_path=r'C:\Users\田澤龍之介\Desktop\myo-sdk-win-0.9.0\bin')
hub = myo.Hub()
start = time.time()
listener = ApiDeviceListener()
temp = []         ## ☚ ここではないかと思います.
    
try:
    listener = ApiDeviceListener()
    hub.run(listener, 2000)
    while True:
        data = listener.get_emg_data()
        if time.time() - start >= 1:
            response = np.argmax(np.bincount(temp))
            print("Predicted gesture: {0}".format(response))
            temp = []
            start = time.time()
        if len(data) > 0:
            tmp = []
            for v in listener.get_emg_data():
                tmp.append(v[1])
            tmp = list(np.stack(tmp).flatten())
            if len(tmp) >= 64:
                pred = sess.run(y_pred_cls, feed_dict={x: np.array([tmp])})
                temp.append(pred[0])
        time.sleep(0.01)
finally:
    #hub.shutdown()
    sess.close()

以前に解答していただき,1秒間隔や行列値の個数(64個以上の)の部分を変えたのですが,依然と同じエラーがでます.条件の替え方あるいは他の部分の問題でしょうか?
以下に解答してい頂いた後に実践したこととmyoの値を習得できるコードを添付いたします.取得できるコードを入れてみたりましたが,いい方法はないでしょうか?何度も,大変申し訳ありません.よろしくお願いいたします.

変えた部分

try:
    listener = ApiDeviceListener()
    hub.run(listener, 2000)
    while True:
        data = listener.get_emg_data()
        if time.time() - start >= 10:                 #1
            response = np.argmax(np.bincount(temp))
            print("Predicted gesture: {0}".format(response))
            temp = []
            start = time.time()
        if len(data) > 0:             #0
            tmp = []
            for v in listener.get_emg_data():
                tmp.append(v[1])
            tmp = list(np.stack(tmp).flatten())
            if len(tmp) >= 8:                      #64
                pred = sess.run(y_pred_cls, feed_dict={x: np.array([tmp])})
                temp.append(pred[0])
        time.sleep(0.01)             # timesleep(0.01)
finally:
    #hub.shutdown()
    sess.close()

また,以下はmyoのデータが取得できるコードです.(確認済みです)

from collections import deque
from threading import Lock
import myo

class MyListener(myo.DeviceListener):
    """
    classe en écoute d'un myo
    """
    def __init__(self, queue_size=8):
        self.lock = Lock()  # verrouille le thread pour lecture des donnees
        # création de listes optimisées pour seulement ajouter des éléments
        self.emg_data_queue = deque(maxlen=queue_size)
        self.orientation_data_queue = deque(maxlen=queue_size)
        self.acceleration_data_queue = deque(maxlen=queue_size)
        self.gyroscope_data_queue = deque(maxlen=queue_size)
        self.rssi_data_queue = deque(maxlen=100)
        # initialisation d'attribut
        self.pose = myo.Pose.rest  # pose quelconque
        self.connected = False  # non connecté
        self.battery_level = 100  # niveau de batterie maximal
        self.emg_enabled = False  # on acquiert pas les EMG
        self.locked = False  # myo non vérouillé
        self.rssi = None  # aucune valeur de force du signal bluetooth
        self.emg = None  # données null des emg
        self.device_name = None  # pas de nom du myo
        self.device = None
        self.myo_firmware = None
        self.arm = None
        self.x_direction = None
        self.sync = None

    def on_paired(self, event):
        """
        méthode appelée si le myo est appareillé
        """
        if __name__ == '__main__':
            print('paired')
        else:
            pass

    def on_unpaired(self, event):
        """
        méthode appelée si le myo n'est plus appareillé
        """
        if __name__ == '__main__':
            print('unpaired')
        else:
            pass

    def on_connected(self, event):
        """
        méthode appelé si le myo est connecté
        """
        self.device = event.device  # sauvegarde de l'instance au myo arm
        event.device.unlock()  # demande de desappareiller
        event.device.lock()  # demande d'appareiller (génère des vibrations)
        event.device.stream_emg(True)  # lance l'acquisition des emg
        self.connected = True  # mise à jour du flag de connection du myo
        self.device_name = event.device_name  # on récupère le petit nom du myo
        # on récupère également le numéro du firmware (non exploité dans l'UI)
        self.myo_firmware = '.'.join(map(str, event.firmware_version[:-1]))

    def on_disconnected(self, event):
        """
        méthode appelée si le myo est déconnecté
        """
        self.connected = False  # flag mis à jour

    def on_arm_synced(self, event):
        """
        méthode appelé si un bras est synchronisé
        pas vraiment compris !!!
        """
        self.sync = True
        self.arm = event.arm  # informe de la latéralité du bras détecté
        # informe de l'orientation du bracelet
        # (vers le poignet ou vers le coude)
        self.x_direction = event.x_direction
        if __name__ == '__main__':
            print(self.x_direction)
        else:
            pass

    def on_arm_unsynced(self, event):
        """
        méthode appelée si le bras est désynchronisé
        pas vraiment compris !!!
        """
        self.sync = False
        if __name__ == '__main__':
            print(f'arm unsynced : {event.arm}')
        else:
            pass

    def on_unlocked(self, event):
        """
        méthode appelée si le myo est dévérouillé
        """
        self.locked = False  # flag mis à jour

    def on_locked(self, event):
        """
        méthode appelée si le myo est vérouillé
        """
        self.locked = True  # flag mis à jour

    def on_pose(self, event):
        """
        méthode appelée dès qu'une pose gestuelle est reconnue

            a) Spread
            b) Fist
            c) Wave in
            d) Wave out
            e) Double Tap
            f) Rest
        """
        self.pose = event.pose  # attribut mis à jour

    def on_orientation(self, event):
        """
        méthode appelée pour récupérer

            a) orientation
            b) gyroscope
            c) accéléromètre
            d) associé à un timestamp
        """
        with self.lock:
            self.orientation_data_queue.append((event.timestamp,
                                                event.orientation))
            self.gyroscope_data_queue.append((event.timestamp,
                                              event.gyroscope))
            self.acceleration_data_queue.append((event.timestamp,
                                                 event.acceleration))

    def on_rssi(self, event):
        """
        méthode appelée suite à la réponse d'une requête "request_rssi()"
        """
        with self.lock:
            # mise à jour de la liste
            self.rssi_data_queue.append(-event.rssi)

    def on_battery_level(self, event):
        """
        méthode appelée dès que le niveau de batterie évolue
        """
        self.battery_level = event.battery_level  # mise à jour de l'attribut

    def on_emg(self, event):
        """
        méthode appelée pour réceptionner les données EMG
        avec son timestamp
        """
        with self.lock:
            self.emg_data_queue.append((event.timestamp,
                                        event.emg))

    def on_warmup_completed(self, event):
        """
        méthode appelée quand le myo arm est "chaud"

        c'est à partir de ce moment que les données sont les plus stables
        mais ça reste à vérifier

        pas vraiment pris en compte dans ce code (à faire évoluer)
        """
        event.device.stream_emg(True)  # lancement de l'acquisition EMG
        self.emg_enabled = True  # mise à jour du flag

    def get_emg_data(self):
        """
        méthode pour récupérer les données EMGs
        """
        with self.lock:
            return list(self.emg_data_queue)

    def get_orientation_data(self):
        """
        méthode pour récupérer les données d'orientation
        """
        with self.lock:
            return list(self.orientation_data_queue)

    def get_gyroscope_data(self):
        """
        méthode pour récupérer les données du gyroscope
        """
        with self.lock:
            return list(self.gyroscope_data_queue)

    def get_accelerometor_data(self):
        """
        méthode pour récupérer les données de l'accéléromètre
        """
        with self.lock:
            return list(self.acceleration_data_queue)



if __name__ == '__main__':
    # permet de tester sans interface graphique
    #import os
    from time import sleep
    myo.init(bin_path=r'C:\Users\ユーザー名\Desktop\myo-sdk-win-0.9.0\bin')
    HUB = myo.Hub()
    LISTENER = MyListener()
    with HUB.run_in_background(LISTENER.on_event):
        while True:
            print(LISTENER.emg_data_queue)
            sleep(0.1)
1

1 件の回答 1

1

原因は、class ApiDeviceListener(myo.DeviceListener):のプログラムが、参考記事のclass MyListener(myo.DeviceListener):の機能・性能、引いてはmyo armbandの機能・性能をシミュレート出来ていないからでしょう。
発生している問題状況は質問では以下のように見えます。

  1. ValueError: attempt to get argmax of an empty sequenceが発生している
  2. argmaxresponse = np.argmax(np.bincount(temp))でのみ呼ばれている
  3. ということはnp.bincount(temp)が0個のデータしかない(つまりtempが空)のだろう
  4. 主な処理の直前でtemp = []に設定している

ここまでが質問者さんの考えたことでしょう。
さらに以下のように原因を遡って考えられます。

  1. response = np.argmax(np.bincount(temp))の処理はstartから1秒経過後に呼ばれる。
    1度呼ばれるとstartが再設定されるので、常に1秒以上の間隔で呼ばれる。
    そして呼ばれて処理が終わるとtempは空に初期化される
  2. 他にtempを設定・変更しているのが1ヵ所あり、whileループの下の方のtemp.append(pred[0])
  3. 上記 6. の処理が呼ばれるのは、別のtmpというリストにデータが64個以上ある時
  4. tmpにデータを詰めているのは、その3行上のfor v in listener.get_emg_data():ループで、listener.get_emg_data()で読み取ったデータをtmp.append(v[1])で追加している
  5. 上記 8. の処理が呼ばれるのは、dataというリストにデータが1個以上ある時
  6. dataにデータを詰めているのは、whileループの最初のdata = listener.get_emg_data()
  7. whileループは最後に0.01秒のディレイがあるので最大でも1秒間に100回まわる(色々な処理が行われるので実際にはそれ以下)
  8. つまり最大1秒100回まわる処理のどこかで0.01秒の間に65個以上(whileの先頭とforの中を合わせて)のデータがlistener.get_emg_data()から通知される必要がある
  9. listener.get_emg_data()は、class ApiDeviceListener(myo.DeviceListener):のメソッドである
  10. 参考記事の元々の処理は、myo armbandからのデータを受信してキューに詰め、それを取り出して通知しているだけ

ということで、質問者さんの作ったclass ApiDeviceListener(myo.DeviceListener):のプログラムで、上記のタイミングと内容・個数に合わせてデータを通知出来るようにしておかないと、問題の状況が発生することになります。

  • 1秒未満の間隔で、何処かの0.01秒の間に、65個以上のデータを用意して通知する

結果や精度を気にせずに、プログラムが一通り動作するかだけを確認するなら、1秒という間隔や、65個以上という数の条件を一時的に緩和してみれば良いでしょう。
(ただし数を変えると他のところで別の何か問題が出るかもしれませんが。その時は頑張って数合わせをしましょう。)

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。