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 23from __future__ import annotations24 25import heapq26import logging27import os28import sys29import time30import typing31import warnings32from contextlib import suppress33 34from urwid import display, signals35from urwid.command_map import Command, command_map36from urwid.display.common import INPUT_DESCRIPTORS_CHANGED37from urwid.util import StoppingContext, is_mouse_event38from urwid.widget import PopUpTarget39 40from .abstract_loop import ExitMainLoop41from .select_loop import SelectEventLoop42 43if typing.TYPE_CHECKING:44    from collections.abc import Callable, Iterable45 46    from typing_extensions import Self47 48    from urwid.display import BaseScreen49    from urwid.widget import Widget50 51    from .abstract_loop import EventLoop52 53    _T = typing.TypeVar("_T")54 55 56IS_WINDOWS = sys.platform == "win32"57PIPE_BUFFER_READ_SIZE = 4096  # can expect this much on Linux, so try for that58 59__all__ = ("CantUseExternalLoop", "MainLoop")60 61 62class CantUseExternalLoop(Exception):63    pass64 65 66class MainLoop:67    """68    This is the standard main loop implementation for a single interactive69    session.70 71    :param widget: the topmost widget used for painting the screen, stored as72                   :attr:`widget` and may be modified. Must be a box widget.73    :type widget: widget instance74 75    :param palette: initial palette for screen76    :type palette: iterable of palette entries77 78    :param screen: screen to use, default is a new :class:`raw_display.Screen`79                   instance; stored as :attr:`screen`80    :type screen: display module screen instance81 82    :param handle_mouse: ``True`` to ask :attr:`.screen` to process mouse events83    :type handle_mouse: bool84 85    :param input_filter: a function to filter input before sending it to86                   :attr:`.widget`, called from :meth:`.input_filter`87    :type input_filter: callable88 89    :param unhandled_input: a function called when input is not handled by90                            :attr:`.widget`, called from :meth:`.unhandled_input`91    :type unhandled_input: callable92 93    :param event_loop: if :attr:`.screen` supports external an event loop it may be94                       given here, default is a new :class:`SelectEventLoop` instance;95                       stored as :attr:`.event_loop`96    :type event_loop: event loop instance97 98    :param pop_ups: `True` to wrap :attr:`.widget` with a :class:`PopUpTarget`99                    instance to allow any widget to open a pop-up anywhere on the screen100    :type pop_ups: boolean101 102 103    .. attribute:: screen104 105        The screen object this main loop uses for screen updates and reading input106 107    .. attribute:: event_loop108 109        The event loop object this main loop uses for waiting on alarms and IO110    """111 112    def __init__(113        self,114        widget: Widget,115        palette: Iterable[116            tuple[str, str] | tuple[str, str, str] | tuple[str, str, str, str] | tuple[str, str, str, str, str, str]117        ] = (),118        screen: BaseScreen | None = None,119        handle_mouse: bool = True,120        input_filter: Callable[[list[str], list[int]], list[str]] | None = None,121        unhandled_input: Callable[[str | tuple[str, int, int, int]], bool | None] | None = None,122        event_loop: EventLoop | None = None,123        pop_ups: bool = False,124    ):125        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)126        self._widget = widget127        self.handle_mouse = handle_mouse128        self._pop_ups = False  # only initialize placeholder129        self.pop_ups = pop_ups  # triggers property setting side-effect130 131        if not screen:132            screen = display.raw.Screen()133 134        if palette:135            screen.register_palette(palette)136 137        self.screen: BaseScreen = screen138        self.screen_size: tuple[int, int] | None = None139 140        self._unhandled_input = unhandled_input141        self._input_filter = input_filter142 143        if not hasattr(screen, "hook_event_loop") and event_loop is not None:144            raise NotImplementedError(f"screen object passed {screen!r} does not support external event loops")145        if event_loop is None:146            event_loop = SelectEventLoop()147        self.event_loop: EventLoop = event_loop148 149        if hasattr(self.screen, "signal_handler_setter"):150            # Tell the screen what function it must use to set151            # signal handlers152            self.screen.signal_handler_setter = self.event_loop.set_signal_handler153 154        self._watch_pipes: dict[int, tuple[Callable[[], typing.Any], int]] = {}155 156    @property157    def widget(self) -> Widget:158        """159        Property for the topmost widget used to draw the screen.160        This must be a box widget.161        """162        return self._widget163 164    @widget.setter165    def widget(self, widget: Widget) -> None:166        self._widget = widget167        if self.pop_ups:168            self._topmost_widget.original_widget = self._widget169        else:170            self._topmost_widget = self._widget171 172    def _set_widget(self, widget: Widget) -> None:173        warnings.warn(174            f"method `{self.__class__.__name__}._set_widget` is deprecated, "175            f"please use `{self.__class__.__name__}.widget` property."176            "API will be removed in version 4.0.",177            DeprecationWarning,178            stacklevel=2,179        )180        self.widget = widget181 182    @property183    def pop_ups(self) -> bool:184        return self._pop_ups185 186    @pop_ups.setter187    def pop_ups(self, pop_ups: bool) -> None:188        self._pop_ups = pop_ups189        if pop_ups:190            self._topmost_widget = PopUpTarget(self._widget)191        else:192            self._topmost_widget = self._widget193 194    def _set_pop_ups(self, pop_ups: bool) -> None:195        warnings.warn(196            f"method `{self.__class__.__name__}._set_pop_ups` is deprecated, "197            f"please use `{self.__class__.__name__}.pop_ups` property."198            "API will be removed in version 4.0.",199            DeprecationWarning,200            stacklevel=2,201        )202        self.pop_ups = pop_ups203 204    def set_alarm_in(self, sec: float, callback: Callable[[Self, _T], typing.Any], user_data: _T = None):205        """206        Schedule an alarm in *sec* seconds that will call *callback* from the207        within the :meth:`run` method.208 209        :param sec: seconds until alarm210        :type sec: float211        :param callback: function to call with two parameters: this main loop212                         object and *user_data*213        :type callback: callable214        :param user_data: optional user data to pass to the callback215        :type user_data: object216        """217        self.logger.debug(f"Setting alarm in {sec!r} seconds with callback {callback!r}")218 219        def cb() -> None:220            callback(self, user_data)221 222        return self.event_loop.alarm(sec, cb)223 224    def set_alarm_at(self, tm: float, callback: Callable[[Self, _T], typing.Any], user_data: _T = None):225        """226        Schedule an alarm at *tm* time that will call *callback* from the227        within the :meth:`run` function. Returns a handle that may be passed to228        :meth:`remove_alarm`.229 230        :param tm: time to call callback e.g. ``time.time() + 5``231        :type tm: float232        :param callback: function to call with two parameters: this main loop233                         object and *user_data*234        :type callback: callable235        :param user_data: optional user data to pass to the callback236        :type user_data: object237        """238        sec = tm - time.time()239        self.logger.debug(f"Setting alarm in {sec!r} seconds with callback {callback!r}")240 241        def cb() -> None:242            callback(self, user_data)243 244        return self.event_loop.alarm(sec, cb)245 246    def remove_alarm(self, handle) -> bool:247        """248        Remove an alarm. Return ``True`` if *handle* was found, ``False``249        otherwise.250        """251        return self.event_loop.remove_alarm(handle)252 253    if not IS_WINDOWS:254 255        def watch_pipe(self, callback: Callable[[bytes], bool | None]) -> int:256            """257            Create a pipe for use by a subprocess or thread to trigger a callback258            in the process/thread running the main loop.259 260            :param callback: function taking one parameter to call from within the process/thread running the main loop261            :type callback: callable262 263            This method returns a file descriptor attached to the write end of a pipe.264            The read end of the pipe is added to the list of files :attr:`event_loop` is watching.265            When data is written to the pipe the callback function will be called266            and passed a single value containing data read from the pipe.267 268            This method may be used any time you want to update widgets from another thread or subprocess.269 270            Data may be written to the returned file descriptor with ``os.write(fd, data)``.271            Ensure that data is less than 512 bytes (or 4K on Linux)272            so that the callback will be triggered just once with the complete value of data passed in.273 274            If the callback returns ``False`` then the watch will be removed from :attr:`event_loop`275            and the read end of the pipe will be closed.276            You are responsible for closing the write end of the pipe with ``os.close(fd)``.277            """278            import fcntl279 280            pipe_rd, pipe_wr = os.pipe()281            fcntl.fcntl(pipe_rd, fcntl.F_SETFL, os.O_NONBLOCK)282            watch_handle = None283 284            def cb() -> None:285                data = os.read(pipe_rd, PIPE_BUFFER_READ_SIZE)286                if callback(data) is False:287                    self.event_loop.remove_watch_file(watch_handle)288                    os.close(pipe_rd)289 290            watch_handle = self.event_loop.watch_file(pipe_rd, cb)291            self._watch_pipes[pipe_wr] = (watch_handle, pipe_rd)292            return pipe_wr293 294        def remove_watch_pipe(self, write_fd: int) -> bool:295            """296            Close the read end of the pipe and remove the watch created by :meth:`watch_pipe`.297 298            ..note:: You are responsible for closing the write end of the pipe.299 300            Returns ``True`` if the watch pipe exists, ``False`` otherwise301            """302            try:303                watch_handle, pipe_rd = self._watch_pipes.pop(write_fd)304            except KeyError:305                return False306 307            if not self.event_loop.remove_watch_file(watch_handle):308                return False309            os.close(pipe_rd)310            return True311 312    def watch_file(self, fd: int, callback: Callable[[], typing.Any]):313        """314        Call *callback* when *fd* has some data to read. No parameters are315        passed to callback.316 317        Returns a handle that may be passed to :meth:`remove_watch_file`.318        """319        self.logger.debug(f"Setting watch file descriptor {fd!r} with {callback!r}")320        return self.event_loop.watch_file(fd, callback)321 322    def remove_watch_file(self, handle):323        """324        Remove a watch file. Returns ``True`` if the watch file325        exists, ``False`` otherwise.326        """327        return self.event_loop.remove_watch_file(handle)328 329    def run(self) -> None:330        """331        Start the main loop handling input events and updating the screen. The332        loop will continue until an :exc:`ExitMainLoop` exception is raised.333 334        If you would prefer to manage the event loop yourself, don't use this335        method.  Instead, call :meth:`start` before starting the event loop,336        and :meth:`stop` once it's finished.337        """338        with suppress(ExitMainLoop):339            self._run()340 341    def _test_run(self):342        """343        >>> w = _refl("widget")  # _refl prints out function calls344        >>> w.render_rval = "fake canvas"  # *_rval is used for return values345        >>> scr = _refl("screen")346        >>> scr.get_input_descriptors_rval = [42]347        >>> scr.get_cols_rows_rval = (20, 10)348        >>> scr.started = True349        >>> scr._urwid_signals = {}350        >>> evl = _refl("event_loop")351        >>> evl.enter_idle_rval = 1352        >>> evl.watch_file_rval = 2353        >>> ml = MainLoop(w, [], scr, event_loop=evl)354        >>> ml.run()  # doctest:+ELLIPSIS355        screen.start()356        screen.set_mouse_tracking()357        screen.unhook_event_loop(...)358        screen.hook_event_loop(...)359        event_loop.enter_idle(<bound method MainLoop.entering_idle...>)360        event_loop.run()361        event_loop.remove_enter_idle(1)362        screen.unhook_event_loop(...)363        screen.stop()364        >>> ml.draw_screen()  # doctest:+ELLIPSIS365        screen.get_cols_rows()366        widget.render((20, 10), focus=True)367        screen.draw_screen((20, 10), 'fake canvas')368        """369 370    def start(self) -> StoppingContext:371        """372        Sets up the main loop, hooking into the event loop where necessary.373        Starts the :attr:`screen` if it hasn't already been started.374 375        If you want to control starting and stopping the event loop yourself,376        you should call this method before starting, and call `stop` once the377        loop has finished.  You may also use this method as a context manager,378        which will stop the loop automatically at the end of the block:379 380            with main_loop.start():381                ...382 383        Note that some event loop implementations don't handle exceptions384        specially if you manage the event loop yourself.  In particular, the385        Twisted and asyncio loops won't stop automatically when386        :exc:`ExitMainLoop` (or anything else) is raised.387        """388 389        self.logger.debug(f"Starting event loop {self.event_loop.__class__.__name__!r} to manage display.")390 391        self.screen.start()392 393        if self.handle_mouse:394            self.screen.set_mouse_tracking()395 396        if not hasattr(self.screen, "hook_event_loop"):397            raise CantUseExternalLoop(f"Screen {self.screen!r} doesn't support external event loops")398 399        with suppress(NameError):400            signals.connect_signal(self.screen, INPUT_DESCRIPTORS_CHANGED, self._reset_input_descriptors)401 402        # watch our input descriptors403        self._reset_input_descriptors()404        self.idle_handle = self.event_loop.enter_idle(self.entering_idle)405 406        # the screen is redrawn automatically after input and alarms,407        # however, there can be none of those at the start,408        # so draw the initial screen here unconditionally409        self.event_loop.alarm(0, self.entering_idle)410 411        return StoppingContext(self)412 413    def stop(self) -> None:414        """415        Cleans up any hooks added to the event loop.  Only call this if you're416        managing the event loop yourself, after the loop stops.417        """418 419        self.event_loop.remove_enter_idle(self.idle_handle)420        del self.idle_handle421        signals.disconnect_signal(self.screen, INPUT_DESCRIPTORS_CHANGED, self._reset_input_descriptors)422        self.screen.unhook_event_loop(self.event_loop)423 424        self.screen.stop()425 426    def _reset_input_descriptors(self) -> None:427        self.screen.unhook_event_loop(self.event_loop)428        self.screen.hook_event_loop(self.event_loop, self._update)429 430    def _run(self) -> None:431        try:432            self.start()433        except CantUseExternalLoop:434            try:435                self._run_screen_event_loop()436                return437            finally:438                self.screen.stop()439 440        try:441            self.event_loop.run()442        except:443            self.screen.stop()  # clean up screen control444            raise445        self.stop()446 447    def _update(self, keys: list[str], raw: list[int]) -> None:448        """449        >>> w = _refl("widget")450        >>> w.selectable_rval = True451        >>> w.mouse_event_rval = True452        >>> scr = _refl("screen")453        >>> scr.get_cols_rows_rval = (15, 5)454        >>> evl = _refl("event_loop")455        >>> ml = MainLoop(w, [], scr, event_loop=evl)456        >>> ml._input_timeout = "old timeout"457        >>> ml._update(["y"], [121])  # doctest:+ELLIPSIS458        screen.get_cols_rows()459        widget.selectable()460        widget.keypress((15, 5), 'y')461        >>> ml._update([("mouse press", 1, 5, 4)], [])462        widget.mouse_event((15, 5), 'mouse press', 1, 5, 4, focus=True)463        >>> ml._update([], [])464        """465        if keys := self.input_filter(keys, raw):466            self.process_input(keys)467            if "window resize" in keys:468                self.screen_size = None469 470    def _run_screen_event_loop(self) -> None:471        """472        This method is used when the screen does not support using external event loops.473 474        The alarms stored in the SelectEventLoop in :attr:`event_loop` are modified by this method.475        """476        # pylint: disable=protected-access  # special case for alarms handling477        self.logger.debug(f"Starting screen {self.screen!r} event loop")478 479        next_alarm = None480 481        while True:482            self.draw_screen()483 484            if not next_alarm and self.event_loop._alarms:485                next_alarm = heapq.heappop(self.event_loop._alarms)486 487            keys: list[str] = []488            raw: list[int] = []489            while not keys:490                if next_alarm:491                    sec = max(0.0, next_alarm[0] - time.time())492                    self.screen.set_input_timeouts(sec)493                else:494                    self.screen.set_input_timeouts(None)495                keys, raw = self.screen.get_input(True)496                if not keys and next_alarm and next_alarm[0] - time.time() <= 0:497                    break498 499            if keys := self.input_filter(keys, raw):500                self.process_input(keys)501 502            while next_alarm:503                if (next_alarm[0] - time.time()) > 0:504                    break505                _tm, _tie_break, callback = next_alarm506                callback()507 508                if self.event_loop._alarms:509                    next_alarm = heapq.heappop(self.event_loop._alarms)510                else:511                    next_alarm = None512 513            if "window resize" in keys:514                self.screen_size = None515 516    def _test_run_screen_event_loop(self):517        """518        >>> w = _refl("widget")519        >>> scr = _refl("screen")520        >>> scr.get_cols_rows_rval = (10, 5)521        >>> scr.get_input_rval = [], []522        >>> ml = MainLoop(w, screen=scr)523        >>> def stop_now(loop, data):524        ...     raise ExitMainLoop()525        >>> handle = ml.set_alarm_in(0, stop_now)526        >>> try:527        ...     ml._run_screen_event_loop()528        ... except ExitMainLoop:529        ...     pass530        screen.get_cols_rows()531        widget.render((10, 5), focus=True)532        screen.draw_screen((10, 5), None)533        screen.set_input_timeouts(0.0)534        screen.get_input(True)535        """536 537    def process_input(self, keys: Iterable[str | tuple[str, int, int, int]]) -> bool:538        """539        This method will pass keyboard input and mouse events to :attr:`widget`.540        This method is called automatically from the :meth:`run` method when541        there is input, but may also be called to simulate input from the user.542 543        *keys* is a list of input returned from :attr:`screen`'s get_input()544        or get_input_nonblocking() methods.545 546        Returns ``True`` if any key was handled by a widget or the547        :meth:`unhandled_input` method.548        """549        self.logger.debug(f"Processing input: keys={keys!r}")550        if not self.screen_size:551            self.screen_size = self.screen.get_cols_rows()552 553        something_handled = False554 555        for key in keys:556            if key == "window resize":557                continue558 559            if isinstance(key, str):560                if self._topmost_widget.selectable():561                    if handled_key := self._topmost_widget.keypress(self.screen_size, key):562                        key = handled_key  # noqa: PLW2901563 564                    else:565                        something_handled = True566                        continue567 568            elif is_mouse_event(key):569                event, button, col, row = key570                if hasattr(self._topmost_widget, "mouse_event") and self._topmost_widget.mouse_event(571                    self.screen_size,572                    event,573                    button,574                    col,575                    row,576                    focus=True,577                ):578                    something_handled = True579                    continue580 581            else:582                raise TypeError(f"{key!r} is not str | tuple[str, int, int, int]")583 584            if key:585                if command_map[key] == Command.REDRAW_SCREEN:586                    self.screen.clear()587                    something_handled = True588                else:589                    something_handled |= bool(self.unhandled_input(key))590            else:591                something_handled = True592 593        return something_handled594 595    def _test_process_input(self):596        """597        >>> w = _refl("widget")598        >>> w.selectable_rval = True599        >>> scr = _refl("screen")600        >>> scr.get_cols_rows_rval = (10, 5)601        >>> ml = MainLoop(w, [], scr)602        >>> ml.process_input(["enter", ("mouse drag", 1, 14, 20)])603        screen.get_cols_rows()604        widget.selectable()605        widget.keypress((10, 5), 'enter')606        widget.mouse_event((10, 5), 'mouse drag', 1, 14, 20, focus=True)607        True608        """609 610    def input_filter(self, keys: list[str], raw: list[int]) -> list[str]:611        """612        This function is passed each all the input events and raw keystroke613        values. These values are passed to the *input_filter* function614        passed to the constructor. That function must return a list of keys to615        be passed to the widgets to handle. If no *input_filter* was616        defined this implementation will return all the input events.617        """618        if self._input_filter:619            return self._input_filter(keys, raw)620        return keys621 622    def unhandled_input(self, data: str | tuple[str, int, int, int]) -> bool | None:623        """624        This function is called with any input that was not handled by the625        widgets, and calls the *unhandled_input* function passed to the626        constructor. If no *unhandled_input* was defined then the input627        will be ignored.628 629        *input* is the keyboard or mouse input.630 631        The *unhandled_input* function should return ``True`` if it handled632        the input.633        """634        if self._unhandled_input:635            return self._unhandled_input(data)636        return False637 638    def entering_idle(self) -> None:639        """640        This method is called whenever the event loop is about to enter the641        idle state. :meth:`draw_screen` is called here to update the642        screen when anything has changed.643        """644        if self.screen.started:645            self.draw_screen()646        else:647            self.logger.debug(f"No redrawing screen: {self.screen!r} is not started.")648 649    def draw_screen(self) -> None:650        """651        Render the widgets and paint the screen. This method is called652        automatically from :meth:`entering_idle`.653 654        If you modify the widgets displayed outside of handling input or655        responding to an alarm you will need to call this method yourself656        to repaint the screen.657        """658        if not self.screen_size:659            self.screen_size = self.screen.get_cols_rows()660            self.logger.debug(f"Screen size recalculated: {self.screen_size!r}")661 662        canvas = self._topmost_widget.render(self.screen_size, focus=True)663        self.screen.draw_screen(self.screen_size, canvas)664 665 666def _refl(name: str, rval=None, loop_exit=False):667    """668    This function is used to test the main loop classes.669 670    >>> scr = _refl("screen")671    >>> scr.function("argument")672    screen.function('argument')673    >>> scr.callme(when="now")674    screen.callme(when='now')675    >>> scr.want_something_rval = 42676    >>> x = scr.want_something()677    screen.want_something()678    >>> x679    42680 681    """682 683    class Reflect:684        def __init__(self, name: str, rval=None):685            self._name = name686            self._rval = rval687 688        def __call__(self, *argl, **argd):689            args = ", ".join([repr(a) for a in argl])690            if args and argd:691                args = f"{args}, "692            args += ", ".join([f"{k}={v!r}" for k, v in argd.items()])693            print(f"{self._name}({args})")694            if loop_exit:695                raise ExitMainLoop()696            return self._rval697 698        def __getattr__(self, attr):699            if attr.endswith("_rval"):700                raise AttributeError()701            # print(self._name+"."+attr)702            if hasattr(self, f"{attr}_rval"):703                return Reflect(f"{self._name}.{attr}", getattr(self, f"{attr}_rval"))704            return Reflect(f"{self._name}.{attr}")705 706    return Reflect(name)707 708 709def _test():710    import doctest711 712    doctest.testmod()713 714 715if __name__ == "__main__":716    _test()717