Mirrors_Framework/ODrive
1
0
Fork 0
mirror of https://github.com/odriverobotics/ODrive.git synced 2026-09-20 22:55:00 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

clean up fiber port

Browse Source
This commit is contained in:
Oskar Weigl
2018-06-08 20:18:47 -07:00
parent 4841ff03fb
commit 2b0164af74
7 changed files with 24 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Firmware/fibre/python/fibre/discovery.py
+4 -4
View File
@@ -11,7 +11,7 @@ import fibre.protocol
import fibre.utils
import fibre.remote_object
import fibre.serial_transport
from fibre.utils import Event
from fibre.utils import Event, Logger
from fibre.protocol import ChannelBrokenException
# Load all installed transport layers
@@ -90,7 +90,7 @@ def find_all(path, serial_number,
obj.__dict__['_json_data'] = json_data['members']
obj.__dict__['_json_crc'] = json_crc16
device_serial_number = odrive.utils.get_serial_number_str(obj)
device_serial_number = fibre.utils.get_serial_number_str(obj)
if serial_number != None and device_serial_number != serial_number:
logger.debug("Ignoring device with serial number {}".format(device_serial_number))
return
@@ -111,7 +111,7 @@ def find_all(path, serial_number,
def find_any(path="usb", serial_number=None,
search_cancellation_token=None, channel_termination_token=None,
timeout=None, printer=noprint):
timeout=None, logger=Logger(verbose=False)):
"""
Blocks until the first matching Fibre node is connected and then returns that node
"""
@@ -120,7 +120,7 @@ def find_any(path="usb", serial_number=None,
def did_discover_object(obj):
result[0] = obj
done_signal.set()
find_all(path, serial_number, did_discover_object, done_signal, channel_termination_token, printer)
find_all(path, serial_number, did_discover_object, done_signal, channel_termination_token, logger)
try:
done_signal.wait(timeout=timeout)
finally:
Firmware/fibre/python/fibre/usbbulk_transport.py
+1 -1
View File
@@ -182,7 +182,7 @@ def discover_channels(path, serial_number, callback, cancellation_token, channel
return True
while not cancellation_token.is_set():
# logger.debug("USB discover loop")
logger.debug("USB discover loop")
devices = usb.core.find(find_all=True, custom_match=device_matcher)
for usb_device in devices:
try:
Firmware/fibre/python/fibre/utils.py
+6 -1
View File
@@ -17,8 +17,13 @@ except ModuleNotFoundError:
sys.stdout.flush()
pass
## Threading utils ##
def get_serial_number_str(device):
if hasattr(device, 'serial_number'):
return format(device.serial_number, 'x').upper()
else:
return "[unknown serial number]"
## Threading utils ##
class Event():
"""
Alternative to threading.Event(), enhanced by the subscribe() function
tools/odrive/configuration.py
+5 -5
View File
@@ -2,15 +2,15 @@
import json
import os
import tempfile
import odrive.remote_object
import fibre.remote_object
from odrive.utils import OperationAbortedException
def get_dict(obj, is_config_object):
result = {}
for (k,v) in obj._remote_attributes.items():
if isinstance(v, odrive.remote_object.RemoteProperty) and is_config_object:
if isinstance(v, fibre.remote_object.RemoteProperty) and is_config_object:
result[k] = v.get_value()
elif isinstance(v, odrive.remote_object.RemoteObject):
elif isinstance(v, fibre.remote_object.RemoteObject):
sub_dict = get_dict(v, k == 'config')
if sub_dict != {}:
result[k] = sub_dict
@@ -24,7 +24,7 @@ def set_dict(obj, path, config_dict):
errors.append("Could not restore {}: property not found on device".format(name))
continue
remote_attribute = obj._remote_attributes[k]
if isinstance(remote_attribute, odrive.remote_object.RemoteObject):
if isinstance(remote_attribute, fibre.remote_object.RemoteObject):
errors += set_dict(remote_attribute, name, v)
else:
try:
@@ -34,7 +34,7 @@ def set_dict(obj, path, config_dict):
return errors
def get_temp_config_filename(device):
serial_number = odrive.utils.get_serial_number_str(device)
serial_number = fibre.utils.get_serial_number_str(device)
safe_serial_number = ''.join(filter(str.isalnum, serial_number))
return os.path.join(tempfile.gettempdir(), 'odrive-config-{}.json'.format(safe_serial_number))
tools/odrive/dfu.py
+2 -2
View File
@@ -402,7 +402,7 @@ def update_device(device, firmware, logger, cancellation_token):
dfudev.jump_to_application(0x08000000)
logger.info("Waiting for the device to reappear...")
device = odrive.discovery.find_any("usb", serial_number,
device = odrive.find_any("usb", serial_number,
cancellation_token, cancellation_token, timeout=30)
if did_backup_config:
@@ -432,7 +432,7 @@ def launch_dfu(args, logger, cancellation_token):
# Scan for ODrives not in DFU mode
# We only scan on USB because DFU is only implemented over USB
devices[1] = odrive.discovery.find_any("usb", serial_number,
devices[1] = odrive.find_any("usb", serial_number,
find_odrive_cancellation_token, cancellation_token)
find_odrive_cancellation_token.set()
tools/odrive/utils.py
-6
View File
@@ -157,12 +157,6 @@ def setup_udev_rules(logger):
subprocess.run(["udevadm", "trigger"], check=True)
logger.info('udev rules configured successfully')
def get_serial_number_str(device):
if hasattr(device, 'serial_number'):
return format(device.serial_number, 'x').upper()
else:
return "[unknown serial number]"
def yes_no_prompt(question, default=None):
if default is None:
question += " [y/n] "
tools/odrivetool
+6 -7
View File
@@ -14,7 +14,6 @@ sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(
import fibre.discovery
from fibre import Logger, Event
import odrive
import odrive.discovery
from odrive.utils import OperationAbortedException
from odrive.configuration import *
@@ -150,7 +149,7 @@ try:
elif args.command == 'liveplotter':
from odrive.utils import start_liveplotter
print("Waiting for ODrive...")
my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number,
my_odrive = odrive.find_any(path=args.path, serial_number=args.serial_number,
search_cancellation_token=app_shutdown_token,
channel_termination_token=app_shutdown_token)
@@ -162,7 +161,7 @@ try:
elif args.command == 'drv-status':
from odrive.utils import print_drv_regs
print("Waiting for ODrive...")
my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number,
my_odrive = odrive.find_any(path=args.path, serial_number=args.serial_number,
search_cancellation_token=app_shutdown_token,
channel_termination_token=app_shutdown_token)
print_drv_regs("Motor 0", my_odrive.axis0.motor)
@@ -171,7 +170,7 @@ try:
elif args.command == 'rate-test':
from odrive.utils import rate_test
print("Waiting for ODrive...")
my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number,
my_odrive = odrive.find_any(path=args.path, serial_number=args.serial_number,
search_cancellation_token=app_shutdown_token,
channel_termination_token=app_shutdown_token)
rate_test(my_odrive)
@@ -182,14 +181,14 @@ try:
elif args.command == 'generate-code':
from odrive.code_generator import generate_code
my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number,
my_odrive = odrive.find_any(path=args.path, serial_number=args.serial_number,
channel_termination_token=app_shutdown_token)
generate_code(my_odrive, args.template, args.output)
elif args.command == 'backup-config':
from odrive.configuration import backup_config
print("Waiting for ODrive...")
my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number,
my_odrive = odrive.find_any(path=args.path, serial_number=args.serial_number,
search_cancellation_token=app_shutdown_token,
channel_termination_token=app_shutdown_token)
backup_config(my_odrive, args.file, logger)
@@ -197,7 +196,7 @@ try:
elif args.command == 'restore-config':
from odrive.configuration import restore_config
print("Waiting for ODrive...")
my_odrive = odrive.discovery.find_any(path=args.path, serial_number=args.serial_number,
my_odrive = odrive.find_any(path=args.path, serial_number=args.serial_number,
search_cancellation_token=app_shutdown_token,
channel_termination_token=app_shutdown_token)
restore_config(my_odrive, args.file, logger)