

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

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

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

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

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

[code]

    ValueError: attempt to get argmax of an empty sequence
    
[/code]

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

[code]

    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()
    
[/code]

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

変えた部分

[code]

    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()
    
[/code]

また，以下はmyoのデータが取得できるコードです．（確認済みです）

[code]

    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)
    
[/code]


CC BY-SA 4.0
2020-08-27T08:47:35.287
69919
2023-04-19T17:05:25.653
2021-12-31T10:40:55.330
3060
41671
question
python tensorflow numpy
attempt to get argmax of an empty sequence argmaxの値が空のままで，実行できない状態です．


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

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

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

  5. `response = np.argmax(np.bincount(temp))`の処理は`start`から1秒経過後に呼ばれる。  
1度呼ばれると`start`が再設定されるので、常に1秒以上の間隔で呼ばれる。  
そして呼ばれて処理が終わると`temp`は空に初期化される

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

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

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

* * *

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


CC BY-SA 4.0
2020-08-31T09:34:02.860
70014
2020-08-31T12:17:45.143
2020-08-31T12:17:45.143
26370
26370
69919
answer