Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use poll_events method in pyatom

Best Python code snippet using pyatom_python

screens.py

Source:screens.py

...6def instructions():7    while True:8        next_frame()9        draw_image(pic_instructions, position=(0, 0), anchor=(0, 0))10        for event in poll_events():11            if type(event) is CloseEvent:12                terminate()13            if type(event) is MouseDownEvent:14                return15            if type(event) is KeyDownEvent:16                return17def main_menu():18    while True:19        next_frame()20        draw_image(pic_init_screen, position=(0, 0), anchor=(0, 0))21        for event in poll_events():22            if type(event) is CloseEvent:23                terminate()24            if type(event) is MouseDownEvent:25                if event.button == 'LEFT':26                    if 204 <= event.x <= 598:27                        y = 599 - event.y28                        if 343 <= y <= 430:29                            return MENU_NEW_GAME30                        if 463 <= y <= 549:31                            return MENU_INSTRUCTIONS32def loss(state):33    while True:34        next_frame()35        if state.saved_clone is None:36            draw_image(pic_loss, position=(0, 0), anchor=(0, 0) )37        else:38            draw_image(pic_loss_clone, position=(0, 0), anchor=(0, 0) )39        for event in poll_events():40            if type(event) is CloseEvent:41                terminate()42            if type(event) is KeyDownEvent:43                return44def pause_screen(state):45    while True:46        next_frame()47        draw_image(pic_pause, position=(0, 0), anchor=(0, 0) )48        for event in poll_events():49            if type(event) is CloseEvent:50                terminate()51            if type(event) is KeyDownEvent:52                return53def report(state):54    while True:55        next_frame()56        draw_image(pic_report, position=(0, 0), anchor=(0, 0) )57        offset_y, step_y = 270, 2658        59        seconds = int(state.game_duration)60        hours = seconds // 360061        seconds %= 360062        minutes = seconds // 6063        seconds %= 6064        duration = 'CelkovÃ© trvanie hry: {}:{:02d}:{:02d}'.format(hours,minutes,seconds)65        draw_text(duration, 'Arial', 16, position=(50,offset_y), color=COLOR_BLACK)66        offset_y -= step_y67        disinfections = 'PoÄet dezinfekciÃ rÃºk: {}'.format(state.disinfections)68        draw_text(disinfections, 'Arial', 16, position=(50,offset_y), color=COLOR_BLACK)69        offset_y -= step_y70        if state.stage >= STAGE_CLONE:71            clonings = 'PoÄet potrebnÃ½ch naklonovanÃ: {}'.format(state.clonings)72            draw_text(clonings, 'Arial', 16, position=(50,offset_y), color=COLOR_BLACK)73            offset_y -= step_y74        else:75            clonings = 'PoÄet (eÅ¡te nevieÅ¡ Äoho): {}'.format(state.clonings)76            draw_text(clonings, 'Arial', 16, position=(50,offset_y), color=COLOR_BLACK)77            offset_y -= step_y78        offset_y -= step_y79        draw_text('BONUSY:', 'Arial', 16, position=(50,offset_y), color=COLOR_BLACK)80        offset_y -= step_y81        82        draw_text('Biela ÄokolÃ¡da: ' + ('zÃskanÃ¡' if state.saw_fake_easter_egg else 'nezÃskanÃ¡'), 'Arial', 16, position=(50,offset_y), color=COLOR_BLACK)83        offset_y -= step_y84        85        draw_text('ÄokolÃ¡da: ' + ('zÃskanÃ¡' if state.should_win_pro else 'nezÃskanÃ¡'), 'Arial', 16, position=(50,offset_y), color=COLOR_BLACK)86        offset_y -= step_y87        88        for event in poll_events():89            if type(event) is CloseEvent:90                terminate()91            if type(event) is KeyDownEvent:92                return93def win(state):94    while True:95        next_frame()96        draw_image(pic_win, position=(0, 0), anchor=(0, 0) )97        for event in poll_events():98            if type(event) is CloseEvent:99                terminate()100            if type(event) is KeyDownEvent:101                win_pro(state)102                return103def win_fake(state):104    while True:105        next_frame()106        draw_image(pic_win_fake, position=(0, 0), anchor=(0, 0) )107        for event in poll_events():108            if type(event) is CloseEvent:109                terminate()110            if type(event) is KeyDownEvent:111                return112def win_pro(state):113    if state.should_win_pro:114        verification_string = '_'.join( str(state.item_counts[s]) for s in FINGERPRINT)115        verification_string = hashlib.sha256(verification_string.encode()).hexdigest()116        while True:117            next_frame()118            draw_image(pic_win_pro, position=(0, 0), anchor=(0, 0) )119            draw_text('verifikaÄnÃ½ reÅ¥azec:', 'Arial', 16, position=(10,30), color=(0,0,0,1) )120            draw_text(verification_string, 'Arial', 12, position=(10,10), color=(0,0,0,1) )121            for event in poll_events():122                if type(event) is CloseEvent:123                    terminate()124                if type(event) is KeyDownEvent:...

cog_alert.py

Source:cog_alert.py

...15                         data_def_constructor=Alert)16        self.bot = bot17        self.poll_events.start()18    @tasks.loop(seconds=conf.ALERT_POLL_INTERVAL)19    async def poll_events(self):20        if await self.data.check_next_event(self.bot):21            # Save if an alert was sent22            self.save()23        if self.data.next_event is None:24            log('No pending Alerts - Polling Stopping')25            self.poll_events.cancel()26    @poll_events.before_loop27    async def before_poll_alerts(self):28        await self.bot.wait_until_ready()29        log('Poll Alerts Loop Started')30    ##########################################################################31    # BASE GROUP32    @commands.group(**conf.BASE_GROUP)33    async def base(self, ctx):...

poll_patcher.py

Source:poll_patcher.py

1"""A patch to the select.poll object."""2from .base_patcher import BasePatcher3import select as select_lib4class MockPollObject(object):5    """A mock poll object."""6    def __init__(self, clock, event_pool):7        """Initialize the object."""8        self.clock = clock9        self.event_pool = event_pool10        self.poll_events = {}11    def register(self, fd, eventmask=None):12        """Register a file descriptor with the fake polling object."""13        if eventmask is None:14            eventmask = (select_lib.POLLIN |15                         select_lib.POLLOUT |16                         select_lib.POLLPRI)17        self.poll_events[fd] = eventmask18    def modify(self, fd, eventmask):19        """Modify an already registered fd's event mask."""20        if fd not in self.poll_events:21            raise IOError()22        self.poll_events[fd] = eventmask23    def unregister(self, fd):24        """Remove a file descriptor tracked by the fake polling object."""25        if fd not in self.poll_events:26            raise KeyError(fd)27        self.poll_events.pop(fd)28    def poll(self, timeout=None):29        """Poll the set of registered file descriptors.30        `timeout` is a value in milliseconds.31        """32        timestamp, fd_events = \33            self._get_earliest_events_for_waited_fds(timeout)34        if timestamp == float('inf'):35            raise ValueError('No relevant future events were set for infinite '36                             'timout')37        for fd, events in fd_events:38            self.event_pool.remove_events_from_fds(39                timestamp,40                [(fd, event) for event in events])41        self.clock.time = timestamp42        def _crunch_events(_event_set):43            out = 044            for _event in _event_set:45                out |= _event46            return out47        return [(fd, _crunch_events(events)) for fd, events in fd_events]48    def _get_earliest_events_for_waited_fds(self, timeout=None):49        """Return a list of [(fd, set(events)), ...]."""50        if timeout is None or timeout < 0:51            timeout = float('inf')52        else:53            timeout = timeout / 1000.054        timeout_timestamp = self.clock.time + timeout55        def _is_relevant_fd_event(fd, evt):56            return fd in self.poll_events and self.poll_events[fd] & evt57        # fd_events is a list of [(fd, set(events)), ...].58        ts, fd_events = self.event_pool.get_next_event(_is_relevant_fd_event)59        if ts is None or ts > timeout_timestamp:60            return timeout_timestamp, []61        else:62            return ts, fd_events63class PollPatcher(BasePatcher):64    """Patcher for select.poll."""65    def __init__(self, *args, **kwargs):66        """Create the patch."""67        super(PollPatcher, self).__init__(*args, **kwargs)68    def get_patched_module(self):69        """Return the actual module obect to be patched."""70        return select_lib71    def get_patch_actions(self):72        """Return generator containing all patches to do."""73        return [('poll', select_lib.poll, self._mock_poll)]74    def _mock_poll(self):...

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.