diff --git a/Firmware/CHANGELOG.md b/Firmware/CHANGELOG.md index 36d89bc4..4a026b9a 100644 --- a/Firmware/CHANGELOG.md +++ b/Firmware/CHANGELOG.md @@ -15,6 +15,13 @@ Please add a note of your changes below this heading if you make a Pull Request. * Update CubeMX generated STM platform code to version 1.19.0 * Remove `UUID_0`, `UUID_1` and `UUID_2` from USB protocol. Use `serial_number` instead. * Freertos memory pool (task stacks, etc) now uses Core Coupled Memory. +* Refactor python tools + * `explore_odrive.py` now supports controlling multiple ODrives concurrently (`odrv0`, `odrv1`, ...) + * No need to restart `explore_odrive.py` when devices get disconnected and reconnected + * The command line arguments of `explore_odrive.py` have changed. See `explore_odrive.py --help` for more details. + * ODrive accesses from within python tools are now thread-safe + * Liveplotter does no longer steal focus and closes as expected + * (experimental: start liveplotter from `explore_odrive.py` by typing `start_liveplotter(lambda: odrv0.motor0.encoder.encoder_state)`) ### Fixed * malloc now fails if we run out of memory (before it would always succeed even if we are out of ram...) diff --git a/Firmware/README.md b/Firmware/README.md index 2888193f..6c7124e8 100644 --- a/Firmware/README.md +++ b/Firmware/README.md @@ -165,7 +165,7 @@ pip install pyusb pyserial 6. Open the bash prompt in the `ODrive/tools/` folder. 7. Run `python3 demo.py` or `python3 explore_odrive.py`. - `demo.py` is a very simple script which will make motor 0 turn back and forth. Use this as an example if you want to control the ODrive yourself programatically. -- `explore_odrive.py` drops you into an interactive python shell where you can explore and edit the parameters that are available on your device. For instance `my_odrive.motor0.pos_setpoint = 10000` makes motor0 move to position 10000. To connect over serial instead of USB run `./tools/explore_odrive.py --discover serial`. +- `explore_odrive.py` drops you into an interactive python shell where you can explore and edit the parameters that are available on your device. For instance `my_odrive.motor0.pos_setpoint = 10000` makes motor0 move to position 10000. To connect over serial instead of USB run `./tools/explore_odrive.py --path serial`. ### From Arduino [See ODrive Arduino Library](https://github.com/madcowswe/ODriveArduino) diff --git a/tools/demo.py b/tools/demo.py index 4a06c260..41360333 100755 --- a/tools/demo.py +++ b/tools/demo.py @@ -5,12 +5,15 @@ Example usage of the ODrive python library to monitor and control ODrive devices from __future__ import print_function -import odrive.core +import odrive.discovery import time import math # Find a connected ODrive (this will block until you connect one) -my_drive = odrive.core.find_any(consider_usb=True, consider_serial=False, printer=print) +my_drive = odrive.discovery.find_any() + +# Find an ODrive that is connected on the serial port /dev/ttyUSB0 +#my_drive = odrive.discovery.find_any("serial:/dev/ttyUSB0") # The above call returns a python object with a dynamically generated type. The # type hierarchy will correspond to the endpoint list in `MotorControl/protocol.cpp`. diff --git a/tools/dfu.py b/tools/dfu.py index 7ad27bf8..cd8553c1 100755 --- a/tools/dfu.py +++ b/tools/dfu.py @@ -12,7 +12,7 @@ import struct import dfuse import usb.core import usb.util -import odrive.core +import odrive.discovery # We are interactively printing status messages, so flush by default import functools @@ -142,18 +142,6 @@ def jump_to_application(dfudev, address): if status[1] != dfuse.DfuState.DFU_MANIFEST: raise RuntimeError("An error occured. Device Status: {}".format(status[1])) -def str_to_uuid(uuid): - uuid = bytearray.fromhex(uuid.replace('-', '')) - return struct.unpack('>I', uuid[0:4]), struct.unpack('>I', uuid[4:8]), struct.unpack('>I', uuid[8:12]) - -def uuid_to_str(uuid0, uuid1, uuid2): - return "{:08X}-{:08X}-{:08X}".format(struct.pack('>I', uuid0), struct.pack('>I', uuid1), struct.pack('>I', uuid2)) - -def uuid_to_serial(uuid0, uuid1, uuid2): - return (struct.pack('>I', uuid0 + uuid2) + struct.pack('>I', uuid1)[0:2]).hex().upper() - - -### THREADS ### def show_deferred_message(message, cancellation_token): """ @@ -170,43 +158,25 @@ def show_deferred_message(message, cancellation_token): t.daemon = True t.start() -def put_odrive_into_dfu_mode_thread(cancellation_token): +def put_odrive_into_dfu_mode(my_drive): """ - Waits for an ODrive with a matching serial number and puts - it into DFU mode once it's found. The thread continues to put - matching devices into DFU mode until cancellation_token - is set. + Puts the specified device into DFU mode """ - global app_cancellation_token - while not cancellation_token.is_set(): - constraints = {} if serial_number == None else {'serial_number': serial_number} - my_drive = odrive.core.find_any(consider_usb=True, consider_serial=False, - cancellation_token=cancellation_token, - **constraints) - if cancellation_token.is_set(): - return - if not hasattr(my_drive, "enter_dfu_mode"): - print("The firmware on device {} does not support DFU. You need to \n" - "flash the firmware once using STLink (`make flash`), after that \n" - "DFU with this script should work fine." - .format(my_drive.__channel__.usb_device.serial_number)) - # Terminate script, otherwise it would try to reconnect to the same - # incompatible device - app_cancellation_token.set() # TODO: implement a more sensible discorvery mechanism to fix this - return - print("Putting device {} into DFU mode...".format(my_drive.__channel__.usb_device.serial_number)) - try: - my_drive.enter_dfu_mode() - except usb.core.USBError as ex: - pass # this is expected because the device reboots - if platform.system() == "Windows": - show_deferred_message("Still waiting for the device to reappear.\n" - "Use the Zadig utility to set the driver of 'STM32 BOOTLOADER' to libusb-win32.", - cancellation_token) - # If we immediately continue we might still pick up the device that was - # just rebooted. This isn't an issue but will display a distracting - # error message. - time.sleep(1) + if not hasattr(my_drive, "enter_dfu_mode"): + print("The firmware on device {} does not support DFU. You need to \n" + "flash the firmware once using STLink (`make flash`), after that \n" + "DFU with this script should work fine." + .format(my_drive.__channel__.usb_device.serial_number)) + return + print("Putting device {} into DFU mode...".format(my_drive.__channel__.usb_device.serial_number)) + try: + my_drive.enter_dfu_mode() + except odrive.protocol.ChannelBrokenException as ex: + pass # this is expected because the device reboots + if platform.system() == "Windows": + show_deferred_message("Still waiting for the device to reappear.\n" + "Use the Zadig utility to set the driver of 'STM32 BOOTLOADER' to libusb-win32.", + find_odrive_cancellation_token) ### BEGINNING OF APPLICATION ### @@ -230,12 +200,7 @@ hexfile = IntelHex(args.file) #for start, end in hexfile.segments(): # print(" {:08X} to {:08X}".format(start, end - 1)) -if args.uuid != None: - serial_number = uuid_to_serial(*str_to_uuid(args.uuid)) -elif args.serial_number != None: - serial_number = args.serial_number -else: - serial_number = None +serial_number = args.serial_number app_cancellation_token = threading.Event() @@ -244,7 +209,8 @@ try: print("Waiting for ODrive...") # Scan for ODrives not in DFU mode and put them into DFU mode once they appear - threading.Thread(target=put_odrive_into_dfu_mode_thread, args=(find_odrive_cancellation_token,)).start() + # We only scan on USB because DFU is only possible over USB + odrive.discovery.find_all("usb", serial_number, put_odrive_into_dfu_mode, find_odrive_cancellation_token) # Poll libUSB until a device in DFU mode is found while not app_cancellation_token.is_set(): diff --git a/tools/drv_status.py b/tools/drv_status.py old mode 100644 new mode 100755 index fda6c292..bc62c3ed --- a/tools/drv_status.py +++ b/tools/drv_status.py @@ -5,12 +5,14 @@ Example usage of the ODrive python library to monitor and control ODrive devices from __future__ import print_function -import odrive.core +import odrive.discovery import time import math # Find a connected ODrive (this will block until you connect one) -my_drive = odrive.core.find_any(consider_usb=True, consider_serial=False, printer=print) +print("Waiting for ODrive...") +my_drive = odrive.discovery.find_any() +print("connected") # Print DRV device regs for Motor 0 fault = my_drive.motor0.gate_driver.drv_fault diff --git a/tools/explore_odrive.py b/tools/explore_odrive.py index 9efaebb7..ccc252d4 100755 --- a/tools/explore_odrive.py +++ b/tools/explore_odrive.py @@ -3,37 +3,16 @@ Load an odrive object to play with in the IPython interactive shell. """ -import odrive.core import argparse import sys import platform +import threading +import odrive.discovery +from odrive.utils import start_liveplotter -# Check if IPython is installed -try: - import IPython - embed_ipython = True -except: - embed_ipython = False - - print("Warning: you don't have IPython installed.") - print("If you want to have an improved interactive console with pretty colors,") - print("you should install IPython\n") - - # Ensure interactive mode - if not bool(getattr(sys, 'ps1', sys.flags.interactive)): - print("You're not running in interactive mode. Run python -i explore_odrive.py") - print('') - sys.exit(1) - - # Enable tab complete if possible - try: - import readline - readline.parse_and_bind("tab: complete") - except: - sudo_prefix = "" if platform.system() == "Windows" else "sudo " - print("Warning: could not enable tab-complete. User experience will suffer.\n" - "Run `{}pip install readline` and then restart this script to fix this." - .format(sudo_prefix)) +# Flush stdout by default +import functools +print = functools.partial(print, flush=True) # some enums described in the README @@ -48,25 +27,32 @@ CTRL_MODE_VELOCITY_CONTROL = 2, CTRL_MODE_POSITION_CONTROL = 3 -# Parse arguments -parser = argparse.ArgumentParser(description='Load an odrive object to play with in the IPython interactive shell.') +## Parse arguments ## +parser = argparse.ArgumentParser(description='Load an odrive object to play with in the IPython interactive shell.', + formatter_class=argparse.RawTextHelpFormatter) parser.add_argument("-v", "--verbose", action="store_true", help="print debug information") -group = parser.add_mutually_exclusive_group() -group.add_argument("-d", "--discover", metavar="CHANNELS", action="store", - help="Automatically discover ODrives. Takes a comma-separated list (without spaces) " - "to indicate which connection types should be considered. Possible values are " - "usb and serial. For example \"--discover usb,serial\" indicates " - "that USB and serial ports should be scanned for ODrives. " - "If none of the below options are specified, --discover usb is assumed.") -group.add_argument("-u", "--usb", metavar="BUS:DEVICE", action="store", - help="Specifies the USB port on which the device is connected. " - "For example \"001:014\" means bus 001, device 014. The numbers can be obtained " - "using `lsusb`.") -group.add_argument("-s", "--serial", metavar="PORT", action="store", - help="Specifies the serial port on which the device is connected. " - "For example \"/dev/ttyUSB0\". Use `ls /dev/tty*` to find your port name.") -parser.set_defaults(discover="usb") +parser.add_argument("-p", "--path", metavar="PATH", action="store", + help="The path(s) where ODrive(s) should be discovered.\n" + "By default the script will connect to any ODrive on USB.\n\n" + "To select a specific USB device:\n" + " --path usb:BUS:DEVICE\n" + "usbwhere BUS and DEVICE are the bus and device numbers as shown in `lsusb`.\n\n" + "To select a specific serial port:\n" + " --path serial:PATH\n" + "where PATH is the path of the serial port. For example \"/dev/ttyUSB0\".\n" + "You can use `ls /dev/tty*` to find the correct port.\n\n" + "You can combine USB and serial specs by separating them with a comma (no space!)\n" + "Example:\n" + " --path usb,serial:/dev/ttyUSB0\n" + "means \"discover any USB device or a serial device on /dev/ttyUSB0\"") +parser.add_argument("--no-ipython", action="store_true", + help="Use the regular Python shell\n" + "instead of the IPython shell,\n" + "even if IPython is installed\n") +parser.add_argument("-s", "--serial-number", action="store", + help="The serial number of the device. If omitted, any device is accepted.\n") +parser.set_defaults(path="usb") args = parser.parse_args() if (args.verbose): @@ -74,41 +60,151 @@ if (args.verbose): else: printer = lambda x: None + +## Interactive console utils ## + +COLOR_RED = '\x1b[91;1m' +COLOR_CYAN = '\x1b[96;1m' +COLOR_RESET = '\x1b[0m' + +def print_on_second_last_line(text, **kwargs): + """ + Prints a text on the second last line. + This can be used to print a message above the command + prompt. If the command prompt spans multiple lines, + there will be glitches. + """ + # Escape character sequence: + # ESC 7: store cursor position + # ESC 1A: move cursor up by one + # ESC 1S: scroll entire viewport by one + # ESC 1L: insert 1 line at cursor position + # (print text) + # ESC 8: restore old cursor position + kwargs['end'] = '' + kwargs['flush'] = True + print('\x1b7\x1b[1A\x1b[1S\x1b[1L' + text + '\x1b8', **kwargs) + +def print_banner(): + print('ODrive control utility v0.4') + print('Please connect your ODrive.') + print('Type help() for help.') + +def print_help(): + print('') + if len(discovered_devices) == 0: + print('Connect your ODrive to {} and power it up.'.format(args.path)) + print('After that, the following message should appear:') + print(' "Connected to ODrive [serial number] as odrv0"') + print('') + print('Once the ODrive is connected, type "odrv0." and press ') + else: + print('Type "odrv0." and press ') + print('This will present you with all the properties that you can reference') + print('') + print('For example: "odrv0.motor0.encoder.pll_pos"') + print('will print the current encoder position on motor 0') + print('and "odrv0.motor0.pos_setpoint = 10000"') + print('will send motor0 to 10000') + print('') + +interactive_variables = {} +interactive_variables["help"] = print_help + + +## Device discovery ## + +discovered_devices = [] + +def did_discover_device(odrive): + """ + Handles the discovery of new devices by displaying a + message and making the device available to the interactive + console + """ + serial_number = odrive.serial_number if hasattr(odrive, 'serial_number') else "[unknown serial number]" + if serial_number in discovered_devices: + verb = "Reconnected" + index = discovered_devices.index(serial_number) + else: + verb = "Connected" + discovered_devices.append(serial_number) + index = len(discovered_devices) - 1 + interactive_name = "odrv" + str(index) + + # Publish new ODrive to interactive console + interactive_variables[interactive_name] = odrive + globals()[interactive_name] = odrive # Add to globals so tab complete works + print_on_second_last_line(COLOR_CYAN + "{} to ODrive {:012X} as {}".format(verb, serial_number, interactive_name) + COLOR_RESET) + + # Subscribe to disappearance of the device + odrive.__channel__._channel_broken.subscribe(lambda: did_lose_device(interactive_name)) + +def did_lose_device(interactive_name): + """ + Handles the disappearance of a device by displaying + a message. + """ + if not app_shutdown_token.is_set(): + print_on_second_last_line(COLOR_RED + "Oh no {} disappeared".format(interactive_name) + COLOR_RESET) + # Connect to device -if not args.usb is None: - try: - bus = int(args.usb.split(":")[0]) - address = int(args.usb.split(":")[1]) - except (ValueError, IndexError): - print("the --usb argument must look something like this: \"001:014\"") - sys.exit(1) - try: - my_odrive = odrive.core.open_usb(bus, address, printer=printer) - except odrive.protocol.DeviceInitException as ex: - print(str(ex)) - sys.exit(1) -elif not args.serial is None: - my_odrive = odrive.core.open_serial(args.serial, printer=printer) +printer("Waiting for device...") +app_shutdown_token = threading.Event() +odrive.discovery.find_all(args.path, args.serial_number, + did_discover_device, app_shutdown_token, + printer=printer) + + +## Launch interactive shell ## + +# Check if IPython is installed +if args.no_ipython: + use_ipython = False else: - print("Waiting for device...") - consider_usb = 'usb' in args.discover.split(',') - consider_serial = 'serial' in args.discover.split(',') - my_odrive = odrive.core.find_any(consider_usb, consider_serial, printer=printer) -print("Connected!") + try: + import IPython + use_ipython = True + except: + print("Warning: you don't have IPython installed.") + print("If you want to have an improved interactive console with pretty colors,") + print("you should install IPython\n") + use_ipython = False + +# If IPython is installed, embed IPython shell, otherwise embed regular shell +if use_ipython: + #interactive_variables["lalala"] = print_help + help = print_help # Override help function + console = IPython.terminal.embed.InteractiveShellEmbed(local_ns=interactive_variables, banner1='') + console.runcode = console.run_code # hack to make IPython look like the regular console + interact = console +else: + + # Enable tab complete if possible + try: + import rlcompleter + import readline + readline.parse_and_bind("tab: complete") + except: + sudo_prefix = "" if platform.system() == "Windows" else "sudo " + print("Warning: could not enable tab-complete. User experience will suffer.\n" + "Run `{}pip install readline` and then restart this script to fix this." + .format(sudo_prefix)) + + import code + console = code.InteractiveConsole(locals=interactive_variables) + interact = lambda: console.interact(banner='') + +# install hook to hide ChannelBrokenException +console.runcode('import sys') +console.runcode('superexcepthook = sys.excepthook') +console.runcode('def newexcepthook(ex_class,ex,trace):\n' + ' if ex_class.__module__ + "." + ex_class.__name__ != "odrive.protocol.ChannelBrokenException":\n' + ' superexcepthook(ex_class,ex,trace)') +console.runcode('sys.excepthook=newexcepthook') -print('') -print('ODRIVE EXPLORER') -print('') -print('You can now type "my_odrive." and press ') -print('This will present you with all the properties that you can reference') -print('') -print('For example: "my_odrive.motor0.encoder.pll_pos"') -print('will print the current encoder position on motor 0') -print('and "my_odrive.motor0.pos_setpoint = 10000"') -print('will send motor0 to 10000') -print('') - -# If IPython is installed, embed shell, otherwise drop into interactive stock python shell -if embed_ipython: - IPython.embed() +# Launch shell +print_banner() +interact() +app_shutdown_token.set() diff --git a/tools/liveplotter.py b/tools/liveplotter.py index 1c4ba849..95c50c24 100755 --- a/tools/liveplotter.py +++ b/tools/liveplotter.py @@ -4,49 +4,18 @@ Liveplotter """ import time -import odrive.core -import matplotlib.pyplot as plt -import numpy as np import threading +import odrive.discovery +from odrive.utils import start_liveplotter data_rate = 100 plot_rate = 10 num_samples = 1000 -my_odrive = odrive.core.find_any() +my_odrive = odrive.discovery.find_any() -plt.ion() -global vals -vals = [] +# If you want to plot different values, change them here. +# You can plot any number of values concurrently. +start_liveplotter(lambda: [my_odrive.motor0.encoder.pll_pos, + my_odrive.motor1.encoder.pll_pos]) -# Make sure the script terminates when the user closes the plotter -cancellation_token = threading.Event() -def handle_close(evt): - cancellation_token.set() -fig = plt.figure() -fig.canvas.mpl_connect('close_event', handle_close) - -def fetch_data(): - global vals - global cancellation_token - while not cancellation_token.is_set(): - vals.append(my_odrive.motor0.timing_log.TIMING_LOG_FOC_CURRENT) - if len(vals) > num_samples: - vals = vals[-num_samples:] - time.sleep(1/data_rate) - -# TODO: use animation for better UI performance, see: -# https://matplotlib.org/examples/animation/simple_anim.html -def plot_data(): - global vals - global cancellation_token - while not cancellation_token.is_set(): - plt.clf() - plt.plot(vals) - #time.sleep(1/plot_rate) - fig.canvas.flush_events() - -fetch_thread = threading.Thread(target=fetch_data, daemon=True) -fetch_thread.start() - -plot_data() diff --git a/tools/odrive/core.py b/tools/odrive/core.py deleted file mode 100644 index bb2bd8e1..00000000 --- a/tools/odrive/core.py +++ /dev/null @@ -1,334 +0,0 @@ -""" -Provides functions for the discovery of ODrive devices -""" - -import sys -import time -import json -import usb.core -import usb.util -import serial -import serial.tools.list_ports -import odrive.util -import odrive.usbbulk_transport -import odrive.serial_transport -import re -import time -import os -import odrive.protocol -import itertools -import struct -import functools - -def noprint(x): - pass - - -class SimpleDeviceProperty(property): - """ - Used internally by dynamically created objects to translate - property assignments and fetches into endpoint operations on the - object's associated channel - """ - def __init__(self, channel, id, type, struct_format, can_read, can_write): - self._channel = channel - self._id = id - self._type = type - self._struct_format = struct_format - property.__init__(self, - self.fget if can_read else None, - self.fset if can_write else None) - - def fget(self, obj): - size = struct.calcsize(self._struct_format) - buffer = self._channel.remote_endpoint_operation(self._id, None, True, size) - return struct.unpack(self._struct_format, buffer)[0] - - def fset(self, obj, value): - value = self._type(value) - buffer = struct.pack(self._struct_format, value) - # TODO: Currenly we wait for an ack here. Settle on the default guarantee. - self._channel.remote_endpoint_operation(self._id, buffer, True, 0) - -def call_remote_function(channel, trigger_id, arg_properties, *args): - """ - Used internally by the dynamically created objects to translate - function calls into endpoint operations on the associated channel - """ - if (len(arg_properties) != len(args)): - raise TypeError("expected {} arguments but have {}".format(len(arg_properties), len(args))) - for i in range(len(args)): - arg_properties[i].fset(None, args[i]) - channel.remote_endpoint_operation(trigger_id, None, True, 0) - -def setattr_or_raise_if_undefined(self, name, value): - """ - If employed as an object's __setattr__ function, this function - makes sure that an assignment to an undefined attribute doesn't - create a new attribute but instead raises an exception - """ - # We can't use hasattr here because internally it fetches the property - # value, creating unnecessary bus traffic - if name in dir(self): - object.__setattr__(self, name, value) - else: - raise TypeError('Cannot set name %r on object of type %s' % ( - name, self.__class__.__name__)) - -def create_property(name, json_data, channel, printer): - """ - Dynamically creates a property based on a JSON definition - """ - name = name or "[anonymous]" - - type_str = json_data.get("type", None) - if type_str is None: - printer("property {} has no specified type".format(name)) - return None - - if type_str == "float": - property_type = float - struct_format = " num_samples: + vals = vals[-num_samples:] + time.sleep(1/data_rate) + + # TODO: use animation for better UI performance, see: + # https://matplotlib.org/examples/animation/simple_anim.html + def plot_data(): + global vals + + plt.ion() + + # Make sure the script terminates when the user closes the plotter + def did_close(evt): + cancellation_token.set() + fig = plt.figure() + fig.canvas.mpl_connect('close_event', did_close) + + while not cancellation_token.is_set(): + plt.clf() + plt.plot(vals) + #time.sleep(1/plot_rate) + fig.canvas.flush_events() + + threading.Thread(target=fetch_data).start() + threading.Thread(target=plot_data).start() + #plot_data() + +## Exceptions ## + +class TimeoutException(Exception): + pass + +## Threading utils ## + +class Event(): + """ + Alternative to threading.Event(), enhanced by the subscribe() function + that the original fails to provide. + """ + def __init__(self): + self._evt = threading.Event() + self._subscribers = [] + self._mutex = threading.Lock() + + def is_set(self): + return self._evt.is_set() + + def set(self): + """ + Sets the event and invokes all subscribers if the event was + not already set + """ + self._mutex.acquire() + try: + if not self._evt.is_set(): + self._evt.set() + for s in self._subscribers: + s() + finally: + self._mutex.release() + + def subscribe(self, handler): + """ + Invokes the specified handler exactly once as soon as the + specified event is set. If the event is already set, the + handler is invoked immediately. + Returns a function that can be invoked to unsubscribe. + """ + self._mutex.acquire() + try: + self._subscribers.append(handler) + if self._evt.is_set(): + handler() + finally: + self._mutex.release() + return lambda: self.unsubscribe(handler) + + def unsubscribe(self, handler): + self._mutex.acquire() + try: + self._subscribers.pop(self._subscribers.index(handler)) + finally: + self._mutex.release() + + def wait(self, timeout=None): + return self._evt.wait(timeout=timeout) + +def wait_any(*events, timeout=None): + """ + Blocks until any of the specified events are triggered. + Returns the number of the event that was triggerd or raises + a TimeoutException + """ + or_event = threading.Event() + unsubscribe_functions = [] + for event in events: + unsubscribe_functions.append(event.subscribe(lambda: or_event.set())) + or_event.wait(timeout=timeout) + for unsubscribe_function in unsubscribe_functions: + unsubscribe_function() + for i in range(len(events)): + if events[i].is_set(): + return i + raise TimeoutException() diff --git a/tools/rate_test.py b/tools/rate_test.py old mode 100644 new mode 100755 index 191125ed..76f359a6 --- a/tools/rate_test.py +++ b/tools/rate_test.py @@ -1,9 +1,14 @@ +#!/usr/bin/env python3 + import time -import odrive.core +import odrive.discovery import matplotlib.pyplot as plt import numpy as np -myOdrive = odrive.core.find_any() +# Find a connected ODrive (this will block until you connect one) +print("Waiting for ODrive...") +myOdrive = odrive.discovery.find_any() +print("connected") plt.ion()