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"""Tornado IOLoop based urwid EventLoop implementation.23 24Tornado library is required.25"""26 27from __future__ import annotations28 29import asyncio30import functools31import logging32import typing33from contextlib import suppress34 35from tornado import ioloop36 37from .abstract_loop import EventLoop, ExitMainLoop38 39if typing.TYPE_CHECKING:40    from collections.abc import Callable41    from concurrent.futures import Executor42 43    from typing_extensions import Literal, ParamSpec44 45    _Spec = ParamSpec("_Spec")46    _T = typing.TypeVar("_T")47 48__all__ = ("TornadoEventLoop",)49 50 51class TornadoEventLoop(EventLoop):52    """This is an Urwid-specific event loop to plug into its MainLoop.53    It acts as an adaptor for Tornado's IOLoop which does all54    heavy lifting except idle-callbacks.55    """56 57    def __init__(self, loop: ioloop.IOLoop | None = None) -> None:58        super().__init__()59        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)60        if loop:61            self._loop: ioloop.IOLoop = loop62        else:63            try:64                asyncio.get_running_loop()65            except RuntimeError:66                asyncio.set_event_loop(asyncio.new_event_loop())67 68            self._loop = ioloop.IOLoop.current()69 70        self._pending_alarms: dict[object, int] = {}71        self._watch_handles: dict[int, int] = {}  # {<watch_handle> : <file_descriptor>}72        self._max_watch_handle: int = 073        self._exc: BaseException | None = None74 75        self._idle_asyncio_handle: object | None = None76        self._idle_handle: int = 077        self._idle_callbacks: dict[int, Callable[[], typing.Any]] = {}78 79    def _also_call_idle(self, callback: Callable[_Spec, _T]) -> Callable[_Spec, _T]:80        """81        Wrap the callback to also call _entering_idle.82        """83 84        @functools.wraps(callback)85        def wrapper(*args: _Spec.args, **kwargs: _Spec.kwargs) -> _T:86            if not self._idle_asyncio_handle:87                self._idle_asyncio_handle = self._loop.call_later(0, self._entering_idle)88            return callback(*args, **kwargs)89 90        return wrapper91 92    def _entering_idle(self) -> None:93        """94        Call all the registered idle callbacks.95        """96        try:97            for callback in self._idle_callbacks.values():98                callback()99        finally:100            self._idle_asyncio_handle = None101 102    def run_in_executor(103        self,104        executor: Executor,105        func: Callable[_Spec, _T],106        *args: _Spec.args,107        **kwargs: _Spec.kwargs,108    ) -> asyncio.Future[_T]:109        """Run callable in executor.110 111        :param executor: Executor to use for running the function112        :type executor: concurrent.futures.Executor113        :param func: function to call114        :type func: Callable115        :param args: arguments to function (positional only)116        :type args: object117        :param kwargs: keyword arguments to function (keyword only)118        :type kwargs: object119        :return: future object for the function call outcome.120        :rtype: asyncio.Future121        """122        return self._loop.run_in_executor(executor, functools.partial(func, *args, **kwargs))123 124    def alarm(self, seconds: float, callback: Callable[[], typing.Any]):125        @self._also_call_idle126        @functools.wraps(callback)127        def wrapped() -> None:128            with suppress(KeyError):129                del self._pending_alarms[handle]130 131            self.handle_exit(callback)()132 133        handle = self._loop.add_timeout(self._loop.time() + seconds, wrapped)134        self._pending_alarms[handle] = 1135        return handle136 137    def remove_alarm(self, handle: object) -> bool:138        self._loop.remove_timeout(handle)139        try:140            del self._pending_alarms[handle]141        except KeyError:142            return False143 144        return True145 146    def watch_file(self, fd: int, callback: Callable[[], _T]) -> int:147        @self._also_call_idle148        def handler(_fd: int, _events: int) -> None:149            self.handle_exit(callback)()150 151        self._loop.add_handler(fd, handler, ioloop.IOLoop.READ)152        self._max_watch_handle += 1153        handle = self._max_watch_handle154        self._watch_handles[handle] = fd155        return handle156 157    def remove_watch_file(self, handle: int) -> bool:158        if (fd := self._watch_handles.pop(handle, None)) is not None:159            self._loop.remove_handler(fd)160            return True161 162        return False163 164    def enter_idle(self, callback: Callable[[], typing.Any]) -> int:165        """166        Add a callback for entering idle.167 168        Returns a handle that may be passed to remove_idle()169        """170        # XXX there's no such thing as "idle" in most event loops; this fakes171        # it by adding extra callback to the timer and file watch callbacks.172        self._idle_handle += 1173        self._idle_callbacks[self._idle_handle] = callback174        return self._idle_handle175 176    def remove_enter_idle(self, handle: int) -> bool:177        """178        Remove an idle callback.179 180        Returns True if the handle was removed.181        """182        try:183            del self._idle_callbacks[handle]184        except KeyError:185            return False186        return True187 188    def handle_exit(self, f: Callable[_Spec, _T]) -> Callable[_Spec, _T | Literal[False]]:189        @functools.wraps(f)190        def wrapper(*args: _Spec.args, **kwargs: _Spec.kwargs) -> _T | Literal[False]:191            try:192                return f(*args, **kwargs)193            except ExitMainLoop:194                pass  # handled later195            except Exception as exc:196                self._exc = exc197 198            if self._idle_asyncio_handle:199                # clean it up to prevent old callbacks200                # from messing things up if loop is restarted201                self._loop.remove_timeout(self._idle_asyncio_handle)202                self._idle_asyncio_handle = None203 204            self._loop.stop()205            return False206 207        return wrapper208 209    def run(self) -> None:210        self._loop.start()211        if self._exc:212            exc, self._exc = self._exc, None213            raise exc.with_traceback(exc.__traceback__)214