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"""GLib based urwid EventLoop implementation.23 24PyGObject library is required.25"""26 27from __future__ import annotations28 29import functools30import logging31import signal32import typing33 34from gi.repository import GLib35 36from .abstract_loop import EventLoop, ExitMainLoop37 38if typing.TYPE_CHECKING:39    from collections.abc import Callable40    from concurrent.futures import Executor, Future41    from types import FrameType42 43    from typing_extensions import Literal, ParamSpec44 45    _Spec = ParamSpec("_Spec")46    _T = typing.TypeVar("_T")47 48__all__ = ("GLibEventLoop",)49 50 51def _ignore_handler(_sig: int, _frame: FrameType | None = None) -> None:52    return None53 54 55class GLibEventLoop(EventLoop):56    """57    Event loop based on GLib.MainLoop58    """59 60    def __init__(self) -> None:61        super().__init__()62        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)63        self._alarms: list[int] = []64        self._watch_files: dict[int, int] = {}65        self._idle_handle: int = 066        self._glib_idle_enabled = False  # have we called glib.idle_add?67        self._idle_callbacks: dict[int, Callable[[], typing.Any]] = {}68        self._loop = GLib.MainLoop()69        self._exc: BaseException | None = None70        self._enable_glib_idle()71        self._signal_handlers: dict[int, int] = {}72 73    def run_in_executor(74        self,75        executor: Executor,76        func: Callable[_Spec, _T],77        *args: _Spec.args,78        **kwargs: _Spec.kwargs,79    ) -> Future[_T]:80        """Run callable in executor.81 82        :param executor: Executor to use for running the function83        :type executor: concurrent.futures.Executor84        :param func: function to call85        :type func: Callable86        :param args: positional arguments to function87        :type args: object88        :param kwargs: keyword arguments to function89        :type kwargs: object90        :return: future object for the function call outcome.91        :rtype: concurrent.futures.Future92        """93        return executor.submit(func, *args, **kwargs)94 95    def alarm(96        self,97        seconds: float,98        callback: Callable[[], typing.Any],99    ) -> tuple[int, Callable[[], typing.Any]]:100        """101        Call callback() a given time from now.  No parameters are102        passed to callback.103 104        Returns a handle that may be passed to remove_alarm()105 106        seconds -- floating point time to wait before calling callback107        callback -- function to call from event loop108        """109 110        @self.handle_exit111        def ret_false() -> Literal[False]:112            callback()113            self._enable_glib_idle()114            return False115 116        fd = GLib.timeout_add(int(seconds * 1000), ret_false)117        self._alarms.append(fd)118        return (fd, callback)119 120    def set_signal_handler(121        self,122        signum: int,123        handler: Callable[[int, FrameType | None], typing.Any] | int | signal.Handlers,124    ) -> None:125        """126        Sets the signal handler for signal signum.127 128        .. WARNING::129            Because this method uses the `GLib`-specific `unix_signal_add`130            function, its behaviour is different than `signal.signal().`131 132            If `signum` is not `SIGHUP`, `SIGINT`, `SIGTERM`, `SIGUSR1`,133            `SIGUSR2` or `SIGWINCH`, this method performs no actions and134            immediately returns None.135 136            Returns None in all cases (unlike :func:`signal.signal()`).137        ..138 139        signum -- signal number140        handler -- function (taking signum as its single argument),141        or `signal.SIG_IGN`, or `signal.SIG_DFL`142        """143        glib_signals = [144            signal.SIGHUP,145            signal.SIGINT,146            signal.SIGTERM,147            signal.SIGUSR1,148            signal.SIGUSR2,149        ]150 151        # GLib supports SIGWINCH as of version 2.54.152        if not GLib.check_version(2, 54, 0):153            glib_signals.append(signal.SIGWINCH)154 155        if signum not in glib_signals:156            # The GLib event loop supports only the signals listed above157            return158 159        if signum in self._signal_handlers:160            GLib.source_remove(self._signal_handlers.pop(signum))161 162        if handler == signal.Handlers.SIG_IGN:163            handler = _ignore_handler164        elif handler == signal.Handlers.SIG_DFL:165            return166 167        def final_handler(signal_number: int):168            # MyPy False-negative: signal.Handlers casted169            handler(signal_number, None)  # type: ignore[operator]170            return GLib.SOURCE_CONTINUE171 172        source = GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signum, final_handler, signum)173        self._signal_handlers[signum] = source174 175    def remove_alarm(self, handle) -> bool:176        """177        Remove an alarm.178 179        Returns True if the alarm exists, False otherwise180        """181        try:182            self._alarms.remove(handle[0])183            GLib.source_remove(handle[0])184 185        except ValueError:186            return False187 188        return True189 190    def watch_file(self, fd: int, callback: Callable[[], typing.Any]) -> int:191        """192        Call callback() when fd has some data to read.  No parameters193        are passed to callback.194 195        Returns a handle that may be passed to remove_watch_file()196 197        fd -- file descriptor to watch for input198        callback -- function to call when input is available199        """200 201        @self.handle_exit202        def io_callback(source, cb_condition) -> Literal[True]:203            callback()204            self._enable_glib_idle()205            return True206 207        self._watch_files[fd] = GLib.io_add_watch(fd, GLib.IO_IN, io_callback)208        return fd209 210    def remove_watch_file(self, handle: int) -> bool:211        """212        Remove an input file.213 214        Returns True if the input file exists, False otherwise215        """216        if handle in self._watch_files:217            GLib.source_remove(self._watch_files[handle])218            del self._watch_files[handle]219            return True220        return False221 222    def enter_idle(self, callback: Callable[[], typing.Any]) -> int:223        """224        Add a callback for entering idle.225 226        Returns a handle that may be passed to remove_enter_idle()227        """228        self._idle_handle += 1229        self._idle_callbacks[self._idle_handle] = callback230        return self._idle_handle231 232    def _enable_glib_idle(self) -> None:233        if self._glib_idle_enabled:234            return235        GLib.idle_add(self._glib_idle_callback)236        self._glib_idle_enabled = True237 238    def _glib_idle_callback(self):239        for callback in self._idle_callbacks.values():240            callback()241        self._glib_idle_enabled = False242        return False  # ask glib not to call again (or we would be called243 244    def remove_enter_idle(self, handle) -> bool:245        """246        Remove an idle callback.247 248        Returns True if the handle was removed.249        """250        try:251            del self._idle_callbacks[handle]252        except KeyError:253            return False254        return True255 256    def run(self) -> None:257        """258        Start the event loop.  Exit the loop when any callback raises259        an exception.  If ExitMainLoop is raised, exit cleanly.260        """261        try:262            self._loop.run()263        finally:264            if self._loop.is_running():265                self._loop.quit()266        if self._exc:267            # An exception caused us to exit, raise it now268            exc = self._exc269            self._exc = None270            raise exc.with_traceback(exc.__traceback__)271 272    def handle_exit(self, f: Callable[_Spec, _T]) -> Callable[_Spec, _T | Literal[False]]:273        """274        Decorator that cleanly exits the :class:`GLibEventLoop` if275        :exc:`ExitMainLoop` is thrown inside of the wrapped function. Store the276        exception info if some other exception occurs, it will be reraised after277        the loop quits.278 279        *f* -- function to be wrapped280        """281 282        @functools.wraps(f)283        def wrapper(*args: _Spec.args, **kwargs: _Spec.kwargs) -> _T | Literal[False]:284            try:285                return f(*args, **kwargs)286            except ExitMainLoop:287                self._loop.quit()288            except BaseException as exc:289                self._exc = exc290                if self._loop.is_running():291                    self._loop.quit()292            return False293 294        return wrapper295