Refactoring code from original POC phase - integers have been decoded roughly successfully so its time to start building a huge library for this bad boy

collect_data.py

@@ -0,0 +1,33 @@
+
+import asyncio
+import json
+import platform
+
+"""
+This is used for reading and decoding values from t he Kano Harry Potter Coding Wand
+- Wand sends 25 updates to gyro and accel every second
+- Wand sends 3 Dimensional data as 48 bit reverse marshalled integers
+- Wand seems to send smaller single dimensional data in 16bit unsigned integers 
+"""
+
+SPELLS = json.load("spells.json")
+CURR_SPELL = 0
+
+# TODO: RUMBLE
+# TODO: RGB
+
+device_address = "D8:9B:12:D1:08:80"
+
+
+
+if __name__ == "__main__":
+    import os
+
+    os.environ["PYTHONASYNCIODEBUG"] = str(1)
+    address = (
+        device_address  # <--- Change to your device's address here if you are using Windows or Linux
+        if platform.system() != "Darwin"
+        else "243E23AE-4A99-406C-B317-18F1BD7B4CBE"  # <--- Change to your device's address here if you are using macOS
+    )
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(run(address, loop, True))

kano_wand/ble_client.py

@@ -0,0 +1,64 @@
+import logging
+import struct
+
+from converters import BinToInt, BinToFloat
+from .sensor_tracker import *
+from bleak import BleakClient
+from bleak import _logger as logger
+from .utils import *
+
+class KanoBLEClient(object):
+    def __init__(self, wand_address, spells=None):
+        self.wand_address = wand_address
+        self.int_decoder = BinToInt(48)
+        self.float_decoder = BinToFloat(16, 31)
+        self.client = None
+
+
+    async def connect(self, loop, debug=False):
+        if debug:
+            import sys
+            l = logging.getLogger("asyncio")
+            l.setLevel(logging.DEBUG)
+            h = logging.StreamHandler(sys.stdout)
+            h.setLevel(logging.DEBUG)
+            l.addHandler(h)
+            logger.addHandler(h)
+
+        async with BleakClient(self.wand_address, loop=loop) as client:
+            self.client = client
+            x = await client.is_connected()
+            logger.info("Connected: {0}".format(x))
+
+    async def start_recieving_data(self):
+        await start_notify(self.client)
+
+
+    async def stop_recieving_data(self):
+        await stop_notify(self.client)
+
+
+    def sensor_handling(self, sender, data):
+        """Simple notification handler which prints the data received."""
+        # TODO: Convert to a dictionary and a single decode command
+        sender = CHARACTERISTIC_UUIDS[sender]
+        if sender == BUTTON:
+            self.decode_button(data)
+        elif sender == NINE_AXIS:
+            #TODO: refactor the gyro to be local to the sensors file only
+            gyro = Gyro()
+            gyro.x, gyro.y, gyro.z = self.decode_nine_axis(data)
+            sensors.set_gyro(gyro)
+            sensors.append_to_dataframe()
+        elif sender == ACCELEROMETER:
+            accel = Accel()
+            accel.x, accel.y, accel.z = self.decode_nine_axis(data)
+            sensors.set_accel(accel)
+            sensors.append_to_dataframe()
+        elif sender == BATTERY:
+            self.decode_battery(data)
+        elif sender == TEMPERATURE:
+            self.decode_temp(data)
+        elif sender == MAGNETOMETER:
+            self.decode_magnet(data)
+

kano_wand/constants.py

@@ -0,0 +1,15 @@
+BUTTON = 1
+NINE_AXIS = 2
+ACCELEROMETER = 3
+BATTERY = 4
+TEMPERATURE = 5
+MAGNETOMETER = 6
+
+CHARACTERISTIC_UUIDS = {
+    ("64a7000d-f691-4b93-a6f4-0968f5b648f8"): BUTTON,  # Button
+    ("64a7000a-f691-4b93-a6f4-0968f5b648f8"): NINE_AXIS,  # 9 axis
+    ("64a7000c-f691-4b93-a6f4-0968f5b648f8"): ACCELEROMETER,  # Accel
+    ("64a70007-f691-4b93-a6f4-0968f5b648f8"): BATTERY,
+    ("64a70014-f691-4b93-a6f4-0968f5b648f8"): TEMPERATURE,
+    ("64a70021-f691-4b93-a6f4-0968f5b648f8"):MAGNETOMETER
+}  # <--- Change to the characteristic you want to enable notifications from.

kano_wand/sensor_decoders.py

@@ -0,0 +1,33 @@
+import struct
+from converters import BinToInt, BinToFloat
+
+class decoder(object):
+    def __init__(self, int_decoder=BinToInt(48), float_decoder= BinToFloat(16, 31)):
+        self.int_decoder = int_decoder
+        self.float_decoder = float_decoder
+
+    def decode_button(self, data):
+        if int.from_bytes(data, byteorder='big'):
+            print("Button Pressed")
+        else:
+            print("Button Released")
+
+    def decode_nine_axis(self, data):
+        converted = [self.int_decoder.process(x, True) / 10 ** 14 for x in chunker(data, 6)]
+        return converted
+
+    def decode_accelerometer(self, data):
+        converted = [self.int_decoder.process(x, True) / 10 ** 14 for x in chunker(data, 6)]
+        return converted
+
+    def decode_battery(self, data):
+        int.from_bytes(data, byteorder='big')
+
+    def decode_temp(self, data):
+        print(len(data))
+        print(data)
+        print(struct.unpack("h", data))
+
+    def decode_magnet(self, data):
+        print(data)
+        print(len(data))

kano_wand/sensor_tracker.py

@@ -0,0 +1,54 @@
+import pandas as pd
+
+class Gyro(object):
+    def __init__(self):
+        self.x = 0
+        self.y = 0
+        self.z = 0
+
+    def __str__(self):
+        print(self.x, self.y, self.z)
+
+
+class Accel(object):
+    def __init__(self):
+        self.x = 0
+        self.y = 0
+        self.z = 0
+
+    def __str__(self):
+        print(self.x, self.y, self.z)
+
+
+class SensorTracker(object):
+    def __init__(self):
+        self.gyro = None
+        self.gyro_updates = 0
+
+        self.accel = None
+        self.accel_updates = 0
+        self.dataframe = pd.DataFrame(columns=["gyro_x", "gyro_y", "gyro_z", "accel_x", "accel_y", "accel_z"])
+
+    def append_to_dataframe(self):
+        if self.gyro_updates == self.accel_updates:
+            self.dataframe = self.dataframe.append(
+                pd.DataFrame({
+                    "gyro_x": self.gyro.x,
+                    "gyro_y": self.gyro.y,
+                    "gyro_z": self.gyro.z,
+                    "accel_x": self.accel.x,
+                    "accel_y": self.accel.y,
+                    "accel_z": self.accel.z,
+                }, index=[0])
+            )
+
+    def save_df_to_file(self, filename):
+        self.dataframe.to_csv(filename, index=False)
+
+    def set_gyro(self, gyro):
+        self.gyro = gyro
+        self.gyro_updates += 1
+
+    def set_accel(self, accel):
+        self.accel = accel
+        self.accel_updates += 1

kano_wand/testing/discover_and_probe_characteristics.py

@@ -1,21 +1,21 @@
 import asyncio
-from bleak import discover
+from bleak import discover, BleakClient
 from pprint import pprint
 
+
 async def run():
     devices = await discover()
     for d in devices:
         print(d)
 
+
 loop = asyncio.get_event_loop()
 loop.run_until_complete(run())
 
-import asyncio
-from bleak import BleakClient
-
 address = "D8:9B:12:D1:08:80"
 MODEL_NBR_UUID = "64a7000f-f691-4b93-a6f4-0968f5b648f8"
 
+
 async def run(address, loop):
     async with BleakClient(address, loop=loop) as client:
         # await client.connect()
@@ -53,4 +53,4 @@ async def run(address, loop):
 
 
 loop = asyncio.get_event_loop()
-loop.run_until_complete(run(address, loop))
+loop.run_until_complete(run(address, loop))

kano_wand/utils.py

@@ -0,0 +1,13 @@
+from kano_wand.constants import *
+
+async def start_notify(client, notification_handler):
+    for char in CHARACTERISTIC_UUIDS.keys():
+        await client.start_notify(char, notification_handler)
+
+
+async def stop_notify(client):
+    for char in CHARACTERISTIC_UUIDS.keys():
+        await client.stop_notify(char)
+
+def chunker(seq, size):
+    return (seq[pos:pos + size] for pos in range(0, len(seq), size))

spells.json

@@ -0,0 +1,6 @@
+{
+  "0": "unknown",
+  "1": "lumos",
+  "2": "nox",
+  "3": "accio"
+}

subscribe_to_notification.py

@@ -1,241 +0,0 @@
-import logging
-import asyncio
-import platform
-import struct
-import time
-import pandas as pd
-
-from bleak import BleakClient
-from bleak import _logger as logger
-from converters import BinToInt, BinToFloat
-
-"""
-This is used for reading and decoding values from t he Kano Harry Potter Coding Wand
-- Wand sends 25 updates to gyro and accel every second
-- Wand sends 3 Dimensional data as 48 bit reverse marshalled integers
-- Wand seems to send smaller single dimensional data in 16bit unsigned integers 
-"""
-
-BUTTON = 1
-NINE_AXIS = 2
-ACCELEROMETER = 3
-BATTERY = 4
-TEMPERATURE = 5
-MAGNETOMETER = 6
-
-INT_DECODER = BinToInt(48)
-FLOAT_DECODER = BinToFloat(16, 31)
-BUTTON_STATUS = None
-
-CHARACTERISTIC_UUIDS = {
-    ("64a7000d-f691-4b93-a6f4-0968f5b648f8"): BUTTON,  # Button
-    ("64a7000a-f691-4b93-a6f4-0968f5b648f8"): NINE_AXIS,  # 9 axis
-    ("64a7000c-f691-4b93-a6f4-0968f5b648f8"): ACCELEROMETER,  # Accel
-    # ("64a70007-f691-4b93-a6f4-0968f5b648f8"):BATTERY,
-    # ("64a70014-f691-4b93-a6f4-0968f5b648f8"):TEMPERATURE,
-    # ("64a70014-f691-4b93-a6f4-0968f5b648f8"):MAGNETOMETER
-}  # <--- Change to the characteristic you want to enable notifications from.
-
-# TODO: RUMBLE
-# TODO: RGB
-
-device_address = "D8:9B:12:D1:08:80"
-
-
-class Gyro(object):
-    def __init__(self):
-        self.x = 0
-        self.y = 0
-        self.z = 0
-
-    def __str__(self):
-        print(self.x, self.y, self.z)
-
-
-class Accel(object):
-    def __init__(self):
-        self.x = 0
-        self.y = 0
-        self.z = 0
-
-    def __str__(self):
-        print(self.x, self.y, self.z)
-
-
-class Sensors(object):
-    def __init__(self):
-        self.gyro = None
-        self.gyro_updates = 0
-
-        self.accel = None
-        self.accel_updates = 0
-        self.dataframe = pd.DataFrame(columns=["gyro_x", "gyro_y", "gyro_z", "accel_x", "accel_y", "accel_z"])
-
-    def append_to_dataframe(self):
-        # print("Printing the gyro and accel", self.gyro, self.accel)
-        # print(self.dataframe)
-        # print(self.gyro_updates, self.accel_updates)
-        if self.gyro_updates == self.accel_updates:
-            # print({
-            #     "gyro_x": self.gyro.x,
-            #     "gyro_y": self.gyro.y,
-            #     "gyro_z": self.gyro.z,
-            #     "accel_x": self.accel.x,
-            #     "accel_y": self.accel.y,
-            #     "accel_z": self.accel.z,
-            # })
-            # try:
-            #
-            # except Exception as e:
-            #     print("ERROR HAS OCCURRED")
-            #     print(e)
-            self.dataframe = self.dataframe.append(
-                pd.DataFrame({
-                    "gyro_x": self.gyro.x,
-                    "gyro_y": self.gyro.y,
-                    "gyro_z": self.gyro.z,
-                    "accel_x": self.accel.x,
-                    "accel_y": self.accel.y,
-                    "accel_z": self.accel.z,
-                }, index=[0])
-            )
-
-    def save_df_to_file(self, filename):
-        self.dataframe.to_csv(filename, index=False)
-
-    def set_gyro(self, gyro):
-        self.gyro = gyro
-        self.gyro_updates += 1
-
-    def set_accel(self, accel):
-        self.accel = accel
-        self.accel_updates += 1
-
-
-def chunker(seq, size):
-    return (seq[pos:pos + size] for pos in range(0, len(seq), size))
-
-
-count = 0
-
-
-def decode_button(data):
-    global sensors, count
-    if int.from_bytes(data, byteorder='big'):
-        print("Button Pressed")
-        sensors = Sensors()
-        count += 1
-    else:
-        print("Button Released")
-        sensors.save_df_to_file(f"dataset{count}.csv")
-
-
-def decode_nine_axis(data):
-    converted = [INT_DECODER.process(x, True) / 10 ** 14 for x in chunker(data, 6)]
-
-    # print(converted, sum(converted))
-    return converted
-
-
-def decode_accelerometer(data):
-    converted = [INT_DECODER.process(x, True) / 10 ** 14 for x in chunker(data, 6)]
-    # print(converted, sum(converted))
-    return converted
-
-
-def decode_battery(data):
-    global BUTTON_STATUS
-    BUTTON_STATUS = int.from_bytes(data, byteorder='big')
-
-
-def decode_temp(data):
-    print(len(data))
-    print(data)
-    print(struct.unpack("h", data))
-
-
-def decode_magnet(data):
-    print(data)
-    print(len(data))
-
-
-count = 0
-
-sensors = Sensors()
-
-
-def notification_handler(sender, data):
-    global count, sensors
-    """Simple notification handler which prints the data received."""
-    # TODO: Convert to a dictionary and a single decode command
-    sender = CHARACTERISTIC_UUIDS[sender]
-    if sender == BUTTON:
-        decode_button(data)
-    elif sender == NINE_AXIS:
-        gyro = Gyro()
-        # print("getting gyro values")
-
-        gyro.x, gyro.y, gyro.z = decode_nine_axis(data)
-        # print("got gyro values")
-        # print(gyro)
-        sensors.set_gyro(gyro)
-        sensors.append_to_dataframe()
-    elif sender == ACCELEROMETER:
-        accel = Accel()
-        accel.x, accel.y, accel.z = decode_nine_axis(data)
-        sensors.set_accel(accel)
-        sensors.append_to_dataframe()
-        # decode_accelerometer(data)
-    elif sender == BATTERY:
-        decode_battery(data)
-    elif sender == TEMPERATURE:
-        decode_temp(data)
-    elif sender == MAGNETOMETER:
-        decode_magnet(data)
-
-
-async def start_notify(client):
-    for char in CHARACTERISTIC_UUIDS.keys():
-        await client.start_notify(char, notification_handler)
-
-
-async def stop_notify(client):
-    for char in CHARACTERISTIC_UUIDS.keys():
-        await client.stop_notify(char)
-
-
-async def run(address, loop, debug=False):
-    if debug:
-        import sys
-
-        # loop.set_debug(True)
-        l = logging.getLogger("asyncio")
-        l.setLevel(logging.DEBUG)
-        h = logging.StreamHandler(sys.stdout)
-        h.setLevel(logging.DEBUG)
-        l.addHandler(h)
-        logger.addHandler(h)
-
-    async with BleakClient(address, loop=loop) as client:
-        x = await client.is_connected()
-        logger.info("Connected: {0}".format(x))
-        await start_notify(client)
-
-        # await asyncio.sleep(60.0, loop=loop)
-        while True:
-            time.sleep(1)
-
-        await stop_notify(client)
-
-
-if __name__ == "__main__":
-    import os
-
-    os.environ["PYTHONASYNCIODEBUG"] = str(1)
-    address = (
-        device_address  # <--- Change to your device's address here if you are using Windows or Linux
-        if platform.system() != "Darwin"
-        else "243E23AE-4A99-406C-B317-18F1BD7B4CBE"  # <--- Change to your device's address here if you are using macOS
-    )
-    loop = asyncio.get_event_loop()
-    loop.run_until_complete(run(address, loop, True))