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

make fancy terminal features work on Windows

Browse Source 
Add Windows support for the following terminal features:
 - colored output
 - output on the second last line

On Unix systems, VT100 escape codes are used to achieve this,
functionality however Windows <10 doesn't interpret VT100 escape codes.
For normal colored output, we use the colorama module to abstract this
away. To print text on the second-last line we call the appropriate
Win32 API functions directly (using the win32console module).
This commit is contained in:
Samuel Sadok
2018-03-25 15:57:50 -07:00
parent 0784bd3413
commit ae8acae7ef
4 changed files with 105 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/explore_odrive.py
+6 -25
View File
@@ -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()
tools/odrive/serial_transport.py
+1 -1
View File
@@ -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)
tools/odrive/usbbulk_transport.py
+6 -2
View File
@@ -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))
tools/odrive/utils.py
+92 -10
View File
@@ -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)