"""
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