KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _listener method in ATX

Best Python code snippet using ATX

multiprocessing.py

Source:multiprocessing.py

1"""Version of multiprocessing that doesn't use fork.2Necessary because of grpcio bug.3"""4import base645import importlib6import logging7import multiprocessing.connection8import os9import pickle10import socket11from queue import Empty12import subprocess13import traceback14import sys15class Receiver(object):16    def __init__(self):17        self._listener = socket.socket(getattr(socket, 'AF_UNIX'))18        try:19            self._listener.setsockopt(socket.SOL_SOCKET,20                                      socket.SO_REUSEADDR, 1)21            address = multiprocessing.connection.arbitrary_address('AF_UNIX')22            self._listener.bind(address)23            self._listener.listen(1)24            self.address = self._listener.getsockname()25        except OSError:26            self._listener.close()27            raise28        self._sockets = [self._listener]29    def recv(self, timeout=None):30        done = False31        while not done:32            done = True33            for sock in multiprocessing.connection.wait(self._sockets,34                                                        timeout):35                if sock == self._listener:36                    s, addr = self._listener.accept()37                    conn = multiprocessing.connection.Connection(s.detach())38                    self._sockets.append(conn)39                    done = False40                else:41                    try:42                        msg = sock.recv()43                    except EOFError:44                        self._sockets.remove(sock)45                    else:46                        return msg47        raise Empty48    def send(self, msg):49        for sock in self._sockets:50            if sock != self._listener:51                sock.send(msg)52    def close(self):53        try:54            self._listener.close()55        finally:56            os.unlink(self.address)57def run_process(target, tag, msg_queue, **kwargs):58    """Call a Python function by name in a subprocess.59    :param target: Fully-qualified name of function to call.60    :param tag: Tag to add to logger to identify that process.61    :return: A `subprocess.Popen` object.62    """63    assert isinstance(msg_queue, Receiver)64    data = msg_queue.address, kwargs65    proc = subprocess.Popen(66        [67            sys.executable,68            '-c',69            'from d3m_ta2_nyu.multiprocessing import _invoke; _invoke(%r, %r)' % (70                tag, target71            ),72            base64.b64encode(pickle.dumps(data)),73        ],74        stdin=subprocess.PIPE, stderr=subprocess.PIPE)75    return proc76def _invoke(tag, target):77    """Invoked in the subprocess to setup logging and start the function.78    Arguments are read from ``sys.argv``.79    """80    data = pickle.loads(base64.b64decode(sys.argv[1]))81    address, kwargs = data82    tag = '{}-{}'.format(tag, os.getpid())83    logging.getLogger().handlers = []84    logging.basicConfig(85        level=logging.INFO,86        format="%(asctime)s:%(levelname)s:{}:%(name)s:%(message)s".format(tag),87        stream=sys.stdout)88    msg_queue = multiprocessing.connection.Client(address)89    module, function = target.rsplit('.', 1)90    module = importlib.import_module(module)91    function = getattr(module, function)92    try:93        function(msg_queue=msg_queue, **kwargs)94    except Exception:95        logging.exception("Uncaught exception in subprocess %s", tag)96        error = traceback.format_exc()97        sys.stderr.write(error)...

aria2_download.py

Source:aria2_download.py

1from aria2p import API2from aria2p.client import ClientException3from bot import aria24from bot.helper.ext_utils.bot_utils import *5from bot.helper.mirror_utils.status_utils.aria_download_status import AriaDownloadStatus6from bot.helper.telegram_helper.message_utils import *7from .download_helper import DownloadHelper8class AriaDownloadHelper(DownloadHelper):9    def __init__(self, listener):10        super().__init__()11        self.gid = None12        self._listener = listener13        self._resource_lock = threading.Lock()14    def __onDownloadStarted(self, api, gid):15        with self._resource_lock:16            if self.gid == gid:17                download = api.get_download(gid)18                self.name = download.name19                update_all_messages()20    def __onDownloadComplete(self, api: API, gid):21        with self._resource_lock:22            if self.gid == gid:23                if api.get_download(gid).followed_by_ids:24                    self.gid = api.get_download(gid).followed_by_ids[0]25                    with download_dict_lock:26                        download_dict[self._listener.uid] = AriaDownloadStatus(self.gid, self._listener)27                        download_dict[self._listener.uid].is_torrent =True28                    update_all_messages()29                    LOGGER.info(f'Changed gid from {gid} to {self.gid}')30                else:31                    self._listener.onDownloadComplete()32    def __onDownloadPause(self, api, gid):33        if self.gid == gid:34            LOGGER.info("Called onDownloadPause")35            download = api.get_download(gid)36            error = download.error_message37            self._listener.onDownloadError(error)38    def __onDownloadStopped(self, api, gid):39        if self.gid == gid:40            LOGGER.info("Called on_download_stop")41            download = api.get_download(gid)42            error = download.error_message43            self._listener.onDownloadError(error)44    def __onDownloadError(self, api, gid):45        with self._resource_lock:46            if self.gid == gid:47                download = api.get_download(gid)48                error = download.error_message49                LOGGER.info(f"Download Error: {error}")50                self._listener.onDownloadError(error)51    def add_download(self, link: str, path):52        try:53            if is_magnet(link):54                download = aria2.add_magnet(link, {'dir': path})55            else:56                download = aria2.add_uris([link], {'dir': path})57        except ClientException as err:58            self._listener.onDownloadError(err.message)59            return60        self.gid = download.gid61        with download_dict_lock:62            download_dict[self._listener.uid] = AriaDownloadStatus(self.gid, self._listener)63        if download.error_message:64            self._listener.onDownloadError(download.error_message)65            return66        LOGGER.info(f"Started: {self.gid} DIR:{download.dir} ")67        aria2.listen_to_notifications(threaded=True, on_download_start=self.__onDownloadStarted,68                                      on_download_error=self.__onDownloadError,69                                      on_download_pause=self.__onDownloadPause,70                                      on_download_stop=self.__onDownloadStopped,...

mouse_listener.py

Source:mouse_listener.py

1from pynput.mouse import Listener2# TODO this is too complicated for something so simple, remove this class3class MouseListener:4    def __init__(self, on_click_function):5        self.on_click = on_click_function6    def start_listening(self):7        # non-blocking8        # a Listener can only be started once, so I must create a new one every time9        if hasattr(self, "_listener") and self._listener.running is True:10            self._listener.stop()11        self._listener = Listener(on_click=self.on_click)12        self._listener.start()13    def stop_listening(self):14        if hasattr(self, "_listener") and self._listener.running is True:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.