Python interpreter console tkinter widget

python_console.py

import code
import hashlib
import queue
import sys
import threading
import tkinter as tk
import traceback
from tkinter.scrolledtext import ScrolledText


class Pipe:
    """mock stdin stdout or stderr"""

    def __init__(self):
        self.buffer = queue.Queue()
        self.reading = False

    def write(self, data):
        self.buffer.put(data)

    def flush(self):
        pass

    def readline(self):
        self.reading = True
        line = self.buffer.get()
        self.reading = False
        return line


class Console(tk.Frame):
    """A tkinter widget which behaves like an interpreter"""

    def __init__(self, parent, _locals, exit_callback):
        super().__init__(parent)

        self.text = ConsoleText(self, wrap=tk.WORD)
        self.text.pack(fill=tk.BOTH, expand=True)

        self.shell = code.InteractiveConsole(_locals)

        # make the enter key call the self.enter function
        self.text.bind("<Return>", self.enter)
        self.prompt_flag = True
        self.command_running = False
        self.exit_callback = exit_callback

        # replace all input and output
        sys.stdout = Pipe()
        sys.stderr = Pipe()
        sys.stdin = Pipe()

        def loop():
            self.read_from_pipe(sys.stdout, "stdout")
            self.read_from_pipe(sys.stderr, "stderr", foreground='red')

            self.after(50, loop)

        self.after(50, loop)

    def prompt(self):
        """Add a '>>> ' to the console"""
        self.prompt_flag = True

    def read_from_pipe(self, pipe: Pipe, tag_name, **kwargs):
        """Method for writing data from the replaced stdout and stderr to the console widget"""

        # write the >>>
        if self.prompt_flag and not self.command_running:
            self.text.prompt()
            self.prompt_flag = False

        # get data from buffer
        string_parts = []
        while not pipe.buffer.empty():
            part = pipe.buffer.get()
            string_parts.append(part)

        # write to console
        str_data = ''.join(string_parts)
        if str_data:
            if self.command_running:
                insert_position = "end-1c"
            else:
                insert_position = "prompt_end"

            self.text.write(str_data, tag_name, insert_position, **kwargs)

    def enter(self, e):
        """The <Return> key press handler"""

        if sys.stdin.reading:
            # if stdin requested, then put data in stdin instead of running a new command
            line = self.text.consume_last_line()
            line = line + '\n'
            sys.stdin.buffer.put(line)
            return

        # don't run multiple commands simultaneously
        if self.command_running:
            return

        # get the command text
        command = self.text.read_last_line()
        try:
            # compile it
            compiled = code.compile_command(command)
            is_complete_command = compiled is not None
        except (SyntaxError, OverflowError, ValueError):
            # if there is an error compiling the command, print it to the console
            self.text.consume_last_line()
            self.prompt()
            traceback.print_exc()
            return

        # if it is a complete command
        if is_complete_command:
            # consume the line and run the command
            self.text.consume_last_line()

            self.prompt()
            self.command_running = True

            def run_command():
                try:
                    self.shell.runcode(compiled)
                except SystemExit:
                    self.after(0, self.exit_callback)

                self.command_running = False

            threading.Thread(target=run_command).start()


class ConsoleText(ScrolledText):
    """
    A Text widget which handles some application logic,
    e.g. having a line of input at the end with everything else being uneditable
    """

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

        # make edits that occur during on_text_change not cause it to trigger again
        def on_modified(event):
            flag = self.edit_modified()
            if flag:
                self.after(10, self.on_text_change(event))
            self.edit_modified(False)

        self.bind("<<Modified>>", on_modified)

        # store info about what parts of the text have what colour
        # used when colour info is lost and needs to be re-applied
        self.console_tags = []

        # the position just before the prompt (>>>)
        # used when inserting command output and errors
        self.mark_set("prompt_end", 1.0)

        # keep track of where user input/commands start and the committed text ends
        self.committed_hash = None
        self.committed_text_backup = ""
        self.commit_all()

    def prompt(self):
        """Insert a prompt"""
        self.mark_set("prompt_end", 'end-1c')
        self.mark_gravity("prompt_end", tk.LEFT)
        self.write(">>> ", "prompt", foreground="blue")
        self.mark_gravity("prompt_end", tk.RIGHT)

    def commit_all(self):
        """Mark all text as committed"""
        self.commit_to('end-1c')

    def commit_to(self, pos):
        """Mark all text up to a certain position as committed"""
        if self.index(pos) in (self.index("end-1c"), self.index("end")):
            # don't let text become un-committed
            self.mark_set("committed_text", "end-1c")
            self.mark_gravity("committed_text", tk.LEFT)
        else:
            # if text is added before the last prompt (">>> "), update the stored position of the tag
            for i, (tag_name, _, _) in reversed(list(enumerate(self.console_tags))):
                if tag_name == "prompt":
                    tag_ranges = self.tag_ranges("prompt")
                    self.console_tags[i] = ("prompt", tag_ranges[-2], tag_ranges[-1])
                    break

        # update the hash and backup
        self.committed_hash = self.get_committed_text_hash()
        self.committed_text_backup = self.get_committed_text()

    def get_committed_text_hash(self):
        """Get the hash of the committed area - used for detecting an attempt to edit it"""
        return hashlib.md5(self.get_committed_text().encode()).digest()

    def get_committed_text(self):
        """Get all text marked as committed"""
        return self.get(1.0, "committed_text")

    def write(self, string, tag_name, pos='end-1c', **kwargs):
        """Write some text to the console"""

        # get position of the start of the text being added
        start = self.index(pos)

        # insert the text
        self.insert(pos, string)
        self.see(tk.END)

        # commit text
        self.commit_to(pos)

        # color text
        self.tag_add(tag_name, start, pos)
        self.tag_config(tag_name, **kwargs)

        # save color in case it needs to be re-colured
        self.console_tags.append((tag_name, start, self.index(pos)))

    def on_text_change(self, event):
        """If the text is changed, check if the change is part of the committed text, and if it is revert the change"""
        if self.get_committed_text_hash() != self.committed_hash:
            # revert change
            self.mark_gravity("committed_text", tk.RIGHT)
            self.replace(1.0, "committed_text", self.committed_text_backup)
            self.mark_gravity("committed_text", tk.LEFT)

            # re-apply colours
            for tag_name, start, end in self.console_tags:
                self.tag_add(tag_name, start, end)

    def read_last_line(self):
        """Read the user input, i.e. everything written after the committed text"""
        return self.get("committed_text", "end-1c")

    def consume_last_line(self):
        """Read the user input as in read_last_line, and mark it is committed"""
        line = self.read_last_line()
        self.commit_all()
        return line
        

if __name__ == '__main__':
    root = tk.Tk()
    root.config(background="red")
    main_window = Console(root, locals(), root.destroy)
    main_window.pack(fill=tk.BOTH, expand=True)
    root.mainloop()

@olisolomons
I assumed the same with import mttkinter as tk but mttkinter seems to be a kind of wrapper.
I debugged the code now (sometimes i forget how useful this is ^^) and recognized an interesting behavior.
With your old implementation:

• The first time when i run my module with input my input is recognized as an empty string.

• The second time (without termination the mainloop) my input is recognized correctly.

With your new implementation:

• First input is recognized, second is again empty string.

There must be a bug somewhere, but im already happy that i come closer and closer to a result.
Btw. my last input stays in the console, it wont be deleted. That might cause some problems.

@muesgit That does sound rather strange... As I said earlier:

and [I] wouldn't be able to debug without seeing more code (preferably something I can run myself)

Something like this, though it doesn't have to be completely minimal. Or you can try to debug yourself if you'd rather not send me anything 🤷

@olisolomons
I created a minimal code and this indeed helped me already to precise, where the error happens.
It is not in your code.
Here is the minimal code:

All Modules are in one folder

Module 1 (scratch_4.py)

from console import Console
import mttkinter as tkinter
import tkinter as tk
from tkinter import ttk
import threading
from scratches import scratch_5


def threadstart():
    input_thread = threading.Thread(target=lambda: scratch_5.main(True))
    input_thread.start()


root = tk.Tk()
root.geometry('500x500')

run_button = tk.Button(root, text='Start Input', command=threadstart)
run_button.grid(row=0, column=0, columnspan=2, sticky='nsew')

console_container = ttk.LabelFrame(root, text='User Interaction')
console_container.grid(column=0, row=1, sticky='nsew', padx=10, pady=10)

console = Console(console_container, locals(), console_container.destroy)
console.grid()

root.mainloop()

Module 2 (scratch_5.py)

from scratches import scratch_6


def main(a):
    if a == True:
        user_input = scratch_6.main()
    print(user_input)


if __name__ == '__main__':
    boolean = True
    main(boolean)

Module 3 (scratch_6.py)

input_dict = {
    'text': 'Please Enter Something'
}


def main():
    user_input = input(input_dict['text'])
    print('Your input is: ', user_input)
    return user_input


if __name__ == '__main__':
    main()

You dont even need module 1 to see that when you start module 2 and type something in only None is returned.
If you start module 3 by itself the expected value is returned.

Edit: It is actually strange because in my original project there is no issue like that and the structure is the same.

@muesgit Everything all sorted then?

@olisolomons
Its not completely solved, but maybe not the right place to solve it here.

@muesgit Good luck!

@olisolomons Seems to work now, dont know why. Anyways many many thanks for your effort.