From 2b3df47e853f34eb9de40b212d2e0aab13120ac5 Mon Sep 17 00:00:00 2001 From: Samuel Sadok Date: Tue, 15 Sep 2020 13:30:40 +0200 Subject: [PATCH] Squashed 'tools/odrive/pyfibre/' content from commit f9ebfdd3 git-subtree-dir: tools/odrive/pyfibre git-subtree-split: f9ebfdd3b4c9d5b907ed145f0285b67c7f0f7eb6 --- .gitignore | 16 ++ fibre/__init__.py | 4 + fibre/libfibre.py | 678 ++++++++++++++++++++++++++++++++++++++++++++++ fibre/protocol.py | 359 ++++++++++++++++++++++++ fibre/shell.py | 148 ++++++++++ fibre/utils.py | 238 ++++++++++++++++ setup.py | 88 ++++++ 7 files changed, 1531 insertions(+) create mode 100644 .gitignore create mode 100644 fibre/__init__.py create mode 100644 fibre/libfibre.py create mode 100644 fibre/protocol.py create mode 100644 fibre/shell.py create mode 100644 fibre/utils.py create mode 100644 setup.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 00000000..826d149c --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ + +# Python Distribution / packaging +.Python +/dist/ +/*.egg-info/ +/MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt diff --git a/fibre/__init__.py b/fibre/__init__.py new file mode 100644 index 00000000..216f2e06 --- /dev/null +++ b/fibre/__init__.py @@ -0,0 +1,4 @@ + +from .utils import Event, Logger, TimeoutError +from .shell import launch_shell +from .libfibre import find_all, find_any, ObjectLostError diff --git a/fibre/libfibre.py b/fibre/libfibre.py new file mode 100644 index 00000000..038f51d7 --- /dev/null +++ b/fibre/libfibre.py @@ -0,0 +1,678 @@ +#!/bin/python + +from ctypes import * +import asyncio +import os +from itertools import count, takewhile +import struct +from types import MethodType +import concurrent +import threading +import time +from fibre.utils import Logger, Event +import platform + +lib_names = { + ('Linux', 'x86_64'): 'libfibre-linux-amd64.so', + ('Linux', 'armv7l'): 'libfibre-linux-armhf.so', + ('Windows', 'AMD64'): 'libfibre-windows-amd64.dll', + ('Darwin', 'x86_64'): 'libfibre-macos-x86.dylib' +} + +system_desc = (platform.system(), platform.machine()) + +lib_dir = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))), + 'cpp') + +def test_path(path): + return path if os.path.isfile(path) else None + +lib_path = (test_path(os.path.join(lib_dir, 'libfibre.so')) or + test_path(os.path.join(lib_dir, 'libfibre.dll')) or + (test_path(os.path.join(lib_dir, lib_names[system_desc])) if (system_desc in lib_names) else None)) + +if lib_path is None: + raise ModuleNotFoundError("This package has no precompiled libfibre for your platform ({} {}). " + "Go to fibre/cpp/ and run `make` to compile libfibre for your platform.".format(*system_desc)) + +lib = windll.LoadLibrary(lib_path) if os.name == 'nt' else cdll.LoadLibrary(lib_path) + + +# libfibre definitions --------------------------------------------------------# + +PostSignature = CFUNCTYPE(c_void_p, CFUNCTYPE(None, c_void_p), POINTER(c_int)) +RegisterEventSignature = CFUNCTYPE(c_int, c_int, c_uint32, CFUNCTYPE(None, c_void_p), POINTER(c_int)) +DeregisterEventSignature = CFUNCTYPE(c_int, c_int) +CallLaterSignature = CFUNCTYPE(c_void_p, c_float, CFUNCTYPE(None, c_void_p), POINTER(c_int)) +CancelTimerSignature = CFUNCTYPE(c_int, c_void_p) +ConstructObjectSignature = CFUNCTYPE(None, c_void_p, c_void_p, c_void_p, c_void_p, c_size_t) +DestroyObjectSignature = CFUNCTYPE(None, c_void_p, c_void_p) + +OnFoundObjectSignature = CFUNCTYPE(None, c_void_p, c_void_p) +OnStoppedSignature = CFUNCTYPE(None, c_void_p, c_int) + +OnAttributeAddedSignature = CFUNCTYPE(None, c_void_p, c_void_p, c_void_p, c_size_t, c_void_p, c_void_p, c_size_t) +OnAttributeRemovedSignature = CFUNCTYPE(None, c_void_p, c_void_p) +OnFunctionAddedSignature = CFUNCTYPE(None, c_void_p, c_void_p, c_void_p, c_size_t, POINTER(c_char_p), POINTER(c_char_p), POINTER(c_char_p), POINTER(c_char_p)) +OnFunctionRemovedSignature = CFUNCTYPE(None, c_void_p, c_void_p) + +OnCallCompletedSignature = CFUNCTYPE(None, c_void_p, c_int, c_char_p) + +kFibreOk = 0 +kFibreCancelled = 1 +kFibreClosed = 2 +kFibreInvalidArgument = 3 +kFibreInternalError = 4 + +class LibFibreVersion(Structure): + _fields_ = [ + ("major", c_uint16), + ("minor", c_uint16), + ("patch", c_uint16), + ] + + def __repr__(self): + return "{}.{}.{}".format(self.major, self.minor, self.patch) + +libfibre_get_version = lib.libfibre_get_version +libfibre_get_version.argtypes = [] +libfibre_get_version.restype = POINTER(LibFibreVersion) + +version = libfibre_get_version().contents +if version.major != 0: + raise Exception("Incompatible libfibre version: {}".format(version)) + +libfibre_open = lib.libfibre_open +libfibre_open.argtypes = [PostSignature, RegisterEventSignature, DeregisterEventSignature, CallLaterSignature, CancelTimerSignature, ConstructObjectSignature, DestroyObjectSignature, c_void_p] +libfibre_open.restype = c_void_p + +libfibre_close = lib.libfibre_close +libfibre_close.argtypes = [c_void_p] +libfibre_close.restype = None + +libfibre_start_discovery = lib.libfibre_start_discovery +libfibre_start_discovery.argtypes = [c_void_p, c_char_p, c_size_t, c_void_p, OnFoundObjectSignature, OnStoppedSignature, c_void_p] +libfibre_start_discovery.restype = c_void_p + +libfibre_stop_discovery = lib.libfibre_stop_discovery +libfibre_stop_discovery.argtypes = [c_void_p, c_void_p] +libfibre_stop_discovery.restype = None + +libfibre_subscribe_to_interface = lib.libfibre_subscribe_to_interface +libfibre_subscribe_to_interface.argtypes = [c_void_p, OnAttributeAddedSignature, OnAttributeRemovedSignature, OnFunctionAddedSignature, OnFunctionRemovedSignature, c_void_p] +libfibre_subscribe_to_interface.restype = None + +libfibre_get_attribute = lib.libfibre_get_attribute +libfibre_get_attribute.argtypes = [c_void_p, c_void_p, POINTER(c_void_p)] +libfibre_get_attribute.restype = c_int + +libfibre_start_call = lib.libfibre_start_call +libfibre_start_call.argtypes = [c_void_p, c_void_p, c_char_p, c_size_t, c_char_p, c_size_t, c_void_p, OnCallCompletedSignature, c_void_p] +libfibre_start_call.restype = None + +libfibre_cancel_call = lib.libfibre_cancel_call +libfibre_cancel_call.argtypes = [c_void_p] +libfibre_cancel_call.restype = None + + +# libfibre wrapper ------------------------------------------------------------# + +class ObjectLostError(Exception): + def __init__(self): + super(Exception, self).__init__("the object disappeared") + +def _get_exception(status): + if status == kFibreOk: + return None + elif status == kFibreCancelled: + return asyncio.CancelledError() + elif status == kFibreClosed: + return ObjectLostError() + elif status == kFibreInvalidArgument: + return ArgumentError() + elif status == kFibreInternalError: + return Exception("internal libfibre error") + else: + return Exception("unknown libfibre error {}".format(status)) + +class StructCodec(): + """ + Generic serializer/deserializer based on struct pack + """ + def __init__(self, struct_format, target_type): + self._struct_format = struct_format + self._target_type = target_type + def get_length(self): + return struct.calcsize(self._struct_format) + def serialize(self, libfibre, value): + value = self._target_type(value) + return struct.pack(self._struct_format, value) + def deserialize(self, libfibre, buffer): + value = struct.unpack(self._struct_format, buffer) + value = value[0] if len(value) == 1 else value + return self._target_type(value) + +class ObjectPtrCodec(): + """ + Serializer/deserializer for an object reference + + libfibre transcodes object references internally from/to something that can + be sent over the wire and understood by the remote instance. + """ + def get_length(self): + return struct.calcsize("P") + def serialize(self, libfibre, value): + if value is None: + return struct.pack("P", 0) + elif isinstance(value, RemoteObject): + return struct.pack("P", value._obj_handle) + else: + raise TypeError("Expected value of type RemoteObject or None but got '{}'. An example for a RemoteObject is this expression: odrv0.axis0.controller._input_pos_property".format(type(value).__name__)) + def deserialize(self, libfibre, buffer): + handle = struct.unpack("P", buffer)[0] + return None if handle == 0 else libfibre._objects[handle] + + +codecs = { + 'int8': StructCodec(" " + print_arglist(self._outputs) if len(self._outputs) == 1 else + " -> (" + print_arglist(self._outputs) + ")") + +class RemoteAttribute(object): + def __init__(self, libfibre, attr_handle, intf_handle, intf_name, magic_getter, magic_setter): + self._libfibre = libfibre + self._attr_handle = attr_handle + self._intf_handle = intf_handle + self._intf_name = intf_name + self._magic_getter = magic_getter + 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] + + def __get__(self, instance, owner): + if not instance: + return self + + if self._magic_getter: + 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) + else: + raise Exception("this attribute cannot be written to") + + +class RemoteObject(object): + """ + Base class for interfaces of remote objects. + """ + __sealed__ = False + + def __init__(self, libfibre, obj_handle): + self.__class__._refcount += 1 + + self._libfibre = libfibre + self._obj_handle = obj_handle + self._on_lost = concurrent.futures.Future() # TODO: maybe we can do this with conc + + # Ensure that assignments to undefined attributes raise an exception + self.__sealed__ = True + + def __setattr__(self, key, value): + if self.__sealed__ and not hasattr(self, key): + raise AttributeError("Attribute {} not found".format(key)) + object.__setattr__(self, key, value) + + #def __del__(self): + # print("unref") + # libfibre_unref_obj(self._obj_handle) + + def _dump(self, indent, depth): + if self._obj_handle is None: + return "[object lost]" + + try: + if depth <= 0: + return "..." + lines = [] + for key in dir(self.__class__): + if key.startswith('_'): + continue + class_member = getattr(self.__class__, key) + if isinstance(class_member, RemoteFunction): + lines.append(indent + class_member._dump(key)) + elif isinstance(class_member, RemoteAttribute): + val = getattr(self, key) + if isinstance(val, RemoteObject) and not class_member._magic_getter: + lines.append(indent + key + (": " if depth == 1 else ":\n") + val._dump(indent + " ", depth - 1)) + else: + if isinstance(val, RemoteObject) and class_member._magic_getter: + val_str = get_user_name(val) + else: + val_str = str(val) + property_type = str(class_member._get_obj(self).__class__.read._outputs[0][1]) + lines.append(indent + key + ": " + val_str + " (" + property_type + ")") + else: + lines.append(indent + key + ": " + str(type(val))) + except: + return "[failed to dump object]" + + return "\n".join(lines) + + def __str__(self): + return self._dump("", depth=2) + + def __repr__(self): + return self.__str__() + + def _destroy(self): + libfibre = self._libfibre + on_lost = self._on_lost + + self._libfibre = None + self._obj_handle = None + self._on_lost = None + + self.__class__._refcount -= 1 + if self.__class__._refcount == 0: + libfibre.interfaces.pop(self.__class__._handle) + + on_lost.set_result(True) + + +class LibFibre(): + def __init__(self): + self.loop = asyncio.get_event_loop() + + # We must keep a reference to these function objects so they don't get + # garbage collected. + self.c_post = PostSignature(self._post) + self.c_register_event = RegisterEventSignature(self._register_event) + self.c_deregister_event = DeregisterEventSignature(self._deregister_event) + self.c_call_later = CallLaterSignature(self._call_later) + self.c_cancel_timer = CancelTimerSignature(self._cancel_timer) + self.c_construct_object = ConstructObjectSignature(self._construct_object) + self.c_destroy_object = DestroyObjectSignature(self._destroy_object) + self.c_on_found_object = OnFoundObjectSignature(self._on_found_object) + self.c_on_discovery_stopped = OnStoppedSignature(self._on_discovery_stopped) + self.c_on_attribute_added = OnAttributeAddedSignature(self._on_attribute_added) + self.c_on_attribute_removed = OnAttributeRemovedSignature(self._on_attribute_removed) + self.c_on_function_added = OnFunctionAddedSignature(self._on_function_added) + self.c_on_function_removed = OnFunctionRemovedSignature(self._on_function_removed) + + self.timer_map = {} + self.eventfd_map = {} + self.interfaces = {} # key: libfibre handle, value: python class + self.discovery_processes = {} # key: ID, value: python dict + self._objects = {} # key: libfibre handle, value: pyhton class + + self.ctx = c_void_p(libfibre_open( + self.c_post, + self.c_register_event, self.c_deregister_event, + self.c_call_later, self.c_cancel_timer, + self.c_construct_object, self.c_destroy_object, None)) + + def _post(self, callback, ctx): + self.loop.call_soon_threadsafe(callback, ctx) + + def _register_event(self, event_fd, events, callback, ctx): + self.eventfd_map[event_fd] = events + if (events & 1): + self.loop.add_reader(event_fd, callback, ctx) + if (events & 4): + self.loop.add_writer(event_fd, callback, ctx) + if (events & 0xfffffffa): + raise Exception("unsupported event mask " + str(events)) + return 0 + + def _deregister_event(self, event_fd): + events = self.eventfd_map.pop(event_fd) + if (events & 1): + self.loop.remove_reader(event_fd) + if (events & 4): + self.loop.remove_writer(event_fd) + return 0 + + def _call_later(self, delay, callback, ctx): + timer_id = insert_with_new_id(self.timer_map, self.loop.call_later(delay, callback, ctx)) + return timer_id + + def _cancel_timer(self, timer_id): + self.timer_map.pop(timer_id).cancel() + return 0 + + def _load_py_intf(self, name, intf_handle): + """ + Creates a new python type for the specified libfibre interface handle or + returns the existing python type if one was already create before. + + Behind the scenes the python type will react to future events coming + from libfibre, such as functions/attributes being added/removed. + """ + if intf_handle in self.interfaces: + return self.interfaces[intf_handle] + else: + if name is None: + name = "anonymous_interface_" + str(intf_handle) + py_intf = self.interfaces[intf_handle] = type(name, (RemoteObject,), {'_handle': intf_handle, '_refcount': 0}) + #exit(1) + libfibre_subscribe_to_interface(intf_handle, self.c_on_attribute_added, self.c_on_attribute_removed, self.c_on_function_added, self.c_on_function_removed, intf_handle) + return py_intf + + def _construct_object(self, ctx, obj, intf, name, name_length): + #increment_libfibre_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 _destroy_object(self, ctx, obj): + py_obj = self._objects.pop(obj) + py_obj._destroy() + #decrement_lib_refcount() + + def _on_found_object(self, ctx, obj): + py_obj = self._objects[obj] + # notify the subscriber + asyncio.ensure_future(self.discovery_processes[ctx]['callback'](py_obj)) + + def _on_discovery_stopped(self, ctx, result): + print("discovery stopped") + + def _on_attribute_added(self, ctx, attr, name, name_length, subintf, subintf_name, subintf_name_length): + name = string_at(name, name_length).decode('utf-8') + subintf_name = None if subintf_name is None else string_at(subintf_name, subintf_name_length).decode('utf-8') + intf = self.interfaces[ctx] + + magic_getter = not subintf_name is None and subintf_name.startswith("fibre.Property<") and subintf_name.endswith(">") + magic_setter = not subintf_name is None and subintf_name.startswith("fibre.Property") + + setattr(intf, name, RemoteAttribute(self, attr, subintf, subintf_name, magic_getter, magic_setter)) + if magic_getter or magic_setter: + setattr(intf, "_" + name + "_property", RemoteAttribute(self, attr, subintf, subintf_name, False, False)) + + def _on_attribute_removed(self, ctx, attr): + print("attribute removed") + + def _on_function_added(self, ctx, func, name, name_length, input_names, input_codecs, output_names, output_codecs): + name = string_at(name, name_length).decode('utf-8') + inputs = list(decode_arg_list(input_names, input_codecs)) + outputs = list(decode_arg_list(output_names, output_codecs)) + intf = self.interfaces[ctx] + setattr(intf, name, RemoteFunction(self, func, inputs, outputs)) + + def _on_function_removed(self, ctx, func): + print("function removed") + + def start_discovery(self, path, on_obj_discovered, cancellation_token): + buf = path.encode('ascii') + + discovery = { + 'handle': c_void_p(0), + 'callback': on_obj_discovered + } + discovery_id = insert_with_new_id(self.discovery_processes, discovery) + + cancellation_token.subscribe(lambda: libfibre_stop_discovery(self.ctx, discovery['handle'])) + libfibre_start_discovery(self.ctx, buf, len(buf), byref(discovery['handle']), self.c_on_found_object, self.c_on_discovery_stopped, discovery_id) + + + +libfibre = None + +def run_event_loop(): + global libfibre + global terminate_libfibre + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + terminate_libfibre = loop.create_future() + libfibre = LibFibre() + + libfibre.loop.run_until_complete(terminate_libfibre) + + libfibre_close(libfibre.ctx) + + # Detach all objects that still exist + # TODO: the proper way would be either of these + # - provide a libfibre function to destroy an object on-demand which we'd + # call before libfibre_close(). + # - have libfibre_close() report the destruction of all objects + + while len(libfibre._objects): + libfibre._objects.pop(list(libfibre._objects.keys())[0])._destroy() + assert(len(libfibre.interfaces) == 0) + + libfibre = None + + +lock = threading.Lock() +libfibre_refcount = 0 +libfibre_thread = None + +def increment_libfibre_refcount(): + global libfibre_refcount + global libfibre_thread + + with lock: + libfibre_refcount += 1 + #print("inc refcount to {}".format(libfibre_refcount)) + + if libfibre_refcount == 1: + libfibre_thread = threading.Thread(target = run_event_loop) + libfibre_thread.start() + + while libfibre is None: + time.sleep(0.1) + +def decrement_lib_refcount(): + global libfibre_refcount + global libfibre_thread + + with lock: + #print("dec refcount from {}".format(libfibre_refcount)) + libfibre_refcount -= 1 + + if libfibre_refcount == 0: + libfibre.loop.call_soon_threadsafe(lambda: terminate_libfibre.set_result(True)) + + # It's unlikely that releasing fibre from a fibre callback is ok. If + # there is a valid scenario for this then we can remove the assert. + assert(libfibre_thread != threading.current_thread()) + + libfibre_thread.join() + libfibre_thread = None + + +def find_all(path, serial_number, + on_object_discovered, + search_cancellation_token, + channel_termination_token, + logger): + """ + 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_libfibre_refcount() + channel_termination_token.subscribe(lambda: decrement_lib_refcount()) + if serial_number is None or (await fibre.utils.get_serial_number_str(obj)) == serial_number: + result = on_object_discovered(obj) + if not result is None: + await result + + increment_libfibre_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 find_any(path="usb", serial_number=None, + search_cancellation_token=None, channel_termination_token=None, + timeout=None, logger=Logger(verbose=False), find_multiple=False): + """ + Blocks until the first matching Fibre object is connected and then returns that object + """ + result = [] + done_signal = Event(search_cancellation_token) + def did_discover_object(obj): + result.append(obj) + if find_multiple: + if len(result) >= int(find_multiple): + done_signal.set() + else: + done_signal.set() + + find_all(path, serial_number, did_discover_object, done_signal, channel_termination_token, logger) + + try: + done_signal.wait(timeout=timeout) + except TimeoutError: + if not find_multiple: + return None + finally: + done_signal.set() # terminate find_all + + if find_multiple: + return result + else: + return result[0] if len(result) > 0 else None + + +def get_user_name(obj): + """ + Can be overridden by the application to return the user-facing name of an + object. + """ + return "[anonymous object]" diff --git a/fibre/protocol.py b/fibre/protocol.py new file mode 100644 index 00000000..d7d38e30 --- /dev/null +++ b/fibre/protocol.py @@ -0,0 +1,359 @@ +# See protocol.hpp for an overview of the protocol + +import time +import struct +import sys +import threading +import traceback +#import fibre.utils +from fibre.utils import Event, wait_any, TimeoutError + +import abc +if sys.version_info >= (3, 4): + ABC = abc.ABC +else: + ABC = abc.ABCMeta('ABC', (), {}) + +if sys.version_info < (3, 3): + from monotonic import monotonic + time.monotonic = monotonic + +SYNC_BYTE = 0xAA +CRC8_INIT = 0x42 +CRC16_INIT = 0x1337 +PROTOCOL_VERSION = 1 + +CRC8_DEFAULT = 0x37 # this must match the polynomial in the C++ implementation +CRC16_DEFAULT = 0x3d65 # this must match the polynomial in the C++ implementation + +MAX_PACKET_SIZE = 128 + +# For more information on the CRC algorithm refer to protocol.md + +def calc_crc(remainder, value, polynomial, bitwidth): + topbit = (1 << (bitwidth - 1)) + + # Bring the next byte into the remainder. + remainder ^= (value << (bitwidth - 8)) + for bitnumber in range(0,8): + if (remainder & topbit): + remainder = (remainder << 1) ^ polynomial + else: + remainder = (remainder << 1) + + return remainder & ((1 << bitwidth) - 1) + +def calc_crc8(remainder, value): + if isinstance(value, bytearray) or isinstance(value, bytes) or isinstance(value, list): + for byte in value: + if not isinstance(byte,int): + byte = ord(byte) + remainder = calc_crc(remainder, byte, CRC8_DEFAULT, 8) + else: + remainder = calc_crc(remainder, byte, CRC8_DEFAULT, 8) + return remainder + +def calc_crc16(remainder, value): + if isinstance(value, bytearray) or isinstance(value, bytes) or isinstance(value, list): + for byte in value: + if not isinstance(byte, int): + byte = ord(byte) + remainder = calc_crc(remainder, byte, CRC16_DEFAULT, 16) + else: + remainder = calc_crc(remainder, value, CRC16_DEFAULT, 16) + return remainder + + +class DeviceInitException(Exception): + pass + +class ChannelDamagedException(Exception): + """ + Raised when the channel is temporarily broken and a + resend of the message might be successful + """ + pass + +class ObjectLostError(Exception): + """ + Raised when the channel is permanently broken + """ + pass + + +class StreamSource(ABC): + @abc.abstractmethod + def get_bytes(self, n_bytes, deadline): + pass + +class StreamSink(ABC): + @abc.abstractmethod + def process_bytes(self, bytes): + pass + +class PacketSource(ABC): + @abc.abstractmethod + def get_packet(self, deadline): + pass + +class PacketSink(ABC): + @abc.abstractmethod + def process_packet(self, packet): + pass + + +class StreamToPacketSegmenter(StreamSink): + def __init__(self, output): + self._header = [] + self._packet = [] + self._packet_length = 0 + self._output = output + + def process_bytes(self, bytes): + """ + Processes an arbitrary number of bytes. If one or more full packets are + are received, they are sent to this instance's output PacketSink. + Incomplete packets are buffered between subsequent calls to this function. + """ + + for byte in bytes: + if (len(self._header) < 3): + # Process header byte + self._header.append(byte) + if (len(self._header) == 1) and (self._header[0] != SYNC_BYTE): + self._header = [] + elif (len(self._header) == 2) and (self._header[1] & 0x80): + self._header = [] # TODO: support packets larger than 128 bytes + elif (len(self._header) == 3) and calc_crc8(CRC8_INIT, self._header): + self._header = [] + elif (len(self._header) == 3): + self._packet_length = self._header[1] + 2 + else: + # Process payload byte + self._packet.append(byte) + + # If both header and packet are fully received, hand it on to the packet processor + if (len(self._header) == 3) and (len(self._packet) == self._packet_length): + if calc_crc16(CRC16_INIT, self._packet) == 0: + self._output.process_packet(self._packet[:-2]) + self._header = [] + self._packet = [] + self._packet_length = 0 + + +class StreamBasedPacketSink(PacketSink): + def __init__(self, output): + self._output = output + + def process_packet(self, packet): + if (len(packet) >= MAX_PACKET_SIZE): + raise NotImplementedError("packet larger than 127 currently not supported") + + header = bytearray() + header.append(SYNC_BYTE) + header.append(len(packet)) + header.append(calc_crc8(CRC8_INIT, header)) + + self._output.process_bytes(header) + self._output.process_bytes(packet) + + # append CRC in big endian + crc16 = calc_crc16(CRC16_INIT, packet) + self._output.process_bytes(struct.pack('>H', crc16)) + +class PacketFromStreamConverter(PacketSource): + def __init__(self, input): + self._input = input + + def get_packet(self, deadline): + """ + Requests bytes from the underlying input stream until a full packet is + received or the deadline is reached, in which case None is returned. A + deadline before the current time corresponds to non-blocking mode. + """ + while True: + header = bytes() + + # TODO: sometimes this call hangs, even though the device apparently sent something + header = header + self._input.get_bytes_or_fail(1, deadline) + if (header[0] != SYNC_BYTE): + #print("sync byte mismatch") + continue + + header = header + self._input.get_bytes_or_fail(1, deadline) + if (header[1] & 0x80): + #print("packet too large") + continue # TODO: support packets larger than 128 bytes + + header = header + self._input.get_bytes_or_fail(1, deadline) + if calc_crc8(CRC8_INIT, header) != 0: + #print("crc8 mismatch") + continue + + packet_length = header[1] + 2 + #print("wait for {} bytes".format(packet_length)) + packet = self._input.get_bytes_or_fail(packet_length, deadline) + if calc_crc16(CRC16_INIT, packet) != 0: + #print("crc16 mismatch") + continue + return packet[:-2] + + +class Channel(PacketSink): + # Choose these parameters to be sensible for a specific transport layer + _resend_timeout = 5.0 # [s] + _send_attempts = 5 + + def __init__(self, name, input, output, cancellation_token, logger): + """ + Params: + input: A PacketSource where this channel will source packets from on + demand. Alternatively packets can be provided to this channel + directly by calling process_packet on this instance. + output: A PacketSink where this channel will put outgoing packets. + """ + self._name = name + self._input = input + self._output = output + self._logger = logger + self._outbound_seq_no = 0 + self._interface_definition_crc = 0 + self._expected_acks = {} + self._responses = {} + self._my_lock = threading.Lock() + self._channel_broken = Event(cancellation_token) + self.start_receiver_thread(Event(self._channel_broken)) + + def start_receiver_thread(self, cancellation_token): + """ + Starts the receiver thread that processes incoming messages. + The thread quits as soon as the channel enters a broken state. + """ + def receiver_thread(): + error_ctr = 0 + try: + while (not cancellation_token.is_set() and not self._channel_broken.is_set() + and error_ctr < 10): + # Set an arbitrary deadline because the get_packet function + # currently doesn't support a cancellation_token + deadline = time.monotonic() + 1.0 + try: + response = self._input.get_packet(deadline) + except TimeoutError: + continue # try again + except ChannelDamagedException: + error_ctr += 1 + continue # try again + if (error_ctr > 0): + error_ctr -= 1 + # Process response + # This should not throw an exception, otherwise the channel breaks + self.process_packet(response) + #print("receiver thread is exiting") + except Exception: + self._logger.debug("receiver thread is exiting: " + traceback.format_exc()) + finally: + self._channel_broken.set() + t = threading.Thread(target=receiver_thread) + t.daemon = True + t.start() + + def remote_endpoint_operation(self, endpoint_id, input, expect_ack, output_length): + if input is None: + input = bytearray(0) + if (len(input) >= 128): + raise Exception("packet larger than 127 currently not supported") + + if (expect_ack): + endpoint_id |= 0x8000 + + self._my_lock.acquire() + try: + self._outbound_seq_no = ((self._outbound_seq_no + 1) & 0x7fff) + seq_no = self._outbound_seq_no + finally: + self._my_lock.release() + seq_no |= 0x80 # FIXME: we hardwire one bit of the seq-no to 1 to avoid conflicts with the ascii protocol + packet = struct.pack(' 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='') + + # 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)) + + 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") + + + # Launch shell + print_banner() + logger._skip_bottom_line = True + interact() + app_shutdown_token.set() diff --git a/fibre/utils.py b/fibre/utils.py new file mode 100644 index 00000000..4a6f8965 --- /dev/null +++ b/fibre/utils.py @@ -0,0 +1,238 @@ + +import sys +import time +import threading +import platform +import subprocess +import os + +try: + if platform.system() == 'Windows': + import win32console + # TODO: we should win32console anyway so we could just omit colorama + import colorama + colorama.init() +except ImportError: + print("Could not init terminal features.") + sys.stdout.flush() + pass + +if sys.version_info < (3, 3): + class TimeoutError(Exception): + pass +else: + TimeoutError = TimeoutError + +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 + that the original fails to provide. + @param Trigger: if supplied, the newly created event will be triggered + as soon as the trigger event becomes set + """ + def __init__(self, trigger=None): + self._evt = threading.Event() + self._subscribers = [] + self._mutex = threading.Lock() + if not trigger is None: + trigger.subscribe(lambda: self.set()) + + 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. + The subscribers are called in the reverse order in which they subscribed. + Returns a function that can be invoked to unsubscribe. + """ + if handler is None: + raise TypeError + self._mutex.acquire() + try: + self._subscribers.insert(0, handler) + if self._evt.is_set(): + handler() + finally: + self._mutex.release() + return 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): + if not self._evt.wait(timeout=timeout): + raise TimeoutError() + + def trigger_after(self, timeout): + """ + Triggers the event after the specified timeout. + This function returns immediately. + """ + def delayed_trigger(): + if not self.wait(timeout=timeout): + self.set() + threading.Thread(target=delayed_trigger) + t.daemon = True + t.start() + + +def wait_any(timeout=None, *events): + """ + Blocks until any of the specified events are triggered. + Returns the index of the event that was triggerd or raises + a TimeoutError + Param timeout: A timeout in seconds + """ + or_event = threading.Event() + subscriptions = [] + for event in events: + subscriptions.append((event, event.subscribe(lambda: or_event.set()))) + or_event.wait(timeout=timeout) + for event, sub in subscriptions: + event.unsubscribe(sub) + for i in range(len(events)): + if events[i].is_set(): + return i + raise TimeoutError() + + +## Log utils ## + +class Logger(): + """ + Logs messages to stdout + """ + + COLOR_DEFAULT = 0 + COLOR_GREEN = 1 + COLOR_CYAN = 2 + COLOR_YELLOW = 3 + COLOR_RED = 4 + + _VT100Colors = { + COLOR_GREEN: '\x1b[92;1m', + COLOR_CYAN: '\x1b[96;1m', + COLOR_YELLOW: '\x1b[93;1m', + COLOR_RED: '\x1b[91;1m', + COLOR_DEFAULT: '\x1b[0m' + } + + _Win32Colors = { + COLOR_GREEN: 0x0A, + COLOR_CYAN: 0x0B, + COLOR_YELLOW: 0x0E, + COLOR_RED: 0x0C, + COLOR_DEFAULT: 0x07 + } + + def __init__(self, verbose=True): + self._prefix = '' + self._skip_bottom_line = False # If true, messages are printed one line above the cursor + self._verbose = verbose + self._print_lock = threading.Lock() + if platform.system() == 'Windows': + self._stdout_buf = win32console.GetStdHandle(win32console.STD_OUTPUT_HANDLE) + + def indent(self, prefix=' '): + indented_logger = Logger() + indented_logger._prefix = self._prefix + prefix + return indented_logger + + def print_on_second_last_line(self, text, color): + """ + 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. + If the printed text spans multiple lines there will also + be glitches (though this could be fixed). + """ + + if platform.system() == 'Windows': + # Windows <10 doesn't understand VT100 escape codes and the colorama + # also doesn't support the specific escape codes we need so we use the + # native Win32 API. + info = self._stdout_buf.GetConsoleScreenBufferInfo() + cursor_pos = info['CursorPosition'] + scroll_rect=win32console.PySMALL_RECTType( + Left=0, Top=1, + Right=info['Window'].Right, + Bottom=cursor_pos.Y-1) + scroll_dest = win32console.PyCOORDType(scroll_rect.Left, scroll_rect.Top-1) + self._stdout_buf.ScrollConsoleScreenBuffer( + scroll_rect, scroll_rect, scroll_dest, # clipping rect is same as scroll rect + u' ', Logger._Win32Colors[color]) # fill with empty cells with the desired color attributes + line_start = win32console.PyCOORDType(0, cursor_pos.Y-1) + self._stdout_buf.WriteConsoleOutputCharacter(text, line_start) + + else: + # Assume we're in a terminal that interprets VT100 escape codes. + # TODO: test on macOS + + # 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 + + self._print_lock.acquire() + sys.stdout.write('\x1b7\x1b[1A\x1b[1S\x1b[1L') + sys.stdout.write(Logger._VT100Colors[color] + text + Logger._VT100Colors[Logger.COLOR_DEFAULT]) + sys.stdout.write('\x1b8') + sys.stdout.flush() + self._print_lock.release() + + def print_colored(self, text, color): + if self._skip_bottom_line: + self.print_on_second_last_line(text, color) + else: + # On Windows, colorama does the job of interpreting the VT100 escape sequences + self._print_lock.acquire() + sys.stdout.write(Logger._VT100Colors[color] + text + Logger._VT100Colors[Logger.COLOR_DEFAULT] + '\n') + sys.stdout.flush() + self._print_lock.release() + + def debug(self, text): + if self._verbose: + self.print_colored(self._prefix + text, Logger.COLOR_DEFAULT) + def success(self, text): + self.print_colored(self._prefix + text, Logger.COLOR_GREEN) + def info(self, text): + self.print_colored(self._prefix + text, Logger.COLOR_DEFAULT) + def notify(self, text): + self.print_colored(self._prefix + text, Logger.COLOR_CYAN) + def warn(self, text): + self.print_colored(self._prefix + text, Logger.COLOR_YELLOW) + def error(self, text): + # TODO: write to stderr + self.print_colored(self._prefix + text, Logger.COLOR_RED) diff --git a/setup.py b/setup.py new file mode 100644 index 00000000..85f266f3 --- /dev/null +++ b/setup.py @@ -0,0 +1,88 @@ +""" +This script is used to deploy the Fibre python library to PyPi +so that users can install them easily with +"pip install fibre" + +To install the package and its dependencies locally, run: + sudo pip install -r requirements.txt + +To build and package the python tools into a tar archive: + python setup.py sdist + +Warning: Before you proceed, be aware that you can upload a +specific version only once ever. After that you need to increment +the hotfix number. Deleting the release manually on the PyPi +website does not help. + +Use TestPyPi while developing. + +To build, package and upload the python tools to TestPyPi, run: + python setup.py sdist upload -r pypitest +To make a real release ensure you're at the release commit +and then run the above command without the "test" (so just "pypi"). + +To install a prerelease version from test index: + sudo pip install --pre --index-url https://test.pypi.org/simple/ --no-cache-dir fibre + + +PyPi access requires that you have set up ~/.pypirc with your +PyPi credentials and that your account has the rights +to publish packages with the name fibre. +""" + +# TODO: add additional y/n prompt to prevent from erroneous upload + +from setuptools import setup +import os +import sys + +# Change this if you already uploaded the current +# version but need to release a hotfix +hotfix = 0 + +#creating_package = "sdist" in sys.argv +# +## Load version from Git tag +#import odrive.version +#version = odrive.version.get_version_str(git_only=creating_package) +# +#if creating_package and (hotfix > 0 or not version[-1].isdigit()): +# # Add this for hotfixes +# version += "-" + str(hotfix) +# +# +## If we're currently creating the package we need to autogenerate +## a file that contains the version string +#if creating_package: +# version_file_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'odrive', 'version.txt') +# with open(version_file_path, mode='w') as version_file: +# version_file.write(version) +# +## TODO: find a better place for this +#if not creating_package: +# import platform +# if platform.system() == 'Linux': +# import odrive.utils +# odrive.utils.setup_udev_rules(odrive.utils.Logger()) + +setup( + name = 'fibre', + packages = ['fibre'], + #scripts = ['..fibre', 'odrivetool.bat', 'odrive_demo.py'], + version = '0.0.1dev0', + description = 'Abstraction layer for painlessly building object oriented distributed systems that just work', + author = 'Samuel Sadok', + author_email = 'samuel.sadok@bluewin.ch', + license='MIT', + url = 'https://github.com/samuelsadok/fibre', + keywords = ['communication', 'transport-layer', 'rpc'], + install_requires = [], + #package_data={'': ['version.txt']}, + classifiers = [], +) + +# TODO: include README + +## clean up +#if creating_package: +# os.remove(version_file_path)