codekingpro/portable-devtools

1# Urwid main loop code using Python-3.5 features (Trio, Curio, etc)2#    Copyright (C) 2018 Toshio Kuratomi3#    Copyright (C) 2019 Tamas Nepusz4#5#    This library is free software; you can redistribute it and/or6#    modify it under the terms of the GNU Lesser General Public7#    License as published by the Free Software Foundation; either8#    version 2.1 of the License, or (at your option) any later version.9#10#    This library is distributed in the hope that it will be useful,11#    but WITHOUT ANY WARRANTY; without even the implied warranty of12#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU13#    Lesser General Public License for more details.14#15#    You should have received a copy of the GNU Lesser General Public16#    License along with this library; if not, write to the Free Software17#    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA18#19# Urwid web site: https://urwid.org/20 21"""Trio Runner based urwid EventLoop implementation.22 23Trio library is required.24"""25 26from __future__ import annotations27 28import logging29import sys30import typing31 32import trio33 34from .abstract_loop import EventLoop, ExitMainLoop35 36if sys.version_info < (3, 11):37    from exceptiongroup import BaseExceptionGroup  # pylint: disable=redefined-builtin  # backport38 39if typing.TYPE_CHECKING:40    import io41    from collections.abc import Awaitable, Callable, Hashable, Mapping42 43    from typing_extensions import Concatenate, ParamSpec44 45    _Spec = ParamSpec("_Spec")46 47__all__ = ("TrioEventLoop",)48 49 50class _TrioIdleCallbackInstrument(trio.abc.Instrument):51    """IDLE callbacks emulation helper."""52 53    __slots__ = ("idle_callbacks",)54 55    def __init__(self, idle_callbacks: Mapping[Hashable, Callable[[], typing.Any]]):56        self.idle_callbacks = idle_callbacks57 58    def before_io_wait(self, timeout: float) -> None:59        if timeout > 0:60            for idle_callback in self.idle_callbacks.values():61                idle_callback()62 63 64class TrioEventLoop(EventLoop):65    """66    Event loop based on the ``trio`` module.67 68    ``trio`` is an async library for Python 3.5 and later.69    """70 71    def __init__(self) -> None:72        """Constructor."""73        super().__init__()74        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)75 76        self._idle_handle = 077        self._idle_callbacks: dict[int, Callable[[], typing.Any]] = {}78        self._pending_tasks: list[tuple[Callable[_Spec, Awaitable], trio.CancelScope, _Spec.args]] = []79 80        self._nursery: trio.Nursery | None = None81 82        self._sleep = trio.sleep83        self._wait_readable = trio.lowlevel.wait_readable84 85    def alarm(86        self,87        seconds: float,88        callback: Callable[[], typing.Any],89    ) -> trio.CancelScope:90        """Calls `callback()` a given time from now.91 92        :param seconds: time in seconds to wait before calling the callback93        :type seconds: float94        :param callback: function to call from the event loop95        :type callback: Callable[[], typing.Any]96        :return: a handle that may be passed to `remove_alarm()`97        :rtype: trio.CancelScope98 99        No parameters are passed to the callback.100        """101        return self._start_task(self._alarm_task, seconds, callback)102 103    def enter_idle(self, callback: Callable[[], typing.Any]) -> int:104        """Calls `callback()` when the event loop enters the idle state.105 106        There is no such thing as being idle in a Trio event loop so we107        simulate it by repeatedly calling `callback()` with a short delay.108        """109        self._idle_handle += 1110        self._idle_callbacks[self._idle_handle] = callback111        return self._idle_handle112 113    def remove_alarm(self, handle: trio.CancelScope) -> bool:114        """Removes an alarm.115 116        Parameters:117            handle: the handle of the alarm to remove118        """119        return self._cancel_scope(handle)120 121    def remove_enter_idle(self, handle: int) -> bool:122        """Removes an idle callback.123 124        Parameters:125            handle: the handle of the idle callback to remove126        """127        try:128            del self._idle_callbacks[handle]129        except KeyError:130            return False131        return True132 133    def remove_watch_file(self, handle: trio.CancelScope) -> bool:134        """Removes a file descriptor being watched for input.135 136        Parameters:137            handle: the handle of the file descriptor callback to remove138 139        Returns:140            True if the file descriptor was watched, False otherwise141        """142        return self._cancel_scope(handle)143 144    def _cancel_scope(self, scope: trio.CancelScope) -> bool:145        """Cancels the given Trio cancellation scope.146 147        Returns:148            True if the scope was cancelled, False if it was cancelled already149            before invoking this function150        """151        existed = not scope.cancel_called152        scope.cancel()153        return existed154 155    def run(self) -> None:156        """Starts the event loop. Exits the loop when any callback raises an157        exception. If ExitMainLoop is raised, exits cleanly.158        """159 160        emulate_idle_callbacks = _TrioIdleCallbackInstrument(self._idle_callbacks)161 162        try:163            trio.run(self._main_task, instruments=[emulate_idle_callbacks])164        except BaseException as exc:165            self._handle_main_loop_exception(exc)166 167    async def run_async(self) -> None:168        """Starts the main loop and blocks asynchronously until the main loop exits.169 170        This allows one to embed an urwid app in a Trio app even if the Trio event loop is already running.171        Example::172 173            with trio.open_nursery() as nursery:174                event_loop = urwid.TrioEventLoop()175 176                # [...launch other async tasks in the nursery...]177 178                loop = urwid.MainLoop(widget, event_loop=event_loop)179                with loop.start():180                    await event_loop.run_async()181 182                nursery.cancel_scope.cancel()183        """184 185        emulate_idle_callbacks = _TrioIdleCallbackInstrument(self._idle_callbacks)186 187        try:188            trio.lowlevel.add_instrument(emulate_idle_callbacks)189            try:190                await self._main_task()191            finally:192                trio.lowlevel.remove_instrument(emulate_idle_callbacks)193        except BaseException as exc:194            self._handle_main_loop_exception(exc)195 196    def watch_file(197        self,198        fd: int | io.IOBase,199        callback: Callable[[], typing.Any],200    ) -> trio.CancelScope:201        """Calls `callback()` when the given file descriptor has some data202        to read. No parameters are passed to the callback.203 204        Parameters:205            fd: file descriptor to watch for input206            callback: function to call when some input is available207 208        Returns:209            a handle that may be passed to `remove_watch_file()`210        """211        return self._start_task(self._watch_task, fd, callback)212 213    async def _alarm_task(214        self,215        scope: trio.CancelScope,216        seconds: float,217        callback: Callable[[], typing.Any],218    ) -> None:219        """Asynchronous task that sleeps for a given number of seconds and then220        calls the given callback.221 222        Parameters:223            scope: the cancellation scope that can be used to cancel the task224            seconds: the number of seconds to wait225            callback: the callback to call226        """227        with scope:228            await self._sleep(seconds)229            callback()230 231    def _handle_main_loop_exception(self, exc: BaseException) -> None:232        """Handles exceptions raised from the main loop, catching ExitMainLoop233        instead of letting it propagate through.234 235        Note that since Trio may collect multiple exceptions from tasks into an ExceptionGroup,236        we cannot simply use a try..catch clause, we need a helper function like this.237        """238        self._idle_callbacks.clear()239        if isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:240            exc = exc.exceptions[0]241 242        if isinstance(exc, ExitMainLoop):243            return244 245        raise exc.with_traceback(exc.__traceback__) from None246 247    async def _main_task(self) -> None:248        """Main Trio task that opens a nursery and then sleeps until the user249        exits the app by raising ExitMainLoop.250        """251        try:252            async with trio.open_nursery() as self._nursery:253                self._schedule_pending_tasks()254                await trio.sleep_forever()255        finally:256            self._nursery = None257 258    def _schedule_pending_tasks(self) -> None:259        """Schedules all pending asynchronous tasks that were created before260        the nursery to be executed on the nursery soon.261        """262        for task, scope, args in self._pending_tasks:263            self._nursery.start_soon(task, scope, *args)264        self._pending_tasks.clear()265 266    def _start_task(267        self,268        task: Callable[Concatenate[trio.CancelScope, _Spec], Awaitable],269        *args: _Spec.args,270    ) -> trio.CancelScope:271        """Starts an asynchronous task in the Trio nursery managed by the272        main loop. If the nursery has not started yet, store a reference to273        the task and the arguments so we can start the task when the nursery274        is open.275 276        Parameters:277            task: a Trio task to run278 279        Returns:280            a cancellation scope for the Trio task281        """282        scope = trio.CancelScope()283        if self._nursery:284            self._nursery.start_soon(task, scope, *args)285        else:286            self._pending_tasks.append((task, scope, args))287        return scope288 289    async def _watch_task(290        self,291        scope: trio.CancelScope,292        fd: int | io.IOBase,293        callback: Callable[[], typing.Any],294    ) -> None:295        """Asynchronous task that watches the given file descriptor and calls296        the given callback whenever the file descriptor becomes readable.297 298        Parameters:299            scope: the cancellation scope that can be used to cancel the task300            fd: the file descriptor to watch301            callback: the callback to call302        """303        with scope:304            # We check for the scope being cancelled before calling305            # wait_readable because if callback cancels the scope, fd might be306            # closed and calling wait_readable with a closed fd does not work.307            while not scope.cancel_called:308                await self._wait_readable(fd)309                callback()310