<commit_before>import asyncio
import json
import logging
import signal
import websockets

from netdumplings.exceptions import InvalidDumpling
from netdumplings.shared import (
    validate_dumpling, ND_CLOSE_MSGS, DEFAULT_SHIFTY_HOST,
    DEFAULT_SHIFTY_OUT_PORT
)


class DumplingEater:
    """
    Base helper class for Python-based dumpling eaters.

    Connects to `nd-shifty` and listens for any dumplings made by the provided
    list of chefs (or all chefs if ``chefs`` is ``None``).  Can be given
    callables for any of the following events:

    ``on_connect(websocket_uri, websocket_obj)``
        invoked when the connection to `nd-shifty` is made.

    ``on_dumpling(dumpling)``
        invoked whenever a dumpling is emitted from `nd-shifty`.

    ``on_connection_lost(e)``
        invoked when the connection to `nd-shifty` is closed.

    :param name: Name of the dumpling eater. Is ideally unique per eater.
    :param shifty: Address where `nd-shifty` is sending dumplings from.
    :param chefs: List of chef names whose dumplings this eater wants to
        receive. ``None`` means get all chefs' dumplings.
    :param on_connect: Called when connection to shifty is made. Is passed two
        parameters: the shifty websocket URI (string) and websocket
        object (:class:`websockets.client.WebSocketClientProtocol`).
    :param on_dumpling: Called whenever a dumpling is received. Is passed the
        dumpling as a Python dict.
    :param on_connection_lost: Called when connection to `nd-shifty` is lost.
        Is passed the associated exception object.
    """
    def __init__(
            self,
            name='nameless_eater',
            shifty='{}:{}'.format(
                DEFAULT_SHIFTY_HOST, DEFAULT_SHIFTY_OUT_PORT
            ),
            *,
            chefs=None,
            on_connect=None,
            on_dumpling=None,
            on_connection_lost=None
    ):
        self.name = name
        self.chefs = chefs
        self.shifty_uri = "ws://{0}".format(shifty)

        # Configure handlers. If we're not provided with handlers then we
        # fall back on the default handlers or the handlers provided by a
        # subclass.
        self.on_connect = (
            on_connect if on_connect is not None else self.on_connect
        )
        self.on_dumpling = (
            on_dumpling if on_dumpling is not None else self.on_dumpling
        )
        self.on_connection_lost = (
            on_connection_lost if on_connection_lost is not None
            else self.on_connection_lost
        )

        self._was_connected = False
        self._logger_name = "{}.{}".format(__name__, self.name)
        self.logger = logging.getLogger(self._logger_name)

    async def _grab_dumplings(self, dumpling_count=None):
        """
        Receives all dumplings from shifty and looks for any dumplings which
        were created by the chef(s) we're interested in.  All those dumplings
        are then passed to the on_dumpling handler.

        :param dumpling_count: Number of dumplings to eat.  None means eat
            forever.
        """
        dumplings_eaten = 0

        websocket = await websockets.client.connect(self.shifty_uri)
        self._was_connected = True

        self.logger.info("{0}: Connected to shifty at {1}".format(
            self.name, self.shifty_uri))

        try:
            # Announce ourselves to shifty.
            await websocket.send(json.dumps({'eater_name': self.name}))

            if self.on_connect:
                await self.on_connect(self.shifty_uri, websocket)

            while True:
                # Eat a single dumpling.
                dumpling_json = await websocket.recv()

                # Validate the dumpling.  Note that invalid dumplings will
                # probably be stripped out by shifty (the DumplingHub).
                try:
                    dumpling = validate_dumpling(dumpling_json)
                except InvalidDumpling as e:
                    self.logger.error("{0}: Invalid dumpling: {1}".format(
                        self.name, e))
                    continue

                dumpling_chef = dumpling['metadata']['chef']

                self.logger.debug("{0}: Received dumpling from {1}".format(
                    self.name, dumpling_chef))

                # Call the on_dumpling handler if this dumpling is from a
                # chef that we've registered interest in.
                if self.chefs is None or dumpling_chef in self.chefs:
                    self.logger.debug(
                        "{0}: Calling dumpling handler {1}".format(
                            self.name, self.on_dumpling))

                    dumplings_eaten += 1
                    await self.on_dumpling(dumpling)

                # Stop eating dumplings if we've reached our threshold.
                if dumpling_count is not None and \
                        dumplings_eaten >= dumpling_count:
                    await websocket.close(*ND_CLOSE_MSGS['eater_full'])
                    break
        except asyncio.CancelledError:
            self.logger.warning(
                "{0}: Connection to shifty cancelled; closing...".format(
                    self.name))

            try:
                await websocket.close(*ND_CLOSE_MSGS['conn_cancelled'])
            except websockets.exceptions.InvalidState:
                pass
        except websockets.exceptions.ConnectionClosed as e:
            self.logger.warning("{0}: Lost connection to shifty: {1}".format(
                self.name, e))

            if self.on_connection_lost:
                await self.on_connection_lost(e)

    @staticmethod
    def _interrupt_handler():
        """
        Signal handler.  Cancels all running async tasks.
        """
        tasks = asyncio.Task.all_tasks()
        for task in tasks:
            task.cancel()

    def run(self, dumpling_count=None):
        """
        Run the dumpling eater.

        :param dumpling_count: Number of dumplings to eat.  ``None`` means eat
            forever.
        """
        self.logger.info("{0}: Running dumpling eater".format(self.name))

        # Check that we have a valid callable on_dumpling handler.
        if not hasattr(self.on_dumpling, '__call__'):
            self.logger.error(
                "{0}: on_dumpling handler is not callable".format(self.name))<commit_msg>Replace hasattr __call__ check with callable()<commit_after>import asyncio
import json
import logging
import signal
import websockets

from netdumplings.exceptions import InvalidDumpling
from netdumplings.shared import (
    validate_dumpling, ND_CLOSE_MSGS, DEFAULT_SHIFTY_HOST,
    DEFAULT_SHIFTY_OUT_PORT
)


class DumplingEater:
    """
    Base helper class for Python-based dumpling eaters.

    Connects to `nd-shifty` and listens for any dumplings made by the provided
    list of chefs (or all chefs if ``chefs`` is ``None``).  Can be given
    callables for any of the following events:

    ``on_connect(websocket_uri, websocket_obj)``
        invoked when the connection to `nd-shifty` is made.

    ``on_dumpling(dumpling)``
        invoked whenever a dumpling is emitted from `nd-shifty`.

    ``on_connection_lost(e)``
        invoked when the connection to `nd-shifty` is closed.

    :param name: Name of the dumpling eater. Is ideally unique per eater.
    :param shifty: Address where `nd-shifty` is sending dumplings from.
    :param chefs: List of chef names whose dumplings this eater wants to
        receive. ``None`` means get all chefs' dumplings.
    :param on_connect: Called when connection to shifty is made. Is passed two
        parameters: the shifty websocket URI (string) and websocket
        object (:class:`websockets.client.WebSocketClientProtocol`).
    :param on_dumpling: Called whenever a dumpling is received. Is passed the
        dumpling as a Python dict.
    :param on_connection_lost: Called when connection to `nd-shifty` is lost.
        Is passed the associated exception object.
    """
    def __init__(
            self,
            name='nameless_eater',
            shifty='{}:{}'.format(
                DEFAULT_SHIFTY_HOST, DEFAULT_SHIFTY_OUT_PORT
            ),
            *,
            chefs=None,
            on_connect=None,
            on_dumpling=None,
            on_connection_lost=None
    ):
        self.name = name
        self.chefs = chefs
        self.shifty_uri = "ws://{0}".format(shifty)

        # Configure handlers. If we're not provided with handlers then we
        # fall back on the default handlers or the handlers provided by a
        # subclass.
        self.on_connect = (
            on_connect if on_connect is not None else self.on_connect
        )
        self.on_dumpling = (
            on_dumpling if on_dumpling is not None else self.on_dumpling
        )
        self.on_connection_lost = (
            on_connection_lost if on_connection_lost is not None
            else self.on_connection_lost
        )

        self._was_connected = False
        self._logger_name = "{}.{}".format(__name__, self.name)
        self.logger = logging.getLogger(self._logger_name)

    async def _grab_dumplings(self, dumpling_count=None):
        """
        Receives all dumplings from shifty and looks for any dumplings which
        were created by the chef(s) we're interested in.  All those dumplings
        are then passed to the on_dumpling handler.

        :param dumpling_count: Number of dumplings to eat.  None means eat
            forever.
        """
        dumplings_eaten = 0

        websocket = await websockets.client.connect(self.shifty_uri)
        self._was_connected = True

        self.logger.info("{0}: Connected to shifty at {1}".format(
            self.name, self.shifty_uri))

        try:
            # Announce ourselves to shifty.
            await websocket.send(json.dumps({'eater_name': self.name}))

            if self.on_connect:
                await self.on_connect(self.shifty_uri, websocket)

            while True:
                # Eat a single dumpling.
                dumpling_json = await websocket.recv()

                # Validate the dumpling.  Note that invalid dumplings will
                # probably be stripped out by shifty (the DumplingHub).
                try:
                    dumpling = validate_dumpling(dumpling_json)
                except InvalidDumpling as e:
                    self.logger.error("{0}: Invalid dumpling: {1}".format(
                        self.name, e))
                    continue

                dumpling_chef = dumpling['metadata']['chef']

                self.logger.debug("{0}: Received dumpling from {1}".format(
                    self.name, dumpling_chef))

                # Call the on_dumpling handler if this dumpling is from a
                # chef that we've registered interest in.
                if self.chefs is None or dumpling_chef in self.chefs:
                    self.logger.debug(
                        "{0}: Calling dumpling handler {1}".format(
                            self.name, self.on_dumpling))

                    dumplings_eaten += 1
                    await self.on_dumpling(dumpling)

                # Stop eating dumplings if we've reached our threshold.
                if dumpling_count is not None and \
                        dumplings_eaten >= dumpling_count:
                    await websocket.close(*ND_CLOSE_MSGS['eater_full'])
                    break
        except asyncio.CancelledError:
            self.logger.warning(
                "{0}: Connection to shifty cancelled; closing...".format(
                    self.name))

            try:
                await websocket.close(*ND_CLOSE_MSGS['conn_cancelled'])
            except websockets.exceptions.InvalidState:
                pass
        except websockets.exceptions.ConnectionClosed as e:
            self.logger.warning("{0}: Lost connection to shifty: {1}".format(
                self.name, e))

            if self.on_connection_lost:
                await self.on_connection_lost(e)

    @staticmethod
    def _interrupt_handler():
        """
        Signal handler.  Cancels all running async tasks.
        """
        tasks = asyncio.Task.all_tasks()
        for task in tasks:
            task.cancel()

    def run(self, dumpling_count=None):
        """
        Run the dumpling eater.

        :param dumpling_count: Number of dumplings to eat.  ``None`` means eat
            forever.
        """
        self.logger.info("{0}: Running dumpling eater".format(self.name))

        if not callable(self.on_dumpling):
            self.logger.error(
                "{0}: on_dumpling handler is not callable".format(self.name))
{'fraction_non_alphanumeric': 0.07868518082686123, 'fraction_numerical': 0.0025103946026516044, 'mean_word_length': 1.94275161588181, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, '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': '2242248', 'n_tokens_mistral': 3770, 'n_tokens_neox': 3497, 'n_words': 1040}