# -*- coding: utf-8 -*-
#
# Copyright © Spyder Project Contributors
# Licensed under the terms of the MIT License
# (see spyder/__init__.py for details)
"""
IPython Console plugin based on QtConsole
"""
# pylint: disable=C0103
# pylint: disable=R0903
# pylint: disable=R0911
# pylint: disable=R0201
# Standard library imports
import atexit
import os
import os.path as osp
import uuid
import sys
import traceback
# Third party imports
from jupyter_client.connect import find_connection_file
from jupyter_core.paths import jupyter_config_dir, jupyter_runtime_dir
from qtconsole.client import QtKernelClient
from qtconsole.manager import QtKernelManager
from qtpy import PYQT5
from qtpy.compat import getopenfilename
from qtpy.QtCore import Qt, Signal, Slot
from qtpy.QtWidgets import (QApplication, QCheckBox, QDialog, QDialogButtonBox,
QFormLayout, QGridLayout, QGroupBox, QHBoxLayout,
QLabel, QLineEdit, QMessageBox, QPushButton,
QTabWidget, QVBoxLayout, QWidget)
from traitlets.config.loader import Config, load_pyconfig_files
from zmq.ssh import tunnel as zmqtunnel
try:
import pexpect
except ImportError:
pexpect = None
# Local imports
from spyder import dependencies
from spyder.config.base import _, get_conf_path, get_home_dir
from spyder.config.main import CONF
from spyder.plugins import SpyderPluginWidget
from spyder.plugins.configdialog import PluginConfigPage
from spyder.py3compat import is_string, PY2, to_text_string
from spyder.utils.ipython.kernelspec import SpyderKernelSpec
from spyder.utils.ipython.style import create_qss_style
from spyder.utils.qthelpers import create_action, MENU_SEPARATOR
from spyder.utils import icon_manager as ima
from spyder.utils import encoding, programs, sourcecode
from spyder.utils.programs import get_temp_dir
from spyder.utils.misc import get_error_match, remove_backslashes
from spyder.widgets.findreplace import FindReplace
from spyder.widgets.ipythonconsole import ClientWidget
from spyder.widgets.tabs import Tabs
# Dependencies
SYMPY_REQVER = '>=0.7.3'
dependencies.add("sympy", _("Symbolic mathematics in the IPython Console"),
required_version=SYMPY_REQVER, optional=True)
CYTHON_REQVER = '>=0.21'
dependencies.add("cython", _("Run Cython files in the IPython Console"),
required_version=CYTHON_REQVER, optional=True)
QTCONSOLE_REQVER = ">=4.2.0"
dependencies.add("qtconsole", _("Integrate the IPython console"),
required_version=QTCONSOLE_REQVER)
IPYTHON_REQVER = ">=4.0;<6.0" if PY2 else ">=4.0"
dependencies.add("IPython", _("IPython interactive python environment"),
required_version=IPYTHON_REQVER)
MATPLOTLIB_REQVER = '>=2.0.0'
dependencies.add("matplotlib", _("Display 2D graphics in the IPython Console"),
required_version=MATPLOTLIB_REQVER, optional=True)
#------------------------------------------------------------------------------
# Existing kernels
#------------------------------------------------------------------------------
# Replacing pyzmq openssh_tunnel method to work around the issue
# https://github.com/zeromq/pyzmq/issues/589 which was solved in pyzmq
# https://github.com/zeromq/pyzmq/pull/615
def _stop_tunnel(cmd):
pexpect.run(cmd)
def openssh_tunnel(self, lport, rport, server, remoteip='127.0.0.1',
keyfile=None, password=None, timeout=0.4):
if pexpect is None:
raise ImportError("pexpect unavailable, use paramiko_tunnel")
ssh="ssh "
if keyfile:
ssh += "-i " + keyfile
if ':' in server:
server, port = server.split(':')
ssh += " -p %s" % port
cmd = "%s -O check %s" % (ssh, server)
(output, exitstatus) = pexpect.run(cmd, withexitstatus=True)
if not exitstatus:
pid = int(output[output.find("(pid=")+5:output.find(")")])
cmd = "%s -O forward -L 127.0.0.1:%i:%s:%i %s" % (
ssh, lport, remoteip, rport, server)
(output, exitstatus) = pexpect.run(cmd, withexitstatus=True)
if not exitstatus:
atexit.register(_stop_tunnel, cmd.replace("-O forward",
"-O cancel",
1))
return pid
cmd = "%s -f -S none -L 127.0.0.1:%i:%s:%i %s sleep %i" % (
ssh, lport, remoteip, rport, server, timeout)
# pop SSH_ASKPASS from env
env = os.environ.copy()
env.pop('SSH_ASKPASS', None)
ssh_newkey = 'Are you sure you want to continue connecting'
tunnel = pexpect.spawn(cmd, env=env)
failed = False
while True:
try:
i = tunnel.expect([ssh_newkey, '[Pp]assword:'], timeout=.1)
if i==0:
host = server.split('@')[-1]
question = _("The authenticity of host %s can't be "
"established. Are you sure you want to continue "
"connecting?") % host
reply = QMessageBox.question(self, _('Warning'), question,
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No)
if reply == QMessageBox.Yes:
tunnel.sendline('yes')
continue
else:
tunnel.sendline('no')
raise RuntimeError(
_("The authenticity of the host can't be established"))
if i==1 and password is not None:
tunnel.sendline(password)
except pexpect.TIMEOUT:
continue
except pexpect.EOF:
if tunnel.exitstatus:
raise RuntimeError(_("Tunnel '%s' failed to start") % cmd)
else:
return tunnel.pid
else:
if failed or password is None:
raise RuntimeError(_("Could not connect to remote host"))
# TODO: Use this block when pyzmq bug #620 is fixed
# # Prompt a passphrase dialog to the user for a second attempt
# password, ok = QInputDialog.getText(self, _('Password'),
# _('Enter password for: ') + server,
# echo=QLineEdit.Password)
# if ok is False:
# raise RuntimeError('Could not connect to remote host.')
tunnel.sendline(password)
failed = True
class KernelConnectionDialog(QDialog):
"""Dialog to connect to existing kernels (either local or remote)."""
def __init__(self, parent=None):
super(KernelConnectionDialog, self).__init__(parent)
self.setWindowTitle(_('Connect to an existing kernel'))
main_label = QLabel(_(
"Please select the JSON connection file (e.g. "
"kernel-3764.json) or enter the 4-digit ID (e.g. "
"3764) of the existing kernel to connect to, "
"and enter the SSH host name and credentials if a remote kernel."))
main_label.setWordWrap(True)
main_label.setAlignment(Qt.AlignJustify)
# Connection file
cf_label = QLabel(_('Kernel ID/Connection file:'))
self.cf = QLineEdit()
self.cf.setPlaceholderText(_('ID number or path to connection file'))
self.cf.setMinimumWidth(250)
cf_open_btn = QPushButton(_('Browse'))
cf_open_btn.clicked.connect(self.select_connection_file)
cf_layout = QHBoxLayout()
cf_layout.addWidget(cf_label)
cf_layout.addWidget(self.cf)
cf_layout.addWidget(cf_open_btn)
# Remote kernel checkbox
self.rm_cb = QCheckBox(_('This is a remote kernel (via SSH)'))
# Descriptive text explaining SSH credentials
credential_label = QLabel(_(
"Note: If connecting to a remote kernel, only the "
"SSH keyfile or the Password field need to be completed, "
"unless the keyfile is protected with a passphrase."))
credential_label.setWordWrap(True)
# SSH connection
self.hn = QLineEdit()
self.hn.setPlaceholderText(_('username@hostname:port'))
self.pw = QLineEdit()
self.pw.setPlaceholderText(_('Remote user password or '
'SSH keyfile passphrase'))
self.pw.setEchoMode(QLineEdit.Password)
self.kf = QLineEdit()
self.kf.setPlaceholderText(_('Path to SSH keyfile (optional)'))
kf_open_btn = QPushButton(_('Browse'))
kf_open_btn.clicked.connect(self.select_ssh_key)
kf_layout = QHBoxLayout()
kf_layout.addWidget(self.kf)
kf_layout.addWidget(kf_open_btn)
ssh_form = QFormLayout()
ssh_form.addRow(_('Host name:'), self.hn)
ssh_form.addRow(_('Password:'), self.pw)
ssh_form.addRow(_('SSH keyfile:'), kf_layout)
# Ok and Cancel buttons
self.accept_btns = QDialogButtonBox(
QDialogButtonBox.Ok | QDialogButtonBox.Cancel,
Qt.Horizontal, self)
self.accept_btns.accepted.connect(self.accept)
self.accept_btns.rejected.connect(self.reject)
# Dialog layout
layout = QVBoxLayout(self)
layout.addWidget(main_label)
layout.addLayout(cf_layout)
layout.addWidget(self.rm_cb)
layout.addWidget(credential_label)
layout.addLayout(ssh_form)
layout.addWidget(self.accept_btns)
# Remote kernel checkbox enables the ssh_connection_form
def ssh_set_enabled(state):
for wid in [credential_label, self.hn, self.pw,
self.kf, kf_open_btn]:
wid.setEnabled(state)
for i in range(ssh_form.rowCount()):
ssh_form.itemAt(2 * i).widget().setEnabled(state)
ssh_set_enabled(self.rm_cb.checkState())
self.rm_cb.stateChanged.connect(ssh_set_enabled)
def select_connection_file(self):
cf = getopenfilename(self, _('Select kernel connection file'),
jupyter_runtime_dir(), '*.json;;*.*')[0]
self.cf.setText(cf)
def select_ssh_key(self):
kf = getopenfilename(self, _('Select SSH keyfile'),
get_home_dir(), '*.pem;;*.*')[0]
self.kf.setText(kf)
@staticmethod
def get_connection_parameters(parent=None, dialog=None):
if not dialog:
dialog = KernelConnectionDialog(parent)
result = dialog.exec_()
is_remote = bool(dialog.rm_cb.checkState())
accepted = result == QDialog.Accepted
if is_remote:
falsy_to_none = lambda arg: arg if arg else None
return (dialog.cf.text(), # connection file
falsy_to_none(dialog.hn.text()), # host name
falsy_to_none(dialog.kf.text()), # ssh key file
falsy_to_none(dialog.pw.text()), # ssh password
accepted) # ok
else:
path = dialog.cf.text()
_dir, filename = osp.dirname(path), osp.basename(path)
if _dir == '' and not filename.endswith('.json'):
path = osp.join(jupyter_runtime_dir(), 'kernel-'+path+'.json')
return (path, None, None, None, accepted)
#------------------------------------------------------------------------------
# Config page
#------------------------------------------------------------------------------
class IPythonConsoleConfigPage(PluginConfigPage):
def __init__(self, plugin, parent):
PluginConfigPage.__init__(self, plugin, parent)
self.get_name = lambda: _("IPython console")
def setup_page(self):
newcb = self.create_checkbox
# Interface Group
interface_group = QGroupBox(_("Interface"))
banner_box = newcb(_("Display initial banner"), 'show_banner',
tip=_("This option lets you hide the message shown at\n"
"the top of the console when it's opened."))
pager_box = newcb(_("Use a pager to display additional text inside "
"the console"), 'use_pager',
tip=_("Useful if you don't want to fill the "
"console with long help or completion texts.\n"
"Note: Use the Q key to get out of the "
"pager."))
calltips_box = newcb(_("Display balloon tips"), 'show_calltips')
ask_box = newcb(_("Ask for confirmation before closing"),
'ask_before_closing')
reset_namespace_box = newcb(
_("Ask for confirmation before removing all user-defined "
"variables"),
'show_reset_namespace_warning',
tip=_("This option lets you hide the warning message shown\n"
"when resetting the namespace from Spyder."))
show_time_box = newcb(_("Show elapsed time"), 'show_elapsed_time')
ask_restart_box = newcb(
_("Ask for confirmation before restarting"),
'ask_before_restart',
tip=_("This option lets you hide the warning message shown\n"
"when restarting the kernel."))
interface_layout = QVBoxLayout()
interface_layout.addWidget(banner_box)
interface_layout.addWidget(pager_box)
interface_layout.addWidget(calltips_box)
interface_layout.addWidget(ask_box)
interface_layout.addWidget(reset_namespace_box)
interface_layout.addWidget(show_time_box)
interface_layout.addWidget(ask_restart_box)
interface_group.setLayout(interface_layout)
comp_group = QGroupBox(_("Completion Type"))
comp_label = QLabel(_("Decide what type of completion to use"))
comp_label.setWordWrap(True)
completers = [(_("Graphical"), 0), (_("Terminal"), 1), (_("Plain"), 2)]
comp_box = self.create_combobox(_("Completion:")+" ", completers,
'completion_type')
comp_layout = QVBoxLayout()
comp_layout.addWidget(comp_label)
comp_layout.addWidget(comp_box)
comp_group.setLayout(comp_layout)
# Source Code Group
source_code_group = QGroupBox(_("Source code"))
buffer_spin = self.create_spinbox(
_("Buffer: "), _(" lines"),
'buffer_size', min_=-1, max_=1000000, step=100,
tip=_("Set the maximum number of lines of text shown in the\n"
"console before truncation. Specifying -1 disables it\n"
"(not recommended!)"))
source_code_layout = QVBoxLayout()
source_code_layout.addWidget(buffer_spin)
source_code_group.setLayout(source_code_layout)
# --- Graphics ---
# Pylab Group
pylab_group = QGroupBox(_("Support for graphics (Matplotlib)"))
pylab_box = newcb(_("Activate support"), 'pylab')
autoload_pylab_box = newcb(_("Automatically load Pylab and NumPy "
"modules"),
'pylab/autoload',
tip=_("This lets you load graphics support "
"without importing \nthe commands to do "
"plots. Useful to work with other\n"
"plotting libraries different to "
"Matplotlib or to develop \nGUIs with "
"Spyder."))
autoload_pylab_box.setEnabled(self.get_option('pylab'))
pylab_box.toggled.connect(autoload_pylab_box.setEnabled)
pylab_layout = QVBoxLayout()
pylab_layout.addWidget(pylab_box)
pylab_layout.addWidget(autoload_pylab_box)
pylab_group.setLayout(pylab_layout)
# Pylab backend Group
inline = _("Inline")
automatic = _("Automatic")
backend_group = QGroupBox(_("Graphics backend"))
bend_label = QLabel(_("Decide how graphics are going to be displayed "
"in the console. If unsure, please select "
"%s to put graphics inside the "
"console or %s to interact with "
"them (through zooming and panning) in a "
"separate window.") % (inline, automatic))
bend_label.setWordWrap(True)
backends = [(inline, 0), (automatic, 1), ("Qt5", 2), ("Qt4", 3)]
if sys.platform == 'darwin':
backends.append( ("OS X", 4) )
if sys.platform.startswith('linux'):
backends.append( ("Gtk3", 5) )
backends.append( ("Gtk", 6) )
if PY2:
backends.append( ("Wx", 7) )
backends.append( ("Tkinter", 8) )
backends = tuple(backends)
backend_box = self.create_combobox( _("Backend:")+" ", backends,
'pylab/backend', default=0,
tip=_("This option will be applied the "
"next time a console is opened."))
backend_layout = QVBoxLayout()
backend_layout.addWidget(bend_label)
backend_layout.addWidget(backend_box)
backend_group.setLayout(backend_layout)
backend_group.setEnabled(self.get_option('pylab'))
pylab_box.toggled.connect(backend_group.setEnabled)
# Inline backend Group
inline_group = QGroupBox(_("Inline backend"))
inline_label = QLabel(_("Decide how to render the figures created by "
"this backend"))
inline_label.setWordWrap(True)
formats = (("PNG", 0), ("SVG", 1))
format_box = self.create_combobox(_("Format:")+" ", formats,
'pylab/inline/figure_format',
default=0)
resolution_spin = self.create_spinbox(
_("Resolution:")+" ", " "+_("dpi"),
'pylab/inline/resolution', min_=50, max_=999, step=0.1,
tip=_("Only used when the format is PNG. Default is "
"72"))
width_spin = self.create_spinbox(
_("Width:")+" ", " "+_("inches"),
'pylab/inline/width', min_=2, max_=20, step=1,
tip=_("Default is 6"))
height_spin = self.create_spinbox(
_("Height:")+" ", " "+_("inches"),
'pylab/inline/height', min_=1, max_=20, step=1,
tip=_("Default is 4"))
bbox_inches_box = newcb(
_("Use a tight layout for inline plots"),
'pylab/inline/bbox_inches',
tip=_("Sets bbox_inches to \"tight\" when\n"
"plotting inline with matplotlib.\n"
"When enabled, can cause discrepancies\n"
"between the image displayed inline and\n"
"that created using savefig."))
inline_v_layout = QVBoxLayout()
inline_v_layout.addWidget(inline_label)
inline_layout = QGridLayout()
inline_layout.addWidget(format_box.label, 1, 0)
inline_layout.addWidget(format_box.combobox, 1, 1)
inline_layout.addWidget(resolution_spin.plabel, 2, 0)
inline_layout.addWidget(resolution_spin.spinbox, 2, 1)
inline_layout.addWidget(resolution_spin.slabel, 2, 2)
inline_layout.addWidget(width_spin.plabel, 3, 0)
inline_layout.addWidget(width_spin.spinbox, 3, 1)
inline_layout.addWidget(width_spin.slabel, 3, 2)
inline_layout.addWidget(height_spin.plabel, 4, 0)
inline_layout.addWidget(height_spin.spinbox, 4, 1)
inline_layout.addWidget(height_spin.slabel, 4, 2)
inline_layout.addWidget(bbox_inches_box, 5, 0, 1, 4)
inline_h_layout = QHBoxLayout()
inline_h_layout.addLayout(inline_layout)
inline_h_layout.addStretch(1)
inline_v_layout.addLayout(inline_h_layout)
inline_group.setLayout(inline_v_layout)
inline_group.setEnabled(self.get_option('pylab'))
pylab_box.toggled.connect(inline_group.setEnabled)
# --- Startup ---
# Run lines Group
run_lines_group = QGroupBox(_("Run code"))
run_lines_label = QLabel(_("You can run several lines of code when "
"a console is started. Please introduce "
"each one separated by commas, for "
"example:
"
"import os, import sys"))
run_lines_label.setWordWrap(True)
run_lines_edit = self.create_lineedit(_("Lines:"), 'startup/run_lines',
'', alignment=Qt.Horizontal)
run_lines_layout = QVBoxLayout()
run_lines_layout.addWidget(run_lines_label)
run_lines_layout.addWidget(run_lines_edit)
run_lines_group.setLayout(run_lines_layout)
# Run file Group
run_file_group = QGroupBox(_("Run a file"))
run_file_label = QLabel(_("You can also run a whole file at startup "
"instead of just some lines (This is "
"similar to have a PYTHONSTARTUP file)."))
run_file_label.setWordWrap(True)
file_radio = newcb(_("Use the following file:"),
'startup/use_run_file', False)
run_file_browser = self.create_browsefile('', 'startup/run_file', '')
run_file_browser.setEnabled(False)
file_radio.toggled.connect(run_file_browser.setEnabled)
run_file_layout = QVBoxLayout()
run_file_layout.addWidget(run_file_label)
run_file_layout.addWidget(file_radio)
run_file_layout.addWidget(run_file_browser)
run_file_group.setLayout(run_file_layout)
# ---- Advanced settings ----
# Enable Jedi completion
jedi_group = QGroupBox(_("Jedi completion"))
jedi_label = QLabel(_("Enable Jedi-based Tab completion "
"in the IPython console; similar to the "
"greedy completer, but without evaluating "
"the code.
"
"Warning: Slows down your console "
"when working with large dataframes!"))
jedi_label.setWordWrap(True)
jedi_box = newcb(_("Use Jedi completion in the IPython console"),
"jedi_completer",
tip="Warning: "
"Slows down your console when working with "
"large dataframes!
"
"Allows completion of nested lists etc.")
jedi_layout = QVBoxLayout()
jedi_layout.addWidget(jedi_label)
jedi_layout.addWidget(jedi_box)
jedi_group.setLayout(jedi_layout)
# Greedy completer group
greedy_group = QGroupBox(_("Greedy completion"))
greedy_label = QLabel(_("Enable Tab completion on elements "
"of lists, results of function calls, etc, "
"without assigning them to a variable, "
"like li[0].<Tab> or "
"ins.meth().<Tab>
"
"Warning: Due to a bug, IPython's "
"greedy completer requires a leading "
"<Space> for some completions; "
"e.g. np.sin(<Space>np.<Tab>"
" works while np.sin(np.<Tab> "
" doesn't."))
greedy_label.setWordWrap(True)
greedy_box = newcb(_("Use greedy completion in the IPython console"),
"greedy_completer",
tip="Warning: It can be unsafe because the "
"code is actually evaluated when you press "
"Tab.")
greedy_layout = QVBoxLayout()
greedy_layout.addWidget(greedy_label)
greedy_layout.addWidget(greedy_box)
greedy_group.setLayout(greedy_layout)
# Autocall group
autocall_group = QGroupBox(_("Autocall"))
autocall_label = QLabel(_("Autocall makes IPython automatically call "
"any callable object even if you didn't "
"type explicit parentheses.
"
"For example, if you type str 43 it "
"becomes str(43) automatically."))
autocall_label.setWordWrap(True)
smart = _('Smart')
full = _('Full')
autocall_opts = ((_('Off'), 0), (smart, 1), (full, 2))
autocall_box = self.create_combobox(
_("Autocall: "), autocall_opts, 'autocall', default=0,
tip=_("On %s mode, Autocall is not applied if "
"there are no arguments after the callable. On "
"%s mode, all callable objects are "
"automatically called (even if no arguments are "
"present).") % (smart, full))
autocall_layout = QVBoxLayout()
autocall_layout.addWidget(autocall_label)
autocall_layout.addWidget(autocall_box)
autocall_group.setLayout(autocall_layout)
# Sympy group
sympy_group = QGroupBox(_("Symbolic Mathematics"))
sympy_label = QLabel(_("Perfom symbolic operations in the console "
"(e.g. integrals, derivatives, vector calculus, "
"etc) and get the outputs in a beautifully "
"printed style (it requires the Sympy module)."))
sympy_label.setWordWrap(True)
sympy_box = newcb(_("Use symbolic math"), "symbolic_math",
tip=_("This option loads the Sympy library to work "
"with.
Please refer to its documentation to "
"learn how to use it."))
sympy_layout = QVBoxLayout()
sympy_layout.addWidget(sympy_label)
sympy_layout.addWidget(sympy_box)
sympy_group.setLayout(sympy_layout)
# Prompts group
prompts_group = QGroupBox(_("Prompts"))
prompts_label = QLabel(_("Modify how Input and Output prompts are "
"shown in the console."))
prompts_label.setWordWrap(True)
in_prompt_edit = self.create_lineedit(_("Input prompt:"),
'in_prompt', '',
_('Default is
'
'In [<span class="in-prompt-number">'
'%i</span>]:'),
alignment=Qt.Horizontal)
out_prompt_edit = self.create_lineedit(_("Output prompt:"),
'out_prompt', '',
_('Default is
'
'Out[<span class="out-prompt-number">'
'%i</span>]:'),
alignment=Qt.Horizontal)
prompts_layout = QVBoxLayout()
prompts_layout.addWidget(prompts_label)
prompts_g_layout = QGridLayout()
prompts_g_layout.addWidget(in_prompt_edit.label, 0, 0)
prompts_g_layout.addWidget(in_prompt_edit.textbox, 0, 1)
prompts_g_layout.addWidget(out_prompt_edit.label, 1, 0)
prompts_g_layout.addWidget(out_prompt_edit.textbox, 1, 1)
prompts_layout.addLayout(prompts_g_layout)
prompts_group.setLayout(prompts_layout)
# Windows adjustments
windows_group = QGroupBox(_("Windows adjustments"))
hide_cmd_windows = newcb(
_("Hide command line output windows "
"generated by the subprocess module."),
'hide_cmd_windows')
windows_layout = QVBoxLayout()
windows_layout.addWidget(hide_cmd_windows)
windows_group.setLayout(windows_layout)
# --- Tabs organization ---
tabs = QTabWidget()
tabs.addTab(self.create_tab(interface_group, comp_group,
source_code_group), _("Display"))
tabs.addTab(self.create_tab(pylab_group, backend_group, inline_group),
_("Graphics"))
tabs.addTab(self.create_tab(run_lines_group, run_file_group),
_("Startup"))
tabs.addTab(self.create_tab(jedi_group, greedy_group, autocall_group,
sympy_group, prompts_group,
windows_group), _("Advanced Settings"))
vlayout = QVBoxLayout()
vlayout.addWidget(tabs)
self.setLayout(vlayout)
#------------------------------------------------------------------------------
# Plugin widget
#------------------------------------------------------------------------------
class IPythonConsole(SpyderPluginWidget):
"""
IPython Console plugin
This is a widget with tabs where each one is a ClientWidget
"""
CONF_SECTION = 'ipython_console'
CONFIGWIDGET_CLASS = IPythonConsoleConfigPage
DISABLE_ACTIONS_WHEN_HIDDEN = False
# Signals
focus_changed = Signal()
edit_goto = Signal((str, int, str), (str, int, str, bool))
# Error messages
permission_error_msg = _("The directory {} is not writable and it is "
"required to create IPython consoles. Please "
"make it writable.")
def __init__(self, parent, testing=False, test_dir=None,
test_no_stderr=False):
"""Ipython Console constructor."""
if PYQT5:
SpyderPluginWidget.__init__(self, parent, main = parent)
else:
SpyderPluginWidget.__init__(self, parent)
self.tabwidget = None
self.menu_actions = None
self.master_clients = 0
self.clients = []
self.filenames = []
self.mainwindow_close = False
self.create_new_client_if_empty = True
# Attrs for testing
self.testing = testing
self.test_dir = test_dir
self.test_no_stderr = test_no_stderr
# Initialize plugin
self.initialize_plugin()
# Create temp dir on testing to save kernel errors
if self.test_dir is not None:
if not osp.isdir(osp.join(test_dir)):
os.makedirs(osp.join(test_dir))
layout = QVBoxLayout()
self.tabwidget = Tabs(self, self.menu_actions, rename_tabs=True,
split_char='/', split_index=0)
if hasattr(self.tabwidget, 'setDocumentMode')\
and not sys.platform == 'darwin':
# Don't set document mode to true on OSX because it generates
# a crash when the console is detached from the main window
# Fixes Issue 561
self.tabwidget.setDocumentMode(True)
self.tabwidget.currentChanged.connect(self.refresh_plugin)
self.tabwidget.tabBar().tabMoved.connect(self.move_tab)
self.tabwidget.tabBar().sig_change_name.connect(
self.rename_tabs_after_change)
self.tabwidget.set_close_function(self.close_client)
if sys.platform == 'darwin':
tab_container = QWidget()
tab_container.setObjectName('tab-container')
tab_layout = QHBoxLayout(tab_container)
tab_layout.setContentsMargins(0, 0, 0, 0)
tab_layout.addWidget(self.tabwidget)
layout.addWidget(tab_container)
else:
layout.addWidget(self.tabwidget)
# Find/replace widget
self.find_widget = FindReplace(self)
self.find_widget.hide()
self.register_widget_shortcuts(self.find_widget)
layout.addWidget(self.find_widget)
self.setLayout(layout)
# Accepting drops
self.setAcceptDrops(True)
#------ SpyderPluginMixin API ---------------------------------------------
def update_font(self):
"""Update font from Preferences"""
font = self.get_plugin_font()
for client in self.clients:
client.set_font(font)
def apply_plugin_settings(self, options):
"""Apply configuration file's plugin settings"""
font_n = 'plugin_font'
font_o = self.get_plugin_font()
help_n = 'connect_to_oi'
help_o = CONF.get('help', 'connect/ipython_console')
color_scheme_n = 'color_scheme_name'
color_scheme_o = CONF.get('color_schemes', 'selected')
show_time_n = 'show_elapsed_time'
show_time_o = self.get_option(show_time_n)
reset_namespace_n = 'show_reset_namespace_warning'
reset_namespace_o = self.get_option(reset_namespace_n)
ask_before_restart_n = 'ask_before_restart'
ask_before_restart_o = self.get_option(ask_before_restart_n)
for client in self.clients:
control = client.get_control()
if font_n in options:
client.set_font(font_o)
if help_n in options and control is not None:
control.set_help_enabled(help_o)
if color_scheme_n in options:
client.set_color_scheme(color_scheme_o)
if show_time_n in options:
client.show_time_action.setChecked(show_time_o)
client.set_elapsed_time_visible(show_time_o)
if reset_namespace_n in options:
client.reset_warning = reset_namespace_o
if ask_before_restart_n in options:
client.ask_before_restart = ask_before_restart_o
def toggle_view(self, checked):
"""Toggle view"""
if checked:
self.dockwidget.show()
self.dockwidget.raise_()
# Start a client in case there are none shown
if not self.clients:
if self.main.is_setting_up:
self.create_new_client(give_focus=False)
else:
self.create_new_client(give_focus=True)
else:
self.dockwidget.hide()
#------ SpyderPluginWidget API --------------------------------------------
def get_plugin_title(self):
"""Return widget title"""
return _('IPython console')
def get_plugin_icon(self):
"""Return widget icon"""
return ima.icon('ipython_console')
def get_focus_widget(self):
"""
Return the widget to give focus to when
this plugin's dockwidget is raised on top-level
"""
client = self.tabwidget.currentWidget()
if client is not None:
return client.get_control()
def closing_plugin(self, cancelable=False):
"""Perform actions before parent main window is closed"""
self.mainwindow_close = True
for client in self.clients:
client.shutdown()
client.remove_stderr_file()
client.close()
return True
def refresh_plugin(self):
"""Refresh tabwidget"""
client = None
if self.tabwidget.count():
# Give focus to the control widget of the selected tab
client = self.tabwidget.currentWidget()
control = client.get_control()
control.setFocus()
buttons = [[b, -7] for b in client.get_toolbar_buttons()]
buttons = sum(buttons, [])[:-1]
widgets = [client.create_time_label()] + buttons
else:
control = None
widgets = []
self.find_widget.set_editor(control)
self.tabwidget.set_corner_widgets({Qt.TopRightCorner: widgets})
if client:
sw = client.shellwidget
self.main.variableexplorer.set_shellwidget_from_id(id(sw))
self.main.help.set_shell(sw)
self.update_tabs_text()
self.update_plugin_title.emit()
def get_plugin_actions(self):
"""Return a list of actions related to plugin."""
create_client_action = create_action(
self,
_("Open an &IPython console"),
icon=ima.icon('ipython_console'),
triggered=self.create_new_client,
context=Qt.WidgetWithChildrenShortcut)
self.register_shortcut(create_client_action, context="ipython_console",
name="New tab")
restart_action = create_action(self, _("Restart kernel"),
icon=ima.icon('restart'),
triggered=self.restart_kernel,
context=Qt.WidgetWithChildrenShortcut)
self.register_shortcut(restart_action, context="ipython_console",
name="Restart kernel")
connect_to_kernel_action = create_action(self,
_("Connect to an existing kernel"), None, None,
_("Open a new IPython console connected to an existing kernel"),
triggered=self.create_client_for_kernel)
rename_tab_action = create_action(self, _("Rename tab"),
icon=ima.icon('rename'),
triggered=self.tab_name_editor)
# Add the action to the 'Consoles' menu on the main window
main_consoles_menu = self.main.consoles_menu_actions
main_consoles_menu.insert(0, create_client_action)
main_consoles_menu += [MENU_SEPARATOR, restart_action,
connect_to_kernel_action]
# Plugin actions
self.menu_actions = [create_client_action, MENU_SEPARATOR,
restart_action, connect_to_kernel_action,
MENU_SEPARATOR, rename_tab_action]
return self.menu_actions
def register_plugin(self):
"""Register plugin in Spyder's main window"""
self.main.add_dockwidget(self)
self.focus_changed.connect(self.main.plugin_focus_changed)
self.edit_goto.connect(self.main.editor.load)
self.edit_goto[str, int, str, bool].connect(
lambda fname, lineno, word, processevents:
self.main.editor.load(fname, lineno, word,
processevents=processevents))
self.main.editor.breakpoints_saved.connect(self.set_spyder_breakpoints)
self.main.editor.run_in_current_ipyclient.connect(self.run_script)
self.main.workingdirectory.set_current_console_wd.connect(
self.set_current_client_working_directory)
self.tabwidget.currentChanged.connect(self.update_working_directory)
self.main.explorer.open_interpreter.connect(
self.create_client_from_path)
self.main.explorer.run.connect(lambda fname: self.run_script(
fname, osp.dirname(fname), '', False, False, False, True))
self.main.projects.open_interpreter.connect(
self.create_client_from_path)
self.main.projects.run.connect(lambda fname: self.run_script(
fname, osp.dirname(fname), '', False, False, False, True))
self._remove_old_stderr_files()
#------ Public API (for clients) ------------------------------------------
def get_clients(self):
"""Return clients list"""
return [cl for cl in self.clients if isinstance(cl, ClientWidget)]
def get_focus_client(self):
"""Return current client with focus, if any"""
widget = QApplication.focusWidget()
for client in self.get_clients():
if widget is client or widget is client.get_control():
return client
def get_current_client(self):
"""Return the currently selected client"""
client = self.tabwidget.currentWidget()
if client is not None:
return client
def get_current_shellwidget(self):
"""Return the shellwidget of the current client"""
client = self.get_current_client()
if client is not None:
return client.shellwidget
def run_script(self, filename, wdir, args, debug, post_mortem,
current_client, clear_variables):
"""Run script in current or dedicated client"""
norm = lambda text: remove_backslashes(to_text_string(text))
# Run Cython files in a dedicated console
is_cython = osp.splitext(filename)[1] == '.pyx'
if is_cython:
current_client = False
# Select client to execute code on it
is_new_client = False
if current_client:
client = self.get_current_client()
else:
client = self.get_client_for_file(filename)
if client is None:
self.create_client_for_file(filename, is_cython=is_cython)
client = self.get_current_client()
is_new_client = True
if client is not None:
# Internal kernels, use runfile
if client.get_kernel() is not None:
line = "%s('%s'" % ('debugfile' if debug else 'runfile',
norm(filename))
if args:
line += ", args='%s'" % norm(args)
if wdir:
line += ", wdir='%s'" % norm(wdir)
if post_mortem:
line += ", post_mortem=True"
line += ")"
else: # External kernels, use %run
line = "%run "
if debug:
line += "-d "
line += "\"%s\"" % to_text_string(filename)
if args:
line += " %s" % norm(args)
try:
if client.shellwidget._executing:
# Don't allow multiple executions when there's
# still an execution taking place
# Fixes issue 7293
pass
elif client.shellwidget._reading:
client.shellwidget._append_html(
_("
Please exit from debugging before trying to "
"run a file in this console.\n
"),
before_prompt=True)
return
elif current_client:
self.execute_code(line, current_client, clear_variables)
else:
if is_new_client:
client.shellwidget.silent_execute('%clear')
else:
client.shellwidget.execute('%clear')
client.shellwidget.sig_prompt_ready.connect(
lambda: self.execute_code(line, current_client,
clear_variables))
except AttributeError:
pass
self.switch_to_plugin()
else:
#XXX: not sure it can really happen
QMessageBox.warning(self, _('Warning'),
_("No IPython console is currently available to run %s."
"
Please open a new one and try again."
) % osp.basename(filename), QMessageBox.Ok)
def set_current_client_working_directory(self, directory):
"""Set current client working directory."""
shellwidget = self.get_current_shellwidget()
if shellwidget is not None:
shellwidget.set_cwd(directory)
def set_working_directory(self, dirname):
"""Set current working directory.
In the workingdirectory and explorer plugins.
"""
if dirname:
self.main.workingdirectory.chdir(dirname, refresh_explorer=True,
refresh_console=False)
def update_working_directory(self):
"""Update working directory to console cwd."""
shellwidget = self.get_current_shellwidget()
if shellwidget is not None:
shellwidget.get_cwd()
def execute_code(self, lines, current_client=True, clear_variables=False):
"""Execute code instructions."""
sw = self.get_current_shellwidget()
if sw is not None:
if sw._reading:
pass
else:
if not current_client:
# Clear console and reset namespace for
# dedicated clients
# See issue 5748
try:
sw.sig_prompt_ready.disconnect()
except TypeError:
pass
sw.reset_namespace(warning=False, silent=True)
elif current_client and clear_variables:
sw.reset_namespace(warning=False, silent=True)
# Needed to handle an error when kernel_client is none
# See issue 6308
try:
sw.execute(to_text_string(lines))
except AttributeError:
pass
self.activateWindow()
self.get_current_client().get_control().setFocus()
def write_to_stdin(self, line):
sw = self.get_current_shellwidget()
if sw is not None:
# Needed to handle an error when kernel_client is None
# See issue 7578
try:
sw.write_to_stdin(line)
except AttributeError:
pass
@Slot()
@Slot(bool)
@Slot(str)
@Slot(bool, str)
@Slot(bool, bool)
@Slot(bool, str, bool)
def create_new_client(self, give_focus=True, filename='', is_cython=False):
"""Create a new client"""
self.master_clients += 1
client_id = dict(int_id=to_text_string(self.master_clients),
str_id='A')
cf = self._new_connection_file()
show_elapsed_time = self.get_option('show_elapsed_time')
reset_warning = self.get_option('show_reset_namespace_warning')
ask_before_restart = self.get_option('ask_before_restart')
client = ClientWidget(self, id_=client_id,
history_filename=get_conf_path('history.py'),
config_options=self.config_options(),
additional_options=self.additional_options(),
interpreter_versions=self.interpreter_versions(),
connection_file=cf,
menu_actions=self.menu_actions,
show_elapsed_time=show_elapsed_time,
reset_warning=reset_warning,
ask_before_restart=ask_before_restart)
# Change stderr_dir if requested
if self.test_dir is not None:
client.stderr_dir = self.test_dir
self.add_tab(client, name=client.get_name(), filename=filename)
if cf is None:
error_msg = self.permission_error_msg.format(jupyter_runtime_dir())
client.show_kernel_error(error_msg)
return
# Check if ipykernel is present in the external interpreter.
# Else we won't be able to create a client
if not CONF.get('main_interpreter', 'default'):
pyexec = CONF.get('main_interpreter', 'executable')
has_spyder_kernels = programs.is_module_installed(
'spyder_kernels',
interpreter=pyexec,
version='<1.0.0')
if not has_spyder_kernels:
client.show_kernel_error(
_("Your Python environment or installation doesn't "
"have the spyder-kernels module or the "
"right version of it installed. "
"Without this module is not possible for "
"Spyder to create a console for you.
"
"You can install it by running in a system terminal:"
"
"
"conda install spyder-kernels=0.*"
"
or
"
"pip install spyder-kernels==0.*"))
return
self.connect_client_to_kernel(client, is_cython=is_cython)
if client.shellwidget.kernel_manager is None:
return
self.register_client(client)
@Slot()
def create_client_for_kernel(self):
"""Create a client connected to an existing kernel"""
connect_output = KernelConnectionDialog.get_connection_parameters(self)
(connection_file, hostname, sshkey, password, ok) = connect_output
if not ok:
return
else:
self._create_client_for_kernel(connection_file, hostname, sshkey,
password)
def connect_client_to_kernel(self, client, is_cython=False):
"""Connect a client to its kernel"""
connection_file = client.connection_file
stderr_handle = None if self.test_no_stderr else client.stderr_handle
km, kc = self.create_kernel_manager_and_kernel_client(
connection_file,
stderr_handle,
is_cython=is_cython)
# An error occurred if this is True
if is_string(km) and kc is None:
client.shellwidget.kernel_manager = None
client.show_kernel_error(km)
return
# This avoids a recurrent, spurious NameError when running our
# tests in our CIs
if not self.testing:
kc.started_channels.connect(
lambda c=client: self.process_started(c))
kc.stopped_channels.connect(
lambda c=client: self.process_finished(c))
kc.start_channels(shell=True, iopub=True)
shellwidget = client.shellwidget
shellwidget.kernel_manager = km
shellwidget.kernel_client = kc
@Slot(object, object)
def edit_file(self, filename, line):
"""Handle %edit magic petitions."""
if encoding.is_text_file(filename):
# The default line number sent by ipykernel is always the last
# one, but we prefer to use the first.
self.edit_goto.emit(filename, 1, '')
def config_options(self):
"""
Generate a Trailets Config instance for shell widgets using our
config system
This lets us create each widget with its own config
"""
# ---- Jupyter config ----
try:
full_cfg = load_pyconfig_files(['jupyter_qtconsole_config.py'],
jupyter_config_dir())
# From the full config we only select the JupyterWidget section
# because the others have no effect here.
cfg = Config({'JupyterWidget': full_cfg.JupyterWidget})
except:
cfg = Config()
# ---- Spyder config ----
spy_cfg = Config()
# Make the pager widget a rich one (i.e a QTextEdit)
spy_cfg.JupyterWidget.kind = 'rich'
# Gui completion widget
completion_type_o = self.get_option('completion_type')
completions = {0: "droplist", 1: "ncurses", 2: "plain"}
spy_cfg.JupyterWidget.gui_completion = completions[completion_type_o]
# Pager
pager_o = self.get_option('use_pager')
if pager_o:
spy_cfg.JupyterWidget.paging = 'inside'
else:
spy_cfg.JupyterWidget.paging = 'none'
# Calltips
calltips_o = self.get_option('show_calltips')
spy_cfg.JupyterWidget.enable_calltips = calltips_o
# Buffer size
buffer_size_o = self.get_option('buffer_size')
spy_cfg.JupyterWidget.buffer_size = buffer_size_o
# Prompts
in_prompt_o = self.get_option('in_prompt')
out_prompt_o = self.get_option('out_prompt')
if in_prompt_o:
spy_cfg.JupyterWidget.in_prompt = in_prompt_o
if out_prompt_o:
spy_cfg.JupyterWidget.out_prompt = out_prompt_o
# Style
color_scheme = CONF.get('color_schemes', 'selected')
style_sheet = create_qss_style(color_scheme)[0]
spy_cfg.JupyterWidget.style_sheet = style_sheet
spy_cfg.JupyterWidget.syntax_style = color_scheme
# Merge QtConsole and Spyder configs. Spyder prefs will have
# prevalence over QtConsole ones
cfg._merge(spy_cfg)
return cfg
def interpreter_versions(self):
"""Python and IPython versions used by clients"""
if CONF.get('main_interpreter', 'default'):
from IPython.core import release
versions = dict(
python_version = sys.version.split("\n")[0].strip(),
ipython_version = release.version
)
else:
import subprocess
versions = {}
pyexec = CONF.get('main_interpreter', 'executable')
py_cmd = "%s -c 'import sys; print(sys.version.split(\"\\n\")[0])'" % \
pyexec
ipy_cmd = "%s -c 'import IPython.core.release as r; print(r.version)'" \
% pyexec
for cmd in [py_cmd, ipy_cmd]:
try:
proc = programs.run_shell_command(cmd)
output, _err = proc.communicate()
except subprocess.CalledProcessError:
output = ''
output = output.decode().split('\n')[0].strip()
if 'IPython' in cmd:
versions['ipython_version'] = output
else:
versions['python_version'] = output
return versions
def additional_options(self):
"""
Additional options for shell widgets that are not defined
in JupyterWidget config options
"""
options = dict(
pylab=self.get_option('pylab'),
autoload_pylab=self.get_option('pylab/autoload'),
sympy=self.get_option('symbolic_math'),
show_banner=self.get_option('show_banner')
)
return options
def register_client(self, client, give_focus=True):
"""Register new client"""
client.configure_shellwidget(give_focus=give_focus)
# Local vars
shellwidget = client.shellwidget
control = shellwidget._control
page_control = shellwidget._page_control
# Create new clients with Ctrl+T shortcut
shellwidget.new_client.connect(self.create_new_client)
# For tracebacks
control.go_to_error.connect(self.go_to_error)
shellwidget.sig_pdb_step.connect(
lambda fname, lineno, shellwidget=shellwidget:
self.pdb_has_stopped(fname, lineno, shellwidget))
# To handle %edit magic petitions
shellwidget.custom_edit_requested.connect(self.edit_file)
# Set shell cwd according to preferences
cwd_path = ''
if CONF.get('workingdir', 'console/use_project_or_home_directory'):
cwd_path = get_home_dir()
if (self.main.projects is not None and
self.main.projects.get_active_project() is not None):
cwd_path = self.main.projects.get_active_project_path()
elif CONF.get('workingdir', 'console/use_fixed_directory'):
cwd_path = CONF.get('workingdir', 'console/fixed_directory')
if osp.isdir(cwd_path) and self.main is not None:
shellwidget.set_cwd(cwd_path)
if give_focus:
# Syncronice cwd with explorer and cwd widget
shellwidget.get_cwd()
# Connect text widget to Help
if self.main.help is not None:
control.set_help(self.main.help)
control.set_help_enabled(CONF.get('help', 'connect/ipython_console'))
# Connect client to our history log
if self.main.historylog is not None:
self.main.historylog.add_history(client.history_filename)
client.append_to_history.connect(
self.main.historylog.append_to_history)
# Set font for client
client.set_font( self.get_plugin_font() )
# Connect focus signal to client's control widget
control.focus_changed.connect(lambda: self.focus_changed.emit())
shellwidget.sig_change_cwd.connect(self.set_working_directory)
# Update the find widget if focus changes between control and
# page_control
self.find_widget.set_editor(control)
if page_control:
page_control.focus_changed.connect(lambda: self.focus_changed.emit())
control.visibility_changed.connect(self.refresh_plugin)
page_control.visibility_changed.connect(self.refresh_plugin)
page_control.show_find_widget.connect(self.find_widget.show)
def close_client(self, index=None, client=None, force=False):
"""Close client tab from index or widget (or close current tab)"""
if not self.tabwidget.count():
return
if client is not None:
index = self.tabwidget.indexOf(client)
# if index is not found in tabwidget it's because this client was
# already closed and the call was performed by the exit callback
if index == -1:
return
if index is None and client is None:
index = self.tabwidget.currentIndex()
if index is not None:
client = self.tabwidget.widget(index)
# Needed to handle a RuntimeError. See issue 5568.
try:
# Close client
client.stop_button_click_handler()
except RuntimeError:
pass
# Disconnect timer needed to update elapsed time
try:
client.timer.timeout.disconnect(client.show_time)
except (RuntimeError, TypeError):
pass
# Check if related clients or kernels are opened
# and eventually ask before closing them
if not self.mainwindow_close and not force:
close_all = True
if self.get_option('ask_before_closing'):
close = QMessageBox.question(self, self.get_plugin_title(),
_("Do you want to close this console?"),
QMessageBox.Yes | QMessageBox.No)
if close == QMessageBox.No:
return
if len(self.get_related_clients(client)) > 0:
close_all = QMessageBox.question(self, self.get_plugin_title(),
_("Do you want to close all other consoles connected "
"to the same kernel as this one?"),
QMessageBox.Yes | QMessageBox.No)
client.shutdown()
if close_all == QMessageBox.Yes:
self.close_related_clients(client)
# if there aren't related clients we can remove stderr_file
related_clients = self.get_related_clients(client)
if len(related_clients) == 0 and osp.exists(client.stderr_file):
client.remove_stderr_file()
client.close()
# Note: client index may have changed after closing related widgets
self.tabwidget.removeTab(self.tabwidget.indexOf(client))
self.clients.remove(client)
# This is needed to prevent that hanged consoles make reference
# to an index that doesn't exist. See issue 4881
try:
self.filenames.pop(index)
except IndexError:
pass
self.update_tabs_text()
# Create a new client if the console is about to become empty
if not self.tabwidget.count() and self.create_new_client_if_empty:
self.create_new_client()
self.update_plugin_title.emit()
def get_client_index_from_id(self, client_id):
"""Return client index from id"""
for index, client in enumerate(self.clients):
if id(client) == client_id:
return index
def get_related_clients(self, client):
"""
Get all other clients that are connected to the same kernel as `client`
"""
related_clients = []
for cl in self.get_clients():
if cl.connection_file == client.connection_file and \
cl is not client:
related_clients.append(cl)
return related_clients
def close_related_clients(self, client):
"""Close all clients related to *client*, except itself"""
related_clients = self.get_related_clients(client)
for cl in related_clients:
self.close_client(client=cl, force=True)
def restart(self):
"""
Restart the console
This is needed when we switch projects to update PYTHONPATH
and the selected interpreter
"""
self.master_clients = 0
self.create_new_client_if_empty = False
for i in range(len(self.clients)):
client = self.clients[-1]
try:
client.shutdown()
except Exception as e:
QMessageBox.warning(self, _('Warning'),
_("It was not possible to restart the IPython console "
"when switching to this project. The error was
"
"{0}").format(e), QMessageBox.Ok)
self.close_client(client=client, force=True)
self.create_new_client(give_focus=False)
self.create_new_client_if_empty = True
def pdb_has_stopped(self, fname, lineno, shellwidget):
"""Python debugger has just stopped at frame (fname, lineno)"""
# This is a unique form of the edit_goto signal that is intended to
# prevent keyboard input from accidentally entering the editor
# during repeated, rapid entry of debugging commands.
self.edit_goto[str, int, str, bool].emit(fname, lineno, '', False)
self.activateWindow()
shellwidget._control.setFocus()
def set_spyder_breakpoints(self):
"""Set Spyder breakpoints into all clients"""
for cl in self.clients:
cl.shellwidget.set_spyder_breakpoints()
@Slot(str)
def create_client_from_path(self, path):
"""Create a client with its cwd pointing to path."""
self.create_new_client()
sw = self.get_current_shellwidget()
sw.set_cwd(path)
def create_client_for_file(self, filename, is_cython=False):
"""Create a client to execute code related to a file."""
# Create client
self.create_new_client(filename=filename, is_cython=is_cython)
# Don't increase the count of master clients
self.master_clients -= 1
# Rename client tab with filename
client = self.get_current_client()
client.allow_rename = False
tab_text = self.disambiguate_fname(filename)
self.rename_client_tab(client, tab_text)
def get_client_for_file(self, filename):
"""Get client associated with a given file."""
client = None
for idx, cl in enumerate(self.get_clients()):
if self.filenames[idx] == filename:
self.tabwidget.setCurrentIndex(idx)
client = cl
break
return client
def set_elapsed_time(self, client):
"""Set elapsed time for slave clients."""
related_clients = self.get_related_clients(client)
for cl in related_clients:
if cl.timer is not None:
client.create_time_label()
client.t0 = cl.t0
client.timer.timeout.connect(client.show_time)
client.timer.start(1000)
break
#------ Public API (for kernels) ------------------------------------------
def ssh_tunnel(self, *args, **kwargs):
if os.name == 'nt':
return zmqtunnel.paramiko_tunnel(*args, **kwargs)
else:
return openssh_tunnel(self, *args, **kwargs)
def tunnel_to_kernel(self, connection_info, hostname, sshkey=None,
password=None, timeout=10):
"""
Tunnel connections to a kernel via ssh.
Remote ports are specified in the connection info ci.
"""
lports = zmqtunnel.select_random_ports(4)
rports = (connection_info['shell_port'], connection_info['iopub_port'],
connection_info['stdin_port'], connection_info['hb_port'])
remote_ip = connection_info['ip']
for lp, rp in zip(lports, rports):
self.ssh_tunnel(lp, rp, hostname, remote_ip, sshkey, password,
timeout)
return tuple(lports)
def create_kernel_spec(self, is_cython=False):
"""Create a kernel spec for our own kernels"""
# Before creating our kernel spec, we always need to
# set this value in spyder.ini
CONF.set('main', 'spyder_pythonpath',
self.main.get_spyder_pythonpath())
return SpyderKernelSpec(is_cython=is_cython)
def create_kernel_manager_and_kernel_client(self, connection_file,
stderr_handle,
is_cython=False):
"""Create kernel manager and client."""
# Kernel spec
kernel_spec = self.create_kernel_spec(is_cython=is_cython)
# Kernel manager
try:
kernel_manager = QtKernelManager(connection_file=connection_file,
config=None, autorestart=True)
except Exception:
error_msg = _("The error is:
"
"{}").format(traceback.format_exc())
return (error_msg, None)
kernel_manager._kernel_spec = kernel_spec
# Catch any error generated when trying to start the kernel
# See issue 7302
try:
kernel_manager.start_kernel(stderr=stderr_handle)
except Exception:
error_msg = _("The error is:
"
"{}").format(traceback.format_exc())
return (error_msg, None)
# Kernel client
kernel_client = kernel_manager.client()
# Increase time to detect if a kernel is alive
# See Issue 3444
kernel_client.hb_channel.time_to_dead = 18.0
return kernel_manager, kernel_client
def restart_kernel(self):
"""Restart kernel of current client."""
client = self.get_current_client()
if client is not None:
self.switch_to_plugin()
client.restart_kernel()
def connect_external_kernel(self, shellwidget):
"""
Connect an external kernel to the Variable Explorer and Help, if
it is a Spyder kernel.
"""
sw = shellwidget
kc = shellwidget.kernel_client
if self.main.help is not None:
self.main.help.set_shell(sw)
if self.main.variableexplorer is not None:
self.main.variableexplorer.add_shellwidget(sw)
sw.set_namespace_view_settings()
sw.refresh_namespacebrowser()
kc.stopped_channels.connect(lambda :
self.main.variableexplorer.remove_shellwidget(id(sw)))
#------ Public API (for tabs) ---------------------------------------------
def add_tab(self, widget, name, filename=''):
"""Add tab"""
self.clients.append(widget)
index = self.tabwidget.addTab(widget, name)
self.filenames.insert(index, filename)
self.tabwidget.setCurrentIndex(index)
if self.dockwidget and not self.main.is_setting_up:
self.switch_to_plugin()
self.activateWindow()
widget.get_control().setFocus()
self.update_tabs_text()
def move_tab(self, index_from, index_to):
"""
Move tab (tabs themselves have already been moved by the tabwidget)
"""
filename = self.filenames.pop(index_from)
client = self.clients.pop(index_from)
self.filenames.insert(index_to, filename)
self.clients.insert(index_to, client)
self.update_tabs_text()
self.update_plugin_title.emit()
def disambiguate_fname(self, fname):
"""Generate a file name without ambiguation."""
files_path_list = [filename for filename in self.filenames
if filename]
return sourcecode.disambiguate_fname(files_path_list, fname)
def update_tabs_text(self):
"""Update the text from the tabs."""
# This is needed to prevent that hanged consoles make reference
# to an index that doesn't exist. See issue 4881
try:
for index, fname in enumerate(self.filenames):
client = self.clients[index]
if fname:
self.rename_client_tab(client,
self.disambiguate_fname(fname))
else:
self.rename_client_tab(client, None)
except IndexError:
pass
def rename_client_tab(self, client, given_name):
"""Rename client's tab"""
index = self.get_client_index_from_id(id(client))
if given_name is not None:
client.given_name = given_name
self.tabwidget.setTabText(index, client.get_name())
def rename_tabs_after_change(self, given_name):
"""Rename tabs after a change in name."""
client = self.get_current_client()
# Prevent renames that want to assign the same name of
# a previous tab
repeated = False
for cl in self.get_clients():
if id(client) != id(cl) and given_name == cl.given_name:
repeated = True
break
# Rename current client tab to add str_id
if client.allow_rename and not u'/' in given_name and not repeated:
self.rename_client_tab(client, given_name)
else:
self.rename_client_tab(client, None)
# Rename related clients
if client.allow_rename and not u'/' in given_name and not repeated:
for cl in self.get_related_clients(client):
self.rename_client_tab(cl, given_name)
def tab_name_editor(self):
"""Trigger the tab name editor."""
index = self.tabwidget.currentIndex()
self.tabwidget.tabBar().tab_name_editor.edit_tab(index)
#------ Public API (for help) ---------------------------------------------
def go_to_error(self, text):
"""Go to error if relevant"""
match = get_error_match(to_text_string(text))
if match:
fname, lnb = match.groups()
# This is needed to fix issue spyder-ide/spyder#9217
try:
self.edit_goto.emit(osp.abspath(fname), int(lnb), '')
except ValueError:
pass
@Slot()
def show_intro(self):
"""Show intro to IPython help"""
from IPython.core.usage import interactive_usage
self.main.help.show_rich_text(interactive_usage)
@Slot()
def show_guiref(self):
"""Show qtconsole help"""
from qtconsole.usage import gui_reference
self.main.help.show_rich_text(gui_reference, collapse=True)
@Slot()
def show_quickref(self):
"""Show IPython Cheat Sheet"""
from IPython.core.usage import quick_reference
self.main.help.show_plain_text(quick_reference)
#------ Private API -------------------------------------------------------
def _new_connection_file(self):
"""
Generate a new connection file
Taken from jupyter_client/console_app.py
Licensed under the BSD license
"""
# Check if jupyter_runtime_dir exists (Spyder addition)
if not osp.isdir(jupyter_runtime_dir()):
try:
os.makedirs(jupyter_runtime_dir())
except (IOError, OSError):
return None
cf = ''
while not cf:
ident = str(uuid.uuid4()).split('-')[-1]
cf = os.path.join(jupyter_runtime_dir(), 'kernel-%s.json' % ident)
cf = cf if not os.path.exists(cf) else ''
return cf
def process_started(self, client):
if self.main.help is not None:
self.main.help.set_shell(client.shellwidget)
if self.main.variableexplorer is not None:
self.main.variableexplorer.add_shellwidget(client.shellwidget)
def process_finished(self, client):
if self.main.variableexplorer is not None:
self.main.variableexplorer.remove_shellwidget(
id(client.shellwidget))
def _create_client_for_kernel(self, connection_file, hostname, sshkey,
password):
# Verifying if the connection file exists
try:
cf_path = osp.dirname(connection_file)
cf_filename = osp.basename(connection_file)
# To change a possible empty string to None
cf_path = cf_path if cf_path else None
connection_file = find_connection_file(filename=cf_filename,
path=cf_path)
except (IOError, UnboundLocalError):
QMessageBox.critical(self, _('IPython'),
_("Unable to connect to "
"%s") % connection_file)
return
# Getting the master id that corresponds to the client
# (i.e. the i in i/A)
master_id = None
given_name = None
external_kernel = False
slave_ord = ord('A') - 1
kernel_manager = None
for cl in self.get_clients():
if connection_file in cl.connection_file:
if cl.get_kernel() is not None:
kernel_manager = cl.get_kernel()
connection_file = cl.connection_file
if master_id is None:
master_id = cl.id_['int_id']
given_name = cl.given_name
new_slave_ord = ord(cl.id_['str_id'])
if new_slave_ord > slave_ord:
slave_ord = new_slave_ord
# If we couldn't find a client with the same connection file,
# it means this is a new master client
if master_id is None:
self.master_clients += 1
master_id = to_text_string(self.master_clients)
external_kernel = True
# Set full client name
client_id = dict(int_id=master_id,
str_id=chr(slave_ord + 1))
# Creating the client
show_elapsed_time = self.get_option('show_elapsed_time')
reset_warning = self.get_option('show_reset_namespace_warning')
ask_before_restart = self.get_option('ask_before_restart')
client = ClientWidget(self,
id_=client_id,
given_name=given_name,
history_filename=get_conf_path('history.py'),
config_options=self.config_options(),
additional_options=self.additional_options(),
interpreter_versions=self.interpreter_versions(),
connection_file=connection_file,
menu_actions=self.menu_actions,
hostname=hostname,
external_kernel=external_kernel,
slave=True,
show_elapsed_time=show_elapsed_time,
reset_warning=reset_warning,
ask_before_restart=ask_before_restart)
# Change stderr_dir if requested
if self.test_dir is not None:
client.stderr_dir = self.test_dir
# Create kernel client
kernel_client = QtKernelClient(connection_file=connection_file)
# This is needed for issue spyder-ide/spyder#9304
try:
kernel_client.load_connection_file()
except Exception as e:
QMessageBox.critical(self, _('Connection error'),
_("An error occurred while trying to load "
"the kernel connection file. The error "
"was:\n\n") + to_text_string(e))
return
if hostname is not None:
try:
connection_info = dict(ip = kernel_client.ip,
shell_port = kernel_client.shell_port,
iopub_port = kernel_client.iopub_port,
stdin_port = kernel_client.stdin_port,
hb_port = kernel_client.hb_port)
newports = self.tunnel_to_kernel(connection_info, hostname,
sshkey, password)
(kernel_client.shell_port,
kernel_client.iopub_port,
kernel_client.stdin_port,
kernel_client.hb_port) = newports
except Exception as e:
QMessageBox.critical(self, _('Connection error'),
_("Could not open ssh tunnel. The "
"error was:\n\n") + to_text_string(e))
return
# Assign kernel manager and client to shellwidget
kernel_client.start_channels()
client.shellwidget.kernel_client = kernel_client
client.shellwidget.kernel_manager = kernel_manager
if external_kernel:
client.shellwidget.sig_is_spykernel.connect(
self.connect_external_kernel)
client.shellwidget.is_spyder_kernel()
# Set elapsed time, if possible
if not external_kernel:
self.set_elapsed_time(client)
# Adding a new tab for the client
self.add_tab(client, name=client.get_name())
# Register client
self.register_client(client)
def _remove_old_stderr_files(self):
"""
Remove stderr files left by previous Spyder instances.
This is only required on Windows because we can't
clean up stderr files while Spyder is running on it.
"""
if os.name == 'nt':
tmpdir = get_temp_dir()
for fname in os.listdir(tmpdir):
if osp.splitext(fname)[1] == '.stderr':
try:
os.remove(osp.join(tmpdir, fname))
except Exception:
pass