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"""Asyncio based urwid EventLoop implementation."""23 24from __future__ import annotations25 26import asyncio27import functools28import logging29import sys30import typing31 32from .abstract_loop import EventLoop, ExitMainLoop33 34if typing.TYPE_CHECKING:35    from collections.abc import Callable36    from concurrent.futures import Executor37 38    from typing_extensions import ParamSpec39 40    _Spec = ParamSpec("_Spec")41    _T = typing.TypeVar("_T")42 43__all__ = ("AsyncioEventLoop",)44 45 46class AsyncioEventLoop(EventLoop):47    """48    Event loop based on the standard library ``asyncio`` module.49 50    .. warning::51        Under Windows, AsyncioEventLoop globally enforces WindowsSelectorEventLoopPolicy52        as a side-effect of creating a class instance.53        Original event loop policy is restored in destructor method.54 55    .. note::56        If you make any changes to the urwid state outside of it57        handling input or responding to alarms (for example, from asyncio.Task58        running in background), and wish the screen to be59        redrawn, you must call :meth:`MainLoop.draw_screen` method of the60        main loop manually.61 62        A good way to do this:63            asyncio.get_event_loop().call_soon(main_loop.draw_screen)64    """65 66    def __init__(self, *, loop: asyncio.AbstractEventLoop | None = None, **kwargs) -> None:67        super().__init__()68        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)69 70        if sys.version_info[:2] < (3, 11):71            self._event_loop_policy_altered: bool = False72            self._original_event_loop_policy: asyncio.AbstractEventLoopPolicy | None = None73 74            if loop:75                self._loop: asyncio.AbstractEventLoop = loop76            else:77                self._original_event_loop_policy = asyncio.get_event_loop_policy()78                if sys.platform == "win32" and not isinstance(79                    self._original_event_loop_policy, asyncio.WindowsSelectorEventLoopPolicy80                ):81                    self.logger.debug("Set WindowsSelectorEventLoopPolicy as asyncio event loop policy")82                    asyncio.set_event_loop_policy(83                        asyncio.WindowsSelectorEventLoopPolicy()  # pylint: disable=deprecated-class84                    )85                    self._event_loop_policy_altered = True86                else:87                    self._event_loop_policy_altered = False88 89                self._loop = asyncio.get_event_loop()90 91        else:92            self._runner: asyncio.Runner | None = None93 94            if loop:95                self._loop: asyncio.AbstractEventLoop = loop96            else:97                try:98                    self._loop = asyncio.get_running_loop()99                except RuntimeError:100                    self._runner = asyncio.Runner(loop_factory=asyncio.SelectorEventLoop)101                    self._loop = self._runner.get_loop()102 103        self._exc: BaseException | None = None104 105        self._idle_asyncio_handle: asyncio.TimerHandle | None = None106        self._idle_handle: int = 0107        self._idle_callbacks: dict[int, Callable[[], typing.Any]] = {}108 109    def __del__(self) -> None:110        if sys.version_info[:2] < (3, 11):111            if self._event_loop_policy_altered:112                asyncio.set_event_loop_policy(self._original_event_loop_policy)  # Restore default event loop policy113        elif self._runner is not None:114            self._runner.close()115 116    def _also_call_idle(self, callback: Callable[_Spec, _T]) -> Callable[_Spec, _T]:117        """118        Wrap the callback to also call _entering_idle.119        """120 121        @functools.wraps(callback)122        def wrapper(*args: _Spec.args, **kwargs: _Spec.kwargs) -> _T:123            if not self._idle_asyncio_handle:124                self._idle_asyncio_handle = self._loop.call_later(0, self._entering_idle)125            return callback(*args, **kwargs)126 127        return wrapper128 129    def _entering_idle(self) -> None:130        """131        Call all the registered idle callbacks.132        """133        try:134            for callback in self._idle_callbacks.values():135                callback()136        finally:137            self._idle_asyncio_handle = None138 139    def run_in_executor(140        self,141        executor: Executor | None,142        func: Callable[_Spec, _T],143        *args: _Spec.args,144        **kwargs: _Spec.kwargs,145    ) -> asyncio.Future[_T]:146        """Run callable in executor.147 148        :param executor: Executor to use for running the function. Default asyncio executor is used if None.149        :type executor: concurrent.futures.Executor | None150        :param func: function to call151        :type func: Callable152        :param args: arguments to function (positional only)153        :type args: object154        :param kwargs: keyword arguments to function (keyword only)155        :type kwargs: object156        :return: future object for the function call outcome.157        :rtype: asyncio.Future158        """159        return self._loop.run_in_executor(executor, functools.partial(func, *args, **kwargs))160 161    def alarm(self, seconds: float, callback: Callable[[], typing.Any]) -> asyncio.TimerHandle:162        """163        Call callback() a given time from now.  No parameters are164        passed to callback.165 166        Returns a handle that may be passed to remove_alarm()167 168        seconds -- time in seconds to wait before calling callback169        callback -- function to call from event loop170        """171        return self._loop.call_later(seconds, self._also_call_idle(callback))172 173    def remove_alarm(self, handle) -> bool:174        """175        Remove an alarm.176 177        Returns True if the alarm exists, False otherwise178        """179        existed = not handle.cancelled()180        handle.cancel()181        return existed182 183    def watch_file(self, fd: int, callback: Callable[[], typing.Any]) -> int:184        """185        Call callback() when fd has some data to read.  No parameters186        are passed to callback.187 188        Returns a handle that may be passed to remove_watch_file()189 190        fd -- file descriptor to watch for input191        callback -- function to call when input is available192        """193        self._loop.add_reader(fd, self._also_call_idle(callback))194        return fd195 196    def remove_watch_file(self, handle: int) -> bool:197        """198        Remove an input file.199 200        Returns True if the input file exists, False otherwise201        """202        return self._loop.remove_reader(handle)203 204    def enter_idle(self, callback: Callable[[], typing.Any]) -> int:205        """206        Add a callback for entering idle.207 208        Returns a handle that may be passed to remove_enter_idle()209        """210        # XXX there's no such thing as "idle" in most event loops; this fakes211        # it by adding extra callback to the timer and file watch callbacks.212        self._idle_handle += 1213        self._idle_callbacks[self._idle_handle] = callback214        return self._idle_handle215 216    def remove_enter_idle(self, handle: int) -> bool:217        """218        Remove an idle callback.219 220        Returns True if the handle was removed.221        """222        try:223            del self._idle_callbacks[handle]224        except KeyError:225            return False226        return True227 228    def _exception_handler(self, loop: asyncio.AbstractEventLoop, context):229        if exc := context.get("exception"):230            loop.stop()231 232            if self._idle_asyncio_handle:233                # clean it up to prevent old callbacks234                # from messing things up if loop is restarted235                self._idle_asyncio_handle.cancel()236                self._idle_asyncio_handle = None237 238            if not isinstance(exc, ExitMainLoop):239                # Store the exc_info so we can re-raise after the loop stops240                self._exc = exc241        else:242            loop.default_exception_handler(context)243 244    def run(self) -> None:245        """Start the event loop.246 247        Exit the loop when any callback raises an exception.248        If ExitMainLoop is raised, exit cleanly.249        """250        self._loop.set_exception_handler(self._exception_handler)251        self._loop.run_forever()252        if self._exc:253            exc = self._exc254            self._exc = None255            raise exc.with_traceback(exc.__traceback__)256