diff --git a/.idea/kano-wand-smarthome.iml b/.idea/kano-wand-smarthome.iml index b9bfdfd..7e7895b 100644 --- a/.idea/kano-wand-smarthome.iml +++ b/.idea/kano-wand-smarthome.iml @@ -3,6 +3,9 @@ + + + diff --git a/KanoWandAsync/__init__.py b/KanoWandAsync/__init__.py deleted file mode 100644 index 2a0a720..0000000 --- a/KanoWandAsync/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .shop import Shop -from .wand import Wand -from .constants import * \ No newline at end of file diff --git a/KanoWandAsync/constants.py b/KanoWandAsync/constants.py deleted file mode 100644 index 7d6aca7..0000000 --- a/KanoWandAsync/constants.py +++ /dev/null @@ -1,38 +0,0 @@ -from enum import Enum - -class INFO(Enum): - """Enum containing info UUIDs""" - SERVICE = '64a70010-f691-4b93-a6f4-0968f5b648f8' - ORGANIZATION_CHAR = '64a7000b-f691-4b93-a6f4-0968f5b648f8' - SOFTWARE_CHAR = '64a70013-f691-4b93-a6f4-0968f5b648f8' - HARDWARE_CHAR = '64a70001-f691-4b93-a6f4-0968f5b648f8' - -class IO(Enum): - """Enum containing _IO UUIDs""" - SERVICE = '64a70012-f691-4b93-a6f4-0968f5b648f8' - BATTERY_CHAR = '64a70007-f691-4b93-a6f4-0968f5b648f8' - USER_BUTTON_CHAR = '64a7000d-f691-4b93-a6f4-0968f5b648f8' - VIBRATOR_CHAR = '64a70008-f691-4b93-a6f4-0968f5b648f8' - LED_CHAR = '64a70009-f691-4b93-a6f4-0968f5b648f8' - KEEP_ALIVE_CHAR = '64a7000f-f691-4b93-a6f4-0968f5b648f8' - -class SENSOR(Enum): - """Enum containing sensor UUIDs""" - SERVICE = '64a70011-f691-4b93-a6f4-0968f5b648f8' - TEMP_CHAR = '64a70014-f691-4b93-a6f4-0968f5b648f8' - QUATERNIONS_CHAR = '64a70002-f691-4b93-a6f4-0968f5b648f8' - # RAW_CHAR = '64a7000a-f691-4b93-a6f4-0968f5b648f8' - # MOTION_CHAR = '64a7000c-f691-4b93-a6f4-0968f5b648f8' - MAGN_CALIBRATE_CHAR = '64a70021-f691-4b93-a6f4-0968f5b648f8' - QUATERNIONS_RESET_CHAR = '64a70004-f691-4b93-a6f4-0968f5b648f8' - - -class PATTERN(Enum): - """Enum for wand vibration patterns""" - REGULAR = 1 - SHORT = 2 - BURST = 3 - LONG = 4 - SHORT_LONG = 5 - SHORT_SHORT = 6 - BIG_PAUSE = 7 \ No newline at end of file diff --git a/KanoWandAsync/shop.py b/KanoWandAsync/shop.py deleted file mode 100644 index ff3ba09..0000000 --- a/KanoWandAsync/shop.py +++ /dev/null @@ -1,64 +0,0 @@ -from bleak import discover -from .wand import Wand - -class Shop(object): - """ - A scanner class to connect to wands - """ - def __init__(self, shop_loop, wand_class=Wand, debug=False): - """ - Create a new scanner - - Keyword Arguments: - wand_class {class} -- Class to use when connecting to wand (default: {Wand}) - debug {bool} -- Print debug messages (default: {False}) - """ - self.shop_loop = shop_loop - self.wand_class = wand_class - self.debug = debug - self._name = None - self._prefix = None - self._mac = None - - async def scan(self, prefix="Kano-Wand", mac=None, timeout=2.0, connect=False): - """ - Scan for devices - - Keyword Arguments: - name {str} -- Name of the device to scan for (default: {None}) - prefix {str} -- Prefix of name of device to scan for (default: {"Kano-Wand"}) - mac {str} -- MAC Address of the device to scan for (default: {None}) - timeout {float} -- Timeout before returning from scan (default: {1.0}) - connect {bool} -- Connect to the wands automatically (default: {False}) - - Returns {Wand[]} -- Array of wand objects - """ - - if self.debug: - print("Scanning for {} seconds...".format(timeout)) - try: - prefix_check = not (prefix is None) - mac_check = not (mac is None) - assert prefix_check or mac_check - except AssertionError as e: - print("Either a name, prefix, or mac address must be provided to find a wand") - raise e - if prefix is not None: - self._prefix = prefix - elif mac is not None: - self._mac = mac - self.wands = [] - devices = await discover(timeout= timeout) - print(devices) - if self._prefix: - devices = list(filter(lambda x : x.name.startswith(self._prefix), devices)) - if self._mac: - devices = list(filter(lambda x : x.address == self.mac, devices)) - print(devices) - - self.wands = [self.wand_class(d.address, d.name, self.shop_loop) for d in devices] - print(self.wands) - if connect: - for wand in self.wands: - await wand.connect() - return self.wands diff --git a/KanoWandAsync/wand.py b/KanoWandAsync/wand.py deleted file mode 100644 index d32c62d..0000000 --- a/KanoWandAsync/wand.py +++ /dev/null @@ -1,422 +0,0 @@ -import asyncio -import numpy as np - -from .constants import * -from bleak import BleakClient - - -class Wand(object): - """ - A wand class to interact with the Kano wand - """ - - def __init__(self, device_addr, name, bot_loop, debug=True): - """ - Create a new wand - - Arguments: - device {bluepy.ScanEntry} -- Device information - - Keyword Arguments: - debug {bool} -- Print debug messages (default: {False}) - """ - self.debug = debug - self._dev = BleakClient(device_addr) - self.name = name - self.bot_loop = bot_loop - - if debug: - print("Wand: {}\n\rWand Mac: {}".format(self.name, self._dev.address)) - - # Notification stuff - self.connected = False - self._position_callbacks = {} - self._position_subscribed = False - self._button_callbacks = {} - self._button_subscribed = False - self._temperature_callbacks = {} - self._temperature_subscribed = False - self._battery_callbacks = {} - self._battery_subscribed = False - self._notification_thread = None - self._position_notification_handle = 41 - self._button_notification_handle = 33 - self._temp_notification_handle = 56 - self._battery_notification_handle = 23 - - async def connect(self): - if self.debug: - print("Connecting to {}...".format(self.name)) - - connected = await self._dev.connect() - if not connected: - raise Exception("ERROR NOT CONNECTED TO THE DEVICE") - - self.connected = True - await self.post_connect() - - if self.debug: - print("Connected to {}".format(self.name)) - - async def post_connect(self): - """ - Do anything necessary after connecting - """ - pass - - async def disconnect(self): - - await self._dev.disconnect() - self.connected = False - await self.post_disconnect() - - if self.debug: - print("Disconnected from {}".format(self.name)) - - async def post_disconnect(self): - """ - Do anything necessary after disconnecting - """ - pass - - async def get_organization(self): - """ - Get organization of device - Returns {str} -- Organization name - """ - result = await self._dev.read_gatt_char(INFO.ORGANIZATION_CHAR.value) - return result.decode("utf-8") - - async def get_software_version(self): - """ - Get software version - Returns {str} -- Version number - """ - result = await self._dev.read_gatt_char(INFO.SOFTWARE_CHAR.value) - return result.decode("utf-8") - - async def get_hardware_version(self): - - """ - Get hardware version - Returns {str} -- Hardware version - """ - result = await self._dev.read_gatt_char(INFO.HARDWARE_CHAR.value) - return result.decode("utf-8") - - async def get_battery(self): - """ - Get battery level (currently only returns 0) - Returns {str} -- Battery level - """ - result = await self._dev.read_gatt_char(IO.BATTERY_CHAR.value) - print(f"battery is {result}") - print(f"battery is {result.decode('utf-8')}") - return result.decode("utf-8") - - async def get_button(self): - """ - Get current button status - Returns {bool} -- Button pressed status - """ - data = await self._dev.read_gatt_char(IO.USER_BUTTON_CHAR.value) - return data[0] == 1 - - async def get_temperature(self): - """ - Get temperature - - Returns {str} -- Battery level - """ - # with self._lock: - # if not hasattr(self, "_temperature_handle"): - # handle = self._sensor_service.getCharacteristics(SENSOR.TEMP_CHAR.value)[0] - # self._temperature_handle = handle.getHandle() - result = await self._dev.read_gatt_char(SENSOR.TEMP_CHAR.value) - print(f"temp is {result}") - return result - - async def keep_alive(self): - """ - Keep the wand's connection active - - Returns {bytes} -- Status - """ - if self.debug: - print("Keeping wand alive.") - return await self._dev.write_gatt_char(IO.KEEP_ALIVE_CHAR.value, bytes([1]), response=True) - - async def vibrate(self, pattern=PATTERN.REGULAR): - """ - Vibrate wand with pattern - - Keyword Arguments: - pattern {kano_wand.PATTERN} -- Vibration pattern (default: {PATTERN.REGULAR}) - - Returns {bytes} -- Status - """ - if isinstance(pattern, PATTERN): - message = [pattern.value] - else: - message = [pattern] - if self.debug: - print("Setting Vibration to {}".format(message)) - return await self._dev.write_gatt_char(IO.VIBRATOR_CHAR.value, bytes(message), response=True) - - async def set_led(self, color="0x2185d0", on=True): - """ - Set the LED's color - - Keyword Arguments: - color {str} -- Color hex code (default: {"0x2185d0"}) - on {bool} -- Whether light is on or off (default: {True}) - - Returns {bytes} -- Status - """ - message = [] - if on: - message.append(1) - else: - message.append(0) - - color = int(color.replace("#", ""), 16) - r = (color >> 16) & 255 - g = (color >> 8) & 255 - b = color & 255 - rgb = (((r & 248) << 8) + ((g & 252) << 3) + ((b & 248) >> 3)) - message.append(rgb >> 8) - message.append(rgb & 0xff) - - if self.debug: - print("Setting LED to {}".format(message)) - return await self._dev.write_gatt_char(IO.LED_CHAR.value, bytes(message), response=True) - - async def subscribe_position(self): - """ - Subscribe to position notifications and start thread if necessary - """ - if self.debug: - print("Subscribing to position notification") - self._position_subscribed = True - await self._dev.start_notify(SENSOR.QUATERNIONS_CHAR.value, self.handle_notification) - - async def unsubscribe_position(self, continue_notifications=False): - """ - Unsubscribe to position notifications - - Keyword Arguments: - continue_notifications {bool} -- Keep notification thread running (default: {False}) - """ - if self.debug: - print("Unsubscribing from position notification") - - self._position_subscribed = continue_notifications - await self._dev.stop_notify(SENSOR.QUATERNIONS_CHAR.value) - - async def subscribe_button(self): - """ - Subscribe to button notifications and start thread if necessary - """ - if self.debug: - print("Subscribing to button notification") - - self._button_subscribed = True - await self._dev.start_notify(IO.USER_BUTTON_CHAR.value, self.handle_notification) - - async def unsubscribe_button(self, continue_notifications=False): - """ - Unsubscribe to button notifications - - Keyword Arguments: - continue_notifications {bool} -- Keep notification thread running (default: {False}) - """ - if self.debug: - print("Unsubscribing from button notification") - self._button_subscribed = continue_notifications - await self._dev.stop_notify(IO.USER_BUTTON_CHAR.value) - - async def subscribe_temperature(self): - """ - Subscribe to temperature notifications and start thread if necessary - """ - if self.debug: - print("Subscribing to temperature notification") - - self._temperature_subscribed = True - await self._dev.start_notify(SENSOR.TEMP_CHAR.value, self.handle_notification) - - async def unsubscribe_temperature(self, continue_notifications=False): - """ - Unsubscribe to temperature notifications - - Keyword Arguments: - continue_notifications {bool} -- Keep notification thread running (default: {False}) - """ - if self.debug: - print("Unsubscribing from temperature notification") - - self._temperature_subscribed = continue_notifications - await self._dev.stop_notify(SENSOR.TEMP_CHAR.value) - - # - # - async def subscribe_battery(self): - """ - Subscribe to battery notifications and start thread if necessary - """ - if self.debug: - print("Subscribing to battery notification") - - self._battery_subscribed = True - await self._dev.start_notify(IO.BATTERY_CHAR.value, self.handle_notification) - - async def unsubscribe_battery(self, continue_notifications=False): - """ - Unsubscribe to battery notifications - - Keyword Arguments: - continue_notifications {bool} -- Keep notification thread running (default: {False}) - """ - if self.debug: - print("Unsubscribing from battery notification") - - self._battery_subscribed = continue_notifications - await self._dev.stop_notify(IO.BATTERY_CHAR.value) - - async def _on_position(self, data): - """ - Private function for position notification - - Arguments: - data {bytes} -- Data from device - """ - # I got part of this from Kano's node module and modified it - y = np.int16(np.uint16(int.from_bytes(data[0:2], byteorder='little'))) - x = -1 * np.int16(np.uint16(int.from_bytes(data[2:4], byteorder='little'))) - w = -1 * np.int16(np.uint16(int.from_bytes(data[4:6], byteorder='little'))) - z = np.int16(np.uint16(int.from_bytes(data[6:8], byteorder='little'))) - - if self.debug: - pitch = "Pitch: {}".format(z).ljust(16) - roll = "Roll: {}".format(w).ljust(16) - # print("{}{}(x, y): ({}, {})".format(pitch, roll, x, y)) - - await self.on_position(x, y, z, w) - for callback in self._position_callbacks.values(): - await callback(x, y, z, w) - - async def on_position(self, x, y, pitch, roll): - """ - Function called on position notification - - Arguments: - x {int} -- X position of wand (Between -1000 and 1000) - y {int} -- Y position of wand (Between -1000 and 1000) - pitch {int} -- Pitch of wand (Between -1000 and 1000) - roll {int} -- Roll of wand (Between -1000 and 1000) - """ - pass - - async def reset_position(self): - """ - Reset the quaternains of the wand - """ - if self.debug: - print("resetting the quarternion position") - return await self._dev.write_gatt_char(SENSOR.QUATERNIONS_RESET_CHAR.value, bytes([1]), response=True) - - async def _on_button(self, data): - """ - Private function for button notification - - Arguments: - data {bytes} -- Data from device - """ - val = data[0] == 1 - - if self.debug: - print("Button: {}".format(val)) - - await self.on_button(val) - for callback in self._button_callbacks.values(): - await callback(val) - - async def on_button(self, value): - """ - Function called on button notification - - Arguments: - pressed {bool} -- If button is pressed - """ - pass - - async def _on_temperature(self, data): - """ - Private function for temperature notification - - Arguments: - data {bytes} -- Data from device - """ - val = np.int16(np.uint16(int.from_bytes(data[0:2], byteorder='little'))) - - if self.debug: - print("Temperature: {}".format(val)) - - await self.on_temperature(val) - for callback in self._temperature_callbacks.values(): - await callback(val) - - async def on_temperature(self, value): - """ - Function called on temperature notification - - Arguments: - value {int} -- Temperature of the wand - """ - pass - - async def _on_battery(self, data): - """ - Private function for battery notification - - Arguments: - data {bytes} -- Data from device - """ - val = data[0] - - if self.debug: - print("Battery: {}".format(val)) - print("subscribing to the") - - await self.on_battery(val) - for callback in self._battery_callbacks.values(): - await callback(val) - - async def on_battery(self, value): - """ - Function called on battery notification - - Arguments: - value {int} -- Battery level of the wand - """ - pass - - def handle_notification(self, sender, data): - """ - Handle notifications subscribed to - - Arguments: - cHandle {int} -- Handle of notification - data {bytes} -- Data from device - """ - future = None - if sender == SENSOR.QUATERNIONS_CHAR.value: - future = asyncio.run_coroutine_threadsafe(self._on_position(data), self.bot_loop) - elif sender == IO.USER_BUTTON_CHAR.value: - future = asyncio.run_coroutine_threadsafe(self._on_button(data), self.bot_loop) - elif sender == SENSOR.TEMP_CHAR.value: - future = asyncio.run_coroutine_threadsafe(self._on_temperature(data), self.bot_loop) - elif sender == IO.BATTERY_CHAR.value: - future = asyncio.run_coroutine_threadsafe(self._on_battery(data), self.bot_loop) - if future != None: - future.result() diff --git a/collect_spell_data.py b/collect_spell_data.py index 4e1e3c2..c156a71 100644 --- a/collect_spell_data.py +++ b/collect_spell_data.py @@ -1,11 +1,12 @@ -from KanoWandAsync import Shop, Wand -from KanoWandAsync.constants import * import asyncio import pandas as pd -from uuid import uuid4 +from kanowandasync import Shop, Wand +from kanowandasync.constants import * from pathlib import Path -import os +from uuid import uuid4 + global spell + def save_dataframe(data): df = pd.DataFrame(data) df.columns = ["x", "y", "pitch", "roll"] @@ -38,7 +39,7 @@ class DataCollectWand(Wand): await self.unsubscribe_position() async def on_position(self, x, y, pitch, roll): - # print(x, y, pitch, roll) + self.data.append([x, y, pitch, roll]) def get_prompt_text(spells): diff --git a/kano-wand-test.py b/kano-wand-test.py index 44f5a46..c4097c6 100644 --- a/kano-wand-test.py +++ b/kano-wand-test.py @@ -1,5 +1,5 @@ -from KanoWandAsync import Shop, Wand -from KanoWandAsync.constants import * +from kanowandasync import Shop, Wand +from kanowandasync.constants import * import asyncio # Custom wand class extending the default wand