blob: 7412818e891892e9b05ce870698587e88420471a [file] [log] [blame]
# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
from __future__ import print_function
import atexit
import json
import os
import re
import readline
import select
import socket
import sys
import time
HISTFILE = os.path.expanduser('~/.vtscope_history')
PROMPT = 'vtscope> '
class VTScope(object):
"""VT Scope is a debugging aid for developers of terminal emulators.
VT Scope provides an interactive shell which can load a pre-recorded terminal
session and play it back to one or more clients in a controlled manner.
It is possible to play through to a particular offset in the input or play
until a given number of escape sequences have been encountered.
The next escape sequence is displayed in VT Scope before it is sent to
the target terminal(s), so that you know what's going to be printed before
it happens.
You can connect multiple destination terminals to the scope, in order to
A/B test a known-good terminal with one under development. Clients connect
over a TCP socket to port 8383. VT Scope only listens on the local interface.
Canned VT sessions can be created by enabling logging in xterm.
Sample usage looks like this:
# Open a can of data...
vtscope> open ../test_data/vttest-01.log
Read 16723 bytes from ../test_data/vttest-01.log.
# When the next chunk of data is plain text, the offset, byte count,
# and first 15 bytes are displayed...
Next up: offset 0, 19 chars: "# 20120103.1540..."
# Wait for two clients...
vtscope> accept 2
Listening on
Waiting for client 1/2...
# At this point, open an xterm and type 'nc 8383', then open
# hterm and do the same.
Remote connected by ('', 49464)
Waiting for client 2/2...
Remote connected by ('', 49465)
# Single step through the data...
vtscope> step
# When the next chunk of data is an escape sequence, it is displayed
# with added spaces to make it easier to read.
Next up: offset 19, ESC [ 0 c
# Press ENTER to repeat the previous command.
Next up: offset 23, ESC [ ? 1 l
# Step through multiple escape sequences at a time
vtscope> step 10
Next up: offset 28, ESC [ ? 3 l
Next up: offset 73, ESC [ 0 m
# Start from the beginning of the data...
vtscope> reset
Next up: offset 0, 19 chars: "# 20120103.1540..."
# Seek directly to an offset, reset first if necessary...
vtscope> seek 73
Next up: offset 19, ESC [ 0 c
Next up: offset 73, ESC [ 0 m
# Exit vtscope. Pressing Ctrl-D on a blank line works too.
vtscope> exit
Check the comments in the "cmd_*" methods below for details about specific
# The list of connected terminals.
clients = []
# True if we're running the REPL.
running = False
# The canned data.
data = ''
# The amount of sleep time between each character, in ms.
delay_ms = 0
# The list of header-defined OFFSETs where we might want to stop and view
# the current state.
stops = []
# The current start/end position in the data. The characters between these
# two positions are next up to be sent to the clients.
start_position = 0
end_position = 0
# Patterns for escape sequences we expect to see in the data.
re_escapes = [
# Control Sequence Introducers.
['CSI', re.compile(r'\[.*?[@-~]')],
# Operating System Commands.
['OSC', re.compile(r'\].*?(\x1b\\|\x07)')],
# Privacy Messages.
['PM', re.compile(r'^.*?(\x1b\\|\x07)')],
# Device Control Strings.
['DCS', re.compile(r'P.*?(\x1b\\|\x07)')],
# Application Program Control.
['APC', re.compile(r'_.*?(\x1b\\|\x07)')],
# DEC private sequences.
['DEC', re.compile(r'#[^\x1b]')],
# Character set control.
['CHR', re.compile(r'%[^\x1b]')],
# Graphic character sets.
['SCS', re.compile(r'[()*+-./][^\x1b]')],
# Other escape sequences.
['ESC', re.compile(r'[^\x1b]')],
def run(self):
"""Start the VTScope REPL."""
# Pressing ENTER on a blank line re-executes the previous command.
last_command_line = ''
self.running = True
while self.running:
command_line = raw_input(PROMPT)
except EOFError:
self.running = False
if not command_line:
command_line = last_command_line
command_line = command_line.strip()
last_command_line = command_line
def scan_header(self, header):
"""Scan the header for OFFSET blocks where we might want to stop and view
the current state.
offset_re = re.compile(
self.stops = []
m =
while m:
self.stops.append({'offset': int(,
'lines': int(,
'row': int(,
'column': int(
m =, m.end())
def find_next_chunk(self):
"""Advance start_position and end_position to the next chunk in the
canned data.
self.start_position = self.end_position
if self.start_position >= len(
return ''
if[self.start_position] == '\x1b':
m = None
for (esc_name, pattern) in self.re_escapes:
m = pattern.match(, self.start_position + 1)
if m:
if m:
self.end_position = m.end()
self.end_position = self.start_position + MAX_TEXT
print('Unable to find end of escape sequence.')
sequence =[self.start_position + 1 : self.end_position]
return json.dumps(esc_name + ' ' + ' '.join(sequence))[1:-1]
self.end_position ='\x1b', self.start_position)
if self.end_position == -1:
self.end_position = len(
plaintext =[self.start_position : self.end_position]
if len(plaintext) > MAX_TEXT:
plaintext = plaintext[0:MAX_TEXT] + '...'
return '%s chars: %s' % \
(self.end_position - self.start_position, json.dumps(plaintext))
def show_next_chunk(self):
"""Find the next chunk of data, and display it to the user."""
snippet = self.find_next_chunk()
if snippet:
print('Next up: offset %s, %s' % (self.start_position, snippet))
print('End of data.')
def send(self, str):
"""Broadcast a string to all clients, removing any that appear to
have disconnected."""
for i in xrange(len(self.clients), 0, -1):
fd = self.clients[i - 1]
except IOError:
print('Client #%s disconnected.' % i)
del self.clients[i - 1]
def broadcast_chunk(self):
"""Broadcast the current chunk of data to the connected clients."""
if not self.delay_ms:
self.send([self.start_position : self.end_position])
# If we have a delay, send a character at a time.
for ch in[self.start_position : self.end_position]:
time.sleep(self.delay_ms / 1000.0)
def dispatch_command(self, command_line):
"""Dispatch a command line to an appropriate cmd_* method."""
command_args = command_line.split(' ')
command_name = command_args[0]
command_args = command_args[1:]
if not command_name:
command_function = getattr(self, 'cmd_' + command_name, None)
if not command_function:
print('Unknown command: "%s"' % command_name)
# Change this test to 'if False' to skip over the try/catch, making it
# easier to debug broken cmd_* functions.
if True:
except Exception as ex:
print('Error executing %s: %s: %s' %
(command_name, type(ex), ex))
# Commands start here, in alphabetical order.
def cmd_accept(self, args):
"""Wait for one or more clients to connect.
Usage: accept <client-count>
If <client-count> starts with a '+' as in 'accept +1', then this will
allow additional clients to connect. Otherwise all existing connections
are reset before accepting.
Clients can connect using the the 'nc' (aka netcat) command, with...
$ nc 8383
if not len(args):
print('Missing argument.')
if args[0][0] == '+':
count = len(self.clients) + int(args[0][1:])
count = int(args[0])
self.clients = []
print('Listening on %s:%s' % (LISTEN_HOST, LISTEN_PORT))
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
while len(self.clients) < count:
print('Waiting for client %s/%s...' % (len(self.clients) + 1, count))
(fd, addr) = sock.accept()
print('Remote connected by', addr)
def cmd_bstep(self, args):
"""Step a given number of bytes."""
if len(args) > 0:
count = int(args[0])
count = 1
self.end_position = self.start_position + count
if self.end_position == len(
self.end_position = len(
def cmd_delay(self, args):
"""Set a delay between each character, in milliseconds."""
if len(args):
self.delay_ms = int(args[0])
print('Delay is now: %s' % self.delay_ms)
def cmd_exit(self, args):
"""Exit vtscope.
Usage: exit
self.running = False
def cmd_send(self, args):
"""Send a string to all clients.
You can use ESC to signify the ASCII code for escape, and JSON notation
to specify any other non-printables.
JSON notation uses '\uXXXX' to specify arbitrary codepoints, as well as
common shortcuts such as '\n', '\t', and '\b'.
Whitespace is used only as a separator. To send a literal space you must
encode it as \u0020.
# Send the code to show the cursor (useful if the recorded session
# previously hid it).
send ESC [ ? 25 h
# Same as above.
send ESC [?25h
# Broken, would send the literal characters "ESC" followed by "[?25h".
send ESC[?25h
# Send a tab followed by a backspace.
send \t\b
# Send the text HelloWorld (whitespace is only used as a separator).
send Hello World
# Send the text "Hello World"
send Hello\u0020World
str = ''
for arg in args:
if len(arg) == 1:
str += arg
elif arg == 'ESC':
str += "\x1b"
str += json.loads('"%s"' % arg)
print('Sending %s' % json.dumps(str))
def cmd_stops(self, args):
"""Display a list of the stop offsets.
Usage: stops
if not len(self.stops):
print('No stop offsets found.')
for i in xrange(len(self.stops)):
offset = self.stops[i]
print('#%s offset: %s, lines: %s, cursor: %s,%s' %
(i + 1, offset['offset'], offset['lines'], offset['row'],
def cmd_open(self, args):
"""Open a local file containing canned data.
If the log file has header information then the OFFSETs found in the
header will be available to the 'seek' command. See 'seek' and 'stops'
commands for more information.
Usage: open <local-path>
filename = os.path.expanduser(args[0])
self.position = 0
with open(filename) as f: =
if re.match(r'(#[^\n]*\n)*@@ HEADER_START',
m ='@@ HEADER_END\r?\n',, re.MULTILINE)
if not m:
print('Unable to locate end of header.')
end = m.end()
print('Read %s bytes of header.' % end)
self.scan_header([0 : end]) =[end : ]
print('Read %s bytes of playback.' % len(
def cmd_reset(self, args):
"""Reset the current position in the canned data and display the first
Usage: reset
self.start_position = 0
self.end_position = 0
def cmd_seek(self, args):
"""Seek to a given position in the canned data.
Usage: seek <offset>
If <offset> starts with a pound sign, as in 'seek #1', it will seek to the
stop offset at the given 1-based index. Use the 'stops' command to list
out the stop offsets defined by the log file.
If the resulting offset comes before the current position input will
be replayed from the beginning.
if len(args) < 1:
print('Missing argument')
if not
print('No data.')
if args[0][0] == '#':
index = int(args[0][1:])
if index < 1 or index > len(self.stops):
print('No such stop.')
pos = self.stops[index - 1]['offset']
pos = int(args[0])
if pos > len(
print('Seek past end.')
if pos <= self.start_position:
while self.end_position <= pos:
def cmd_step(self, args):
"""Step over a given number of escape sequences, or 1 if not specified.
Usage: step [<count>]
if self.start_position >= len(
print('Already at end of data.')
if len(args) > 0:
count = int(args[0])
count = 1
while count:
if self.start_position >= len(
count -= 1
if __name__ == '__main__':
except IOError:
atexit.register(lambda: readline.write_history_file(HISTFILE))
vtscope = VTScope()