Made code (including access to raw data) work with v1.0 firmware.

myo.py

@@ -83,19 +83,29 @@ def emg_handler(self, emg, moving):
 
         r, n = self.history_cnt.most_common(1)[0]
         if self.last_pose is None or (n > self.history_cnt[self.last_pose] + 5 and n > Myo.HIST_LEN / 2):
-            self.proc_pose(r)
+            self.on_raw_pose(r)
             self.last_pose = r
 
-    def add_pose_handler(self, h):
+    def add_raw_pose_handler(self, h):
         self.pose_handlers.append(h)
 
-    def proc_pose(self, pose):
+    def on_raw_pose(self, pose):
         for h in self.pose_handlers:
             h(pose)
 
 if __name__ == '__main__':
+    import subprocess
     m = Myo(NNClassifier(), sys.argv[1] if len(sys.argv) >= 2 else None)
-    m.add_pose_handler(print)
+    m.add_raw_pose_handler(print)
+
+    def page(pose):
+        if pose == 5:
+            subprocess.call(['xte', 'key Page_Down'])
+        elif pose == 6:
+            subprocess.call(['xte', 'key Page_Up'])
+
+    m.add_raw_pose_handler(page)
+
     m.connect()
 
     while True:

myo_raw.py

@@ -1,5 +1,6 @@
 from __future__ import print_function
 
+import enum
 import re
 import struct
 import sys
@@ -23,6 +24,25 @@ def multiord(b):
     else:
         return map(ord, b)
 
+class Arm(enum.Enum):
+    UNKNOWN = 0
+    RIGHT = 1
+    LEFT = 2
+
+class XDirection(enum.Enum):
+    UNKNOWN = 0
+    X_TOWARD_WRIST = 1
+    X_TOWARD_ELBOW = 2
+
+class Pose(enum.Enum):
+    REST = 0
+    FIST = 1
+    WAVE_IN = 2
+    WAVE_OUT = 3
+    FINGERS_SPREAD = 4
+    THUMB_TO_PINKY = 5
+    UNKNOWN = 255
+
 
 class Packet(object):
     def __init__(self, ords):
@@ -157,8 +177,11 @@ def __init__(self, tty=None):
             raise ValueError('Myo dongle not found!')
 
         self.bt = BT(tty)
+        self.conn = None
         self.emg_handlers = []
         self.imu_handlers = []
+        self.arm_handlers = []
+        self.pose_handlers = []
 
     def detect_tty(self):
         for p in comports():
@@ -168,8 +191,8 @@ def detect_tty(self):
 
         return None
 
-    def run(self):
-        self.bt.recv_packet()
+    def run(self, timeout=None):
+        self.bt.recv_packet(timeout)
 
     def connect(self):
         ## stop everything from before
@@ -184,7 +207,8 @@ def connect(self):
         while True:
             p = self.bt.recv_packet()
             print('scan response:', p)
-            if p.payload[15:] == b'\x06\x42\x48\x12\x4A\x7F\x2C\x48\x47\xB9\xDE\x04\xA9\x01\x00\x06\xD5':
+
+            if p.payload.endswith(b'\x06\x42\x48\x12\x4A\x7F\x2C\x48\x47\xB9\xDE\x04\xA9\x01\x00\x06\xD5'):
                 addr = list(multiord(p.payload[2:8]))
                 break
         self.bt.end_scan()
@@ -195,40 +219,57 @@ def connect(self):
         self.bt.wait_event(3, 0)
 
         ## get firmware version
-        fw = self.bt.read_attr(self.conn, 0x17)
+        fw = self.read_attr(0x17)
         _, _, _, _, v0, v1, v2, v3 = unpack('BHBBHHHH', fw.payload)
         print('firmware version: %d.%d.%d.%d' % (v0, v1, v2, v3))
 
-        ## don't know what these do; Myo Connect sends them, though we get data
-        ## fine without them
-        self.bt.write_attr(self.conn, 0x19, b'\x01\x02\x00\x00')
-        self.bt.write_attr(self.conn, 0x2f, b'\x01\x00')
-        self.bt.write_attr(self.conn, 0x2c, b'\x01\x00')
-        self.bt.write_attr(self.conn, 0x32, b'\x01\x00')
-        self.bt.write_attr(self.conn, 0x35, b'\x01\x00')
-
-        ## enable EMG data
-        self.bt.write_attr(self.conn, 0x28, b'\x01\x00')
-        ## enable IMU data
-        self.bt.write_attr(self.conn, 0x1d, b'\x01\x00')
-
-        ## Sampling rate of the underlying EMG sensor, capped to 1000. If it's
-        ## less than 1000, emg_hz is correct. If it is greater, the actual
-        ## framerate starts dropping inversely. Also, if this is much less than
-        ## 1000, EMG data becomes slower to respond to changes. In conclusion,
-        ## 1000 is probably a good value.
-        C = 1000
-        emg_hz = 50
-        ## strength of low-pass filtering of EMG data
-        emg_smooth = 100
-
-        imu_hz = 50
-
-        ## send sensor parameters, or we don't get any data
-        self.bt.write_attr(self.conn, 0x19, pack('BBBBHBBBBB', 2, 9, 2, 1, C, emg_smooth, C // emg_hz, imu_hz, 0, 0))
+        self.old = (v0 == 0)
+
+        if self.old:
+            ## don't know what these do; Myo Connect sends them, though we get data
+            ## fine without them
+            self.write_attr(0x19, b'\x01\x02\x00\x00')
+            self.write_attr(0x2f, b'\x01\x00')
+            self.write_attr(0x2c, b'\x01\x00')
+            self.write_attr(0x32, b'\x01\x00')
+            self.write_attr(0x35, b'\x01\x00')
+
+            ## enable EMG data
+            self.write_attr(0x28, b'\x01\x00')
+            ## enable IMU data
+            self.write_attr(0x1d, b'\x01\x00')
+
+            ## Sampling rate of the underlying EMG sensor, capped to 1000. If it's
+            ## less than 1000, emg_hz is correct. If it is greater, the actual
+            ## framerate starts dropping inversely. Also, if this is much less than
+            ## 1000, EMG data becomes slower to respond to changes. In conclusion,
+            ## 1000 is probably a good value.
+            C = 1000
+            emg_hz = 50
+            ## strength of low-pass filtering of EMG data
+            emg_smooth = 100
+
+            imu_hz = 50
+
+            ## send sensor parameters, or we don't get any data
+            self.write_attr(0x19, pack('BBBBHBBBBB', 2, 9, 2, 1, C, emg_smooth, C // emg_hz, imu_hz, 0, 0))
+
+        else:
+            name = self.read_attr(0x03)
+            print('device name: %s' % name.payload)
+
+            ## enable IMU data
+            self.write_attr(0x1d, b'\x01\x00')
+            ## enable on/off arm notifications
+            self.write_attr(0x24, b'\x02\x00')
+
+            # self.write_attr(0x19, b'\x01\x03\x00\x01\x01')
+            self.start_raw()
 
         ## add data handlers
         def handle_data(p):
+            if (p.cls, p.cmd) != (4, 5): return
+
             c, attr, typ = unpack('BHB', p.payload[:4])
             pay = p.payload[5:]
 
@@ -239,25 +280,92 @@ def handle_data(p):
                 ## something
                 emg = vals[:8]
                 moving = vals[8]
-                self.proc_emg(emg, moving)
-            if attr == 0x1c:
+                self.on_emg(emg, moving)
+            elif attr == 0x1c:
                 vals = unpack('10h', pay)
                 quat = vals[:4]
                 acc = vals[4:7]
                 gyro = vals[7:10]
-                self.proc_imu(quat, acc, gyro)
+                self.on_imu(quat, acc, gyro)
+            elif attr == 0x23:
+                typ, val, xdir = unpack('3B', pay)
+
+                if typ == 1: # on arm
+                    self.on_arm(Arm(val), XDirection(xdir))
+                elif typ == 2: # removed from arm
+                    self.on_arm(Arm.UNKNOWN, XDirection.UNKNOWN)
+                elif typ == 3: # pose
+                    self.on_pose(Pose(val))
+            else:
+                print('data with unknown attr: %02X %s' % (attr, p))
 
         self.bt.add_handler(handle_data)
 
 
-    def disconnect(self):
-        self.bt.disconnect(self.conn)
+    def write_attr(self, attr, val):
+        if self.conn is not None:
+            self.bt.write_attr(self.conn, attr, val)
 
+    def read_attr(self, attr):
+        if self.conn is not None:
+            return self.bt.read_attr(self.conn, attr)
+        return None
+
+    def disconnect(self):
+        if self.conn is not None:
+            self.bt.disconnect(self.conn)
+
+    def start_raw(self):
+        '''Sending this sequence for v1.0 firmware seems to enable both raw data and
+        pose notifications.
+        '''
+
+        self.write_attr(0x28, b'\x01\x00')
+        self.write_attr(0x19, b'\x01\x03\x01\x01\x00')
+        self.write_attr(0x19, b'\x01\x03\x01\x01\x01')
+
+    def mc_start_collection(self):
+        '''Myo Connect sends this sequence (or a reordering) when starting data
+        collection for v1.0 firmware; this enables raw data but disables arm and
+        pose notifications.
+        '''
+
+        self.write_attr(0x28, b'\x01\x00')
+        self.write_attr(0x1d, b'\x01\x00')
+        self.write_attr(0x24, b'\x02\x00')
+        self.write_attr(0x19, b'\x01\x03\x01\x01\x01')
+        self.write_attr(0x28, b'\x01\x00')
+        self.write_attr(0x1d, b'\x01\x00')
+        self.write_attr(0x19, b'\x09\x01\x01\x00\x00')
+        self.write_attr(0x1d, b'\x01\x00')
+        self.write_attr(0x19, b'\x01\x03\x00\x01\x00')
+        self.write_attr(0x28, b'\x01\x00')
+        self.write_attr(0x1d, b'\x01\x00')
+        self.write_attr(0x19, b'\x01\x03\x01\x01\x00')
+
+    def mc_end_collection(self):
+        '''Myo Connect sends this sequence (or a reordering) when ending data collection
+        for v1.0 firmware; this reenables arm and pose notifications, but
+        doesn't disable raw data.
+        '''
+
+        self.write_attr(0x28, b'\x01\x00')
+        self.write_attr(0x1d, b'\x01\x00')
+        self.write_attr(0x24, b'\x02\x00')
+        self.write_attr(0x19, b'\x01\x03\x01\x01\x01')
+        self.write_attr(0x19, b'\x09\x01\x00\x00\x00')
+        self.write_attr(0x1d, b'\x01\x00')
+        self.write_attr(0x24, b'\x02\x00')
+        self.write_attr(0x19, b'\x01\x03\x00\x01\x01')
+        self.write_attr(0x28, b'\x01\x00')
+        self.write_attr(0x1d, b'\x01\x00')
+        self.write_attr(0x24, b'\x02\x00')
+        self.write_attr(0x19, b'\x01\x03\x01\x01\x01')
 
     def vibrate(self, length):
         if length in xrange(1, 4):
             ## first byte tells it to vibrate; purpose of second byte is unknown
-            self.bt.write_attr(self.conn, 0x19, pack('3B', 3, 1, length))
+            self.write_attr(0x19, pack('3B', 3, 1, length))
 
 
     def add_emg_handler(self, h):
@@ -266,14 +374,29 @@ def add_emg_handler(self, h):
     def add_imu_handler(self, h):
         self.imu_handlers.append(h)
 
-    def proc_emg(self, emg, moving):
+    def add_pose_handler(self, h):
+        self.pose_handlers.append(h)
+
+    def add_arm_handler(self, h):
+        self.arm_handlers.append(h)
+
+
+    def on_emg(self, emg, moving):
         for h in self.emg_handlers:
             h(emg, moving)
 
-    def proc_imu(self, quat, acc, gyro):
+    def on_imu(self, quat, acc, gyro):
         for h in self.imu_handlers:
             h(quat, acc, gyro)
 
+    def on_pose(self, p):
+        for h in self.pose_handlers:
+            h(p)
+
+    def on_arm(self, arm, xdir):
+        for h in self.arm_handlers:
+            h(arm, xdir)
+
 
 if __name__ == '__main__':
     try:
@@ -332,9 +455,12 @@ def proc_emg(emg, moving, times=[]):
     m.add_emg_handler(proc_emg)
     m.connect()
 
+    m.add_arm_handler(lambda arm, xdir: print('arm', arm, 'xdir', xdir))
+    m.add_pose_handler(lambda p: print('pose', p))
+
     try:
         while True:
-            m.run()
+            m.run(1)
 
             if HAVE_PYGAME:
                 for ev in pygame.event.get():