Communication Framework (Comm)

class Comm:
    """
    Bidirectional communication channel between kernel and frontend.
    
    Enables custom messaging protocols for interactive widgets and
    real-time data exchange beyond standard execution messages.
    """
    
    def __init__(self, target_name=None, data=None, metadata=None,
                 buffers=None, **kwargs):
        """
        Create a new Comm instance.
        
        Parameters:
        - target_name (str, optional): Name of target handler on frontend
        - data (dict, optional): Initial data to send
        - metadata (dict, optional): Message metadata
        - buffers (list, optional): Binary buffer data
        - **kwargs: Additional comm creation options
        """
    
    def send(self, data=None, metadata=None, buffers=None):
        """
        Send data to the frontend via this comm.
        
        Parameters:
        - data (dict, optional): Data to send
        - metadata (dict, optional): Message metadata
        - buffers (list, optional): Binary buffer data
        """
    
    def close(self, data=None, metadata=None, buffers=None):
        """
        Close this comm and optionally send final data.
        
        Parameters:
        - data (dict, optional): Final data to send
        - metadata (dict, optional): Message metadata  
        - buffers (list, optional): Binary buffer data
        """
    
    def on_msg(self, callback):
        """
        Register callback for messages received on this comm.
        
        Parameters:
        - callback (callable): Function to call with received messages
                              Signature: callback(msg)
        """
    
    def on_close(self, callback):
        """
        Register callback for when this comm is closed.
        
        Parameters:
        - callback (callable): Function to call when comm closes
                              Signature: callback(msg)
        """
    
    # Comm attributes
    comm_id: str          # Unique identifier for this comm
    target_name: str      # Target handler name
    kernel: object        # Reference to kernel instance

class CommManager:
    """
    Manager for comm instances and target handlers.
    
    Handles comm creation, registration, message routing, and cleanup
    for the kernel's communication infrastructure.
    """
    
    def register_target(self, target_name, f):
        """
        Register a handler function for a comm target.
        
        Parameters:
        - target_name (str): Name of the target to handle
        - f (callable): Handler function for comm open requests
                       Signature: f(comm, open_msg)
        """
    
    def unregister_target(self, target_name, f):
        """
        Unregister a target handler.
        
        Parameters:
        - target_name (str): Target name to unregister
        - f (callable): Handler function to remove
        """
    
    def register_comm(self, comm):
        """
        Register a comm instance with the manager.
        
        Parameters:
        - comm (Comm): Comm instance to register
        """
    
    def unregister_comm(self, comm):
        """
        Unregister and cleanup a comm instance.
        
        Parameters:
        - comm (Comm): Comm instance to unregister
        """
    
    def get_comm(self, comm_id):
        """
        Get a registered comm by its ID.
        
        Parameters:
        - comm_id (str): Unique comm identifier
        
        Returns:
        Comm or None: The comm instance, or None if not found
        """
    
    # Manager attributes
    comms: dict           # Mapping of comm_id to Comm instances
    targets: dict         # Mapping of target_name to handler functions
    kernel: object        # Reference to kernel instance

from ipykernel.comm import Comm

# Create a comm for bidirectional communication
comm = Comm(target_name='my_widget')

# Send data to frontend
comm.send({
    'action': 'update',
    'value': 42,
    'timestamp': '2023-01-01T00:00:00Z'
})

# Register callback for incoming messages
def handle_message(msg):
    data = msg['content']['data']
    print(f"Received: {data}")
    
    # Echo back a response
    comm.send({
        'action': 'echo',
        'original': data
    })

comm.on_msg(handle_message)

# Close the comm when done
comm.close({'action': 'cleanup'})

from ipykernel.comm import CommManager, Comm

# Get the kernel's comm manager (typically available as kernel.comm_manager)
# comm_manager = kernel.comm_manager

# Register a target handler for new comms
def handle_widget_comm(comm, open_msg):
    """Handle new widget comm connections."""
    print(f"New comm opened: {comm.comm_id}")
    
    # Set up message handler for this comm
    def on_widget_msg(msg):
        data = msg['content']['data']
        
        if data.get('action') == 'get_state':
            # Send current widget state
            comm.send({
                'action': 'state',
                'state': {'value': 100, 'enabled': True}
            })
        elif data.get('action') == 'set_value':
            # Update widget value
            new_value = data.get('value', 0)
            print(f"Widget value updated to: {new_value}")
            
            # Confirm update
            comm.send({
                'action': 'value_updated',
                'value': new_value
            })
    
    comm.on_msg(on_widget_msg)
    
    # Send initial state
    comm.send({
        'action': 'initialized',
        'widget_id': comm.comm_id
    })

# Register the target handler
comm_manager.register_target('my_widget', handle_widget_comm)

from ipykernel.comm import Comm
import threading
import time

class SimpleSlider:
    """Example interactive widget using comms."""
    
    def __init__(self, min_val=0, max_val=100, initial_val=50):
        self.min_val = min_val
        self.max_val = max_val
        self.value = initial_val
        self.observers = []
        
        # Create comm for frontend communication
        self.comm = Comm(target_name='simple_slider')
        self.comm.on_msg(self._handle_frontend_msg)
        
        # Send initial configuration
        self.comm.send({
            'action': 'create',
            'config': {
                'min': self.min_val,
                'max': self.max_val,
                'value': self.value
            }
        })
    
    def _handle_frontend_msg(self, msg):
        """Handle messages from frontend."""
        data = msg['content']['data']
        
        if data.get('action') == 'value_change':
            new_value = data.get('value')
            if self.min_val <= new_value <= self.max_val:
                old_value = self.value
                self.value = new_value
                
                # Notify observers
                for callback in self.observers:
                    callback(old_value, new_value)
    
    def observe(self, callback):
        """Register callback for value changes."""
        self.observers.append(callback)
    
    def set_value(self, value):
        """Programmatically set value."""
        if self.min_val <= value <= self.max_val:
            self.value = value
            self.comm.send({
                'action': 'update_value',
                'value': value
            })

# Usage
slider = SimpleSlider(0, 255, 128)

def on_value_change(old_val, new_val):
    print(f"Slider changed from {old_val} to {new_val}")

slider.observe(on_value_change)

from ipykernel.comm import CommManager

# Access comm manager (typically kernel.comm_manager)
# comm_manager = kernel.comm_manager

# List all active comms
active_comms = list(comm_manager.comms.keys())
print(f"Active comms: {active_comms}")

# Get specific comm
comm_id = active_comms[0] if active_comms else None
if comm_id:
    comm = comm_manager.get_comm(comm_id)
    print(f"Comm target: {comm.target_name}")

# Clean shutdown - close all comms
for comm_id, comm in list(comm_manager.comms.items()):
    comm.close({'action': 'shutdown'})

# Unregister targets
comm_manager.unregister_target('my_widget', handle_widget_comm)

CommMessage = {
    "header": {
        "msg_id": str,
        "msg_type": str,        # 'comm_open', 'comm_msg', 'comm_close'
        "username": str,
        "session": str,
        "date": str,
        "version": str
    },
    "parent_header": dict,
    "metadata": dict,
    "content": {
        "comm_id": str,         # Unique comm identifier
        "target_name": str,     # Target handler name (for open messages)
        "data": dict,           # Custom message data
        "target_module": str    # Optional target module (for open messages)
    },
    "buffers": list           # Optional binary buffers
}