adapter.py

"""Adapters for IPython msg spec versions."""

# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.

import json

from IPython.core.release import kernel_protocol_version_info
from IPython.utils.tokenutil import token_at_cursor


def code_to_line(code, cursor_pos):
    """Turn a multiline code block and cursor position into a single line
    and new cursor position.
    
    For adapting ``complete_`` and ``object_info_request``.
    """
    for line in code.splitlines(True):
        n = len(line)
        if cursor_pos > n:
            cursor_pos -= n
        else:
            break
    return line, cursor_pos


class Adapter(object):
    """Base class for adapting messages
    
    Override message_type(msg) methods to create adapters.
    """
    
    msg_type_map = {}
    
    def update_header(self, msg):
        return msg
    
    def update_metadata(self, msg):
        return msg
    
    def update_msg_type(self, msg):
        header = msg['header']
        msg_type = header['msg_type']
        if msg_type in self.msg_type_map:
            msg['msg_type'] = header['msg_type'] = self.msg_type_map[msg_type]
        return msg
    
    def handle_reply_status_error(self, msg):
        """This will be called *instead of* the regular handler
        
        on any reply with status != ok
        """
        return msg
    
    def __call__(self, msg):
        msg = self.update_header(msg)
        msg = self.update_metadata(msg)
        msg = self.update_msg_type(msg)
        header = msg['header']
        
        handler = getattr(self, header['msg_type'], None)
        if handler is None:
            return msg
        
        # handle status=error replies separately (no change, at present)
        if msg['content'].get('status', None) in {'error', 'aborted'}:
            return self.handle_reply_status_error(msg)
        return handler(msg)

def _version_str_to_list(version):
    """convert a version string to a list of ints
    
    non-int segments are excluded
    """
    v = []
    for part in version.split('.'):
        try:
            v.append(int(part))
        except ValueError:
            pass
    return v

class V5toV4(Adapter):
    """Adapt msg protocol v5 to v4"""
    
    version = '4.1'
    
    msg_type_map = {
        'execute_result' : 'pyout',
        'execute_input' : 'pyin',
        'error' : 'pyerr',
        'inspect_request' : 'object_info_request',
        'inspect_reply' : 'object_info_reply',
    }
    
    def update_header(self, msg):
        msg['header'].pop('version', None)
        return msg
    
    # shell channel
    
    def kernel_info_reply(self, msg):
        content = msg['content']
        content.pop('banner', None)
        for key in ('language_version', 'protocol_version'):
            if key in content:
                content[key] = _version_str_to_list(content[key])
        if content.pop('implementation', '') == 'ipython' \
            and 'implementation_version' in content:
            content['ipython_version'] = content.pop('implmentation_version')
        content.pop('implementation_version', None)
        content.setdefault("implmentation", content['language'])
        return msg
    
    def execute_request(self, msg):
        content = msg['content']
        content.setdefault('user_variables', [])
        return msg
    
    def execute_reply(self, msg):
        content = msg['content']
        content.setdefault('user_variables', {})
        # TODO: handle payloads
        return msg
    
    def complete_request(self, msg):
        content = msg['content']
        code = content['code']
        cursor_pos = content['cursor_pos']
        line, cursor_pos = code_to_line(code, cursor_pos)
        
        new_content = msg['content'] = {}
        new_content['text'] = ''
        new_content['line'] = line
        new_content['block'] = None
        new_content['cursor_pos'] = cursor_pos
        return msg
    
    def complete_reply(self, msg):
        content = msg['content']
        cursor_start = content.pop('cursor_start')
        cursor_end = content.pop('cursor_end')
        match_len = cursor_end - cursor_start
        content['matched_text'] = content['matches'][0][:match_len]
        content.pop('metadata', None)
        return msg
    
    def object_info_request(self, msg):
        content = msg['content']
        code = content['code']
        cursor_pos = content['cursor_pos']
        line, _ = code_to_line(code, cursor_pos)
        
        new_content = msg['content'] = {}
        new_content['oname'] = token_at_cursor(code, cursor_pos)
        new_content['detail_level'] = content['detail_level']
        return msg
    
    def object_info_reply(self, msg):
        """inspect_reply can't be easily backward compatible"""
        msg['content'] = {'found' : False, 'name' : 'unknown'}
        return msg
    
    # iopub channel
    
    def display_data(self, msg):
        content = msg['content']
        content.setdefault("source", "display")
        data = content['data']
        if 'application/json' in data:
            try:
                data['application/json'] = json.dumps(data['application/json'])
            except Exception:
                # warn?
                pass
        return msg
    
    # stdin channel
    
    def input_request(self, msg):
        msg['content'].pop('password', None)
        return msg


class V4toV5(Adapter):
    """Convert msg spec V4 to V5"""
    version = '5.0'
    
    # invert message renames above
    msg_type_map = {v:k for k,v in V5toV4.msg_type_map.items()}
    
    def update_header(self, msg):
        msg['header']['version'] = self.version
        return msg
    
    # shell channel
    
    def kernel_info_reply(self, msg):
        content = msg['content']
        for key in ('language_version', 'protocol_version', 'ipython_version'):
            if key in content:
                content[key] = ".".join(map(str, content[key]))
        
        if content['language'].startswith('python') and 'ipython_version' in content:
            content['implementation'] = 'ipython'
            content['implementation_version'] = content.pop('ipython_version')
        
        content['banner'] = ''
        return msg
    
    def execute_request(self, msg):
        content = msg['content']
        user_variables = content.pop('user_variables', [])
        user_expressions = content.setdefault('user_expressions', {})
        for v in user_variables:
            user_expressions[v] = v
        return msg
    
    def execute_reply(self, msg):
        content = msg['content']
        user_expressions = content.setdefault('user_expressions', {})
        user_variables = content.pop('user_variables', {})
        if user_variables:
            user_expressions.update(user_variables)
        return msg
    
    def complete_request(self, msg):
        old_content = msg['content']
        
        new_content = msg['content'] = {}
        new_content['code'] = old_content['line']
        new_content['cursor_pos'] = old_content['cursor_pos']
        return msg
    
    def complete_reply(self, msg):
        # complete_reply needs more context than we have to get cursor_start and end.
        # use special value of `-1` to indicate to frontend that it should be at
        # the current cursor position.
        content = msg['content']
        new_content = msg['content'] = {'status' : 'ok'}
        new_content['matches'] = content['matches']
        new_content['cursor_start'] = -len(content['matched_text'])
        new_content['cursor_end'] = None
        new_content['metadata'] = {}
        return msg
    
    def inspect_request(self, msg):
        content = msg['content']
        name = content['oname']
        
        new_content = msg['content'] = {}
        new_content['code'] = name
        new_content['cursor_pos'] = len(name)
        new_content['detail_level'] = content['detail_level']
        return msg
    
    def inspect_reply(self, msg):
        """inspect_reply can't be easily backward compatible"""
        content = msg['content']
        new_content = msg['content'] = {'status' : 'ok'}
        found = new_content['found'] = content['found']
        new_content['name'] = content['name']
        new_content['data'] = data = {}
        new_content['metadata'] = {}
        if found:
            lines = []
            for key in ('call_def', 'init_definition', 'definition'):
                if content.get(key, False):
                    lines.append(content[key])
                    break
            for key in ('call_docstring', 'init_docstring', 'docstring'):
                if content.get(key, False):
                    lines.append(content[key])
                    break
            if not lines:
                lines.append("<empty docstring>")
            data['text/plain'] = '\n'.join(lines)
        return msg
    
    # iopub channel
    
    def display_data(self, msg):
        content = msg['content']
        content.pop("source", None)
        data = content['data']
        if 'application/json' in data:
            try:
                data['application/json'] = json.loads(data['application/json'])
            except Exception:
                # warn?
                pass
        return msg
    
    # stdin channel
    
    def input_request(self, msg):
        msg['content'].setdefault('password', False)
        return msg
    


def adapt(msg, to_version=kernel_protocol_version_info[0]):
    """Adapt a single message to a target version
    
    Parameters
    ----------
    
    msg : dict
        An IPython message.
    to_version : int, optional
        The target major version.
        If unspecified, adapt to the current version for IPython.
    
    Returns
    -------
    
    msg : dict
        An IPython message appropriate in the new version.
    """
    header = msg['header']
    if 'version' in header:
        from_version = int(header['version'].split('.')[0])
    else:
        # assume last version before adding the key to the header
        from_version = 4
    adapter = adapters.get((from_version, to_version), None)
    if adapter is None:
        return msg
    return adapter(msg)


# one adapter per major version from,to
adapters = {
    (5,4) : V5toV4(),
    (4,5) : V4toV5(),
}