Source code for mu.modes.python3

"""
The Python3 mode for the Mu editor.

Copyright (c) 2015-2017 Nicholas H.Tollervey and others (see the AUTHORS file).

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 3 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 <http://www.gnu.org/licenses/>.
"""
import sys
import os
import logging
from mu.logic import MODULE_DIR
from mu.modes.base import BaseMode
from mu.modes.api import PYTHON3_APIS, SHARED_APIS, PI_APIS
from mu.resources import load_icon
from mu.interface.panes import CHARTS
from qtconsole.manager import QtKernelManager
from qtconsole.client import QtKernelClient
from PyQt5.QtCore import QObject, QThread, pyqtSignal


logger = logging.getLogger(__name__)


[docs]class KernelRunner(QObject): """ Used to control the iPython kernel in a non-blocking manner so the UI remains responsive. """ kernel_started = pyqtSignal(QtKernelManager, QtKernelClient) kernel_finished = pyqtSignal() # Used to build context with user defined envars when running the REPL. default_envars = os.environ.copy() def __init__(self, cwd, envars): """ Initialise the kernel runner with a target current working directory. """ super().__init__() self.cwd = cwd self.envars = dict(envars)
[docs] def start_kernel(self): """ Create the expected context, start the kernel, obtain a client and emit a signal when both are started. """ logger.info(sys.path) os.chdir(self.cwd) # Ensure the kernel runs with the expected CWD. # Add user defined envars to os.environ so they can be picked up by # the child process running the kernel. logger.info('Starting iPython kernel with user defined envars: ' '{}'.format(self.envars)) for k, v in self.envars.items(): os.environ[k] = v # Ensure the expected paths are in PYTHONPATH of the subprocess so the # kernel and Mu-installed third party applications can be found. if 'PYTHONPATH' not in os.environ: paths = sys.path + [MODULE_DIR, ] os.environ['PYTHONPATH'] = os.pathsep.join(paths) if MODULE_DIR not in os.environ['PYTHONPATH']: # This is needed on Windows to ensure user installed third party # packages are available in the REPL. new_path = os.pathsep.join([os.environ['PYTHONPATH'], MODULE_DIR]) os.environ['PYTHONPATH'] = new_path logger.info("REPL PYTHONPATH: {}".format(os.environ['PYTHONPATH'])) self.repl_kernel_manager = QtKernelManager() self.repl_kernel_manager.start_kernel() self.repl_kernel_client = self.repl_kernel_manager.client() self.kernel_started.emit(self.repl_kernel_manager, self.repl_kernel_client)
[docs] def stop_kernel(self): """ Clean up the context, stop the client connections to the kernel, affect an immediate shutdown of the kernel and emit a "finished" signal. """ os.environ.clear() for k, v in self.default_envars.items(): os.environ[k] = v self.repl_kernel_client.stop_channels() self.repl_kernel_manager.shutdown_kernel(now=True) self.kernel_finished.emit()
[docs]class PythonMode(BaseMode): """ Represents the functionality required by the Python 3 mode. """ name = _('Python 3') description = _('Create code using standard Python 3.') icon = 'python' runner = None has_debugger = True kernel_runner = None stop_kernel = pyqtSignal()
[docs] def actions(self): """ Return an ordered list of actions provided by this module. An action is a name (also used to identify the icon) , description, and handler. """ buttons = [ { 'name': 'run', 'display_name': _('Run'), 'description': _('Run your Python script.'), 'handler': self.run_toggle, 'shortcut': 'F5', }, { 'name': 'debug', 'display_name': _('Debug'), 'description': _('Debug your Python script.'), 'handler': self.debug, 'shortcut': 'F6', }, { 'name': 'repl', 'display_name': _('REPL'), 'description': _('Use the REPL for live coding.'), 'handler': self.toggle_repl, 'shortcut': 'Ctrl+Shift+I', }, ] if CHARTS: buttons.append({ 'name': 'plotter', 'display_name': _('Plotter'), 'description': _('Plot data from your script or the REPL.'), 'handler': self.toggle_plotter, 'shortcut': 'CTRL+Shift+P', }) return buttons
[docs] def api(self): """ Return a list of API specifications to be used by auto-suggest and call tips. """ return SHARED_APIS + PYTHON3_APIS + PI_APIS
[docs] def run_toggle(self, event): """ Handles the toggling of the run button to start/stop a script. """ run_slot = self.view.button_bar.slots['run'] if self.runner: self.stop_script() run_slot.setIcon(load_icon('run')) run_slot.setText(_('Run')) run_slot.setToolTip(_('Run your Python script.')) self.set_buttons(debug=True, modes=True) else: self.run_script() if self.runner: # If the script started, toggle the button state. See #338. run_slot.setIcon(load_icon('stop')) run_slot.setText(_('Stop')) run_slot.setToolTip(_('Stop your Python script.')) self.set_buttons(debug=False, modes=False)
[docs] def run_script(self): """ Run the current script. """ # Grab the Python file. tab = self.view.current_tab if tab is None: logger.debug('There is no active text editor.') self.stop_script() return if tab.path is None: # Unsaved file. self.editor.save() if tab.path: # If needed, save the script. if tab.isModified(): self.editor.save_tab_to_file(tab) envars = self.editor.envars cwd = os.path.dirname(tab.path) self.runner = self.view.add_python3_runner(tab.path, cwd, interactive=True, envars=envars) self.runner.process.waitForStarted() if self.kernel_runner: self.set_buttons(plotter=False) elif self.plotter: self.set_buttons(repl=False)
[docs] def stop_script(self): """ Stop the currently running script. """ logger.debug('Stopping script.') if self.runner: self.runner.process.kill() self.runner.process.waitForFinished() self.runner = None self.view.remove_python_runner() self.set_buttons(plotter=True, repl=True) self.return_focus_to_current_tab()
[docs] def debug(self, event): """ Debug the script using the debug mode. """ logger.info("Starting debug mode.") self.editor.change_mode('debugger') self.editor.mode = 'debugger' self.editor.modes['debugger'].start()
[docs] def toggle_repl(self, event): """ Toggles the REPL on and off """ if self.kernel_runner is None: logger.info('Toggle REPL on.') self.editor.show_status_message(_("Starting iPython REPL.")) self.add_repl() else: logger.info('Toggle REPL off.') self.editor.show_status_message(_("Stopping iPython REPL " "(this may take a short amount " "of time).")) self.remove_repl()
[docs] def add_repl(self): """ Create a new Jupyter REPL session in a non-blocking way. """ self.set_buttons(repl=False) self.kernel_thread = QThread() self.kernel_runner = KernelRunner(cwd=self.workspace_dir(), envars=self.editor.envars) self.kernel_runner.moveToThread(self.kernel_thread) self.kernel_runner.kernel_started.connect(self.on_kernel_start) self.kernel_runner.kernel_finished.connect(self.kernel_thread.quit) self.stop_kernel.connect(self.kernel_runner.stop_kernel) self.kernel_thread.started.connect(self.kernel_runner.start_kernel) self.kernel_thread.finished.connect(self.on_kernel_stop) self.kernel_thread.start()
[docs] def remove_repl(self): """ Remove the Jupyter REPL session. """ self.view.remove_repl() self.set_buttons(repl=False) # Don't block the GUI self.stop_kernel.emit() self.return_focus_to_current_tab()
[docs] def toggle_plotter(self): """ Toggles the plotter on and off. """ if self.plotter is None: logger.info('Toggle plotter on.') self.add_plotter() else: logger.info('Toggle plotter off.') self.remove_plotter()
[docs] def add_plotter(self): """ Add a plotter pane. """ self.view.add_python3_plotter(self) logger.info('Started plotter') self.plotter = True self.set_buttons(debug=False) if self.repl: self.set_buttons(run=False) elif self.runner: self.set_buttons(repl=False)
[docs] def remove_plotter(self): """ Remove the plotter pane, dump data and clean things up. """ self.set_buttons(run=True, repl=True, debug=True) super().remove_plotter()
[docs] def on_data_flood(self): """ Ensure the process (REPL or runner) causing the data flood is stopped *before* the base on_data_flood is called to turn off the plotter and tell the user what to fix. """ self.set_buttons(run=True, repl=True, debug=True) if self.kernel_runner: self.remove_repl() elif self.runner: self.run_toggle(None) super().on_data_flood()
[docs] def on_kernel_start(self, kernel_manager, kernel_client): """ Handles UI update when the kernel runner has started the iPython kernel. """ self.view.add_jupyter_repl(kernel_manager, kernel_client) self.set_buttons(repl=True) if self.runner: self.set_buttons(plotter=False) elif self.plotter: self.set_buttons(run=False, debug=False) self.editor.show_status_message(_("REPL started."))
[docs] def on_kernel_stop(self): """ Handles UI updates for when the kernel runner has shut down the running iPython kernel. """ self.repl_kernel_manager = None self.set_buttons(repl=True, plotter=True, run=True) self.editor.show_status_message(_("REPL stopped.")) self.kernel_runner = None