# -*- coding: utf-8 -*-
import os
import threading
from string import printable
from PyQt5 import QtCore, QtGui, QtWidgets, QtWebKit
from serial import SerialException, serial_for_url
from select import error as SelectError
from socket import error as SocketError
import time
from os.path import getsize
from colourterm import SettingsDialog, ConnectDialog, SComboBox, Highlight, from_utf8, translate, \
create_default_highlights
from colourterm.cwebview import CWebView
from colourterm.xmodem import XMODEM
class MessageType(object):
"""
An enumeration for message types
"""
Info = 1
Question = 2
Warning = 3
Critical = 4
class UiMainWindow(object):
def __init__(self):
"""
Just to satisfy PEP8/PyLint
"""
self.central_widget = None
self.central_layout = None
self.output_browser = None
self.send_layout = None
self.send_combobox = None
self.send_button = None
self.status_bar = None
self.tool_bar = None
self.open_action = None
self.close_action = None
self.capture_action = None
self.xmodem_action = None
self.follow_action = None
self.configure_action = None
self.exit_action = None
self.clear_action = None
self.find_widget = None
self.find_layout = None
self.find_combobox = None
self.find_action = None
def setup_ui(self, main_window):
"""
Set up the user interface
"""
main_window.setObjectName(from_utf8('MainWindow'))
main_window.resize(800, 600)
main_window.setWindowIcon(QtGui.QIcon(':/icons/colourterm-icon.ico'))
self.central_widget = QtWidgets.QWidget(main_window)
self.central_widget.setObjectName(from_utf8('central_widget'))
self.central_layout = QtWidgets.QVBoxLayout(self.central_widget)
self.central_layout.setSpacing(0)
self.central_layout.setContentsMargins(0, 0, 0, 0)
self.central_layout.setObjectName(from_utf8('central_layout'))
self.find_widget = QtWidgets.QWidget(main_window)
self.find_widget.setVisible(False)
self.find_widget.setSizePolicy(QtWidgets.QSizePolicy.Preferred, QtWidgets.QSizePolicy.Minimum)
self.find_widget.setObjectName(from_utf8('find_widget'))
self.find_layout = QtWidgets.QHBoxLayout(self.find_widget)
self.find_layout.setContentsMargins(0, 2, 0, 2)
self.find_layout.setObjectName(from_utf8('find_layout'))
self.find_combobox = SComboBox(self.find_widget)
self.find_combobox.setEditable(True)
self.find_combobox.setEnabled(True)
self.find_combobox.setObjectName(from_utf8('find_combobox'))
self.find_layout.addWidget(self.find_combobox)
self.central_layout.addWidget(self.find_widget)
self.output_browser = CWebView(self.central_widget)
self.output_browser.setHtml('' %
str(QtWidgets.QApplication.palette().color(QtGui.QPalette.Text).name()))
self.output_browser.setSizePolicy(QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Expanding)
self.output_browser.setObjectName(from_utf8('output_browser'))
self.central_layout.addWidget(self.output_browser)
self.send_layout = QtWidgets.QHBoxLayout()
self.send_layout.setSpacing(8)
self.send_layout.setContentsMargins(0, 4, 0, 0)
self.send_layout.setObjectName(from_utf8('sendLayout'))
self.send_combobox = SComboBox(self.central_widget)
self.send_combobox.setEditable(True)
self.send_combobox.setEnabled(False)
self.send_combobox.setObjectName(from_utf8('sendComboBox'))
self.send_layout.addWidget(self.send_combobox)
self.send_button = QtWidgets.QPushButton(self.central_widget)
size_policy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Minimum, QtWidgets.QSizePolicy.Fixed)
size_policy.setHorizontalStretch(0)
size_policy.setVerticalStretch(0)
size_policy.setHeightForWidth(self.send_button.sizePolicy().hasHeightForWidth())
self.send_button.setSizePolicy(size_policy)
self.send_button.setMaximumSize(QtCore.QSize(100, 16777215))
self.send_button.setEnabled(False)
self.send_button.setObjectName(from_utf8('sendButton'))
self.send_layout.addWidget(self.send_button)
self.central_layout.addLayout(self.send_layout)
main_window.setCentralWidget(self.central_widget)
self.status_bar = QtWidgets.QStatusBar(main_window)
self.status_bar.setObjectName(from_utf8('status_bar'))
main_window.setStatusBar(self.status_bar)
self.tool_bar = QtWidgets.QToolBar(main_window)
self.tool_bar.setMovable(False)
self.tool_bar.setToolButtonStyle(QtCore.Qt.ToolButtonTextBesideIcon)
self.tool_bar.setFloatable(False)
self.tool_bar.setObjectName(from_utf8('tool_bar'))
main_window.addToolBar(QtCore.Qt.TopToolBarArea, self.tool_bar)
self.open_action = QtWidgets.QAction(main_window)
connect_icon = QtGui.QIcon()
connect_icon.addPixmap(QtGui.QPixmap(from_utf8(':/toolbar/network-connect.png')),
QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.open_action.setIcon(connect_icon)
self.open_action.setObjectName(from_utf8('open_action'))
self.open_action.setShortcut(QtGui.QKeySequence.Open)
self.close_action = QtWidgets.QAction(main_window)
disconnect_icon = QtGui.QIcon()
disconnect_icon.addPixmap(QtGui.QPixmap(from_utf8(':/toolbar/network-disconnect.png')),
QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.close_action.setIcon(disconnect_icon)
self.close_action.setShortcut(QtGui.QKeySequence.Close)
self.close_action.setObjectName(from_utf8('close_action'))
self.find_action = QtWidgets.QAction(main_window)
find_icon = QtGui.QIcon()
find_icon.addPixmap(QtGui.QPixmap(from_utf8(':/toolbar/find.png')),
QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.find_action.setIcon(find_icon)
self.find_action.setShortcut(QtGui.QKeySequence.Find)
self.find_action.setCheckable(True)
self.find_action.setChecked(False)
self.find_action.setObjectName(from_utf8('find_action'))
self.capture_action = QtWidgets.QAction(main_window)
capture_icon = QtGui.QIcon()
capture_icon.addPixmap(QtGui.QPixmap(from_utf8(':/toolbar/capture-to-disk.png')),
QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.capture_action.setIcon(capture_icon)
self.capture_action.setCheckable(True)
self.capture_action.setChecked(False)
self.capture_action.setObjectName(from_utf8('capture_action'))
self.follow_action = QtWidgets.QAction(main_window)
self.follow_action.setShortcut(QtGui.QKeySequence(QtCore.Qt.CTRL + QtCore.Qt.SHIFT + QtCore.Qt.Key_F))
follow_icon = QtGui.QIcon()
follow_icon.addPixmap(QtGui.QPixmap(from_utf8(':/toolbar/follow-output.png')),
QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.follow_action.setIcon(follow_icon)
self.follow_action.setCheckable(True)
self.follow_action.setChecked(True)
self.follow_action.setObjectName(from_utf8('follow_action'))
self.clear_action = QtWidgets.QAction(main_window)
clear_icon = QtGui.QIcon()
clear_icon.addPixmap(QtGui.QPixmap(from_utf8(':/toolbar/clear.png')),
QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.clear_action.setIcon(clear_icon)
self.clear_action.setObjectName(from_utf8('clear_action'))
self.clear_action.setShortcut(QtGui.QKeySequence(QtCore.Qt.CTRL + QtCore.Qt.Key_Backspace))
self.xmodem_action = QtWidgets.QAction(main_window)
xmodem_icon = QtGui.QIcon()
xmodem_icon.addPixmap(QtGui.QPixmap(from_utf8(':/toolbar/move-up.png')),
QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.xmodem_action.setIcon(xmodem_icon)
self.xmodem_action.setObjectName(from_utf8('xmodem_action'))
self.xmodem_action.setShortcut(QtGui.QKeySequence(QtCore.Qt.CTRL + QtCore.Qt.SHIFT + QtCore.Qt.Key_X))
self.configure_action = QtWidgets.QAction(main_window)
configure_icon = QtGui.QIcon()
configure_icon.addPixmap(QtGui.QPixmap(from_utf8(':/toolbar/configure.png')),
QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.configure_action.setIcon(configure_icon)
self.configure_action.setObjectName(from_utf8('configure_action'))
self.exit_action = QtWidgets.QAction(main_window)
exit_icon = QtGui.QIcon()
exit_icon.addPixmap(QtGui.QPixmap(from_utf8(':/toolbar/application-exit.png')),
QtGui.QIcon.Normal, QtGui.QIcon.Off)
self.exit_action.setIcon(exit_icon)
self.exit_action.setObjectName(from_utf8('exit_action'))
self.tool_bar.addAction(self.open_action)
self.tool_bar.addAction(self.close_action)
self.tool_bar.addSeparator()
self.tool_bar.addAction(self.find_action)
self.tool_bar.addAction(self.capture_action)
self.tool_bar.addAction(self.follow_action)
self.tool_bar.addAction(self.clear_action)
self.tool_bar.addAction(self.xmodem_action)
self.tool_bar.addSeparator()
self.tool_bar.addAction(self.configure_action)
self.tool_bar.addAction(self.exit_action)
self.retranslate_ui(main_window)
def retranslate_ui(self, main_window):
"""
Translate the user interface
"""
main_window.setWindowTitle(translate('MainWindow', 'ColourTerm'))
self.send_button.setText(translate('MainWindow', 'Send'))
self.tool_bar.setWindowTitle(translate('MainWindow', 'Tool Bar'))
self.open_action.setText(translate('MainWindow', 'Open...'))
self.open_action.setToolTip(translate(
'MainWindow', 'Open (%s)' % QtGui.QKeySequence(QtGui.QKeySequence.Open)))
self.close_action.setText(translate('MainWindow', 'Close'))
self.close_action.setToolTip(translate(
'MainWindow', 'Close (%s)' % QtGui.QKeySequence(QtGui.QKeySequence.Close)))
self.find_action.setText(translate('MainWindow', 'Find'))
self.find_action.setToolTip(translate(
'MainWindow', 'Find (%s)' % QtGui.QKeySequence(QtGui.QKeySequence.Find)))
self.capture_action.setText(translate('MainWindow', 'Capture'))
self.capture_action.setToolTip(translate('MainWindow', 'Capture to File'))
self.follow_action.setText(translate('MainWindow', '&Follow'))
self.follow_action.setToolTip(translate('MainWindow', 'Follow (Ctrl+Shift+F)'))
self.clear_action.setText(translate('MainWindow', 'Clear'))
self.clear_action.setToolTip(translate('MainWindow', 'Clear (Ctrl+BkSpace)'))
self.xmodem_action.setText(translate('MainWindow', 'Xmodem'))
self.xmodem_action.setToolTip(translate('MainWindow', 'Send a file via Xmodem (Ctrl+Shift+X)'))
self.configure_action.setText(translate('MainWindow', 'Configure...'))
self.configure_action.setToolTip(translate('MainWindow', 'Configure...'))
self.exit_action.setText(translate('MainWindow', 'Exit'))
self.exit_action.setToolTip(translate('MainWindow', 'Exit (Alt+F4)'))
class MainWindow(QtWidgets.QMainWindow, UiMainWindow):
updateOutput = QtCore.pyqtSignal(str)
showMessage = QtCore.pyqtSignal(str, str, int)
def __init__(self):
super(MainWindow, self).__init__()
self.terminal_lines = []
self.max_lines = 5000
self.setup_ui(self)
self.device = None
self.device_closed = True
self.follow_output = True
self.capture_file = None
self.capture_filename = ''
self.highlights = self.load_highlights()
self.disable_output = False
self.xmodem_send_progress_window = None
if not self.highlights:
self.highlights = create_default_highlights()
self.settings_dialog = SettingsDialog()
self.connect_dialog = ConnectDialog(self)
self.open_action.triggered.connect(self.on_open_action_triggered)
self.close_action.triggered.connect(self.on_close_action_triggered)
self.find_action.triggered.connect(self.on_find_action_toggled)
self.capture_action.toggled.connect(self.on_capture_action_toggled)
self.follow_action.toggled.connect(self.on_follow_action_toggled)
self.clear_action.triggered.connect(self.on_clear_action_triggered)
self.xmodem_action.triggered.connect(self.on_xmodem_action_triggered)
self.configure_action.triggered.connect(self.on_configure_action_triggered)
self.exit_action.triggered.connect(self.close)
self.find_combobox.keyPressed.connect(self.on_find_combobox_key_pressed)
self.send_combobox.keyPressed.connect(self.on_send_combobox_key_pressed)
self.send_button.clicked.connect(self.on_send_button_clicked)
self.output_browser.page().mainFrame().contentsSizeChanged.connect(self.on_contents_size_changed)
self.output_browser.scrolled.connect(self.on_output_browser_scrolled)
self.updateOutput.connect(self.on_update_output)
self.showMessage.connect(self.on_show_message)
def close(self):
if not self.device_closed:
self.device_closed = True
if self.capture_file:
self.capture_file.flush()
self.capture_file.close()
return QtWidgets.QMainWindow.close(self)
def document_body(self):
return self.output_browser.page().mainFrame().documentElement().findFirst('pre')
def receive_text(self):
output = ''
while not self.device_closed:
try:
if self.disable_output:
time.sleep(0.5)
continue
output += self.device.read(1)
except SerialException as e:
self.showMessage.emit('Port Error', 'Error reading from serial port: %s' % e, MessageType.Critical)
self.on_close_action_triggered()
continue
if output.endswith('\r\n'):
self.updateOutput.emit(output.strip('\r\n'))
output = ''
elif output.endswith('\n'):
self.updateOutput.emit(output.strip('\n'))
output = ''
def on_open_action_triggered(self):
self.connect_dialog.update_port_combobox()
settings = QtCore.QSettings()
self.connect_dialog.set_port(settings.value('previous-port', ''))
if self.connect_dialog.exec_() == QtWidgets.QDialog.Accepted:
if not self.device_closed:
self.device_closed = True
self.device.close()
try:
port = self.connect_dialog.get_port()
settings.setValue('previous-port', port)
if isinstance(port, str) and port.startswith('COM'):
try:
# On Windows ports are 0-based, so strip the COM and subtract 1 from the port number
port = int(port[3:]) - 1
except (TypeError, ValueError):
QtWidgets.QMessageBox.critical(self, 'Error opening port', 'Error: Port is not valid')
return
self.device = serial_for_url(
url=port,
baudrate=self.connect_dialog.get_baud(),
bytesize=self.connect_dialog.get_data_bits(),
parity=self.connect_dialog.get_parity(),
stopbits=self.connect_dialog.get_stop_bits(),
timeout=1,
xonxoff=self.connect_dialog.get_software_handshake(),
rtscts=self.connect_dialog.get_hardware_handshake(),
dsrdtr=None,
do_not_open=False
)
self.device_closed = False
if not self.device.isOpen():
self.device.open()
output_thread = threading.Thread(target=self.receive_text)
output_thread.start()
except SerialException as e:
QtWidgets.QMessageBox.critical(self, 'Error opening port', e.args[0])
self.send_combobox.setEnabled(not self.device_closed)
self.send_button.setEnabled(not self.device_closed)
if self.send_combobox.isEnabled():
self.send_combobox.setFocus()
self.status_bar.showMessage('Connected to %s at %s baud' % (self.device.port, self.device.baudrate))
def on_close_action_triggered(self):
self.device_closed = True
if self.device and self.device.isOpen():
self.device.close()
self.send_combobox.setEnabled(not self.device_closed)
self.send_button.setEnabled(not self.device_closed)
self.status_bar.showMessage('')
def on_find_action_toggled(self, enabled):
self.find_widget.setVisible(enabled)
if enabled:
self.find_combobox.setFocus()
def on_capture_action_toggled(self, enabled):
if enabled and not self.capture_file:
if self.capture_filename:
base_dir = os.path.basename(self.capture_filename)
else:
base_dir = ''
self.capture_filename = QtWidgets.QFileDialog.getSaveFileName(self, 'Capture To File', base_dir,
'Text files (*.txt *.log);;All files (*)')
self.capture_file = open(self.capture_filename, 'w')
self.status_bar.showMessage(self.capture_filename)
elif self.capture_file and not enabled:
self.capture_filename = ''
self.capture_file.flush()
self.capture_file.close()
self.capture_file = None
self.status_bar.clearMessage()
def on_follow_action_toggled(self, enabled):
self.follow_output = enabled
if enabled:
self.output_browser.page().mainFrame().scroll(
0, self.output_browser.page().mainFrame().contentsSize().height())
def on_clear_action_triggered(self):
elements = self.output_browser.page().mainFrame().findAllElements('div')
for element in elements:
element.removeFromDocument()
del elements
def xmodem_callback(self, total_packets, success_count, error_count):
if self.xmodem_send_progress_window:
self.xmodem_send_progress_window.setValue(success_count)
print('{} {} {}'.format(total_packets, success_count, error_count))
def on_xmodem_action_triggered(self):
file_dialog = QtWidgets.QFileDialog()
if file_dialog.exec_():
self.disable_output = True
try:
upload_file = file_dialog.selectedFiles()[0]
file_size = getsize(upload_file)
self.device.flushInput()
self.device.flushOutput()
xmodem_transfer = XMODEM(self.getc, self.putc, mode='xmodem1k', pad='\xff')
stream = open(upload_file, 'rb')
self.xmodem_send_progress_window = QtWidgets.QProgressDialog('Sending File...', '', 0, 0)
self.xmodem_send_progress_window.setCancelButton(None)
self.xmodem_send_progress_window.setMinimum(0)
self.xmodem_send_progress_window.setMaximum(file_size/1024)
self.xmodem_send_progress_window.setValue(0)
self.xmodem_send_progress_window.setModal(True)
self.xmodem_send_progress_window.show()
success = xmodem_transfer.send(stream, retry=200, callback=self.xmodem_callback)
print(success)
finally:
if self.xmodem_send_progress_window:
self.xmodem_send_progress_window.close()
self.xmodem_send_progress_window = None
self.disable_output = False
def on_configure_action_triggered(self):
self.settings_dialog.set_highlights(self.highlights)
self.settings_dialog.exec_()
self.highlights = self.settings_dialog.highlights()
self.save_highlights(self.highlights)
self.refresh_output()
def on_find_combobox_key_pressed(self, key):
if key == QtCore.Qt.Key_Return or key == QtCore.Qt.Key_Enter:
self.output_browser.findText(
self.find_combobox.currentText(),
QtWebKit.QWebPage.HighlightAllOccurrences | QtWebKit.QWebPage.FindWrapsAroundDocument
)
elif key == QtCore.Qt.Key_Escape:
if not self.find_combobox.currentText() and self.find_widget.isVisible():
self.find_action.setChecked(not self.find_action.isChecked())
else:
self.find_combobox.clearEditText()
self.output_browser.findText("")
def on_send_combobox_key_pressed(self, key):
if key == QtCore.Qt.Key_Return or key == QtCore.Qt.Key_Enter:
self.on_send_button_clicked()
def on_send_button_clicked(self):
if self.device.isOpen():
output = str(self.send_combobox.currentText())
self.send_combobox.insertItem(0, output)
self.send_combobox.setCurrentIndex(0)
self.send_combobox.clearEditText()
self.device.write(output + '\r\n')
def on_contents_size_changed(self, size):
if self.follow_output:
self.output_browser.page().mainFrame().scroll(0, size.height())
self.output_browser.update()
def on_update_output(self, output):
if self.capture_file:
self.capture_file.write(output + '\n')
self.capture_file.flush()
output = self.style_output(output)
self.document_body().appendInside(output)
def on_output_browser_scrolled(self):
scroll_value = self.output_browser.page().mainFrame().scrollBarValue(QtCore.Qt.Vertical)
scroll_max = self.output_browser.page().mainFrame().scrollBarMaximum(QtCore.Qt.Vertical)
if scroll_value < scroll_max:
self.on_follow_action_toggled(False)
self.follow_action.setChecked(False)
else:
self.on_follow_action_toggled(True)
self.follow_action.setChecked(True)
def on_show_message(self, title, message, type_=MessageType.Info):
if type_ == MessageType.Info:
QtWidgets.QMessageBox.information(self, title, message)
elif type_ == MessageType.Question:
QtWidgets.QMessageBox.question(self, title, message)
elif type_ == MessageType.Warning:
QtWidgets.QMessageBox.warning(self, title, message)
elif type_ == MessageType.Critical:
QtWidgets.QMessageBox.critical(self, title, message)
def refresh_output(self):
elements = self.output_browser.page().mainFrame().findAllElements('div')
lines = [element.toPlainText() for element in elements]
pre = self.output_browser.page().mainFrame().findFirstElement('pre')
pre.setInnerXml('')
for line in lines:
output = self.style_output(line)
self.document_body().appendInside(output)
self.output_browser.page().mainFrame().scroll(0, self.output_browser.page().mainFrame().contentsSize().height())
self.output_browser.update()
def style_output(self, output):
style = 'font-family: \'Hack\', \'Ubuntu Mono\', monospace; '
if not output:
output = ' '
for highlight in self.highlights:
if highlight.regex.search(output):
if highlight.foreground:
style = '%scolor: %s; ' % (style, highlight.foreground)
if highlight.background:
style = '%sbackground-color: %s; ' % (style, highlight.background)
break
if style:
try:
output = '%s
' % (style, self.filter_printable(output))
except TypeError:
output = '%s
' % (style, output)
else:
output = '%s
' % output
return output
def filter_printable(self, output):
printable_output = ''
for char in output:
if char not in printable:
printable_output += '\\x{:02x}'.format(ord(char))
else:
printable_output += str(char, 'utf8')
def save_highlights(self, highlights):
settings = QtCore.QSettings()
settings.setValue('highlights/count', len(highlights))
for index, highlight in enumerate(highlights):
settings.beginGroup('highlight-%s' % index)
settings.setValue('pattern', highlight.pattern)
settings.setValue('foreground', highlight.foreground)
if highlight.background:
settings.setValue('background', highlight.background)
else:
if settings.contains('background'):
settings.remove('background')
settings.endGroup()
def load_highlights(self):
settings = QtCore.QSettings()
highlight_count = settings.value('highlights/count', 0)
highlights = []
for index in range(highlight_count):
settings.beginGroup('highlight-%s' % index)
pattern = settings.value('pattern', '')
foreground = settings.value('foreground', '')
background = None
if settings.contains('background'):
background = settings.value('background', '')
settings.endGroup()
highlights.append(Highlight(pattern, foreground, background))
return highlights
def getc(self, size, timeout=1):
"""
Read a byte (usually a character) from the serial port
:param timeout:
:param size:
"""
try:
data = self.device.read(size)
except SerialException as se:
if 'interrupted system call' in str(se.args[0]).lower():
data = self.device.read(size)
else:
raise Exception(str(se))
except (SelectError, SocketError) as se:
if se.args[0] == 4:
data = self.device.read(size)
else:
raise Exception(str(se))
return data or None
def putc(self, data, timeout=1):
"""
Send a byte (usually a character) to the serial port
:param timeout:
:param data:
"""
try:
self.device.write(data)
except SerialException as se:
self.error('Got a SerialException: {}'.format(se))
if 'interrupted system call' in str(se.args[0]).lower():
self.device.write(data)
else:
raise Exception(str(se))
except (SelectError, SocketError) as se:
if se.args[0] == 4:
self.device.write(data)
else:
self.error('Got a SocketError or SelectError: {}'.format(se))
raise Exception(str(se))
return None