From c7f6250f907020c4b6ea888aeb84d53d4e5ce639 Mon Sep 17 00:00:00 2001 From: Samuel Sadok Date: Mon, 2 Oct 2017 14:52:13 +0200 Subject: [PATCH] add python support for new protocol --- tools/demo.py | 32 +++++++++++++ tools/odrive/core.py | 94 +++++++++++++++++++++++++++++++++++++ tools/odrive/mock_device.py | 64 +++++++++++++++++++++++++ tools/test_bulk.py | 2 +- 4 files changed, 191 insertions(+), 1 deletion(-) create mode 100755 tools/demo.py create mode 100644 tools/odrive/core.py create mode 100644 tools/odrive/mock_device.py mode change 100644 => 100755 tools/test_bulk.py diff --git a/tools/demo.py b/tools/demo.py new file mode 100755 index 00000000..99ecd77e --- /dev/null +++ b/tools/demo.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +""" +Example usage of the ODrive python library to monitor and control ODrive devices +""" + +import odrive.core + +# Find a connected ODrive (this will block until you connect one) +my_drive = odrive.core.find_any() + +# 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`. +# You can also inspect the object using the dir-function: +#print(dir(my_drive)) +#print(dir(my_drive.motor0)) +# TODO: maybe provide an introspection method that dumps the whole type hierarchy at once + +# To read a value, simply read the property +print("Bus voltage is " + str(my_drive.vbus_voltage) + "V") + +# Or to change a value, just assign to the property +my_drive.motor0.pos_setpoint = 3.14 +print("Position setpoint is " + str(my_drive.motor0.pos_setpoint)) + + +# Some more things you can try: + +# Write to a read-only property: +# my_drive.vbus_voltage = 5 # fails with `AttributeError: can't set attribute` + +# Assign an incompatible value: +# my_drive.motor0.pos_setpoint = "I like trains" # fails with `TypeError: expected value of type float` diff --git a/tools/odrive/core.py b/tools/odrive/core.py new file mode 100644 index 00000000..60ca4776 --- /dev/null +++ b/tools/odrive/core.py @@ -0,0 +1,94 @@ +""" +Provides functions for the discovery of ODrive devices +""" + +import sys +import json +import odrive +import odrive.mock_device + +class DeviceProperty(property): + def __init__(self, device, id, type, can_read, can_write): + self._device = device + self._id = id + self._type = type + property.__init__(self, + self.fget if can_read else None, + self.fset if can_write else None) + + def fget(self, obj): + self._device.send("r " + str(self._id) + "\n") + # TODO: message based receive + response = self._device.receive_until('\n') + return self._type(response.strip('\n')) + + def fset(self, obj, value): + if not isinstance(value, self._type): + raise TypeError("expected value of type {}".format(self._type.__name__)) + self._device.send("w " + str(self._id) + " " + str(value) + "\n") + + +def create_object(json_data, namespace, device): + """ + Creates an object that implements the specified JSON type description by + communicating with the provided device object + """ + + # Build property list from JSON + properties = {} + for item in json_data: + name = item.get("name", None) + if name is None: + sys.stderr.write("unnamed property in {}".format(namespace)) + continue + + type_str = item.get("type", None) + if type_str is None: + sys.stderr.write("property {} has no specified type".format(name)) + continue + + if type_str == "tree": + properties[name] = create_object(item["content"], namespace + "." + item["name"], device) + else: + if type_str == "float": + property_type = float + elif type_str == "int": + property_type = int + elif type_str == "bool": + property_type = bool + elif type_str == "uint16": + property_type = int + else: + sys.stderr.write("property {} has unsupported type {}".format(name, type_str)) + continue + + id_str = item.get("id", None) + if id_str is None: + sys.stderr.write("property {} has specified ID".format(name)) + continue + + access_mode = item.get("mode", "rw") + properties[name] = DeviceProperty(device, id_str, property_type, + 'r' in access_mode, + 'w' in access_mode) + + # Create a type from the property list and instantiate it + jit_type = type(namespace, (object,), properties) + new_object = jit_type() + return new_object + + +def find_any(): + """ + Scans for ODrives on all supported interfaces (currently only USB) and + returns the first device that is found. If no device is connected the + function blocks. + """ + # TODO: do device discovery and instantiation in a separate thread + # TODO: test with real device + device = odrive.mock_device.MockDevice() + #device = odrive.usbbulk.poll_odrive_bulk_device() + device.send('j\n') + json_string = device.receive_until('\n') + #print("have JSON: " + json_string) + return create_object(json.loads(json_string), "odrive.usb_device", device) diff --git a/tools/odrive/mock_device.py b/tools/odrive/mock_device.py new file mode 100644 index 00000000..5084ba50 --- /dev/null +++ b/tools/odrive/mock_device.py @@ -0,0 +1,64 @@ +""" +Provides classes for rudimentary testing if you don't have an ODrive. +""" + +JSONDescriptor = ''' +[ + { "name": "vbus_voltage", "id": 0, "type": "float", "mode": "r" }, + { "name": "elec_rad_per_enc", "id": 1, "type": "float", "mode": "r" }, + { "name": "motor0", "id": 2, "type": "tree", "content": [ + { "name": "pos_setpoint", "id": 3, "type": "float" }, + { "name": "pos_gain", "id": 4, "type": "float" }, + { "name": "vel_setpoint", "id": 5, "type": "float" }, + { "name": "current_control", "id": 6, "type": "tree", "content": [ + { "name": "v_current_control_integral_d", "id": 7, "type": "float" }, + { "name": "v_current_control_integral_q", "id": 8, "type": "float" }, + { "name": "Ibus", "id": 9, "type": "float" } + ]}, + { "name": "encoder", "id": 10, "type": "tree", "content": [ + { "name": "phase", "id": 11, "type": "float" }, + { "name": "pll_pos", "id": 12, "type": "float" }, + { "name": "pll_vel", "id": 13, "type": "float" } + ]} + ]} +] +''' + + +class MockDevice(object): + """ + Implements a mock device that supports the 'r', 'w' and 'j' commands + """ + + _rx_buf = "" + _tx_buf = "" + _values = ["0.0"] * 13 + + def execute_cmd(self, cmd): + if cmd[0] == 'j': + self._tx_buf = self._tx_buf + JSONDescriptor.replace('\n', '') + '\n' + elif cmd[0] == 'r': + [_, id] = cmd.split() + self._tx_buf = self._tx_buf + self._values[int(id)] + '\n' + elif cmd[0] == 'w': + [_, id, val] = cmd.split() + self._values[int(id)] = val + else: + # TODO: report error + pass + + def send(self, buffer): + """ + Sends a string to the virtual device + """ + self._rx_buf = self._rx_buf + buffer + while '\n' in self._rx_buf: + [cmd, _, self._rx_buf] = self._rx_buf.partition('\n') + self.execute_cmd(cmd) + + def receive_until(self, char): + """ + Reads a string from the virtual device's TX buffer + """ + [result, char, self._tx_buf] = self._tx_buf.partition(char) + return result + char diff --git a/tools/test_bulk.py b/tools/test_bulk.py old mode 100644 new mode 100755 index 848a0371..6908f631 --- a/tools/test_bulk.py +++ b/tools/test_bulk.py @@ -1,4 +1,4 @@ -#! /usr/bin/env python3 +#!/usr/bin/env python3 import argparse