diff --git a/tools/explore_odrive.py b/tools/explore_odrive.py index ccc252d4..66884d0f 100755 --- a/tools/explore_odrive.py +++ b/tools/explore_odrive.py @@ -8,7 +8,7 @@ import sys import platform import threading import odrive.discovery -from odrive.utils import start_liveplotter +from odrive.utils import start_liveplotter, Logger # Flush stdout by default import functools @@ -63,27 +63,7 @@ else: ## 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) +logger = Logger() def print_banner(): print('ODrive control utility v0.4') @@ -135,7 +115,7 @@ def did_discover_device(odrive): # 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) + logger.info("{} to ODrive {:012X} as {}".format(verb, serial_number, interactive_name)) # Subscribe to disappearance of the device odrive.__channel__._channel_broken.subscribe(lambda: did_lose_device(interactive_name)) @@ -146,7 +126,7 @@ def did_lose_device(interactive_name): a message. """ if not app_shutdown_token.is_set(): - print_on_second_last_line(COLOR_RED + "Oh no {} disappeared".format(interactive_name) + COLOR_RESET) + logger.warn("Oh no {} disappeared".format(interactive_name)) # Connect to device printer("Waiting for device...") @@ -183,7 +163,7 @@ else: # Enable tab complete if possible try: import rlcompleter - import readline + import readline # Works only on Unix readline.parse_and_bind("tab: complete") except: sudo_prefix = "" if platform.system() == "Windows" else "sudo " @@ -206,5 +186,6 @@ console.runcode('sys.excepthook=newexcepthook') # Launch shell print_banner() +logger._skip_bottom_line = True interact() app_shutdown_token.set() diff --git a/tools/odrive/serial_transport.py b/tools/odrive/serial_transport.py index 53ca6850..dd595c6d 100644 --- a/tools/odrive/serial_transport.py +++ b/tools/odrive/serial_transport.py @@ -89,7 +89,7 @@ def discover_channels(path, serial_number, callback, cancellation_token, printer input_stream, output_stream, printer) channel.serial_device = serial_device except serial.serialutil.SerialException: - printer("Serial device init failed. Ignoring this port") + printer("Serial device init failed. Ignoring this port. More info: " + traceback.format_exc()) known_devices.append(port_name) else: known_devices.append(port_name) diff --git a/tools/odrive/usbbulk_transport.py b/tools/odrive/usbbulk_transport.py index edf057e2..e1004ba8 100644 --- a/tools/odrive/usbbulk_transport.py +++ b/tools/odrive/usbbulk_transport.py @@ -6,6 +6,8 @@ import time import usb.core import usb.util import odrive.protocol +import traceback +import platform ODRIVE_VID_PID_PAIRS = [ (0x1209, 0x0D31), @@ -39,7 +41,9 @@ class USBBulkTransport(odrive.protocol.PacketSource, odrive.protocol.PacketSink) # state where there are a few packets in a receive queue but a call # to epr.read() does not return these packet until a new packet arrives. # This undesirable queue can be cleared by resetting the device. - self.dev.reset() + # On windows this would cause file-not-found errors in subsequent dev calls + if platform.system() != 'Windows': + self.dev.reset() try: if self.dev.is_kernel_driver_active(1): @@ -185,7 +189,7 @@ def discover_channels(path, serial_number, callback, cancellation_token, printer usb_device.reset() continue else: - printer("USB device init failed. Ignoring this device") + printer("USB device init failed. Ignoring this device. More info: " + traceback.format_exc()) known_devices.append((usb_device.bus, usb_device.address)) else: known_devices.append((usb_device.bus, usb_device.address)) diff --git a/tools/odrive/utils.py b/tools/odrive/utils.py index 4f070b1e..7762494b 100755 --- a/tools/odrive/utils.py +++ b/tools/odrive/utils.py @@ -6,6 +6,16 @@ Liveplotter import sys import time import threading +import platform + +try: + if platform.system() == 'Windows': + import win32console + import colorama + colorama.init() +except ModuleNotFoundError: + print("Could not init terminal colors") + pass data_rate = 100 plot_rate = 10 @@ -198,27 +208,99 @@ class Logger(): Logs messages to stdout """ - COLOR_GREEN = '\x1b[92;1m' - COLOR_CYAN = '\x1b[96;1m' - COLOR_YELLOW = '\x1b[93;1m' - COLOR_RED = '\x1b[91;1m' - COLOR_RESET = '\x1b[0m' + 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): self._prefix = '' + self._skip_bottom_line = False # If true, messages are printed one line above the cursor + 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 + + sys.stdout.write('\x1b7\x1b[1A\x1b[1S\x1b[1L', end='', flush=True) + sys.stdout.write(Logger._VT100Colors[color] + text + Logger._VT100Colors[Logger.COLOR_DEFAULT]) + sys.stdout.write('\x1b8', end='', flush=True) + sys.stdout.flush() + + 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 + sys.stdout.write(Logger._VT100Colors[color] + text + Logger._VT100Colors[Logger.COLOR_DEFAULT] + '\n') + sys.stdout.flush() + def debug(self, text): - print(self._prefix + text) + self.print_colored(self._prefix + text, Logger.COLOR_DEFAULT) def success(self, text): - print(self._prefix + Logger.COLOR_GREEN + text + Logger.COLOR_RESET) + self.print_colored(self._prefix + text, Logger.COLOR_GREEN) def info(self, text): - print(self._prefix + Logger.COLOR_CYAN + text + Logger.COLOR_RESET) + self.print_colored(self._prefix + text, Logger.COLOR_CYAN) def warn(self, text): - print(self._prefix + Logger.COLOR_YELLOW + text + Logger.COLOR_RESET) + self.print_colored(self._prefix + text, Logger.COLOR_YELLOW) def error(self, text): - print(self._prefix + Logger.COLOR_RED + text + Logger.COLOR_RESET) + # TODO: write to stderr + self.print_colored(self._prefix + text, Logger.COLOR_RED)