<commit_before># Licensed under a 3-clause BSD style license - see LICENSE.rst
"""
Utility functions and classes
"""

from __future__ import (absolute_import, division, print_function,
                        unicode_literals)

import inspect
import traceback

from ...extern.six.moves import queue
from ...extern.six.moves.urllib.error import URLError
from ...extern.six.moves.urllib.request import urlopen
from ...extern.six.moves import xmlrpc_client as xmlrpc
from ...extern.six import StringIO
from ...config import ConfigurationItem

from .constants import SAMP_STATUS_ERROR
from .errors import SAMPProxyError

ALLOW_INTERNET = ConfigurationItem('use_internet', True,
                                   "Whether to allow astropy.vo.samp to use "
                                   "the internet, if available")


def internet_on():
    if not ALLOW_INTERNET():
        return False
    else:
        try:
            urlopen('http://google.com', timeout=1.)
            return True
        except:
            return False

__all__ = ["SAMPMsgReplierWrapper"]

__doctest_skip__ = ['.']


def getattr_recursive(variable, attribute):
    """
    Get attributes recursively.
    """
    if '.' in attribute:
        top, remaining = attribute.split('.', 1)
        return getattr_recursive(getattr(variable, top), remaining)
    else:
        return getattr(variable, attribute)


class _ServerProxyPoolMethod(object):

    # some magic to bind an XML-RPC method to an RPC server.
    # supports "nested" methods (e.g. examples.getStateName)

    def __init__(self, proxies, name):
        self.__proxies = proxies
        self.__name = name

    def __getattr__(self, name):
        return _ServerProxyPoolMethod(self.__proxies, "%s.%s" % (self.__name, name))

    def __call__(self, *args, **kwrds):
        proxy = self.__proxies.get()
        function = getattr_recursive(proxy, self.__name)
        try:
            response = function(*args, **kwrds)
        except xmlrpc.Fault as exc:
            raise SAMPProxyError(exc.faultCode, exc.faultString)
        finally:
            self.__proxies.put(proxy)
        return response


class ServerProxyPool(object):
    """
    A thread-safe pool of `xmlrpc.ServerProxy` objects.
    """

    def __init__(self, size, proxy_class, *args, **keywords):

        self._proxies = queue.Queue(size)
        for i in range(size):
            self._proxies.put(proxy_class(*args, **keywords))

    def __getattr__(self, name):
        # magic method dispatcher
        return _ServerProxyPoolMethod(self._proxies, name)


class SAMPMsgReplierWrapper(object):
    """
    Function decorator that allows to automatically grab errors and returned
    maps (if any) from a function bound to a SAMP call (or notify).

    Parameters
    ----------
    cli : :class:`~astropy.vo.samp.integrated_client.SAMPIntegratedClient` or
          :class:`~astropy.vo.samp.client.SAMPClient`
        SAMP client instance. Decorator initialization, accepting the instance
        of the client that receives the call or notification.
    """

    def __init__(self, cli):
        self.cli = cli

    def __call__(self, f):

        def wrapped_f(*args):

            if get_num_args(f) == 5 or args[2] is None:  # notification

                f(*args)

            else:  # call

                try:
                    result = f(*args)
                    if result:
                        self.cli.hub.reply(self.cli.get_private_key(), args[2],
                                           {"samp.status": SAMP_STATUS_ERROR,
                                            "samp.result": result})
                except:<commit_msg>Fixed wrap in docstring<commit_after># Licensed under a 3-clause BSD style license - see LICENSE.rst
"""
Utility functions and classes
"""

from __future__ import (absolute_import, division, print_function,
                        unicode_literals)

import inspect
import traceback

from ...extern.six.moves import queue
from ...extern.six.moves.urllib.error import URLError
from ...extern.six.moves.urllib.request import urlopen
from ...extern.six.moves import xmlrpc_client as xmlrpc
from ...extern.six import StringIO
from ...config import ConfigurationItem

from .constants import SAMP_STATUS_ERROR
from .errors import SAMPProxyError

ALLOW_INTERNET = ConfigurationItem('use_internet', True,
                                   "Whether to allow astropy.vo.samp to use "
                                   "the internet, if available")


def internet_on():
    if not ALLOW_INTERNET():
        return False
    else:
        try:
            urlopen('http://google.com', timeout=1.)
            return True
        except:
            return False

__all__ = ["SAMPMsgReplierWrapper"]

__doctest_skip__ = ['.']


def getattr_recursive(variable, attribute):
    """
    Get attributes recursively.
    """
    if '.' in attribute:
        top, remaining = attribute.split('.', 1)
        return getattr_recursive(getattr(variable, top), remaining)
    else:
        return getattr(variable, attribute)


class _ServerProxyPoolMethod(object):

    # some magic to bind an XML-RPC method to an RPC server.
    # supports "nested" methods (e.g. examples.getStateName)

    def __init__(self, proxies, name):
        self.__proxies = proxies
        self.__name = name

    def __getattr__(self, name):
        return _ServerProxyPoolMethod(self.__proxies, "%s.%s" % (self.__name, name))

    def __call__(self, *args, **kwrds):
        proxy = self.__proxies.get()
        function = getattr_recursive(proxy, self.__name)
        try:
            response = function(*args, **kwrds)
        except xmlrpc.Fault as exc:
            raise SAMPProxyError(exc.faultCode, exc.faultString)
        finally:
            self.__proxies.put(proxy)
        return response


class ServerProxyPool(object):
    """
    A thread-safe pool of `xmlrpc.ServerProxy` objects.
    """

    def __init__(self, size, proxy_class, *args, **keywords):

        self._proxies = queue.Queue(size)
        for i in range(size):
            self._proxies.put(proxy_class(*args, **keywords))

    def __getattr__(self, name):
        # magic method dispatcher
        return _ServerProxyPoolMethod(self._proxies, name)


class SAMPMsgReplierWrapper(object):
    """
    Function decorator that allows to automatically grab errors and returned
    maps (if any) from a function bound to a SAMP call (or notify).

    Parameters
    ----------
    cli : :class:`~astropy.vo.samp.integrated_client.SAMPIntegratedClient` or :class:`~astropy.vo.samp.client.SAMPClient`
        SAMP client instance. Decorator initialization, accepting the instance
        of the client that receives the call or notification.
    """

    def __init__(self, cli):
        self.cli = cli

    def __call__(self, f):

        def wrapped_f(*args):

            if get_num_args(f) == 5 or args[2] is None:  # notification

                f(*args)

            else:  # call

                try:
                    result = f(*args)
                    if result:
                        self.cli.hub.reply(self.cli.get_private_key(), args[2],
                                           {"samp.status": SAMP_STATUS_ERROR,
                                            "samp.result": result})
                except:
{'fraction_non_alphanumeric': 0.09934497816593886, 'fraction_numerical': 0.0016375545851528383, 'mean_word_length': 2.2004366812227074, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 8}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '778612', 'n_tokens_mistral': 2121, 'n_tokens_neox': 2024, 'n_words': 564}