0
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-13 19:21:50 +02:00

feat: console: various improvements, auto-adapt rows to widget size, Qt Designer plugin

This commit is contained in:
2024-09-04 10:00:29 +02:00
committed by guijar_m
parent adef25f4e2
commit 286ad7196b
4 changed files with 435 additions and 124 deletions

View File

@ -1,24 +1,24 @@
"""
BECConsole is a Qt widget that runs a Bash shell. The widget can be used and
embedded like any other Qt widget.
BECConsole is a Qt widget that runs a Bash shell.
BECConsole is powered by Pyte, a Python based terminal emulator
BECConsole VT100 emulation is powered by Pyte,
(https://github.com/selectel/pyte).
"""
import collections
import fcntl
import html
import os
import pty
import subprocess
import sys
import threading
import pyte
from pyte.screens import History
from qtpy import QtCore, QtGui, QtWidgets
from qtpy.QtCore import QSize, QSocketNotifier, Qt
from qtpy.QtCore import Property as pyqtProperty
from qtpy.QtCore import QSize, QSocketNotifier, Qt, QTimer
from qtpy.QtCore import Signal as pyqtSignal
from qtpy.QtGui import QClipboard, QTextCursor
from qtpy.QtGui import QClipboard, QColor, QPalette, QTextCursor
from qtpy.QtWidgets import QApplication, QHBoxLayout, QScrollBar, QSizePolicy
from bec_widgets.qt_utils.error_popups import SafeSlot as Slot
@ -137,13 +137,52 @@ def QtKeyToAscii(event):
class Screen(pyte.HistoryScreen):
def __init__(self, stdin_fd, numColumns, numLines, historyLength):
super().__init__(numColumns, numLines, historyLength, ratio=1 / numLines)
def __init__(self, stdin_fd, cols, rows, historyLength):
super().__init__(cols, rows, historyLength, ratio=1 / rows)
self._fd = stdin_fd
def write_process_input(self, data):
"""Response to CPR request for example"""
os.write(self._fd, data.encode("utf-8"))
"""Response to CPR request (for example),
this can be for other requests
"""
try:
os.write(self._fd, data.encode("utf-8"))
except (IOError, OSError):
pass
def resize(self, lines, columns):
lines = lines or self.lines
columns = columns or self.columns
if lines == self.lines and columns == self.columns:
return # No changes.
self.dirty.clear()
self.dirty.update(range(lines))
self.save_cursor()
if lines < self.lines:
if lines <= self.cursor.y:
nlines_to_move_up = self.lines - lines
for i in range(nlines_to_move_up):
line = self.buffer[i] # .pop(0)
self.history.top.append(line)
self.cursor_position(0, 0)
self.delete_lines(nlines_to_move_up)
self.restore_cursor()
self.cursor.y -= nlines_to_move_up
else:
self.restore_cursor()
self.lines, self.columns = lines, columns
self.history = History(
self.history.top,
self.history.bottom,
1 / self.lines,
self.history.size,
self.history.position,
)
self.set_margins()
class Backend(QtCore.QObject):
@ -155,17 +194,17 @@ class Backend(QtCore.QObject):
"""
# Signals to communicate with ``_TerminalWidget``.
startWork = pyqtSignal()
dataReady = pyqtSignal(object)
processExited = pyqtSignal()
def __init__(self, fd, numColumns, numLines):
def __init__(self, fd, cols, rows):
super().__init__()
# File descriptor that connects to Bash process.
self.fd = fd
# Setup Pyte (hard coded display size for now).
self.screen = Screen(self.fd, numColumns, numLines, 10000)
self.screen = Screen(self.fd, cols, rows, 10000)
self.stream = pyte.ByteStream()
self.stream.attach(self.screen)
@ -174,12 +213,14 @@ class Backend(QtCore.QObject):
def _fd_readable(self):
"""
Poll the Bash output, run it through Pyte, and notify the main applet.
Poll the Bash output, run it through Pyte, and notify
"""
# Read the shell output until the file descriptor is closed.
try:
out = os.read(self.fd, 2**16)
except OSError:
self.processExited.emit()
self.notifier.setEnabled(False)
return
# Feed output into Pyte's state machine and send the new screen
@ -188,93 +229,256 @@ class Backend(QtCore.QObject):
self.dataReady.emit(self.screen)
class BECConsole(QtWidgets.QScrollArea):
class BECConsole(QtWidgets.QWidget):
"""Container widget for the terminal text area"""
def __init__(self, parent=None, numLines=50, numColumns=125):
ICON_NAME = "terminal"
def __init__(self, parent=None, cols=132):
super().__init__(parent)
self.innerWidget = QtWidgets.QWidget(self)
QHBoxLayout(self.innerWidget)
self.innerWidget.layout().setContentsMargins(0, 0, 0, 0)
self.term = _TerminalWidget(self, cols, rows=43)
self.scroll_bar = QScrollBar(Qt.Vertical, self)
# self.scroll_bar.hide()
layout = QHBoxLayout(self)
layout.addWidget(self.term)
layout.addWidget(self.scroll_bar)
layout.setAlignment(Qt.AlignLeft | Qt.AlignTop)
layout.setContentsMargins(0, 0, 0, 0)
self.setSizePolicy(QSizePolicy.Minimum, QSizePolicy.Expanding)
self.term = _TerminalWidget(self.innerWidget, numLines, numColumns)
self.term.setSizePolicy(QSizePolicy.Fixed, QSizePolicy.Fixed)
self.innerWidget.layout().addWidget(self.term)
pal = QPalette()
self.set_bgcolor(pal.window().color())
self.set_fgcolor(pal.windowText().color())
self.term.set_scroll_bar(self.scroll_bar)
self.set_cmd("bec --nogui")
self.scroll_bar = QScrollBar(Qt.Vertical, self.term)
self.innerWidget.layout().addWidget(self.scroll_bar)
self._check_designer_timer = QTimer()
self._check_designer_timer.timeout.connect(self.check_designer)
self._check_designer_timer.start(1000)
self.term.set_scroll(self.scroll_bar)
def minimumSizeHint(self):
size = self.term.sizeHint()
size.setWidth(size.width() + self.scroll_bar.width())
return size
self.setWidget(self.innerWidget)
def sizeHint(self):
return self.minimumSizeHint()
def start(self, cmd=["bec", "--nogui"], deactivate_ctrl_d=True):
def check_designer(self, calls={"n": 0}):
calls["n"] += 1
if self.term.fd is not None:
# already started
self._check_designer_timer.stop()
elif self.window().windowTitle().endswith("[Preview]"):
# assuming Designer preview -> start
self._check_designer_timer.stop()
self.term.start()
elif calls["n"] >= 3:
# assuming not in Designer -> stop checking
self._check_designer_timer.stop()
def get_rows(self):
return self.term.rows
def set_rows(self, rows):
self.term.rows = rows
self.adjustSize()
self.updateGeometry()
def get_cols(self):
return self.term.cols
def set_cols(self, cols):
self.term.cols = cols
self.adjustSize()
self.updateGeometry()
def get_bgcolor(self):
return QColor.fromString(self.term.bg_color)
def set_bgcolor(self, color):
self.term.bg_color = color.name(QColor.HexRgb)
def get_fgcolor(self):
return QColor.fromString(self.term.fg_color)
def set_fgcolor(self, color):
self.term.fg_color = color.name(QColor.HexRgb)
def get_cmd(self):
return self.term._cmd
def set_cmd(self, cmd):
self.term._cmd = cmd
if self.term.fd is None:
# not started yet
self.term.clear()
self.term.appendHtml(f"<h2>BEC Console - {repr(cmd)}</h2>")
def start(self, deactivate_ctrl_d=True):
self.term.start(deactivate_ctrl_d=deactivate_ctrl_d)
def push(self, text):
"""Push some text to the terminal"""
return self.term.push(text)
cols = pyqtProperty(int, get_cols, set_cols)
rows = pyqtProperty(int, get_rows, set_rows)
bgcolor = pyqtProperty(QColor, get_bgcolor, set_bgcolor)
fgcolor = pyqtProperty(QColor, get_fgcolor, set_fgcolor)
cmd = pyqtProperty(str, get_cmd, set_cmd)
class _TerminalWidget(QtWidgets.QPlainTextEdit):
"""
Start ``Backend`` process and render Pyte output as text.
"""
def __init__(self, parent, numColumns=125, numLines=50, **kwargs):
super().__init__(parent)
def __init__(self, parent, cols=125, rows=50, **kwargs):
# file descriptor to communicate with the subprocess
self.fd = None
self.backend = None
self.lock = threading.Lock()
# command to execute
self._cmd = None
self._cmd = ""
# should ctrl-d be deactivated ? (prevent Python exit)
self._deactivate_ctrl_d = False
# Default colors
pal = QPalette()
self._fg_color = pal.text().color().name()
self._bg_color = pal.base().color().name()
# Specify the terminal size in terms of lines and columns.
self.numLines = numLines
self.numColumns = numColumns
self.output = [""] * numLines
self._rows = rows
self._cols = cols
self.output = collections.deque()
super().__init__(parent)
self.setSizePolicy(QSizePolicy.MinimumExpanding, QSizePolicy.Expanding)
# Disable default scrollbars (we use our own, to be set via .set_scroll_bar())
self.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
self.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
self.scroll_bar = None
# Use Monospace fonts and disable line wrapping.
self.setFont(QtGui.QFont("Courier", 9))
self.setFont(QtGui.QFont("Monospace"))
self.setLineWrapMode(QtWidgets.QPlainTextEdit.NoWrap)
# Disable vertical scrollbar (we use our own, to be set via .set_scroll())
self.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
fmt = QtGui.QFontMetrics(self.font())
self._char_width = fmt.width("w")
self._char_height = fmt.height()
self.setCursorWidth(self._char_width)
# self.setStyleSheet("QPlainTextEdit { color: #ffff00; background-color: #303030; } ");
char_width = fmt.width("w")
self.setCursorWidth(char_width)
def start(self, deactivate_ctrl_d=False):
self.adjustSize()
self.updateGeometry()
self.update_stylesheet()
@property
def bg_color(self):
return self._bg_color
@bg_color.setter
def bg_color(self, hexcolor):
self._bg_color = hexcolor
self.update_stylesheet()
@property
def fg_color(self):
return self._fg_color
@fg_color.setter
def fg_color(self, hexcolor):
self._fg_color = hexcolor
self.update_stylesheet()
def update_stylesheet(self):
self.setStyleSheet(
f"QPlainTextEdit {{ border: 0; color: {self._fg_color}; background-color: {self._bg_color}; }} "
)
@property
def rows(self):
return self._rows
@rows.setter
def rows(self, rows: int):
if self.backend is None:
# not initialized yet, ok to change
self._rows = rows
self.adjustSize()
self.updateGeometry()
else:
raise RuntimeError("Cannot change rows after console is started.")
@property
def cols(self):
return self._cols
@cols.setter
def cols(self, cols: int):
if self.fd is None:
# not initialized yet, ok to change
self._cols = cols
self.adjustSize()
self.updateGeometry()
else:
raise RuntimeError("Cannot change cols after console is started.")
def start(self, deactivate_ctrl_d: bool = False):
self._deactivate_ctrl_d = deactivate_ctrl_d
# Start the Bash process
self.fd = self.forkShell()
self.update_term_size()
# Create the ``Backend`` object
self.backend = Backend(self.fd, self.numColumns, self.numLines)
self.backend.dataReady.connect(self.dataReady)
# Start the Bash process
self.fd = self.fork_shell()
if self.fd:
# Create the ``Backend`` object
self.backend = Backend(self.fd, self.cols, self.rows)
self.backend.dataReady.connect(self.data_ready)
self.backend.processExited.connect(self.process_exited)
else:
self.process_exited()
def process_exited(self):
self.fd = None
self.clear()
self.appendHtml(f"<br><h2>{repr(self._cmd)} - Process exited.</h2>")
self.setReadOnly(True)
def data_ready(self, screen):
"""Handle new screen: redraw, set scroll bar max and slider, move cursor to its position
This method is triggered via a signal from ``Backend``.
"""
self.redraw_screen()
self.adjust_scroll_bar()
self.move_cursor()
def minimumSizeHint(self):
width = self._char_width * self.numColumns
height = self._char_height * self.numLines
return QSize(width, height + 20)
"""Return minimum size for current cols and rows"""
fmt = QtGui.QFontMetrics(self.font())
char_width = fmt.width("w")
char_height = fmt.height()
width = char_width * self.cols
height = char_height * self.rows
return QSize(width, height)
def set_scroll(self, scroll):
self.scroll = scroll
self.scroll.setMinimum(0)
self.scroll.valueChanged.connect(self.scroll_value_change)
def sizeHint(self):
return self.minimumSizeHint()
def scroll_value_change(self, value, old={"value": 0}):
def set_scroll_bar(self, scroll_bar):
self.scroll_bar = scroll_bar
self.scroll_bar.setMinimum(0)
self.scroll_bar.valueChanged.connect(self.scroll_value_change)
def scroll_value_change(self, value, old={"value": -1}):
if self.backend is None:
return
if old["value"] == -1:
old["value"] = self.scroll_bar.maximum()
if value <= old["value"]:
# scroll up
# value is number of lines from the start
@ -288,13 +492,35 @@ class _TerminalWidget(QtWidgets.QPlainTextEdit):
for i in range(nlines):
self.backend.screen.next_page()
old["value"] = value
self.dataReady(self.backend.screen, reset_scroll=False)
self.redraw_screen()
def adjust_scroll_bar(self):
sb = self.scroll_bar
sb.valueChanged.disconnect(self.scroll_value_change)
tmp = len(self.backend.screen.history.top) + len(self.backend.screen.history.bottom)
sb.setMaximum(tmp if tmp > 0 else 0)
sb.setSliderPosition(tmp if tmp > 0 else 0)
# if tmp > 0:
# # show scrollbar, but delayed - prevent recursion with widget size change
# QTimer.singleShot(0, scrollbar.show)
# else:
# QTimer.singleShot(0, scrollbar.hide)
sb.valueChanged.connect(self.scroll_value_change)
def write(self, data):
try:
os.write(self.fd, data)
except (IOError, OSError):
self.process_exited()
@Slot(object)
def keyPressEvent(self, event):
"""
Redirect all keystrokes to the terminal process.
"""
if self.fd is None:
# not started
return
# Convert the Qt key to the correct ASCII code.
if (
self._deactivate_ctrl_d
@ -311,15 +537,17 @@ class _TerminalWidget(QtWidgets.QPlainTextEdit):
# MacOS only: CMD-V handling
self._push_clipboard()
elif code is not None:
os.write(self.fd, code)
self.write(code)
def push(self, text):
"""
Write 'text' to terminal
"""
os.write(self.fd, text.encode("utf-8"))
self.write(text.encode("utf-8"))
def contextMenuEvent(self, event):
if self.fd is None:
return
menu = self.createStandardContextMenu()
for action in menu.actions():
# remove all actions except copy and paste
@ -341,7 +569,20 @@ class _TerminalWidget(QtWidgets.QPlainTextEdit):
clipboard = QApplication.instance().clipboard()
self.push(clipboard.text())
def move_cursor(self):
textCursor = self.textCursor()
textCursor.setPosition(0)
textCursor.movePosition(
QTextCursor.Down, QTextCursor.MoveAnchor, self.backend.screen.cursor.y
)
textCursor.movePosition(
QTextCursor.Right, QTextCursor.MoveAnchor, self.backend.screen.cursor.x
)
self.setTextCursor(textCursor)
def mouseReleaseEvent(self, event):
if self.fd is None:
return
if event.button() == Qt.MiddleButton:
# push primary selection buffer ("mouse clipboard") to terminal
clipboard = QApplication.instance().clipboard()
@ -355,34 +596,34 @@ class _TerminalWidget(QtWidgets.QPlainTextEdit):
# mouse was used to select text -> nothing to do
pass
else:
# a simple 'click', make cursor going to end
textCursor.setPosition(0)
textCursor.movePosition(
QTextCursor.Down, QTextCursor.MoveAnchor, self.backend.screen.cursor.y
)
textCursor.movePosition(
QTextCursor.Right, QTextCursor.MoveAnchor, self.backend.screen.cursor.x
)
self.setTextCursor(textCursor)
self.ensureCursorVisible()
# a simple 'click', move scrollbar to end
self.scroll_bar.setSliderPosition(self.scroll_bar.maximum())
self.move_cursor()
return None
return super().mouseReleaseEvent(event)
def dataReady(self, screenData, reset_scroll=True):
def redraw_screen(self):
"""
Render the new screen as text into the widget.
Render the screen as formatted text into the widget.
"""
screen = self.backend.screen
This method is triggered via a signal from ``Backend``.
"""
with self.lock:
# Clear the widget
# Clear the widget
if screen.dirty:
self.clear()
while len(self.output) < (max(screen.dirty) + 1):
self.output.append("")
while len(self.output) > (max(screen.dirty) + 1):
self.output.pop()
# Prepare the HTML output
for line_no in screenData.dirty:
for line_no in screen.dirty:
line = text = ""
style = old_style = ""
for ch in screenData.buffer[line_no].values():
old_idx = 0
for idx, ch in screen.buffer[line_no].items():
text += " " * (idx - old_idx - 1)
old_idx = idx
style = f"{'background-color:%s;' % ansi_colors.get(ch.bg, ansi_colors['black']) if ch.bg!='default' else ''}{'color:%s;' % ansi_colors.get(ch.fg, ansi_colors['white']) if ch.fg!='default' else ''}{'font-weight:bold;' if ch.bold else ''}{'font-style:italic;' if ch.italics else ''}"
if style != old_style:
if old_style:
@ -396,45 +637,47 @@ class _TerminalWidget(QtWidgets.QPlainTextEdit):
line += f"<span style={repr(style)}>{html.escape(text, quote=True)}</span>"
else:
line += html.escape(text, quote=True)
# do a check at the cursor position:
# it is possible x pos > output line length,
# for example if last escape codes are "cursor forward" past end of text,
# like IPython does for "..." prompt (in a block, like "for" loop or "while" for example)
# In this case, cursor is at 12 but last text output is at 8 -> insert spaces
if line_no == screen.cursor.y:
llen = len(screen.buffer[line_no])
if llen < screen.cursor.x:
line += " " * (screen.cursor.x - llen)
self.output[line_no] = line
# fill the text area with HTML contents in one go
self.appendHtml(f"<pre>{chr(10).join(self.output)}</pre>")
# done updates, all clean
screenData.dirty.clear()
# did updates, all clean
screen.dirty.clear()
# Activate cursor
textCursor = self.textCursor()
textCursor.setPosition(0)
textCursor.movePosition(QTextCursor.Down, QTextCursor.MoveAnchor, screenData.cursor.y)
textCursor.movePosition(QTextCursor.Right, QTextCursor.MoveAnchor, screenData.cursor.x)
self.setTextCursor(textCursor)
self.ensureCursorVisible()
def update_term_size(self):
fmt = QtGui.QFontMetrics(self.font())
char_width = fmt.width("w")
char_height = fmt.height()
self._cols = int(self.width() / char_width)
self._rows = int(self.height() / char_height)
# manage scroll
if reset_scroll:
self.scroll.valueChanged.disconnect(self.scroll_value_change)
tmp = len(self.backend.screen.history.top) + len(self.backend.screen.history.bottom)
self.scroll.setMaximum(tmp if tmp > 0 else 0)
self.scroll.setSliderPosition(len(self.backend.screen.history.top))
self.scroll.valueChanged.connect(self.scroll_value_change)
# def resizeEvent(self, event):
# with self.lock:
# self.numColumns = int(self.width() / self._char_width)
# self.numLines = int(self.height() / self._char_height)
# self.output = [""] * self.numLines
# print("RESIZING TO", self.numColumns, "x", self.numLines)
# self.backend.screen.resize(self.numLines, self.numColumns)
def resizeEvent(self, event):
self.update_term_size()
if self.fd:
self.backend.screen.resize(self._rows, self._cols)
self.redraw_screen()
self.adjust_scroll_bar()
self.move_cursor()
def wheelEvent(self, event):
if not self.fd:
return
y = event.angleDelta().y()
if y > 0:
self.backend.screen.prev_page()
else:
self.backend.screen.next_page()
self.dataReady(self.backend.screen, reset_scroll=False)
self.redraw_screen()
def forkShell(self):
def fork_shell(self):
"""
Fork the current process and execute bec in shell.
"""
@ -443,32 +686,30 @@ class _TerminalWidget(QtWidgets.QPlainTextEdit):
except (IOError, OSError):
return False
if pid == 0:
# Safe way to make it work under BSD and Linux
try:
ls = os.environ["LANG"].split(".")
except KeyError:
ls = []
if len(ls) < 2:
ls = ["en_US", "UTF-8"]
os.putenv("COLUMNS", str(self.cols))
os.putenv("LINES", str(self.rows))
os.putenv("TERM", "linux")
os.putenv("LANG", ls[0] + ".UTF-8")
if not self._cmd:
self._cmd = os.environ["SHELL"]
cmd = self._cmd
if isinstance(cmd, str):
cmd = cmd.split()
try:
os.putenv("COLUMNS", str(self.numColumns))
os.putenv("LINES", str(self.numLines))
os.putenv("TERM", "linux")
os.putenv("LANG", ls[0] + ".UTF-8")
if isinstance(self._cmd, str):
os.execvp(self._cmd, self._cmd)
else:
os.execvp(self._cmd[0], self._cmd)
# print "child_pid", child_pid, sts
os.execvp(cmd[0], cmd)
except (IOError, OSError):
pass
# self.proc_finish(sid)
os._exit(0)
else:
# We are in the parent process.
# Set file control
fcntl.fcntl(fd, fcntl.F_SETFL, os.O_NONBLOCK)
print("Spawned Bash shell (PID {})".format(pid))
return fd
@ -478,17 +719,13 @@ if __name__ == "__main__":
from qtpy import QtGui, QtWidgets
# Terminal size in characters.
numLines = 25
numColumns = 100
# Create the Qt application and QBash instance.
# Create the Qt application and console.
app = QtWidgets.QApplication([])
mainwin = QtWidgets.QMainWindow()
title = "BECConsole ({}x{})".format(numColumns, numLines)
title = "BECConsole"
mainwin.setWindowTitle(title)
console = BECConsole(mainwin, numColumns, numLines)
console = BECConsole(mainwin)
mainwin.setCentralWidget(console)
console.start()

View File

@ -0,0 +1 @@
{'files': ['console.py']}

View File

@ -0,0 +1,58 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
import os
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
import bec_widgets
from bec_widgets.utils.bec_designer import designer_material_icon
from bec_widgets.widgets.console.console import BECConsole
DOM_XML = """
<ui language='c++'>
<widget class='BECConsole' name='bec_console'>
</widget>
</ui>
"""
MODULE_PATH = os.path.dirname(bec_widgets.__file__)
class BECConsolePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
t = BECConsole(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return "BEC Console"
def icon(self):
return designer_material_icon(BECConsole.ICON_NAME)
def includeFile(self):
return "bec_console"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "BECConsole"
def toolTip(self):
return "A terminal-like vt100 widget."
def whatsThis(self):
return self.toolTip()

View File

@ -0,0 +1,15 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.console.console_plugin import BECConsolePlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(BECConsolePlugin())
if __name__ == "__main__": # pragma: no cover
main()