Process and thread repl interfaces with (link to) buffer (consume/excrete)

process.py

import sys
import time

import PyQt5.QtCore
import PyQt5.QtWidgets


app = PyQt5.QtWidgets.QApplication(sys.argv)

text = PyQt5.QtWidgets.QTextEdit()

def append(s):
    scrollbar = text.verticalScrollBar()
    at_end = scrollbar.value() == scrollbar.maximum()

    text.append(s)

    if at_end:
        scrollbar.setValue(scrollbar.maximum())

text.setReadOnly(True)
line = PyQt5.QtWidgets.QLineEdit()
layout = PyQt5.QtWidgets.QVBoxLayout()
label = PyQt5.QtWidgets.QLabel()
layout.addWidget(label)
layout.addWidget(text)
layout.addWidget(line)
widget = PyQt5.QtWidgets.QWidget()
widget.setLayout(layout)
window = PyQt5.QtWidgets.QMainWindow()
window.setCentralWidget(widget)

python = PyQt5.QtCore.QProcess()

def stderr():
    append(bytes(python.readAllStandardError()).decode('utf-8'))

def stdout():
    append(bytes(python.readAllStandardOutput()).decode('utf-8'))

python.readyReadStandardError.connect(stderr)
python.readyReadStandardOutput.connect(stdout)

def text_entered():
    s = line.text() + '\n'
    b = s.encode('utf-8')
    append(s)
    python.write(b)
    line.setText('')

line.returnPressed.connect(text_entered)

python.start(sys.executable, ['-u', '-i'])

line.setFocus()

timer = PyQt5.QtCore.QTimer()
timer.setInterval(0.2)
def update_time():
    label.setText('{:.2f}'.format(time.monotonic()))
timer.timeout.connect(update_time)
timer.start()

line.setText("import time; print('before'); time.sleep(5); print('after')")

window.show()

app.exec()

thread.py

import io
import sys
import time

import PyQt5.QtCore
import PyQt5.QtWidgets


class Stream(PyQt5.QtCore.QObject):
    written = PyQt5.QtCore.pyqtSignal(str)

    def write(self, *args, **kwargs):
        s = io.StringIO()
        result = s.write(*args, **kwargs)
        s.seek(0)
        self.written.emit(s.read())

        return result


# s = Stream()
# s.write('blue')
# s.write('red')
# print(s.read())
# s.write('green')
# print(s.read())
# print(s.read())

class Console(PyQt5.QtCore.QObject):
    stdout_written = PyQt5.QtCore.pyqtSignal(str)
    stderr_written = PyQt5.QtCore.pyqtSignal(str)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.console = None

        self.stdout = Stream()
        self.stdout.written.connect(self.stdout_written)
        self.stderr = Stream()
        self.stderr.written.connect(self.stderr_written)

    @PyQt5.QtCore.pyqtSlot()
    def start(self):
        import code
        self.console = code.InteractiveConsole()

    @PyQt5.QtCore.pyqtSlot(str)
    def runcode(self, code):
        bak_stdout = sys.stdout
        sys.stdout = self.stdout

        bak_stderr = sys.stderr
        sys.stderr = self.stderr

        result = None
        try:
            result = self.console.runcode(code)
        except SyntaxError:
            self.console.showsyntaxerror()
        except:
            self.console.showtraceback()

        sys.stdout = bak_stdout

        sys.stderr = bak_stderr

        return result


class Interpreter(PyQt5.QtCore.QObject):
    _start = PyQt5.QtCore.pyqtSignal()
    _runcode = PyQt5.QtCore.pyqtSignal(str)
    stdout_written = PyQt5.QtCore.pyqtSignal(str)
    stderr_written = PyQt5.QtCore.pyqtSignal(str)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.console = Console()
        self._start.connect(self.console.start)
        self._runcode.connect(self.console.runcode)

        self.console.stdout_written.connect(self.stdout_written)
        self.console.stderr_written.connect(self.stderr_written)

        self.thread = PyQt5.QtCore.QThread()
        self.console.moveToThread(self.thread)

    def start(self):
        self.thread.start()
        self._start.emit()

    def runcode(self, s):
        self._runcode.emit(s)


app = PyQt5.QtWidgets.QApplication(sys.argv)

text = PyQt5.QtWidgets.QTextEdit()

def append(s):
    scrollbar = text.verticalScrollBar()
    at_end = scrollbar.value() == scrollbar.maximum()

    text.append(s)

    if at_end:
        scrollbar.setValue(scrollbar.maximum())

text.setReadOnly(True)
line = PyQt5.QtWidgets.QLineEdit()
layout = PyQt5.QtWidgets.QVBoxLayout()
label = PyQt5.QtWidgets.QLabel()
layout.addWidget(label)
layout.addWidget(text)
layout.addWidget(line)
widget = PyQt5.QtWidgets.QWidget()
widget.setLayout(layout)
window = PyQt5.QtWidgets.QMainWindow()
window.setCentralWidget(widget)

# python = PyQt5.QtCore.QProcess()
python = Interpreter()
python.start()


def stderr(s):
    append(s)


def stdout(s):
    append(s)

python.stdout_written.connect(stdout)
python.stderr_written.connect(stderr)


def text_entered():
    s = line.text() + '\n'
    append(s)
    python.runcode(s)
    line.setText('')

line.returnPressed.connect(text_entered)

line.setFocus()

timer = PyQt5.QtCore.QTimer()
timer.setInterval(0.2)
def update_time():
    label.setText('{:.2f}'.format(time.monotonic()))
timer.timeout.connect(update_time)
timer.start()

line.setText("import time; print('before'); time.sleep(5); print('after')")

window.show()

app.exec()

Probably ought to use this for scrolling to the end.

scrollbar = textedit.verticalScrollBar()
scrollbar.setValue(scrollbar.maximum())

buffer here: https://gist.github.com/daegontaven/a986a2241e15e95be7489f443e79fb7f#file-mvce_main-py-L15

(consume/excrete)