tessl/pypi-ipykernel
IPython Kernel for Jupyter - provides the core communication layer between Jupyter frontends and the Python interpreter
—
Pending
in-process-kernels.mddocs/
In-Process Kernels
Complete in-process kernel implementation for scenarios requiring tight integration without separate processes. Provides kernel functionality within the same Python process as the calling code, enabling synchronous interaction and shared memory access.
Capabilities
In-Process Kernel Manager
Manager for in-process kernel lifecycle and configuration.
class InProcessKernelManager:
"""
Manager for in-process kernels.
Provides kernel lifecycle management within the same Python process
as the client, enabling direct access and synchronous operations.
"""
def start_kernel(self, **kwargs):
"""
Start the in-process kernel.
Parameters:
- **kwargs: Kernel configuration options
"""
def restart_kernel(self, now=False, newports=False, **kwargs):
"""
Restart the in-process kernel.
Parameters:
- now (bool): Whether to restart immediately
- newports (bool): Whether to use new ports (ignored for in-process)
- **kwargs: Additional restart options
"""
def shutdown_kernel(self, now=False, restart=False):
"""
Shutdown the in-process kernel.
Parameters:
- now (bool): Whether to shutdown immediately
- restart (bool): Whether this is part of a restart
"""
def interrupt_kernel(self):
"""Interrupt the in-process kernel."""
def signal_kernel(self, signum):
"""
Send signal to kernel.
Parameters:
- signum (int): Signal number to send
"""
def is_alive(self):
"""
Check if kernel is alive.
Returns:
bool: True if kernel is running
"""
# Manager attributes
kernel: object # The in-process kernel instance
client_class: type # Client class for this managerIn-Process Kernel Client
Client for interacting with in-process kernels.
class InProcessKernelClient:
"""
Client for in-process kernel communication.
Provides methods for executing code, getting completions, and
other kernel interactions within the same process.
"""
def execute(self, code, silent=False, store_history=True,
user_expressions=None, allow_stdin=None, stop_on_error=True):
"""
Execute code in the in-process kernel.
Parameters:
- code (str): Code to execute
- silent (bool): Whether to suppress output
- store_history (bool): Whether to store in history
- user_expressions (dict, optional): Expressions to evaluate
- allow_stdin (bool, optional): Whether to allow stdin
- stop_on_error (bool): Whether to stop on errors
Returns:
str: Message ID for the execution request
"""
def complete(self, code, cursor_pos=None):
"""
Request code completion.
Parameters:
- code (str): Code context for completion
- cursor_pos (int, optional): Cursor position
Returns:
str: Message ID for the completion request
"""
def inspect(self, code, cursor_pos=None, detail_level=0):
"""
Request object inspection.
Parameters:
- code (str): Code containing object to inspect
- cursor_pos (int, optional): Cursor position
- detail_level (int): Level of detail
Returns:
str: Message ID for the inspection request
"""
def history(self, raw=True, output=False, hist_access_type='range', **kwargs):
"""
Request execution history.
Parameters:
- raw (bool): Whether to return raw input
- output (bool): Whether to include output
- hist_access_type (str): Type of history access
- **kwargs: Additional history options
Returns:
str: Message ID for the history request
"""
def kernel_info(self):
"""
Request kernel information.
Returns:
str: Message ID for the kernel info request
"""
def shutdown(self, restart=False):
"""
Request kernel shutdown.
Parameters:
- restart (bool): Whether this is a restart
Returns:
str: Message ID for the shutdown request
"""Blocking In-Process Client
Synchronous client that blocks until responses are received.
class BlockingInProcessKernelClient:
"""
Blocking client for in-process kernels.
Provides synchronous methods that block until responses are received,
making it easier to use in synchronous code.
"""
def execute_interactive(self, code, silent=False, store_history=True,
user_expressions=None, allow_stdin=None,
timeout=None):
"""
Execute code and wait for completion.
Parameters:
- code (str): Code to execute
- silent (bool): Whether to suppress output
- store_history (bool): Whether to store in history
- user_expressions (dict, optional): Expressions to evaluate
- allow_stdin (bool, optional): Whether to allow stdin
- timeout (float, optional): Timeout in seconds
Returns:
dict: Execution result
"""
def get_shell_msg(self, block=True, timeout=None):
"""
Get message from shell channel.
Parameters:
- block (bool): Whether to block until message available
- timeout (float, optional): Timeout in seconds
Returns:
dict: Message from shell channel
"""
def get_iopub_msg(self, block=True, timeout=None):
"""
Get message from IOPub channel.
Parameters:
- block (bool): Whether to block until message available
- timeout (float, optional): Timeout in seconds
Returns:
dict: Message from IOPub channel
"""In-Process Channels
Communication channels for in-process kernel interaction.
class InProcessChannel:
"""
Base channel for in-process kernel communication.
Provides the communication interface between client and kernel
within the same process.
"""
def send(self, msg):
"""
Send message through this channel.
Parameters:
- msg (dict): Message to send
"""
def get_msg(self, block=True, timeout=None):
"""
Get message from this channel.
Parameters:
- block (bool): Whether to block until message available
- timeout (float, optional): Timeout in seconds
Returns:
dict: Received message
"""
def get_msgs(self):
"""
Get all available messages.
Returns:
list: List of all available messages
"""
def msg_ready(self):
"""
Check if messages are available.
Returns:
bool: True if messages are ready
"""
class InProcessHBChannel:
"""
Heartbeat channel for in-process kernels.
Handles heartbeat/ping functionality for kernel liveness checking.
"""
def start(self):
"""Start the heartbeat channel."""
def stop(self):
"""Stop the heartbeat channel."""
def is_beating(self):
"""
Check if heartbeat is active.
Returns:
bool: True if heartbeat is active
"""Usage Examples
Basic In-Process Kernel Usage
from ipykernel.inprocess import InProcessKernelManager, InProcessKernelClient
# Create and start in-process kernel
km = InProcessKernelManager()
km.start_kernel()
# Create client
kc = InProcessKernelClient()
kc.load_connection_info(km.get_connection_info())
# Execute code
msg_id = kc.execute("x = 42; print(x)")
# Get execution results (non-blocking)
while True:
try:
msg = kc.get_shell_msg(timeout=1)
if msg['parent_header']['msg_id'] == msg_id:
print(f"Execution status: {msg['content']['status']}")
break
except:
break
# Cleanup
km.shutdown_kernel()Synchronous In-Process Operations
from ipykernel.inprocess import InProcessKernelManager, BlockingInProcessKernelClient
# Create and start kernel
km = InProcessKernelManager()
km.start_kernel()
# Create blocking client for synchronous operations
kc = BlockingInProcessKernelClient()
kc.load_connection_info(km.get_connection_info())
# Execute code synchronously
result = kc.execute_interactive("import math; math.sqrt(16)")
print(f"Execution result: {result}")
# Get completion suggestions synchronously
completions = kc.complete("mat", 3)
print(f"Completions: {completions}")
# Get object inspection synchronously
inspection = kc.inspect("math.sqrt", 9)
print(f"Inspection: {inspection}")
# Cleanup
km.shutdown_kernel()In-Process Kernel in Class Context
from ipykernel.inprocess import InProcessKernelManager, BlockingInProcessKernelClient
class CalculationEngine:
"""Example calculation engine using in-process kernel."""
def __init__(self):
self.km = InProcessKernelManager()
self.km.start_kernel()
self.kc = BlockingInProcessKernelClient()
self.kc.load_connection_info(self.km.get_connection_info())
# Initialize kernel with utility functions
self._setup_kernel()
def _setup_kernel(self):
"""Setup kernel with utility functions."""
setup_code = """
import math
import statistics
def calculate_stats(data):
return {
'mean': statistics.mean(data),
'median': statistics.median(data),
'stdev': statistics.stdev(data) if len(data) > 1 else 0
}
"""
self.kc.execute_interactive(setup_code)
def execute_calculation(self, code):
"""Execute calculation code in kernel."""
result = self.kc.execute_interactive(code)
return result
def calculate_statistics(self, data):
"""Calculate statistics using kernel."""
code = f"calculate_stats({data})"
result = self.kc.execute_interactive(code)
return result
def shutdown(self):
"""Cleanup kernel resources."""
self.km.shutdown_kernel()
# Usage
engine = CalculationEngine()
# Execute calculations
result = engine.execute_calculation("2 ** 10")
print(f"2^10 = {result}")
# Calculate statistics
stats = engine.calculate_statistics([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
print(f"Statistics: {stats}")
# Cleanup
engine.shutdown()Interactive Data Processing
from ipykernel.inprocess import InProcessKernelManager, BlockingInProcessKernelClient
class DataProcessor:
"""Interactive data processor using in-process kernel."""
def __init__(self):
self.km = InProcessKernelManager()
self.km.start_kernel()
self.kc = BlockingInProcessKernelClient()
self.kc.load_connection_info(self.km.get_connection_info())
# Setup data processing environment
self._setup_environment()
def _setup_environment(self):
"""Setup kernel with data processing libraries."""
setup_code = """
import pandas as pd
import numpy as np
# Global data storage
datasets = {}
def load_dataset(name, data):
datasets[name] = pd.DataFrame(data)
return f"Loaded dataset '{name}' with {len(data)} rows"
def process_dataset(name, operation):
if name not in datasets:
return f"Dataset '{name}' not found"
df = datasets[name]
if operation == 'describe':
return df.describe().to_dict()
elif operation == 'head':
return df.head().to_dict()
else:
return f"Unknown operation: {operation}"
"""
self.kc.execute_interactive(setup_code)
def load_data(self, name, data):
"""Load data into kernel."""
code = f"load_dataset('{name}', {data})"
result = self.kc.execute_interactive(code)
return result
def process_data(self, name, operation):
"""Process data in kernel."""
code = f"process_dataset('{name}', '{operation}')"
result = self.kc.execute_interactive(code)
return result
def execute_custom(self, code):
"""Execute custom data processing code."""
return self.kc.execute_interactive(code)
def shutdown(self):
"""Cleanup resources."""
self.km.shutdown_kernel()
# Usage
processor = DataProcessor()
# Load sample data
data = [
{'name': 'Alice', 'age': 25, 'salary': 50000},
{'name': 'Bob', 'age': 30, 'salary': 60000},
{'name': 'Charlie', 'age': 35, 'salary': 70000}
]
processor.load_data('employees', data)
# Process data
stats = processor.process_data('employees', 'describe')
print(f"Employee statistics: {stats}")
# Custom processing
result = processor.execute_custom("""
df = datasets['employees']
high_earners = df[df['salary'] > 55000]
high_earners.to_dict('records')
""")
print(f"High earners: {result}")
# Cleanup
processor.shutdown()Install with Tessl CLI
npx tessl i tessl/pypi-ipykernel