<commit_before>"""
Core agent classes.
"""
import os
import types
import signal
import sys
import time
import inspect
import multiprocessing
from uuid import uuid4
from datetime import datetime

import pickle
import errno
import zmq
import Pyro4
from Pyro4.errors import PyroError

from .common import address_to_host_port
from .common import format_exception
from .common import unbound_method
from .common import LogLevel
from .common import repeat
from .address import AgentAddress
from .address import AgentAddressKind
from .proxy import Proxy
from .proxy import NSProxy


class Agent():
    """
    A base agent class which is to be served by an AgentProcess.

    An AgentProcess runs a Pyro multiplexed server and serves one Agent
    object.

    Parameters
    ----------
    name : str, default is None
        Name of the Agent.
    host : str, default is None
        Host address where the agent will bind to. When not set, `'127.0.0.1'`
        (localhost) is used.

    Attributes
    ----------
    name : str
        Name of the agent.
    host : str
        Host address where the agent is binding to.
    socket : dict
        A dictionary in which the key is the address or the alias and the
        value is the actual socket.
    adddress : dict
        A dictionary in which the key is the alias and the value is the
        actual address.
    handler : dict
        A dictionary in which the key is the socket and the values are the
        handlers for each socket.
    poll_timeout : int
        Polling timeout, in milliseconds. After this timeout, if no message
        is received, the agent executes de `iddle()` method before going back
        to polling.
    keep_alive : bool
        When set to `True`, the agent will continue executing the main loop.
    running : bool
        Set to `True` if the agent is running (executing the main loop).
    """
    def __init__(self, name=None, host=None):
        self.name = name
        self.host = host
        if not self.host:
            self.host = '127.0.0.1'
        self.socket = {}
        self.address = {}
        self.handler = {}
        self._timer = {}
        self.poll_timeout = 1000
        self.keep_alive = True
        self.running = False
        # Kill parent AgentProcess
        self.kill_agent = False
        self._DEBUG = False

        self.context = zmq.Context()
        self.poller = zmq.Poller()

        # This in-process socket could, eventually, handle safe access to
        # memory from other threads (i.e. when using Pyro proxies).
        socket = self.context.socket(zmq.REP)
        address = 'inproc://loopback'
        socket.bind(address)
        self.register(socket, address, 'loopback', self.handle_loopback)

        self.on_init()

    def on_init(self):
        """
        This user-defined method is to be executed after initialization.
        """
        pass

    def handle_loopback(self, message):
        """
        Handle incoming messages in the loopback socket.
        """
        header, data = message
        if header == 'PING':
            return 'PONG'
        if header == 'STOP':
            self.log_info('Stopping...')
            self.keep_alive = False
            return 'OK'
        if header == 'CLOSE':
            self.log_info('Closing sockets...')
            self.close_sockets()
            return 'OK'
        if header == 'EXECUTE_METHOD':
            method, args, kwargs = data
            try:
                response = getattr(self, method)(*args, **kwargs)
            except Exception as error:
                message = 'Error executing `%s`! (%s)\n' % (method, error)
                message += format_exception()<commit_msg>Remove CLOSE header from loopback handler<commit_after>"""
Core agent classes.
"""
import os
import types
import signal
import sys
import time
import inspect
import multiprocessing
from uuid import uuid4
from datetime import datetime

import pickle
import errno
import zmq
import Pyro4
from Pyro4.errors import PyroError

from .common import address_to_host_port
from .common import format_exception
from .common import unbound_method
from .common import LogLevel
from .common import repeat
from .address import AgentAddress
from .address import AgentAddressKind
from .proxy import Proxy
from .proxy import NSProxy


class Agent():
    """
    A base agent class which is to be served by an AgentProcess.

    An AgentProcess runs a Pyro multiplexed server and serves one Agent
    object.

    Parameters
    ----------
    name : str, default is None
        Name of the Agent.
    host : str, default is None
        Host address where the agent will bind to. When not set, `'127.0.0.1'`
        (localhost) is used.

    Attributes
    ----------
    name : str
        Name of the agent.
    host : str
        Host address where the agent is binding to.
    socket : dict
        A dictionary in which the key is the address or the alias and the
        value is the actual socket.
    adddress : dict
        A dictionary in which the key is the alias and the value is the
        actual address.
    handler : dict
        A dictionary in which the key is the socket and the values are the
        handlers for each socket.
    poll_timeout : int
        Polling timeout, in milliseconds. After this timeout, if no message
        is received, the agent executes de `iddle()` method before going back
        to polling.
    keep_alive : bool
        When set to `True`, the agent will continue executing the main loop.
    running : bool
        Set to `True` if the agent is running (executing the main loop).
    """
    def __init__(self, name=None, host=None):
        self.name = name
        self.host = host
        if not self.host:
            self.host = '127.0.0.1'
        self.socket = {}
        self.address = {}
        self.handler = {}
        self._timer = {}
        self.poll_timeout = 1000
        self.keep_alive = True
        self.running = False
        # Kill parent AgentProcess
        self.kill_agent = False
        self._DEBUG = False

        self.context = zmq.Context()
        self.poller = zmq.Poller()

        # This in-process socket could, eventually, handle safe access to
        # memory from other threads (i.e. when using Pyro proxies).
        socket = self.context.socket(zmq.REP)
        address = 'inproc://loopback'
        socket.bind(address)
        self.register(socket, address, 'loopback', self.handle_loopback)

        self.on_init()

    def on_init(self):
        """
        This user-defined method is to be executed after initialization.
        """
        pass

    def handle_loopback(self, message):
        """
        Handle incoming messages in the loopback socket.
        """
        header, data = message
        if header == 'PING':
            return 'PONG'
        if header == 'STOP':
            self.log_info('Stopping...')
            self.keep_alive = False
            return 'OK'
        if header == 'EXECUTE_METHOD':
            method, args, kwargs = data
            try:
                response = getattr(self, method)(*args, **kwargs)
            except Exception as error:
                message = 'Error executing `%s`! (%s)\n' % (method, error)
                message += format_exception()
{'fraction_non_alphanumeric': 0.07921063450733178, 'fraction_numerical': 0.0052076195696861725, 'mean_word_length': 2.0586756077116513, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '3464455', 'n_tokens_mistral': 1965, 'n_tokens_neox': 1905, 'n_words': 780}