codekingpro/portable-devtools

1# Urwid main loop code2#    Copyright (C) 2004-2012  Ian Ward3#    Copyright (C) 2008 Walter Mundt4#    Copyright (C) 2009 Andrew Psaltis5#6#    This library is free software; you can redistribute it and/or7#    modify it under the terms of the GNU Lesser General Public8#    License as published by the Free Software Foundation; either9#    version 2.1 of the License, or (at your option) any later version.10#11#    This library is distributed in the hope that it will be useful,12#    but WITHOUT ANY WARRANTY; without even the implied warranty of13#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU14#    Lesser General Public License for more details.15#16#    You should have received a copy of the GNU Lesser General Public17#    License along with this library; if not, write to the Free Software18#    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA19#20# Urwid web site: https://urwid.org/21 22"""Select based urwid EventLoop implementation."""23 24from __future__ import annotations25 26import contextlib27import heapq28import logging29import selectors30import time31import typing32from contextlib import suppress33from itertools import count34 35from .abstract_loop import EventLoop, ExitMainLoop36 37if typing.TYPE_CHECKING:38    from collections.abc import Callable, Iterator39    from concurrent.futures import Executor, Future40 41    from typing_extensions import Literal, ParamSpec42 43    _T = typing.TypeVar("_T")44    _Spec = ParamSpec("_Spec")45 46__all__ = ("SelectEventLoop",)47 48 49class SelectEventLoop(EventLoop):50    """51    Event loop based on :func:`selectors.DefaultSelector.select`52    """53 54    def __init__(self) -> None:55        super().__init__()56        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)57        self._alarms: list[tuple[float, int, Callable[[], typing.Any]]] = []58        self._watch_files: dict[int, Callable[[], typing.Any]] = {}59        self._idle_handle: int = 060        self._idle_callbacks: dict[int, Callable[[], typing.Any]] = {}61        self._tie_break: Iterator[int] = count()62        self._did_something: bool = False63 64    def run_in_executor(65        self,66        executor: Executor,67        func: Callable[_Spec, _T],68        *args: _Spec.args,69        **kwargs: _Spec.kwargs,70    ) -> Future[_T]:71        """Run callable in executor.72 73        :param executor: Executor to use for running the function74        :type executor: concurrent.futures.Executor75        :param func: function to call76        :type func: Callable77        :param args: positional arguments to function78        :type args: object79        :param kwargs: keyword arguments to function80        :type kwargs: object81        :return: future object for the function call outcome.82        :rtype: concurrent.futures.Future83        """84        return executor.submit(func, *args, **kwargs)85 86    def alarm(87        self,88        seconds: float,89        callback: Callable[[], typing.Any],90    ) -> tuple[float, int, Callable[[], typing.Any]]:91        """92        Call callback() a given time from now.  No parameters are93        passed to callback.94 95        Returns a handle that may be passed to remove_alarm()96 97        seconds -- floating point time to wait before calling callback98        callback -- function to call from event loop99        """100        tm = time.time() + seconds101        handle = (tm, next(self._tie_break), callback)102        heapq.heappush(self._alarms, handle)103        return handle104 105    def remove_alarm(self, handle: tuple[float, int, Callable[[], typing.Any]]) -> bool:106        """107        Remove an alarm.108 109        Returns True if the alarm exists, False otherwise110        """111        try:112            self._alarms.remove(handle)113            heapq.heapify(self._alarms)114 115        except ValueError:116            return False117 118        return True119 120    def watch_file(self, fd: int, callback: Callable[[], typing.Any]) -> int:121        """122        Call callback() when fd has some data to read.  No parameters123        are passed to callback.124 125        Returns a handle that may be passed to remove_watch_file()126 127        fd -- file descriptor to watch for input128        callback -- function to call when input is available129        """130        self._watch_files[fd] = callback131        return fd132 133    def remove_watch_file(self, handle: int) -> bool:134        """135        Remove an input file.136 137        Returns True if the input file exists, False otherwise138        """139        if handle in self._watch_files:140            del self._watch_files[handle]141            return True142        return False143 144    def enter_idle(self, callback: Callable[[], typing.Any]) -> int:145        """146        Add a callback for entering idle.147 148        Returns a handle that may be passed to remove_idle()149        """150        self._idle_handle += 1151        self._idle_callbacks[self._idle_handle] = callback152        return self._idle_handle153 154    def remove_enter_idle(self, handle: int) -> bool:155        """156        Remove an idle callback.157 158        Returns True if the handle was removed.159        """160        try:161            del self._idle_callbacks[handle]162        except KeyError:163            return False164        return True165 166    def _entering_idle(self) -> None:167        """168        Call all the registered idle callbacks.169        """170        for callback in self._idle_callbacks.values():171            callback()172 173    def run(self) -> None:174        """175        Start the event loop.  Exit the loop when any callback raises176        an exception.  If ExitMainLoop is raised, exit cleanly.177        """178        with contextlib.suppress(ExitMainLoop):179            self._did_something = True180            while True:181                with suppress(InterruptedError):182                    self._loop()183 184    def _loop(self) -> None:185        """186        A single iteration of the event loop187        """188        tm: float | Literal["idle"] | None = None189 190        with selectors.DefaultSelector() as selector:191            for fd, callback in self._watch_files.items():192                selector.register(fd, selectors.EVENT_READ, callback)193 194            if self._alarms or self._did_something:195                timeout = 0.0196 197                if self._alarms:198                    timeout_ = self._alarms[0][0]199                    tm = timeout_200                    timeout = max(timeout, timeout_ - time.time())201 202                if self._did_something and (not self._alarms or (self._alarms and timeout > 0)):203                    timeout = 0.0204                    tm = "idle"205 206                self.logger.debug(f"Waiting for input: timeout={timeout!r}")207                ready = [event for event, _ in selector.select(timeout)]208 209            elif self._watch_files:210                self.logger.debug("Waiting for input: timeout")211                ready = [event for event, _ in selector.select()]212            else:213                ready = []214 215        if not ready:216            if tm == "idle":217                self.logger.debug("No input, entering IDLE")218                self._entering_idle()219                self._did_something = False220            elif tm is not None:221                # must have been a timeout222                tm, _tie_break, alarm_callback = heapq.heappop(self._alarms)223                self.logger.debug(f"No input in timeout, calling scheduled {alarm_callback!r}")224                alarm_callback()225                self._did_something = True226 227        self.logger.debug("Processing input")228        for record in ready:229            record.data()230            self._did_something = True231