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

Merge commit '40b0df60e600d07484b8ef3d5fa2e67db7f88b80' into libfibre

Browse Source
This commit is contained in:
Samuel Sadok
2021-01-11 17:09:08 +01:00
parent ad65ff71c6 40b0df60e6
commit 810435220c
3 changed files with 149 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/odrive/pyfibre/fibre/libfibre.py
+63 -74
View File
@@ -591,27 +591,39 @@ class RemoteAttribute(object):
self._magic_setter = magic_setter
def _get_obj(self, instance):
py_intf = self._libfibre._load_py_intf(self._intf_name, self._intf_handle)
obj_handle = c_void_p(0)
status = libfibre_get_attribute(instance._obj_handle, self._attr_handle, byref(obj_handle))
if status != kFibreOk:
raise _get_exception(status)
return self._libfibre._objects[obj_handle.value]
obj = self._libfibre._load_py_obj(obj_handle.value, self._intf_handle)
if obj in instance._children:
self._libfibre._release_py_obj(obj_handle.value)
else:
# the object will be released when the parent is released
instance._children.add(obj)
return obj
def __get__(self, instance, owner):
if not instance:
return self
if self._magic_getter:
if threading.current_thread() == libfibre_thread:
# read() behaves asynchronously when run on the fibre thread
# which means it returns an awaitable which _must_ be awaited
# (otherwise it's a bug). However hasattr(...) internally calls
# __get__ and does not await the result. Thus the safest thing
# is to just disallow __get__ from run as an async method.
raise Exception("Cannot use magic getter on Fibre thread. Use _[prop_name]_propery.read() instead.")
return self._get_obj(instance).read()
else:
return self._get_obj(instance)
def __set__(self, instance, val):
if self._magic_setter:
self._get_obj(instance).exchange(val)
return self._get_obj(instance).exchange(val)
else:
raise Exception("this attribute cannot be written to")
@@ -624,6 +636,8 @@ class RemoteObject(object):
def __init__(self, libfibre, obj_handle):
self.__class__._refcount += 1
self._refcount = 0
self._children = set()
self._libfibre = libfibre
self._obj_handle = obj_handle
@@ -682,10 +696,15 @@ class RemoteObject(object):
def _destroy(self):
libfibre = self._libfibre
on_lost = self._on_lost
children = self._children
self._libfibre = None
self._obj_handle = None
self._on_lost = None
self._children = set()
for child in children:
libfibre._release_py_obj(child._obj_handle)
self.__class__._refcount -= 1
if self.__class__._refcount == 0:
@@ -729,6 +748,7 @@ class LibFibre():
event_loop.cancel_timer = self.c_cancel_timer
self.ctx = c_void_p(libfibre_open(event_loop))
assert(self.ctx)
def _post(self, callback, ctx):
self.loop.call_soon_threadsafe(callback, ctx)
@@ -786,26 +806,19 @@ class LibFibre():
self._objects[obj_handle] = py_obj
else:
py_obj = self._objects[obj_handle]
# Note: this refcount does not count the python references to the object
# but rather mirrors the libfibre-internal refcount of the object. This
# is so that we can destroy the Python object when libfibre releases it.
py_obj._refcount += 1
return py_obj
def _release_py_obj(self, obj_handle):
py_obj = self._objects[obj_handle]
py_obj._refcount -= 1
if py_obj.refcount <= 0:
if py_obj._refcount <= 0:
self._objects.pop(obj_handle)
#def _construct_object(self, ctx, obj, intf, name, name_length):
# #increment_lib_refcount()
# name = None if name is None else string_at(name, name_length).decode('utf-8')
# py_intf = self._load_py_intf(name, intf)
# assert(not obj in self._objects)
# self._objects[obj] = py_intf(self, obj)
def _free_py_obj(self, ctx, obj):
py_obj = self._objects.pop(obj)
py_obj._destroy()
#decrement_lib_refcount()
py_obj._destroy()
def _on_found_object(self, ctx, obj, intf):
py_obj = self._load_py_obj(obj, intf)
@@ -816,7 +829,7 @@ class LibFibre():
old_future.set_result(None)
def _on_lost_object(self, ctx, obj):
self._free_py_obj(obj)
self._release_py_obj(obj)
def _on_discovery_stopped(self, ctx, result):
print("discovery stopped")
@@ -853,31 +866,11 @@ class LibFibre():
return kFibreBusy
# def start_discovery(self, path, on_obj_discovered, cancellation_token):
# buf = path.encode('ascii')
#
# discovery = {
# 'domain_handle': c_void_p(0),
# 'handle': c_void_p(0),
# 'callback': on_obj_discovered
# }
# discovery_id = insert_with_new_id(self.discovery_processes, discovery)
#
# def stop_discovery():
# libfibre_stop_discovery(discovery['handle'])
# print("closing domain")
#
#
# print("ok ")
# cancellation_token.subscribe(lambda: stop_discovery)
#
# discovery['domain_handle'] = libfibre_open_domain(self.ctx, buf, len(buf))
# assert(discovery['domain_handle'])
# libfibre_start_discovery(discovery['domain_handle'], byref(discovery['handle']), self.c_on_found_object, self.c_on_lost_object, self.c_on_discovery_stopped, discovery_id)
# print("disc handle ", hex(discovery['handle'].value))
#
class Discovery():
"""
All public members of this class are thread-safe.
"""
def __init__(self, domain):
self._domain = domain
self._id = 0
@@ -893,8 +886,19 @@ class Discovery():
def _stop(self):
self._domain._libfibre.discovery_processes.pop(self._id)
libfibre_stop_discovery(self._discovery_handle)
self._future.set_exception(asyncio.CancelledError())
def stop(self):
if threading.current_thread() == libfibre_thread:
self._stop()
else:
run_coroutine_threadsafe(self._domain._libfibre.loop, self._stop)
class _Domain():
"""
All public members of this class are thread-safe.
"""
def __init__(self, libfibre, handle):
self._libfibre = libfibre
self._domain_handle = handle
@@ -917,8 +921,25 @@ class _Domain():
return obj
def discover_one(self):
"""
Blocks until exactly one object is discovered.
"""
return run_coroutine_threadsafe(self._libfibre.loop, self._discover_one)
def run_discovery(self, callback):
"""
Invokes `callback` for every object that is discovered. The callback is
invoked on the libfibre thread and can be an asynchronous function.
Returns a `Discovery` object on which `stop()` can be called to
terminate the discovery.
"""
discovery = run_coroutine_threadsafe(self._libfibre.loop, self._start_discovery)
async def loop():
while True:
obj = await discovery._next()
await callback(obj)
self._libfibre.loop.call_soon_threadsafe(lambda: asyncio.ensure_future(loop()))
return discovery
class Domain():
@@ -1006,38 +1027,6 @@ def decrement_lib_refcount():
libfibre_thread.join()
libfibre_thread = None
#def start_discovery(path, obj_filter,
# on_object_discovered,
# search_cancellation_token,
# channel_termination_token):
# """
# Starts scanning for Fibre objects that match the specified path spec and calls
# the callback for each Fibre object that is found.
#
# This function is non-blocking and thread-safe.
# """
#
# async def on_object_discovered_filter(obj):
# increment_lib_refcount()
# if not channel_termination_token is None:
# channel_termination_token.subscribe(lambda: decrement_lib_refcount())
# if await obj_filter(obj):
# result = on_object_discovered(obj)
# if not result is None:
# await result
# elif not channel_termination_token is None:
# channel_termination_token.set()
#
# increment_lib_refcount()
# search_cancellation_token.subscribe(lambda: decrement_lib_refcount())
#
# libfibre.loop.call_soon_threadsafe(lambda: libfibre.start_discovery(
# path,
# on_object_discovered_filter,
# search_cancellation_token))
def get_user_name(obj):
"""
Can be overridden by the application to return the user-facing name of an
tools/odrive/pyfibre/fibre/shell.py
+80 -79
View File
@@ -4,42 +4,47 @@ import platform
import threading
import fibre
async def did_discover_device(device,
async def discovered_device(device,
interactive_variables, discovered_devices,
branding_short, branding_long,
logger, app_shutdown_token):
mount, shutdown_token, logger):
"""
Handles the discovery of new devices by displaying a
message and making the device available to the interactive
console
"""
serial_number = '{:012X}'.format(await device.serial_number) if hasattr(device, 'serial_number') else "[unknown serial number]"
if serial_number in discovered_devices:
mount_result = await mount(device)
if mount_result is None:
logger.debug("ignoring device")
return
display_name, var_name = mount_result
if display_name in discovered_devices:
verb = "Reconnected"
index = discovered_devices.index(serial_number)
index = discovered_devices.index(display_name)
else:
verb = "Connected"
discovered_devices.append(serial_number)
discovered_devices.append(display_name)
index = len(discovered_devices) - 1
interactive_name = branding_short + str(index)
var_name = var_name + str(index)
# Publish new device to interactive console
interactive_variables[interactive_name] = device
globals()[interactive_name] = device # Add to globals so tab complete works
logger.notify("{} to {} {} as {}".format(verb, branding_long, serial_number, interactive_name))
interactive_variables[var_name] = device
globals()[var_name] = device # Add to globals so tab complete works
logger.notify("{} to {} as {}".format(verb, display_name, var_name))
# Subscribe to disappearance of the device
device._on_lost.add_done_callback(lambda x: did_lose_device(interactive_name, logger, app_shutdown_token))
device._on_lost.add_done_callback(lambda x: lost_device(var_name, shutdown_token, logger))
def did_lose_device(interactive_name, logger, app_shutdown_token):
def lost_device(interactive_name, shutdown_token, logger):
"""
Handles the disappearance of a device by displaying
a message.
"""
if not app_shutdown_token.is_set():
if not shutdown_token[0]:
logger.warn("Oh no {} disappeared".format(interactive_name))
def get_user_name(interactive_variables, obj):
queue = [(k, v) for k, v in interactive_variables.items() if isinstance(v, fibre.libfibre.RemoteObject)]
@@ -57,12 +62,10 @@ def get_user_name(interactive_variables, obj):
return "anonymous_remote_object_" + str(self._obj_handle)
def launch_shell(args,
object_filter,
def launch_shell(args, mount,
interactive_variables,
print_banner, print_help,
logger, app_shutdown_token,
branding_short="dev", branding_long="device"):
logger):
"""
Launches an interactive python or IPython command line
interface.
@@ -72,78 +75,76 @@ def launch_shell(args,
"""
discovered_devices = []
shutdown_token = [False]
globals().update(interactive_variables)
fibre.libfibre.get_user_name = lambda obj: get_user_name(interactive_variables, obj)
# Connect to device
logger.debug("Waiting for {}...".format(branding_long))
fibre.start_discovery(args.path, object_filter,
lambda dev: did_discover_device(dev, interactive_variables, discovered_devices, branding_short, branding_long, logger, app_shutdown_token),
app_shutdown_token,
app_shutdown_token,
logger=logger)
with fibre.Domain(args.path) as domain:
on_discovery = lambda dev: discovered_device(dev, interactive_variables, discovered_devices, mount, shutdown_token, logger)
discovery = domain.run_discovery(on_discovery)
# Check if IPython is installed
if args.no_ipython:
use_ipython = False
else:
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")
# Check if IPython is installed
if args.no_ipython:
use_ipython = False
else:
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
interactive_variables["help"] = lambda: print_help(args, len(discovered_devices) > 0)
interactive_variables["help"] = lambda: print_help(args, len(discovered_devices) > 0)
# If IPython is installed, embed IPython shell, otherwise embed regular shell
if use_ipython:
# Override help function # pylint: disable=W0612
help = lambda: print_help(args, len(discovered_devices) > 0)
# to fix broken "%run -i script.py"
locals()['__name__'] = globals()['__name__']
console = IPython.terminal.embed.InteractiveShellEmbed(banner1='')
# If IPython is installed, embed IPython shell, otherwise embed regular shell
if use_ipython:
# Override help function # pylint: disable=W0612
help = lambda: print_help(args, len(discovered_devices) > 0)
# to fix broken "%run -i script.py"
locals()['__name__'] = globals()['__name__']
console = IPython.terminal.embed.InteractiveShellEmbed(banner1='')
# hack to make IPython look like the regular console
console.runcode = console.run_cell
interact = console
# hack to make IPython look like the regular console
console.runcode = console.run_cell
interact = console
# Catch ObjectLostError (since disconnect is not always an error)
default_exception_hook = console._showtraceback
def filtered_exception_hook(ex_class, ex, trace):
if(ex_class.__module__+'.'+ex_class.__name__ != 'fibre.libfibre.ObjectLostError'):
default_exception_hook(ex_class,ex,trace)
console._showtraceback = filtered_exception_hook
else:
# Enable tab complete if possible
try:
import readline # Works only on Unix
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))
# Catch ObjectLostError (since disconnect is not always an error)
default_exception_hook = console._showtraceback
def filtered_exception_hook(ex_class, ex, trace):
if(ex_class.__module__+'.'+ex_class.__name__ != 'fibre.libfibre.ObjectLostError'):
default_exception_hook(ex_class,ex,trace)
console._showtraceback = filtered_exception_hook
else:
# Enable tab complete if possible
try:
import readline # Works only on Unix
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='')
import code
console = code.InteractiveConsole(locals=interactive_variables)
interact = lambda: console.interact(banner='')
# Catch ObjectLostError (since disconnect is not alway an error)
console.runcode("import sys")
console.runcode("default_exception_hook = sys.excepthook")
console.runcode("def filtered_exception_hook(ex_class, ex, trace):\n"
" if ex_class.__module__ + '.' + ex_class.__name__ != 'fibre.libfibre.ObjectLostError':\n"
" default_exception_hook(ex_class,ex,trace)")
console.runcode("sys.excepthook=filtered_exception_hook")
# Catch ObjectLostError (since disconnect is not alway an error)
console.runcode("import sys")
console.runcode("default_exception_hook = sys.excepthook")
console.runcode("def filtered_exception_hook(ex_class, ex, trace):\n"
" if ex_class.__module__ + '.' + ex_class.__name__ != 'fibre.libfibre.ObjectLostError':\n"
" default_exception_hook(ex_class,ex,trace)")
console.runcode("sys.excepthook=filtered_exception_hook")
# Launch shell
print_banner()
logger._skip_bottom_line = True
interact()
# Launch shell
print_banner()
logger._skip_bottom_line = True
interact()
app_shutdown_token.set()
shutdown_token[0] = True
tools/odrive/pyfibre/fibre/utils.py
+6 -1
View File
@@ -46,15 +46,20 @@ class Event():
Sets the event and invokes all subscribers if the event was
not already set
"""
subscribers = []
self._mutex.acquire()
try:
if not self._evt.is_set():
self._evt.set()
for s in self._subscribers:
s()
subscribers.append(s)
finally:
self._mutex.release()
# Invoke subscribes with the mutex released to prevent deadlocks
for s in subscribers:
s()
def subscribe(self, handler):
"""
Invokes the specified handler exactly once as soon as the