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"""Abstract shared code for urwid EventLoop implementation."""23 24from __future__ import annotations25 26import abc27import logging28import signal29import typing30 31if typing.TYPE_CHECKING:32    import asyncio33    from collections.abc import Callable34    from concurrent.futures import Executor, Future35    from types import FrameType36 37    from typing_extensions import ParamSpec38 39    _T = typing.TypeVar("_T")40    _Spec = ParamSpec("_Spec")41 42__all__ = ("EventLoop", "ExitMainLoop")43 44 45class ExitMainLoop(Exception):46    """47    When this exception is raised within a main loop the main loop48    will exit cleanly.49    """50 51 52class EventLoop(abc.ABC):53    """54    Abstract class representing an event loop to be used by :class:`MainLoop`.55    """56 57    __slots__ = ("logger",)58 59    def __init__(self) -> None:60        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)61 62    def run_in_executor(63        self,64        executor: Executor,65        func: Callable[_Spec, _T],66        *args: _Spec.args,67        **kwargs: _Spec.kwargs,68    ) -> Future[_T] | asyncio.Future[_T]:69        """Run callable in executor if supported.70 71        :param executor: Executor to use for running the function72        :type executor: concurrent.futures.Executor73        :param func: function to call74        :type func: Callable75        :param args: arguments to function (positional only)76        :type args: object77        :param kwargs: keyword arguments to function (keyword only)78        :type kwargs: object79        :return: future object for the function call outcome.80                 (exact future type depends on the event loop type)81        :rtype: concurrent.futures.Future | asyncio.Future82        """83        raise NotImplementedError84 85    @abc.abstractmethod86    def alarm(self, seconds: float, callback: Callable[[], typing.Any]) -> typing.Any:87        """88        Call callback() a given time from now.  No parameters are89        passed to callback.90 91        This method has no default implementation.92 93        Returns a handle that may be passed to remove_alarm()94 95        seconds -- floating point time to wait before calling callback96        callback -- function to call from event loop97        """98 99    @abc.abstractmethod100    def enter_idle(self, callback):101        """102        Add a callback for entering idle.103 104        This method has no default implementation.105 106        Returns a handle that may be passed to remove_idle()107        """108 109    @abc.abstractmethod110    def remove_alarm(self, handle) -> bool:111        """112        Remove an alarm.113 114        This method has no default implementation.115 116        Returns True if the alarm exists, False otherwise117        """118 119    @abc.abstractmethod120    def remove_enter_idle(self, handle) -> bool:121        """122        Remove an idle callback.123 124        This method has no default implementation.125 126        Returns True if the handle was removed.127        """128 129    @abc.abstractmethod130    def remove_watch_file(self, handle) -> bool:131        """132        Remove an input file.133 134        This method has no default implementation.135 136        Returns True if the input file exists, False otherwise137        """138 139    @abc.abstractmethod140    def run(self) -> None:141        """142        Start the event loop.  Exit the loop when any callback raises143        an exception.  If ExitMainLoop is raised, exit cleanly.144 145        This method has no default implementation.146        """147 148    @abc.abstractmethod149    def watch_file(self, fd: int, callback: Callable[[], typing.Any]):150        """151        Call callback() when fd has some data to read.  No parameters152        are passed to callback.153 154        This method has no default implementation.155 156        Returns a handle that may be passed to remove_watch_file()157 158        fd -- file descriptor to watch for input159        callback -- function to call when input is available160        """161 162    def set_signal_handler(163        self,164        signum: int,165        handler: Callable[[int, FrameType | None], typing.Any] | int | signal.Handlers,166    ) -> Callable[[int, FrameType | None], typing.Any] | int | signal.Handlers | None:167        """168        Sets the signal handler for signal signum.169 170        The default implementation of :meth:`set_signal_handler`171        is simply a proxy function that calls :func:`signal.signal()`172        and returns the resulting value.173 174        signum -- signal number175        handler -- function (taking signum as its single argument),176        or `signal.SIG_IGN`, or `signal.SIG_DFL`177        """178        return signal.signal(signum, handler)179