from collections import defaultdict from datetime import UTC, datetime from uuid import uuid4 from IPython.core.interactiveshell import InteractiveShell from IPython.core.pylabtools import ( activate_matplotlib, find_gui_and_backend, import_pylab, mpl_runner, select_figure_formats, ) from matplotlib_inline.backend_inline import configure_inline_support, flush_figures from .events import ( error_from_execution_result, event_from_execution_result, event_from_history_output, ) from .models import CallEvent, CallResult, ShellInfo, ShellStatus class Shell: """Own one configured, in-process IPython shell.""" def __init__(self, name: str): self.name = name self.shell_id = uuid4().hex self.created_at = datetime.now(UTC).isoformat() self.last_used_at: str | None = None self.status: ShellStatus = "ready" self._last_execution_count = 0 self._shell: InteractiveShell | None = None @property def shell(self) -> InteractiveShell: if self._shell is not None: return self._shell return self.init_shell() @classmethod def using_shell(cls, shell: InteractiveShell, name: str) -> Shell: """Create a Shell object that wraps an existing InteractiveShell. The caller owns setup for this low-level adapter. """ obj = cls(name) obj._shell = shell return obj @classmethod def init_instance(cls, name: str = "default") -> Shell: """Create a Shell around IPython's process-wide instance. ``matplotlib-inline`` reaches the shell through IPython's global ``get_ipython()`` lookup. Wrapping ``InteractiveShell.instance()`` keeps those two identities aligned without changing the singleton's existing configuration. """ return cls.using_shell(InteractiveShell.instance(), name) def init_shell(self) -> InteractiveShell: if self._shell is not None: return self._shell self._shell = InteractiveShell() setup_shell(self._shell, plt=True, pylab=True) return self._shell def run_cell( self, code: str, *, call_id: str, ) -> tuple[CallResult, list[CallEvent]]: """Execute one cell and return its result plus ordered output events.""" self.status = "running" try: call_result, events = run_cell_and_collect( self.shell, code, call_id=call_id, ) except Exception: self.status = "error" raise else: self.status = "ready" self.last_used_at = datetime.now(UTC).isoformat() if call_result.execution_count is not None: self._last_execution_count = call_result.execution_count return call_result, events def describe(self) -> ShellInfo: """Return metadata without initializing an unused shell.""" return ShellInfo( shell_id=self.shell_id, name=self.name, status=self.status, execution_count=self._last_execution_count, created_at=self.created_at, last_used_at=self.last_used_at, capabilities=( "stdout", "stderr", "display_data", "execute_result", "error", "matplotlib-inline", ), ) def isolate_output_history(shell: InteractiveShell) -> None: """Give this shell an instance-owned output ledger. IPython declares ``HistoryManager.outputs`` on the class. Shadowing it during shell setup prevents output records from leaking across shells. """ history_manager = shell.history_manager if history_manager is None: raise RuntimeError("InteractiveShell has no history manager") history_manager.outputs = defaultdict(list) def run_cell_and_collect( shell: InteractiveShell, code: str, *, call_id: str, ) -> tuple[CallResult, list[CallEvent]]: """Run one cell and snapshot its output records after execution.""" history_manager = shell.history_manager if history_manager is None: raise RuntimeError("InteractiveShell has no history manager") if "outputs" not in history_manager.__dict__: isolate_output_history(shell) execution = shell.run_cell(code, store_history=True) records = ( tuple(history_manager.outputs.get(execution.execution_count, ())) if execution.execution_count is not None else () ) events = [ event_from_history_output( record, call_id=call_id, sequence=sequence, ) for sequence, record in enumerate(records) ] events = [event for event in events if event is not None] error = error_from_execution_result(execution) if error is not None: error_event = event_from_execution_result( execution, call_id=call_id, sequence=len(events), ) if error_event is not None: events.append(error_event) return ( CallResult( call_id=call_id, execution_count=execution.execution_count, result=execution.result, error=error, ), events, ) def setup_shell( shell: InteractiveShell, *, isolate_output_hist: bool = True, plt: bool = False, pylab: bool = True, ): """Configure an ``InteractiveShell`` without requiring GUI support. ``InteractiveShell.enable_matplotlib`` and ``enable_pylab`` eventually call ``enable_gui``. The base shell intentionally does not implement that method, so the lower-level matplotlib-inline setup is used here instead. """ setup_shell_history_disabled(shell) # ``HistoryManager.outputs`` is class-level in the IPython version used # here. Give every managed shell an instance-local mapping so execution # records cannot leak between Shell objects. if ( isolate_output_hist and shell.history_manager is not None and "outputs" not in shell.history_manager.__dict__ ): isolate_output_history(shell) carefully_setup_shell( shell, activate=plt, pylab=pylab, install_runner=plt, ) def setup_shell_history_disabled(shell: InteractiveShell): """Disable the shell's history manager to mostly disable writing to disk.""" if shell.history_manager is not None: shell.history_manager.enabled = False def carefully_setup_shell( shell: InteractiveShell, *, gui: str = "inline", activate: bool = True, formats: set[str] | None = None, pylab: bool = False, import_all: bool = False, install_runner: bool = False, ) -> tuple[str | None, str, list[str]]: """Add only the requested Matplotlib and pylab support to one shell. This deliberately does not call ``enable_gui``. It performs the useful parts of ``InteractiveShell.enable_matplotlib`` and ``enable_pylab`` for shells whose base class has no GUI implementation. """ import matplotlib selected_gui, backend = find_gui_and_backend( gui, shell.pylab_gui_select, ) wanted_formats = formats or {"svg", "png", "pdf", "jpg"} # select_figure_formats() clears Figure printers, so capture them before # configuring inline support and restore them with our requested formats. existing_formats = figure_formats_for(shell) if activate and matplotlib.get_backend().lower() != backend.lower(): activate_matplotlib(backend) if not inline_support_present(shell, backend): configure_inline_support(shell, backend) select_figure_formats(shell, existing_formats | wanted_formats) if install_runner: execution_magics = shell.magics_manager.registry["ExecutionMagics"] if execution_magics.default_runner is None: execution_magics.default_runner = mpl_runner(shell.safe_execfile) clobbered = enable_pylab_lite(shell, import_all=import_all) if pylab else [] return selected_gui, backend, clobbered def inline_support_present(shell: InteractiveShell, backend: str) -> bool: """Return whether this shell has the requested inline flush support.""" callbacks = shell.events.callbacks.get("post_execute", []) is_inline = backend.lower() in { "inline", "module://matplotlib_inline.backend_inline", } return ( (flush_figures in callbacks) if is_inline else (flush_figures not in callbacks) ) def figure_formats_for(shell: InteractiveShell) -> set[str]: """Return MIME figure formats currently registered on one shell.""" from matplotlib.figure import Figure mime_to_format = { "image/png": "png", "image/jpeg": "jpg", "image/svg+xml": "svg", "application/pdf": "pdf", } return { format_name for mime, format_name in mime_to_format.items() if Figure in shell.display_formatter.formatters[mime].type_printers } def enable_pylab_lite( shell: InteractiveShell, *, import_all: bool = True, ) -> list[str]: """Load pylab names without GUI activation and report overwritten names.""" pylab_ns: dict[str, object] = {} import_pylab(pylab_ns, import_all) ignored = {"__builtins__"} candidates = set(pylab_ns).intersection(shell.user_ns).difference(ignored) clobbered = [ name for name in candidates if shell.user_ns[name] is not pylab_ns[name] ] shell.user_ns.update(pylab_ns) shell.user_ns_hidden.update(pylab_ns) return clobbered _default: Shell | None = None def get_default_shell() -> Shell: """Return the lazily-created, configured process-wide shell.""" global _default if _default is None: _default = Shell.init_instance("default") return _default