<commit_before>"""
KBase narrative service and method API.

The main classes defined here are :class:`Service` and :class:`ServiceMethod`.

See :func:`example` for an example of usage.
"""
__author__ = ["<NAME> <<EMAIL>>", "<NAME> <<EMAIL>>"]
__version__ = "0.0.1"

## Imports
# Stdlib
from collections import deque
import json
import logging
import os
import sys
import time
# Third-party
import IPython.utils.traitlets as trt
# Local
from biokbase.narrative.common import kbtypes

## Logging boilerplate

_log = logging.getLogger(__name__)
# turn on debugging by setting environment variable KBASE_DEBUG
if os.environ.get("KBASE_DEBUG", None):
    _log.setLevel(logging.DEBUG)
else:
    _log.setLevel(logging.WARN)
# set custom log format
_h = logging.StreamHandler()
_h.setFormatter(logging.Formatter("%(levelname)s %(asctime)s %(module)s: %(message)s"))
_log.addHandler(_h)

## Globals


class URLS:
    workspace = "http://kbase.us/services/workspace"
    invocation = "https://kbase.us/services/invocation"
    fba = "https://kbase.us/services/fba_model_services"

## Exceptions


class ServiceError(Exception):
    """Base class for Service errors.
    Should not normally be instantiated directly.
    """

    def __init__(self, errmsg):
        Exception.__init__(self, errmsg)
        self._info = {
            'severity': 'FATAL',
            'type': self.__class__.__name__,
            'msg': str(errmsg)
        }

    def add_info(self, k, v):
        self._info[k] = v

    def as_json(self):
        return json.dumps(self._info)


class DuplicateServiceError(ServiceError):
    pass


class ServiceMethodError(ServiceError):
    """Base class for all ServiceMethod errors"""

    def __init__(self, method, errmsg):
        msg = "in ServiceMethod '{}': {}".format(method.name, errmsg)
        ServiceError.__init__(self, msg)
        self.add_info('method_name', method.name)

class ServiceMethodParameterError(ServiceMethodError):
    """Bad parameter for ServiceMethod."""

    def __init__(self, method, errmsg):
        msg = "bad parameter: " + errmsg
        ServiceMethodError.__init__(self, method, msg)
        self.add_info('details', errmsg)

class ServiceRegistryFormatError(ServiceMethodError):
    """Bad format for Service Registry."""

    def __init__(self, method, errmsg):
        msg = "bad registry format: " + errmsg
        ServiceMethodError.__init__(self, method, msg)
        self.add_info('details', errmsg)

## Utility functions / classes


def is_sequence(arg):
    """Returns True only if input acts like a sequence, but does
    not act like a string.
    """
    return (not hasattr(arg, "strip") and
            hasattr(arg, "__getitem__") or
            hasattr(arg, "__iter__"))


def get_func_desc(fn):
    """Get function description from docstring.
    """
    doc, desc = fn.__doc__, []
    for line in doc.split("\n"):
        line = line.strip()
        if line == "":
            break
        desc.append(line)
    return ' '.join(desc)


def get_func_info(fn):
    """Get params and return from docstring
    """
    doc = fn.__doc__
    params, return_ = {}, {}
    param_order = []
    for line in doc.split("\n"):
        line = line.strip()
        # :param
        if line.startswith(":param"):
            _, name, desc = line.split(":", 2)
            name = name[6:].strip()  # skip 'param '
            params[name] = {'desc': desc.strip()}
            param_order.append(name)
        # :type (of parameter, should be in kbtypes)
        elif line.startswith(":type"):
            _, name, desc = line.split(":", 2)
            name = name[5:].strip()  # skip 'type '
            if not name in params:
                raise ValueError("'type' without 'param' for {}".format(name))
            typeobj = eval(desc.strip())
            params[name]['type'] = typeobj

        # :ui_name (of parameter) - the name that should be displayed in the user interface
        elif line.startswith(":ui_name"):
            _, name, ui_name = line.split(":", 2)
            name = name[8:].strip()  # skip 'ui_name '
            if not name in params:
                raise ValueError("'ui_name' without 'param' for {}".format(name))
            ui_name = ui_name.strip()
            params[name]['ui_name'] = ui_name

        # :return - name of thing to return
        elif line.startswith(":return"):
            _1, _2, desc = line.split(":", 2)
            return_['desc'] = desc

        # :rtype - type of thing to return
        elif line.startswith(":rtype"):
            _1, _2, desc = line.split(":", 2)
            typeobj = eval(desc.strip())
            return_['type'] = typeobj

        # :input_widget - the default visualization widget for this method. 
        # Should be the name as it's invoked in Javascript.
        elif line.startswith(":input_widget"):
            _1, _2, widget = line.split(":", 2)
            return_['input_widget'] = widget.strip()

        # :output_widget - the visualization widget for this method. 
        # Should be the name as it's invoked in Javascript.
        elif line.startswith(":output_widget"):
            _1, _2, widget = line.split(":", 2)
            return_['output_widget'] = widget.strip()

        # :embed - True if the widget should be automatically embedded.
        # so, probably always True, but not necessarily
        elif line.startswith(":embed"):
            _1, _2, embed = line.split(":", 2)
            embed = eval(embed.strip())
            return_['embed_widget'] = embed
    r_params = []
    vis_info = {'input_widget' : None,
                'output_widget' : None,
                'embed_widget' : True }
    for i, name in enumerate(param_order):
        type_ = params[name]['type']
        desc = params[name]['desc']
        ui_name = params[name]['ui_name']
        r_params.append(type_(desc=desc, ui_name=ui_name))
    if return_ is None:
        r_output = None
    else:<commit_msg>Use plain parameter name if ui_name is not given (instead of stack trace)<commit_after>"""
KBase narrative service and method API.

The main classes defined here are :class:`Service` and :class:`ServiceMethod`.

See :func:`example` for an example of usage.
"""
__author__ = ["<NAME> <<EMAIL>>", "<NAME> <<EMAIL>>"]
__version__ = "0.0.1"

## Imports
# Stdlib
from collections import deque
import json
import logging
import os
import sys
import time
# Third-party
import IPython.utils.traitlets as trt
# Local
from biokbase.narrative.common import kbtypes

## Logging boilerplate

_log = logging.getLogger(__name__)
# turn on debugging by setting environment variable KBASE_DEBUG
if os.environ.get("KBASE_DEBUG", None):
    _log.setLevel(logging.DEBUG)
else:
    _log.setLevel(logging.WARN)
# set custom log format
_h = logging.StreamHandler()
_h.setFormatter(logging.Formatter("%(levelname)s %(asctime)s %(module)s: %(message)s"))
_log.addHandler(_h)

## Globals


class URLS:
    workspace = "http://kbase.us/services/workspace"
    invocation = "https://kbase.us/services/invocation"
    fba = "https://kbase.us/services/fba_model_services"

## Exceptions


class ServiceError(Exception):
    """Base class for Service errors.
    Should not normally be instantiated directly.
    """

    def __init__(self, errmsg):
        Exception.__init__(self, errmsg)
        self._info = {
            'severity': 'FATAL',
            'type': self.__class__.__name__,
            'msg': str(errmsg)
        }

    def add_info(self, k, v):
        self._info[k] = v

    def as_json(self):
        return json.dumps(self._info)


class DuplicateServiceError(ServiceError):
    pass


class ServiceMethodError(ServiceError):
    """Base class for all ServiceMethod errors"""

    def __init__(self, method, errmsg):
        msg = "in ServiceMethod '{}': {}".format(method.name, errmsg)
        ServiceError.__init__(self, msg)
        self.add_info('method_name', method.name)

class ServiceMethodParameterError(ServiceMethodError):
    """Bad parameter for ServiceMethod."""

    def __init__(self, method, errmsg):
        msg = "bad parameter: " + errmsg
        ServiceMethodError.__init__(self, method, msg)
        self.add_info('details', errmsg)

class ServiceRegistryFormatError(ServiceMethodError):
    """Bad format for Service Registry."""

    def __init__(self, method, errmsg):
        msg = "bad registry format: " + errmsg
        ServiceMethodError.__init__(self, method, msg)
        self.add_info('details', errmsg)

## Utility functions / classes


def is_sequence(arg):
    """Returns True only if input acts like a sequence, but does
    not act like a string.
    """
    return (not hasattr(arg, "strip") and
            hasattr(arg, "__getitem__") or
            hasattr(arg, "__iter__"))


def get_func_desc(fn):
    """Get function description from docstring.
    """
    doc, desc = fn.__doc__, []
    for line in doc.split("\n"):
        line = line.strip()
        if line == "":
            break
        desc.append(line)
    return ' '.join(desc)


def get_func_info(fn):
    """Get params and return from docstring
    """
    doc = fn.__doc__
    params, return_ = {}, {}
    param_order = []
    for line in doc.split("\n"):
        line = line.strip()
        # :param
        if line.startswith(":param"):
            _, name, desc = line.split(":", 2)
            name = name[6:].strip()  # skip 'param '
            params[name] = {'desc': desc.strip()}
            param_order.append(name)
        # :type (of parameter, should be in kbtypes)
        elif line.startswith(":type"):
            _, name, desc = line.split(":", 2)
            name = name[5:].strip()  # skip 'type '
            if not name in params:
                raise ValueError("'type' without 'param' for {}".format(name))
            typeobj = eval(desc.strip())
            params[name]['type'] = typeobj

        # :ui_name (of parameter) - the name that should be displayed in the user interface
        elif line.startswith(":ui_name"):
            _, name, ui_name = line.split(":", 2)
            name = name[8:].strip()  # skip 'ui_name '
            if not name in params:
                raise ValueError("'ui_name' without 'param' for {}".format(name))
            ui_name = ui_name.strip()
            params[name]['ui_name'] = ui_name

        # :return - name of thing to return
        elif line.startswith(":return"):
            _1, _2, desc = line.split(":", 2)
            return_['desc'] = desc

        # :rtype - type of thing to return
        elif line.startswith(":rtype"):
            _1, _2, desc = line.split(":", 2)
            typeobj = eval(desc.strip())
            return_['type'] = typeobj

        # :input_widget - the default visualization widget for this method. 
        # Should be the name as it's invoked in Javascript.
        elif line.startswith(":input_widget"):
            _1, _2, widget = line.split(":", 2)
            return_['input_widget'] = widget.strip()

        # :output_widget - the visualization widget for this method. 
        # Should be the name as it's invoked in Javascript.
        elif line.startswith(":output_widget"):
            _1, _2, widget = line.split(":", 2)
            return_['output_widget'] = widget.strip()

        # :embed - True if the widget should be automatically embedded.
        # so, probably always True, but not necessarily
        elif line.startswith(":embed"):
            _1, _2, embed = line.split(":", 2)
            embed = eval(embed.strip())
            return_['embed_widget'] = embed
    r_params = []
    vis_info = {'input_widget': None,
                'output_widget': None,
                'embed_widget': True}
    for i, name in enumerate(param_order):
        type_ = params[name]['type']
        desc = params[name]['desc']
        ui_name = params[name].get('ui_name', name)  # use parameter name if no ui_name is given
        r_params.append(type_(desc=desc, ui_name=ui_name))
    if return_ is None:
        r_output = None
    else:
{'fraction_non_alphanumeric': 0.12950778712417757, 'fraction_numerical': 0.003997668026984259, 'mean_word_length': 2.545320342485976, 'pattern_counts': {'":': 34, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7236352', 'n_tokens_mistral': 3623, 'n_tokens_neox': 3570, 'n_words': 998}