### IOPubThread Example Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Shows how to start and send messages using an IOPubThread. Remember to stop and join the thread when finished. ```python from ipymini_zmqthread import IOPubThread iopub = IOPubThread(ctx, "tcp://127.0.0.1:5556", session, qmax=10000, sndhwm=None) iopub.start() iopub.send("status", {"execution_state": "idle"}, parent=None) ... iopub.stop() iopub.join(timeout=1) ``` -------------------------------- ### Install ipymini-zmqthread Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Install the library using pip. ```bash pip install ipymini-zmqthread ``` -------------------------------- ### Install ipymini in editable mode Source: https://github.com/answerdotai/ipymini/blob/main/README.md Install ipymini from the repository root for development purposes. ```bash pip install -e . ``` -------------------------------- ### AsyncRouterThread Example Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Demonstrates starting and stopping an AsyncRouterThread for handling shell messages. Ensure the router is ready before proceeding and handle potential timeouts on join. ```python import zmq from jupyter_client.session import Session from ipymini_zmqthread import AsyncRouterThread ctx = zmq.Context.instance() session = Session(key=b"") router = None def on_msg(msg, idents): router.enqueue(("kernel_info_reply", {"status": "ok"}, msg, idents)) router = AsyncRouterThread( context=ctx, session=session, bind_addr="tcp://127.0.0.1:5555", handler=on_msg, log_label="shell") router.start() router.ready.wait() ... router.stop() router.join(timeout=1) ``` -------------------------------- ### Install ipymini kernel spec (system prefix) Source: https://github.com/answerdotai/ipymini/blob/main/README.md Install the ipymini kernel specification into the current Python environment. ```bash python -m ipymini install --sys-prefix ``` -------------------------------- ### Install ipymini-debug Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/debug/README.md Install the ipymini-debug package using pip. ```bash pip install ipymini-debug ``` -------------------------------- ### Install optional test dependencies Source: https://github.com/answerdotai/ipymini/blob/main/README.md Install optional dependencies for testing ipymini. ```bash pip install -e ".[test]" ``` -------------------------------- ### Install ipymini-term Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/term/README.md Install the ipymini-term package using pip. ```bash pip install ipymini-term ``` -------------------------------- ### Install ipymini from PyPI Source: https://github.com/answerdotai/ipymini/blob/main/README.md Install the ipymini package from the Python Package Index using pip. ```bash pip install ipymini ``` -------------------------------- ### Install ipymini kernel spec (user) Source: https://github.com/answerdotai/ipymini/blob/main/README.md Install the ipymini kernel specification for the current user. ```bash python -m ipymini install --user ``` -------------------------------- ### HeartbeatThread Example Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Illustrates starting and stopping a HeartbeatThread for echo functionality. The thread is designed to receive bytes and send them back. ```python from ipymini_zmqthread import HeartbeatThread hb = HeartbeatThread(ctx, "tcp://127.0.0.1:5557") hb.start() ... hb.stop() hb.join(timeout=1) ``` -------------------------------- ### Quick Start: Execute Code with MiniShell Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Demonstrates basic usage of MiniShell to execute Python code, capture output, and access results. Requires an async context for execution. ```python from ipymini_shell import MiniShell, comm_context def request_input(prompt: str, password: bool) -> str: return "Ada" if not password else "secret" shell = MiniShell(request_input=request_input) def iopub_send(msg_type, parent, content=None, **kwargs): print(msg_type, content) parent = {"header": {"msg_id": "demo"}} with shell.execution_context(allow_stdin=True, silent=False, comm_sender=iopub_send, parent=parent): result = await shell.execute("print('hello')\n1+1", silent=False, store_history=True) print(result["result"]) # last displayhook result bundle print(result["streams"]) # captured stdout/stderr events ``` -------------------------------- ### Start ipymini kernel with custom environment and working directory Source: https://github.com/answerdotai/ipymini/blob/main/README.md Use KernelManager to start the ipymini kernel with specific environment variables and a working directory. ```python from jupyter_client import KernelManager km = KernelManager(kernel_name="ipymini") km.start_kernel(env={"MY_FLAG": "1"}, cwd="/path/to/workdir") ``` -------------------------------- ### Install libzmq on macOS Source: https://github.com/answerdotai/ipymini/blob/main/README.md If you need system ZMQ libraries on macOS, use Homebrew to install libzmq. ```bash brew install libzmq ``` -------------------------------- ### Python Debugging Setup and Message Tracing Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/debug/README.md Configure debug flags from environment variables and set up debug logging. This snippet demonstrates how to trace structured messages. ```python import logging from ipymini_debug import DebugFlags, setup_debug, trace_msg log = logging.getLogger("demo") flags = DebugFlags.from_env("IPYMINI") setup_debug(flags) msg = {"header": {"msg_type": "execute_request", "msg_id": "abc", "subshell_id": "worker-1"}} trace_msg(log, "shell recv", msg, enabled=flags.trace_msgs) ``` -------------------------------- ### StdinRouterThread Example Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Demonstrates the usage of StdinRouterThread for handling input requests and replies. Includes methods for requesting input, interrupting pending requests, stopping, and joining the thread. ```python from ipymini_zmqthread import StdinRouterThread stdin = StdinRouterThread(ctx, "tcp://127.0.0.1:5558", session) stdin.start() value = stdin.request_input("Name: ", password=False, parent=parent_msg, ident=client_idents, timeout=10.0) ... stdin.interrupt_pending() # cancels in-flight waits stdin.stop() stdin.join(timeout=1) ``` -------------------------------- ### List available Jupyter kernels Source: https://github.com/answerdotai/ipymini/blob/main/README.md Check if the ipymini kernel has been successfully installed and is discoverable by Jupyter. ```bash jupyter kernelspec list ``` -------------------------------- ### Initialize and Process DAP Request with Debugger Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/debug/README.md Instantiate the Debugger and process a Debug Adapter Protocol (DAP) 'initialize' request. This shows the expected request shape and how to handle the response and events. ```python from ipymini_debug import Debugger dbg = Debugger(event_callback=None, zmq_context=None, kernel_modules=[]) # Example DAP initialize request shape: req = { "type": "request", "seq": 1, "command": "initialize", "arguments": { "clientID": "demo", "adapterID": "python", "pathFormat": "path", "linesStartAt1": True, "columnsStartAt1": True, "supportsVariableType": True }, } resp, events = dbg.process_request(req) print(resp) print(events) ``` -------------------------------- ### Normal Release Commands Source: https://github.com/answerdotai/ipymini/blob/main/CLAUDE.md Commands to perform normal releases for the project. These commands handle shipping to GitHub and PyPI, and bumping versions. ```bash ship-gh ship-pypi ship-bump ``` -------------------------------- ### Enable Debugging Output via Environment Variables Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/debug/README.md Enable verbose debugging and message tracing by setting environment variables. ```bash export IPYMINI_DEBUG=1 export IPYMINI_DEBUG_MSGS=1 ``` -------------------------------- ### Import ipymini-zmqthread Components Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Import necessary classes from the ipymini_zmqthread library. ```python from ipymini_zmqthread import (AsyncRouterThread, IOPubThread, StdinRouterThread, HeartbeatThread) ``` -------------------------------- ### Run ipymini kernel manually Source: https://github.com/answerdotai/ipymini/blob/main/README.md Execute the ipymini kernel directly from the command line, providing a connection file path. ```bash python -m ipymini -f /path/to/connection.json ``` -------------------------------- ### Run PR Script Source: https://github.com/answerdotai/ipymini/blob/main/CLAUDE.md Use this script to automate the process of creating, committing, and opening a pull request. Ensure your working tree is clean and all changes are staged before execution. ```bash tools/pr.sh "Message" [label] [body|body-file] ``` -------------------------------- ### Thread-local IO Redirection for Input/Getpass Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/term/README.md Redirect input and getpass using thread_local_io context manager. This allows mocking user input for testing. ```python from ipymini_term import MiniStream, thread_local_io import getpass events = [] stdout = MiniStream("stdout", events) stderr = MiniStream("stderr", events) def request_input(prompt: str, password: bool) -> str: if password: return "secret" return "Ada" with thread_local_io(shell=None, stdout=stdout, stderr=stderr, request_input=request_input, allow_stdin=True): name = input("Name: ") pw = getpass.getpass("Password: ") print(f"hi {name}, pwlen={len(pw)}") print("".join(e["text"] for e in events if e["name"] == "stdout")) ``` -------------------------------- ### comm_context Usage Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Illustrates how to use the comm_context manager to attach parent metadata and IOPub sender information for Jupyter comms. ```python with comm_context(sender, parent): ... ``` -------------------------------- ### MiniShell Constructor Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Defines the parameters for initializing a MiniShell instance, including callbacks for input requests and debugging, and optional user namespaces. ```python MiniShell( request_input: Callable, debug_event_callback: Callable|None = None, zmq_context: zmq.Context|None = None, user_ns: dict|None = None, use_singleton: bool = True) ``` -------------------------------- ### HeartbeatThread Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Implements a simple echo server using a REP socket in a dedicated thread. ```APIDOC ## HeartbeatThread ### Description Provides a dedicated thread for a REP socket that echoes received bytes back to the sender. This is commonly used for heartbeat mechanisms. ### Initialization ```python HeartbeatThread( context: zmq.Context, bind_addr: str ) ``` - **context**: The ZeroMQ context instance. - **bind_addr**: The network address to bind the REP socket to. ### Methods - **start()**: Starts the thread. - **stop()**: Signals the thread to stop. - **join(timeout: float | None = None)**: Waits for the thread to terminate. ### Example Usage ```python from ipymini_zmqthread import HeartbeatThread # Assuming ctx is already initialized hb = HeartbeatThread(ctx, "tcp://127.0.0.1:5557") hb.start() # ... heartbeat communication ... hb.stop() hb.join(timeout=1) ``` ``` -------------------------------- ### MiniShell.complete Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Provides code completion suggestions for a given code string at a specific cursor position. ```APIDOC ## MiniShell.complete ### Description Provides code completion suggestions for a given code string at a specific cursor position. ### Method `complete(code, cursor_pos=None)` ### Parameters - **code** (str) - The code string to complete. - **cursor_pos** (int | None) - The cursor position within the code string. ``` -------------------------------- ### MiniShell.is_complete Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Checks if a given code string is syntactically complete. ```APIDOC ## MiniShell.is_complete ### Description Checks if a given code string is syntactically complete. ### Method `is_complete(code)` ### Parameters - **code** (str) - The code string to check. ``` -------------------------------- ### MiniShell.execution_context Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Creates a context manager for per-request IO capture and comm context binding. This is used to manage the environment for executing a single piece of code. ```APIDOC ## MiniShell.execution_context ### Description Creates a context manager for per-request IO capture and comm context binding. This is used to manage the environment for executing a single piece of code. ### Method Context Manager ### Parameters - **allow_stdin** (bool) - Whether to allow standard input. - **silent** (bool) - Whether to suppress output. - **comm_sender** (Callable) - The sender function for comm messages. - **parent** (dict) - Parent message metadata. ``` -------------------------------- ### MiniShell Constructor Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Initializes a MiniShell instance. This class wraps IPython's InteractiveShell to manage code execution, output capture, and error normalization. ```APIDOC ## MiniShell Constructor ### Description Initializes a MiniShell instance. This class wraps IPython's InteractiveShell to manage code execution, output capture, and error normalization. ### Parameters - **request_input** (Callable) - A callable function to handle user input requests. - **debug_event_callback** (Callable | None) - Optional callback for debug events. - **zmq_context** (zmq.Context | None) - Optional ZeroMQ context. - **user_ns** (dict | None) - Optional user namespace. - **use_singleton** (bool) - Whether to use a singleton instance. ``` -------------------------------- ### Run Non-Slow Tests Source: https://github.com/answerdotai/ipymini/blob/main/CLAUDE.md Execute all tests except those marked as slow. This is the default command for running tests. ```bash pytest -q ``` -------------------------------- ### MiniShell.history Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Retrieves the command history. ```APIDOC ## MiniShell.history ### Description Retrieves the command history. ### Method `history(...)` ### Parameters (Details for parameters not provided in the source text.) ``` -------------------------------- ### IPython Display Capture with IPythonCapture Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/term/README.md Capture IPython display events and last-result data using IPythonCapture. This is useful for testing IPython kernel behavior. ```python from IPython.core.interactiveshell import InteractiveShell from ipymini_term import IPythonCapture shell = InteractiveShell.instance() def request_input(prompt: str, password: bool) -> str: return "42" cap = IPythonCapture(shell, request_input=request_input) with cap.capture(allow_stdin=True, silent=False): shell.run_cell("print('hello')\n1+1") payload = cap.consume_payload() snapshot = cap.snapshot() print(snapshot["streams"]) print(snapshot["result"]) # displayhook last result bundle print(payload) ``` -------------------------------- ### Use subshell context manager Source: https://github.com/answerdotai/ipymini/blob/main/README.md Within an ipymini cell, use the subshell context manager to route subsequent execute requests to a temporary subshell. ```python with get_ipython().kernel.subshell(): await something() ``` -------------------------------- ### MiniShell.execute Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Executes a given piece of code. It captures standard output and standard error, normalizes errors, and can optionally store history. ```APIDOC ## MiniShell.execute ### Description Runs code and returns a snapshot dict containing the result, captured streams, and other execution details. ### Method `execute(code, silent=False, store_history=True, user_expressions=None, allow_stdin=False)` ### Parameters - **code** (str) - The code to execute. - **silent** (bool) - Optional. If True, suppresses output. Defaults to False. - **store_history** (bool) - Optional. If True, stores the executed code in history. Defaults to True. - **user_expressions** (dict | None) - Optional dictionary of user expressions to evaluate. - **allow_stdin** (bool) - Optional. If True, allows standard input during execution. Defaults to False. ### Response - **result** (dict) - A dictionary containing execution results, including `result` (last displayhook result bundle) and `streams` (captured stdout/stderr events). ``` -------------------------------- ### MiniShell.debug_request Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Handles Debug Adapter Protocol (DAP) requests. ```APIDOC ## MiniShell.debug_request ### Description Handles Debug Adapter Protocol (DAP) requests. ### Method `debug_request(request_json)` ### Parameters - **request_json** (dict) - The DAP request in JSON format. ``` -------------------------------- ### Capture stdout/stderr Events with MiniStream Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/term/README.md Use MiniStream to capture stdout and stderr events into a list. This is useful for testing or logging terminal output. ```python from ipymini_term import MiniStream events = [] out = MiniStream("stdout", events) err = MiniStream("stderr", events) out.write("hello ") out.write("world\n") err.write(b"oops\n") print(events) # [ # {"name": "stdout", "text": "hello world\n"}, # {"name": "stderr", "text": "oops\n"}, # ] ``` -------------------------------- ### Run All Tests Source: https://github.com/answerdotai/ipymini/blob/main/CLAUDE.md Execute all tests, including those marked as slow. This script also runs non-slow tests, so avoid running 'pytest -q' beforehand. ```bash tools/run_tests.sh ``` -------------------------------- ### IOPubThread Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Manages a PUB socket in a dedicated thread for sending messages via Session.send. ```APIDOC ## IOPubThread ### Description Runs a PUB socket in a dedicated thread and sends messages via `Session.send` from within that thread. This is used for broadcasting kernel status and execution information. ### Initialization ```python IOPubThread( context: zmq.Context, bind_addr: str, session: Session, qmax: int = 10000, sndhwm: int | None = None ) ``` - **context**: The ZeroMQ context instance. - **bind_addr**: The network address to bind the PUB socket to. - **session**: The Jupyter client session object. - **qmax**: Maximum queue size for messages (default: 10000). - **sndhwm**: Send high-water mark for the socket (optional). ### Methods - **start()**: Starts the thread. - **send(msg_type: str, content: dict, parent: dict | None = None)**: Enqueues a message to be sent. - **stop()**: Signals the thread to stop. - **join(timeout: float | None = None)**: Waits for the thread to terminate. ### Example Usage ```python from ipymini_zmqthread import IOPubThread # Assuming ctx, session are already initialized iopub = IOPubThread(ctx, "tcp://127.0.0.1:5556", session, qmax=10000, sndhwm=None) iopub.start() iopub.send("status", {"execution_state": "idle"}, parent=None) # ... other operations ... iopub.stop() iopub.join(timeout=1) ``` ``` -------------------------------- ### MiniShell.set_display_sender Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Sets the callback function for sending display output. ```APIDOC ## MiniShell.set_display_sender ### Description Sets the callback function for sending display output. ### Method `set_display_sender(sender)` ### Parameters - **sender** (Callable) - The callback function to send display output. ``` -------------------------------- ### Run Tests for Specific Module Source: https://github.com/answerdotai/ipymini/blob/main/CLAUDE.md Execute tests belonging to a particular module within the tests directory. Replace 'debug' with the desired module name (e.g., shell, term, zmqthread, kernel). ```bash pytest tests/debug/ ``` ```bash pytest tests/shell/ ``` ```bash pytest tests/term/ ``` ```bash pytest tests/zmqthread/ ``` ```bash pytest tests/kernel/ ``` -------------------------------- ### comm_context Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md A context manager used by comm publish helpers to attach parent metadata and the IOPub sender. ```APIDOC ## comm_context ### Description Attaches the parent header and IOPub sender used by comms. ### Method Context Manager ### Usage ```python with comm_context(sender, parent): ... ``` ### Parameters - **sender** (Callable) - The IOPub sender function. - **parent** (dict) - The parent message metadata. ``` -------------------------------- ### StdinRouterThread Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Handles input requests and replies using a ROUTER socket, providing a blocking request_input API. ```APIDOC ## StdinRouterThread ### Description Routes `input_request`/`input_reply` messages and offers a blocking `request_input(...)` API. It manages a ROUTER socket dedicated to handling user input prompts. ### Initialization ```python StdinRouterThread( context: zmq.Context, bind_addr: str, session: Session ) ``` - **context**: The ZeroMQ context instance. - **bind_addr**: The network address to bind the ROUTER socket to. - **session**: The Jupyter client session object. ### Methods - **start()**: Starts the thread. - **request_input(prompt: str, password: bool = False, parent: dict | None = None, ident: list | None = None, timeout: float | None = None)**: Blocks until user input is received or timeout occurs. - **interrupt_pending()**: Cancels any in-flight waits for input. - **stop()**: Signals the thread to stop. - **join(timeout: float | None = None)**: Waits for the thread to terminate. ### Example Usage ```python from ipymini_zmqthread import StdinRouterThread # Assuming ctx, session, parent_msg, client_idents are initialized stdin = StdinRouterThread(ctx, "tcp://127.0.0.1:5558", session) stdin.start() value = stdin.request_input("Name: ", password=False, parent=parent_msg, ident=client_idents, timeout=10.0) # ... process input value ... stdin.interrupt_pending() # cancels in-flight waits stdin.stop() stdin.join(timeout=1) ``` ``` -------------------------------- ### Process DAP Request using JSON String with Debugger Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/debug/README.md Process a DAP request provided as a JSON string using the Debugger. This convenience method returns a dictionary containing the response and any associated events. ```python import json from ipymini_debug import Debugger dbg = Debugger(event_callback=None, zmq_context=None, kernel_modules=[]) out = dbg.process_request_json(json.dumps(req)) print(out["response"], out["events"]) ``` -------------------------------- ### Generate Cell Filename for Debugging Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/debug/README.md Map Python code strings to stable filenames for use in debug sessions. The generated path typically includes the process ID and a hash. ```python from ipymini_debug import debug_cell_filename path = debug_cell_filename("print('hello')\n") print(path) # e.g. /tmp/ipymini_/.py ``` -------------------------------- ### Run Slow Tests Only Source: https://github.com/answerdotai/ipymini/blob/main/CLAUDE.md Execute only the tests that are marked with the 'slow' tag. Useful for targeted testing of performance-intensive features. ```bash pytest -q -m slow ``` -------------------------------- ### AsyncRouterThread Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/zmqthread/README.md Manages an asyncio loop and a ROUTER socket in a dedicated thread. It's designed to handle asynchronous message routing and processing. ```APIDOC ## AsyncRouterThread ### Description Runs an asyncio loop in a dedicated thread and owns an async ROUTER socket (`zmq.asyncio`). This class is suitable for handling shell or control messages that require asynchronous processing. ### Initialization ```python AsyncRouterThread( context: zmq.Context, session: Session, bind_addr: str, handler: callable, log_label: str ) ``` - **context**: The ZeroMQ context instance. - **session**: The Jupyter client session object. - **bind_addr**: The network address to bind the ROUTER socket to. - **handler**: A callable function to process incoming messages. - **log_label**: A string label for logging purposes. ### Methods - **start()**: Starts the thread and the asyncio loop. - **ready.wait()**: Blocks until the thread is ready. - **stop()**: Signals the thread to stop. - **join(timeout=None)**: Waits for the thread to terminate. ### Example Usage ```python import zmq from jupyter_client.session import Session from ipymini_zmqthread import AsyncRouterThread ctx = zmq.Context.instance() session = Session(key=b"") router = None def on_msg(msg, idents): router.enqueue(("kernel_info_reply", {"status": "ok"}, msg, idents)) router = AsyncRouterThread( context=ctx, session=session, bind_addr="tcp://127.0.0.1:5555", handler=on_msg, log_label="shell" ) router.start() router.ready.wait() # ... process messages ... router.stop() router.join(timeout=1) ``` ``` -------------------------------- ### Set JUPYTER_PATH for development Source: https://github.com/answerdotai/ipymini/blob/main/README.md Configure the JUPYTER_PATH environment variable to include the repository's kernelspec directory for development. ```bash export JUPYTER_PATH=/path/to/ipymini/share/jupyter:$JUPYTER_PATH ``` -------------------------------- ### MiniShell.inspect Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Retrieves introspection information for an object at a specific cursor position. ```APIDOC ## MiniShell.inspect ### Description Retrieves introspection information for an object at a specific cursor position. ### Method `inspect(code, cursor_pos=None, detail_level=0)` ### Parameters - **code** (str) - The code string. - **cursor_pos** (int | None) - The cursor position. - **detail_level** (int) - The level of detail for the introspection information. Defaults to 0. ``` -------------------------------- ### MiniShell.set_stream_sender Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/shell/README.md Sets the callback function for sending stream output (stdout/stderr). ```APIDOC ## MiniShell.set_stream_sender ### Description Sets the callback function for sending stream output (stdout/stderr). ### Method `set_stream_sender(sender)` ### Parameters - **sender** (Callable) - The callback function to send stream output. ``` -------------------------------- ### Override Cell Filename Mapping Source: https://github.com/answerdotai/ipymini/blob/main/ipymini/debug/README.md Set an environment variable to override the default cell filename mapping, useful for testing or deterministic debugging. ```bash export IPYMINI_CELL_NAME=/tmp/my_cell.py ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.