# Copyright (c) 2011-2015 Rusty Wagner # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . from PySide.QtCore import * from PySide.QtGui import * from Fonts import * import code import sys import threading import Threads class PythonConsoleOutput(): def __init__(self, orig, error): self.stdout = orig self.error = error def write(self, data): console = None if "value" in dir(_python_console): console = _python_console.value if console is None: self.stdout.write(data) else: if self.error: Threads.run_on_gui_thread(lambda: console.write_stderr(data)) else: Threads.run_on_gui_thread(lambda: console.write_stdout(data)) class PythonConsoleInput(): def __init__(self, orig): self.stdin = orig def read(self, size): console = None if "value" in dir(_python_console): console = _python_console.value if console is None: return self.stdin.read(size) else: return console.read_stdin(size) def readline(self): console = None if "value" in dir(_python_console): console = _python_console.value if console is None: return self.stdin.readline() else: return console.readline_stdin() class PythonConsoleThread(threading.Thread): def __init__(self, console): threading.Thread.__init__(self) self.console = console self.globals = {"__name__":"__console__", "__doc__":None} self.code = None self.event = threading.Event() self.done = threading.Event() self.exit = False self.interpreter = code.InteractiveInterpreter(self.globals) # There is no way to interrupt a thread that isn't the main thread, so # to avoid not being able to close the app, create the thread as # as daemon thread. self.daemon = True # Set up environment with useful variables and functions self.globals["data"] = Threads.GuiObjectProxy(self.console.view.data) self.globals["exe"] = Threads.GuiObjectProxy(self.console.view.exe) self.globals["view"] = Threads.GuiObjectProxy(self.console.view) self.globals["current_view"] = Threads.GuiObjectProxy(lambda: self.console.view.view) self.globals["change_view"] = Threads.GuiObjectProxy(lambda type: self.console.view.setViewType(type)) self.globals["navigate"] = Threads.GuiObjectProxy(lambda type, pos: self.console.view.navigate(type, pos)) self.globals["create_file"] = Threads.GuiObjectProxy(lambda data: Threads.create_file(data)) self.globals["cursor"] = Threads.GuiObjectProxy(lambda: self.console.view.view.get_cursor_pos()) self.globals["set_cursor"] = Threads.GuiObjectProxy(lambda pos: self.console.view.view.set_cursor_pos(pos)) self.globals["selection_range"] = Threads.GuiObjectProxy(lambda: self.console.view.view.get_selection_range()) self.globals["set_selection_range"] = Threads.GuiObjectProxy(lambda start, end: self.console.view.view.set_selection_range(start, end)) self.globals["selection"] = Threads.GuiObjectProxy(lambda: self.get_selection()) self.globals["replace_selection"] = Threads.GuiObjectProxy(lambda value: self.replace_selection(value)) self.globals["write_at_cursor"] = Threads.GuiObjectProxy(lambda value: self.write_at_cursor(value)) self.globals["undo"] = Threads.GuiObjectProxy(lambda: self.console.view.undo()) self.globals["redo"] = Threads.GuiObjectProxy(lambda: self.console.view.redo()) self.globals["commit"] = Threads.GuiObjectProxy(lambda: self.console.view.commit_undo()) self.globals["copy"] = Threads.GuiObjectProxy(lambda value: self.copy(value)) self.globals["paste"] = Threads.GuiObjectProxy(lambda: self.console.view.view.paste()) self.globals["clipboard"] = Threads.GuiObjectProxy(lambda: self.get_clipboard()) # Helper APIs def get_selection(self): data = self.console.view.view.data range = self.console.view.view.get_selection_range() return data.read(range[0], range[1] - range[0]) def replace_selection(self, value): data = self.console.view.view.data range = self.console.view.view.get_selection_range() if (range[1] - range[0]) == len(value): result = data.write(range[0], value) else: data.remove(range[0], range[1] - range[0]) result = data.insert(range[0], value) self.console.view.view.set_cursor_pos(range[0] + result) return result def write_at_cursor(self, value): data = self.console.view.view.data pos = self.console.view.view.get_cursor_pos() result = data.write(pos, value) self.console.view.view.set_cursor_pos(pos + result) return result def copy(self, data): if type(data) != str: data = str(data) clipboard = QApplication.clipboard() clipboard.clear() mime = QMimeData() mime.setText(data.encode("string_escape").replace("\"", "\\\"")) mime.setData("application/octet-stream", QByteArray(data)) clipboard.setMimeData(mime) def get_clipboard(self): clipboard = QApplication.clipboard() mime = clipboard.mimeData() if mime.hasFormat("application/octet-stream"): return mime.data("application/octet-stream").data() elif mime.hasText(): return mime.text().encode("utf8") else: return None # Thread run loop def run(self): _python_console.value = self.console while not self.exit: self.event.wait() self.event.clear() if self.exit: break if self.code: self.interpreter.runsource(self.code) self.code = None self.done.set() class PythonConsoleLineEdit(QLineEdit): prevHistory = Signal(()) nextHistory = Signal(()) def __init__(self, *args): super(PythonConsoleLineEdit, self).__init__(*args) def event(self, event): if (event.type() == QEvent.KeyPress) and (event.key() == Qt.Key_Tab): self.insert("\t") return True if (event.type() == QEvent.KeyPress) and (event.key() == Qt.Key_Up): self.prevHistory.emit() return True if (event.type() == QEvent.KeyPress) and (event.key() == Qt.Key_Down): self.nextHistory.emit() return True return QLineEdit.event(self, event) class PythonConsole(QWidget): def __init__(self, view): super(PythonConsole, self).__init__(view) self.view = view font = getMonospaceFont() layout = QVBoxLayout() layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) self.output = QTextEdit() self.output.setFont(font) self.output.setReadOnly(True) layout.addWidget(self.output, 1) input_layout = QHBoxLayout() input_layout.setContentsMargins(4, 4, 4, 4) input_layout.setSpacing(4) self.prompt = QLabel(">>>") self.prompt.setFont(font) input_layout.addWidget(self.prompt) self.input = PythonConsoleLineEdit() self.input.setFont(font) self.input.returnPressed.connect(self.process_input) self.input.prevHistory.connect(self.prev_history) self.input.nextHistory.connect(self.next_history) input_layout.addWidget(self.input) layout.addLayout(input_layout, 0) self.setLayout(layout) self.setFocusPolicy(Qt.NoFocus) self.setMinimumSize(100, 100) size_policy = self.sizePolicy() size_policy.setVerticalStretch(1) self.setSizePolicy(size_policy) self.thread = PythonConsoleThread(self) self.thread.start() self.completion_timer = QTimer() self.completion_timer.setInterval(100) self.completion_timer.setSingleShot(False) self.completion_timer.timeout.connect(self.completion_timer_event) self.completion_timer.start() self.source = None self.running = False self.input_requested = False self.input_result = "" self.input_event = threading.Event() self.input_history = [] self.input_history_pos = None def stop(self): self.thread.exit = True self.thread.event.set() # Can't join here, as it might be stuck in user code def process_input(self): self.input_history += [str(self.input.text())] self.input_history_pos = None input = str(self.input.text()) + "\n" self.input.setText("") self.output.textCursor().movePosition(QTextCursor.End) fmt = QTextCharFormat() fmt.setForeground(QBrush(Qt.black)) if len(self.prompt.text()) > 0: self.output.textCursor().insertText(self.prompt.text() + " " + input, fmt) else: self.output.textCursor().insertText(input, fmt) self.output.ensureCursorVisible() if self.input_requested: # Request for data from stdin self.input_requested = False self.input.setEnabled(False) self.input_result = input self.input_event.set() return if self.source is not None: self.source = self.source + input if input != "\n": # Don't end multiline input until a blank line return input = self.source try: result = code.compile_command(input) except: result = False if result is None: if self.source is None: self.source = input else: self.source += input self.prompt.setText("...") return self.source = None self.prompt.setText(">>>") self.thread.code = input self.thread.event.set() self.running = True self.thread.done.wait(0.05) if self.thread.done.is_set(): self.thread.done.clear() self.running = False else: self.input.setEnabled(False) def prev_history(self): if len(self.input_history) == 0: return if self.input_history_pos is None: self.input_history_pos = len(self.input_history) if self.input_history_pos == 0: return self.input_history_pos -= 1 self.input.setText(self.input_history[self.input_history_pos]) def next_history(self): if self.input_history_pos is None: return if (self.input_history_pos + 1) >= len(self.input_history): self.input_history_pos = None self.input.setText("") return self.input_history_pos += 1 self.input.setText(self.input_history[self.input_history_pos]) def completion_timer_event(self): if self.thread.done.is_set(): self.thread.done.clear() self.running = False self.input.setEnabled(True) self.input.setFocus(Qt.OtherFocusReason) self.prompt.setText(">>>") def request_input(self): self.input_requested = True self.input.setEnabled(True) self.input.setFocus(Qt.OtherFocusReason) self.prompt.setText("") def write_stdout(self, data): self.output.textCursor().movePosition(QTextCursor.End) fmt = QTextCharFormat() fmt.setForeground(QBrush(Qt.blue)) self.output.textCursor().insertText(data, fmt) self.output.ensureCursorVisible() def write_stderr(self, data): self.output.textCursor().movePosition(QTextCursor.End) fmt = QTextCharFormat() fmt.setForeground(QBrush(Qt.red)) self.output.textCursor().insertText(data, fmt) self.output.ensureCursorVisible() def read_stdin(self, size): if Threads.is_gui_thread(): raise RuntimeError, "Cannot call read_stdin from GUI thread" if len(self.input_result) == 0: Threads.run_on_gui_thread(self.request_input) self.input_event.wait() self.input_event.clear() if len(self.input_result) > size: result = self.input_result[0:size] self.input_result = self.input_result[size:] return result result = self.input_result self.input_result = "" return def readline_stdin(self): if Threads.is_gui_thread(): raise RuntimeError, "Cannot call readline_stdin from GUI thread" if len(self.input_result) == 0: Threads.run_on_gui_thread(self.request_input) self.input_event.wait() self.input_event.clear() result = self.input_result self.input_result = "" return result sys.stderr = PythonConsoleOutput(sys.stderr, True) sys.stdout = PythonConsoleOutput(sys.stdout, False) sys.stdin = PythonConsoleInput(sys.stdin) _python_console = threading.local()