diff --git a/src/ipython_demo/shell.py b/src/ipython_demo/shell.py deleted file mode 100644 index 5582306..0000000 --- a/src/ipython_demo/shell.py +++ /dev/null @@ -1,194 +0,0 @@ -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, - import_pylab, - mpl_runner, - select_figure_formats, -) -from matplotlib_inline.backend_inline import configure_inline_support - -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() - - 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() - self._last_execution_count = call_result.execution_count or 0 - 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, 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. - """ - history_manager = shell.history_manager - if history_manager is not None: - history_manager.enabled = False - - # ``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 "outputs" not in history_manager.__dict__: - isolate_output_history(shell) - - backend = "module://matplotlib_inline.backend_inline" - - # This is the non-GUI part of enable_matplotlib(). - if plt: - activate_matplotlib(backend) - shell.magics_manager.registry["ExecutionMagics"].default_runner = mpl_runner( - shell.safe_execfile - ) - - # This connects matplotlib-inline to THIS shell. - configure_inline_support(shell, backend) - - # Optional explicit format selection; configure_inline_support() - # also sets up the configured default formats. - select_figure_formats( - shell, - {"svg", "pdf", "png", "jpg"}, - ) - if pylab: - # Match enable_pylab's namespace behavior without its GUI activation: - # hidden names are used by %who/%whos bookkeeping as well as by code. - pylab_ns: dict[str, object] = {} - import_pylab(pylab_ns, import_all=False) # no import * allowed - shell.user_ns.update(pylab_ns) - shell.user_ns_hidden.update(pylab_ns) diff --git a/src/ipython_mcp/app.py b/src/ipython_mcp/app.py index b6a2e1e..bbb4686 100644 --- a/src/ipython_mcp/app.py +++ b/src/ipython_mcp/app.py @@ -5,11 +5,11 @@ from typing import Annotated, Literal from fastmcp import FastMCP from pydantic import Field -import ipython_demo -from ipython_demo.models import AppInfo, CallOptions, CallResponse, ShellInfo +import ipython_shell +from ipython_shell.models import AppInfo, CallOptions, CallResponse, ShellInfo app = FastMCP(name="IPython Demo") -shell_app = ipython_demo.App() +shell_app = ipython_shell.App() @app.tool(title="Get app info") diff --git a/src/ipython_demo/__init__.py b/src/ipython_shell/__init__.py similarity index 94% rename from src/ipython_demo/__init__.py rename to src/ipython_shell/__init__.py index 95cc2ad..4035a00 100644 --- a/src/ipython_demo/__init__.py +++ b/src/ipython_shell/__init__.py @@ -22,6 +22,7 @@ from .models import ( from .shell import ( Shell, isolate_output_history, + carefully_setup_shell, run_cell_and_collect, setup_shell, ) @@ -41,6 +42,7 @@ __all__ = [ "Shell", "ShellInfo", "ShellStatus", + "carefully_setup_shell", "error_from_execution_result", "event_from_execution_result", "event_from_history_output", diff --git a/src/ipython_demo/app.py b/src/ipython_shell/app.py similarity index 98% rename from src/ipython_demo/app.py rename to src/ipython_shell/app.py index d855543..26a383c 100644 --- a/src/ipython_demo/app.py +++ b/src/ipython_shell/app.py @@ -7,7 +7,7 @@ from time import monotonic from typing import Literal from uuid import uuid4 -from ipython_demo.utils import generate_good_names +from ipython_shell.utils import generate_good_names from .models import AppInfo, CallOptions, CallResponse, ShellInfo from .shell import Shell diff --git a/src/ipython_demo/events.py b/src/ipython_shell/events.py similarity index 100% rename from src/ipython_demo/events.py rename to src/ipython_shell/events.py diff --git a/src/ipython_demo/models.py b/src/ipython_shell/models.py similarity index 100% rename from src/ipython_demo/models.py rename to src/ipython_shell/models.py diff --git a/src/ipython_shell/shell.py b/src/ipython_shell/shell.py new file mode 100644 index 0000000..ee65180 --- /dev/null +++ b/src/ipython_shell/shell.py @@ -0,0 +1,318 @@ +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() + self._last_execution_count = call_result.execution_count or 0 + 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 diff --git a/src/ipython_demo/utils.py b/src/ipython_shell/utils.py similarity index 100% rename from src/ipython_demo/utils.py rename to src/ipython_shell/utils.py diff --git a/src/ipython_webapp/app.py b/src/ipython_webapp/app.py index deedd6e..dd43451 100644 --- a/src/ipython_webapp/app.py +++ b/src/ipython_webapp/app.py @@ -2,8 +2,8 @@ from typing import Annotated, Literal from fastapi import Body, Depends, FastAPI, HTTPException, Path, Query -from ipython_demo import App -from ipython_demo.models import CallOptions +from ipython_shell import App +from ipython_shell.models import CallOptions app = FastAPI(title="IPython Demo", version="0.1.0") diff --git a/src/langchain_demo/ipython_wrapper.py b/src/langchain_demo/ipython_wrapper.py index 43a38af..da7ae03 100644 --- a/src/langchain_demo/ipython_wrapper.py +++ b/src/langchain_demo/ipython_wrapper.py @@ -2,8 +2,8 @@ from typing import Literal from langchain.tools import tool -from ipython_demo import App -from ipython_demo.models import AppInfo, CallOptions, CallResponse, ShellInfo +from ipython_shell import App +from ipython_shell.models import AppInfo, CallOptions, CallResponse, ShellInfo app = App() # global, again diff --git a/tests/test_app.py b/tests/test_app.py index c022558..90379f2 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -4,9 +4,9 @@ import re import unittest from unittest.mock import Mock, patch -from ipython_demo import utils -from ipython_demo.app import App -from ipython_demo.models import CallOptions, CallResponse +from ipython_shell import utils +from ipython_shell.app import App +from ipython_shell.models import CallOptions, CallResponse class AppTests(unittest.TestCase): diff --git a/tests/test_info.py b/tests/test_info.py index 55d2b3f..9f883fb 100644 --- a/tests/test_info.py +++ b/tests/test_info.py @@ -2,7 +2,7 @@ import unittest from fastapi.testclient import TestClient -from ipython_demo import App, AppInfo +from ipython_shell import App, AppInfo from ipython_mcp.app import app as mcp_app from ipython_webapp.app import app as web_app from langchain_demo.ipython_wrapper import get_app_info diff --git a/tests/test_my_ideas.py b/tests/test_my_ideas.py index 02a33da..f38bd1c 100644 --- a/tests/test_my_ideas.py +++ b/tests/test_my_ideas.py @@ -4,7 +4,7 @@ from typing import Any from IPython.core.interactiveshell import InteractiveShell -from ipython_demo.shell import Shell +from ipython_shell.shell import Shell class TestMyIdeas(unittest.TestCase): diff --git a/tests/test_serialization.py b/tests/test_serialization.py index 4f16c22..131fe47 100644 --- a/tests/test_serialization.py +++ b/tests/test_serialization.py @@ -5,7 +5,7 @@ import unittest from fastapi.encoders import jsonable_encoder from fastapi.testclient import TestClient -from ipython_demo import App, Shell +from ipython_shell import App, Shell from ipython_webapp.app import app as web_app diff --git a/tests/test_shell.py b/tests/test_shell.py index d8f7578..2069a8a 100644 --- a/tests/test_shell.py +++ b/tests/test_shell.py @@ -8,9 +8,9 @@ from IPython.core.history import HistoryOutput from IPython.core.interactiveshell import InteractiveShell from IPython.utils import capture -import ipython_demo.events as events_module -import ipython_demo.models as models_module -import ipython_demo.shell as shell_module +import ipython_shell.events as events_module +import ipython_shell.models as models_module +import ipython_shell.shell as shell_module PACKAGE_ROOT = Path(__file__).parents[1] / "src" / "ipython_demo" @@ -123,6 +123,37 @@ class ShellTests(unittest.TestCase): shell.magics_manager.registry["ExecutionMagics"].default_runner ) + def test_carefully_setup_shell_preserves_formats_and_is_idempotent(self): + from matplotlib.figure import Figure + from matplotlib_inline.backend_inline import flush_figures + + shell = InteractiveShell() + + shell_module.carefully_setup_shell(shell, formats={"svg"}) + shell_module.carefully_setup_shell(shell, formats={"png"}) + + callbacks = shell.events.callbacks["post_execute"] + self.assertEqual(callbacks.count(flush_figures), 1) + self.assertIn( + Figure, + shell.display_formatter.formatters["image/svg+xml"].type_printers, + ) + self.assertIn( + Figure, + shell.display_formatter.formatters["image/png"].type_printers, + ) + + def test_enable_pylab_lite_reports_clobbered_names(self): + shell = InteractiveShell() + original_np = object() + shell.user_ns["np"] = original_np + + clobbered = shell_module.enable_pylab_lite(shell, import_all=False) + + self.assertIn("np", clobbered) + self.assertIsNot(shell.user_ns["np"], original_np) + self.assertIs(shell.user_ns["np"], shell.user_ns_hidden["np"]) + def test_run_cell_does_not_use_global_capture_output(self): wrapper = shell_module.Shell("test") wrapper.init_shell()