pytermgui.window_manager.compositor

The Compositor class, which is used by the WindowManager to draw onto the terminal.

  1"""The Compositor class, which is used by the WindowManager to draw onto the terminal."""
  2
  3# pylint: disable=too-many-instance-attributes
  4
  5from __future__ import annotations
  6
  7import time
  8from threading import Thread
  9from typing import Iterator, List, Tuple
 10
 11from ..widgets import Widget
 12from ..enums import WidgetChange
 13from ..animations import animator
 14from ..terminal import get_terminal, Terminal
 15
 16from .window import Window
 17
 18PositionedLineList = List[Tuple[Tuple[int, int], str]]
 19
 20
 21class Compositor:
 22    """The class used to draw `pytermgui.window_managers.manager.WindowManager` state.
 23
 24    This class handles turning a list of windows into a drawable buffer (composite),
 25    and then drawing it onto the screen.
 26
 27    Calling its `run` method will start the drawing thread, which will draw the current
 28    window states onto the screen. This routine targets `framerate`, though will likely
 29    not match it perfectly.
 30    """
 31
 32    def __init__(self, windows: list[Window], framerate: int) -> None:
 33        """Initializes the Compositor.
 34
 35        Args:
 36            windows: A list of the windows to be drawn.
 37        """
 38
 39        self._windows = windows
 40        self._is_running = False
 41
 42        self._previous: PositionedLineList = []
 43        self._frametime = 0.0
 44        self._should_redraw: bool = True
 45        self._cache: dict[int, list[str]] = {}
 46
 47        self.fps = 0
 48        self.framerate = framerate
 49
 50    @property
 51    def terminal(self) -> Terminal:
 52        """Returns the current global terminal."""
 53
 54        return get_terminal()
 55
 56    def _draw_loop(self) -> None:
 57        """A loop that draws at regular intervals."""
 58
 59        framecount = 0
 60        last_frame = fps_start_time = time.perf_counter()
 61
 62        while self._is_running:
 63            elapsed = time.perf_counter() - last_frame
 64
 65            if elapsed < self._frametime:
 66                time.sleep(self._frametime - elapsed)
 67                continue
 68
 69            animator.step(elapsed)
 70
 71            last_frame = time.perf_counter()
 72            self.draw()
 73
 74            framecount += 1
 75
 76            if last_frame - fps_start_time >= 1:
 77                self.fps = framecount
 78                fps_start_time = last_frame
 79                framecount = 0
 80
 81    # NOTE: This is not needed at the moment, but might be at some point soon.
 82    # def _get_lines(self, window: Window) -> list[str]:
 83    #     """Gets lines from the window, caching when possible.
 84
 85    #     This also applies the blurred style of the window, if it has no focus.
 86    #     """
 87
 88    #     if window.allow_fullscreen:
 89    #         window.pos = self.terminal.origin
 90    #         window.width = self.terminal.width
 91    #         window.height = self.terminal.height
 92
 93    #     return window.get_lines()
 94
 95    #     if window.has_focus or window.is_noblur:
 96    #         return window.get_lines()
 97
 98    #     _id = id(window)
 99    #     if not window.is_dirty and _id in self._cache:
100    #         return self._cache[_id]
101
102    #     lines: list[str] = []
103    #     for line in window.get_lines():
104    #         if not window.has_focus:
105    #             line = tim.parse("[239]" + strip_ansi(line).replace("[", r"\["))
106
107    #         lines.append(line)
108
109    #     self._cache[_id] = lines
110    #     return lines
111
112    @staticmethod
113    def _iter_positioned(
114        widget: Widget, until: int | None = None
115    ) -> Iterator[tuple[tuple[int, int], str]]:
116        """Iterates through (pos, line) tuples from widget.get_lines()."""
117
118        # get_lines = widget.get_lines
119        # if isinstance(widget, Window):
120        #     get_lines = lambda *_: self._get_lines(widget)  # type: ignore
121
122        if until is None:
123            until = widget.height
124
125        for i, line in enumerate(widget.get_lines()):
126            if i >= until:
127                break
128
129            pos = (widget.pos[0], widget.pos[1] + i)
130
131            yield (pos, line)
132
133    @property
134    def framerate(self) -> int:
135        """The framerate the draw loop runs at.
136
137        Note:
138            This will likely not be matched very accurately, mostly undershooting
139            the given target.
140        """
141
142        return self._framerate
143
144    @framerate.setter
145    def framerate(self, new: int) -> None:
146        """Updates the framerate."""
147
148        self._frametime = 1 / new
149        self._framerate = new
150
151    def clear_cache(self, window: Window) -> None:
152        """Clears the compositor's cache related to the given window."""
153
154        if id(window) in self._cache:
155            del self._cache[id(window)]
156
157    def run(self) -> None:
158        """Runs the compositor draw loop as a thread."""
159
160        self._is_running = True
161        Thread(name="CompositorDrawLoop", target=self._draw_loop).start()
162
163    def stop(self) -> None:
164        """Stops the compositor."""
165
166        self._is_running = False
167
168    def composite(self) -> PositionedLineList:
169        """Creates a composited buffer from the assigned windows.
170
171        Note that this is currently not used."""
172
173        lines = []
174        windows = self._windows
175
176        # Don't unnecessarily print under full screen windows
177        if any(window.allow_fullscreen for window in self._windows):
178            for window in reversed(self._windows):
179                if window.allow_fullscreen:
180                    windows = [window]
181                    break
182
183        size_changes = {WidgetChange.WIDTH, WidgetChange.HEIGHT, WidgetChange.SIZE}
184        for window in reversed(windows):
185            if not window.has_focus:
186                continue
187
188            change = window.get_change()
189
190            if change is None:
191                continue
192
193            if window.is_dirty or change in size_changes:
194                for pos, line in self._iter_positioned(window):
195                    lines.append((pos, line))
196
197                window.is_dirty = False
198                continue
199
200            if change is not None:
201                remaining = window.content_dimensions[1]
202
203                for widget in window.dirty_widgets:
204                    for pos, line in self._iter_positioned(widget, until=remaining):
205                        lines.append((pos, line))
206
207                    remaining -= widget.height
208
209                window.dirty_widgets = []
210                continue
211
212            if window.allow_fullscreen:
213                break
214
215        return lines
216
217    def set_redraw(self) -> None:
218        """Flags compositor for full redraw.
219
220        Note:
221            At the moment the compositor will always redraw the entire screen.
222        """
223
224        self._should_redraw = True
225
226    def draw(self, force: bool = False) -> None:
227        """Writes composited screen to the terminal.
228
229        At the moment this uses full-screen rewrites. There is a compositing
230        implementation in `composite`, but it is currently not performant enough to use.
231
232        Args:
233            force: When set, new composited lines will not be checked against the
234                previous ones, and everything will be redrawn.
235        """
236
237        # if self._should_redraw or force:
238        lines: PositionedLineList = []
239
240        for window in reversed(self._windows):
241            lines.extend(self._iter_positioned(window))
242
243        self._should_redraw = False
244
245        # else:
246        # lines = self.composite()
247
248        if not force and self._previous == lines:
249            return
250
251        buffer = "".join(f"\x1b[{pos[1]};{pos[0]}H{line}" for pos, line in lines)
252
253        self.terminal.clear_stream()
254        self.terminal.write(buffer)
255        self.terminal.flush()
256
257        self._previous = lines
258
259    def redraw(self) -> None:
260        """Force-redraws the buffer."""
261
262        self.draw(force=True)
263
264    def capture(self, title: str, filename: str | None = None) -> None:
265        """Captures the most-recently drawn buffer as `filename`.
266
267        See `pytermgui.exporters.to_svg` for more information.
268        """
269
270        with self.terminal.record() as recording:
271            self.redraw()
272
273        recording.save_svg(title=title, filename=filename)
class Compositor:
 22class Compositor:
 23    """The class used to draw `pytermgui.window_managers.manager.WindowManager` state.
 24
 25    This class handles turning a list of windows into a drawable buffer (composite),
 26    and then drawing it onto the screen.
 27
 28    Calling its `run` method will start the drawing thread, which will draw the current
 29    window states onto the screen. This routine targets `framerate`, though will likely
 30    not match it perfectly.
 31    """
 32
 33    def __init__(self, windows: list[Window], framerate: int) -> None:
 34        """Initializes the Compositor.
 35
 36        Args:
 37            windows: A list of the windows to be drawn.
 38        """
 39
 40        self._windows = windows
 41        self._is_running = False
 42
 43        self._previous: PositionedLineList = []
 44        self._frametime = 0.0
 45        self._should_redraw: bool = True
 46        self._cache: dict[int, list[str]] = {}
 47
 48        self.fps = 0
 49        self.framerate = framerate
 50
 51    @property
 52    def terminal(self) -> Terminal:
 53        """Returns the current global terminal."""
 54
 55        return get_terminal()
 56
 57    def _draw_loop(self) -> None:
 58        """A loop that draws at regular intervals."""
 59
 60        framecount = 0
 61        last_frame = fps_start_time = time.perf_counter()
 62
 63        while self._is_running:
 64            elapsed = time.perf_counter() - last_frame
 65
 66            if elapsed < self._frametime:
 67                time.sleep(self._frametime - elapsed)
 68                continue
 69
 70            animator.step(elapsed)
 71
 72            last_frame = time.perf_counter()
 73            self.draw()
 74
 75            framecount += 1
 76
 77            if last_frame - fps_start_time >= 1:
 78                self.fps = framecount
 79                fps_start_time = last_frame
 80                framecount = 0
 81
 82    # NOTE: This is not needed at the moment, but might be at some point soon.
 83    # def _get_lines(self, window: Window) -> list[str]:
 84    #     """Gets lines from the window, caching when possible.
 85
 86    #     This also applies the blurred style of the window, if it has no focus.
 87    #     """
 88
 89    #     if window.allow_fullscreen:
 90    #         window.pos = self.terminal.origin
 91    #         window.width = self.terminal.width
 92    #         window.height = self.terminal.height
 93
 94    #     return window.get_lines()
 95
 96    #     if window.has_focus or window.is_noblur:
 97    #         return window.get_lines()
 98
 99    #     _id = id(window)
100    #     if not window.is_dirty and _id in self._cache:
101    #         return self._cache[_id]
102
103    #     lines: list[str] = []
104    #     for line in window.get_lines():
105    #         if not window.has_focus:
106    #             line = tim.parse("[239]" + strip_ansi(line).replace("[", r"\["))
107
108    #         lines.append(line)
109
110    #     self._cache[_id] = lines
111    #     return lines
112
113    @staticmethod
114    def _iter_positioned(
115        widget: Widget, until: int | None = None
116    ) -> Iterator[tuple[tuple[int, int], str]]:
117        """Iterates through (pos, line) tuples from widget.get_lines()."""
118
119        # get_lines = widget.get_lines
120        # if isinstance(widget, Window):
121        #     get_lines = lambda *_: self._get_lines(widget)  # type: ignore
122
123        if until is None:
124            until = widget.height
125
126        for i, line in enumerate(widget.get_lines()):
127            if i >= until:
128                break
129
130            pos = (widget.pos[0], widget.pos[1] + i)
131
132            yield (pos, line)
133
134    @property
135    def framerate(self) -> int:
136        """The framerate the draw loop runs at.
137
138        Note:
139            This will likely not be matched very accurately, mostly undershooting
140            the given target.
141        """
142
143        return self._framerate
144
145    @framerate.setter
146    def framerate(self, new: int) -> None:
147        """Updates the framerate."""
148
149        self._frametime = 1 / new
150        self._framerate = new
151
152    def clear_cache(self, window: Window) -> None:
153        """Clears the compositor's cache related to the given window."""
154
155        if id(window) in self._cache:
156            del self._cache[id(window)]
157
158    def run(self) -> None:
159        """Runs the compositor draw loop as a thread."""
160
161        self._is_running = True
162        Thread(name="CompositorDrawLoop", target=self._draw_loop).start()
163
164    def stop(self) -> None:
165        """Stops the compositor."""
166
167        self._is_running = False
168
169    def composite(self) -> PositionedLineList:
170        """Creates a composited buffer from the assigned windows.
171
172        Note that this is currently not used."""
173
174        lines = []
175        windows = self._windows
176
177        # Don't unnecessarily print under full screen windows
178        if any(window.allow_fullscreen for window in self._windows):
179            for window in reversed(self._windows):
180                if window.allow_fullscreen:
181                    windows = [window]
182                    break
183
184        size_changes = {WidgetChange.WIDTH, WidgetChange.HEIGHT, WidgetChange.SIZE}
185        for window in reversed(windows):
186            if not window.has_focus:
187                continue
188
189            change = window.get_change()
190
191            if change is None:
192                continue
193
194            if window.is_dirty or change in size_changes:
195                for pos, line in self._iter_positioned(window):
196                    lines.append((pos, line))
197
198                window.is_dirty = False
199                continue
200
201            if change is not None:
202                remaining = window.content_dimensions[1]
203
204                for widget in window.dirty_widgets:
205                    for pos, line in self._iter_positioned(widget, until=remaining):
206                        lines.append((pos, line))
207
208                    remaining -= widget.height
209
210                window.dirty_widgets = []
211                continue
212
213            if window.allow_fullscreen:
214                break
215
216        return lines
217
218    def set_redraw(self) -> None:
219        """Flags compositor for full redraw.
220
221        Note:
222            At the moment the compositor will always redraw the entire screen.
223        """
224
225        self._should_redraw = True
226
227    def draw(self, force: bool = False) -> None:
228        """Writes composited screen to the terminal.
229
230        At the moment this uses full-screen rewrites. There is a compositing
231        implementation in `composite`, but it is currently not performant enough to use.
232
233        Args:
234            force: When set, new composited lines will not be checked against the
235                previous ones, and everything will be redrawn.
236        """
237
238        # if self._should_redraw or force:
239        lines: PositionedLineList = []
240
241        for window in reversed(self._windows):
242            lines.extend(self._iter_positioned(window))
243
244        self._should_redraw = False
245
246        # else:
247        # lines = self.composite()
248
249        if not force and self._previous == lines:
250            return
251
252        buffer = "".join(f"\x1b[{pos[1]};{pos[0]}H{line}" for pos, line in lines)
253
254        self.terminal.clear_stream()
255        self.terminal.write(buffer)
256        self.terminal.flush()
257
258        self._previous = lines
259
260    def redraw(self) -> None:
261        """Force-redraws the buffer."""
262
263        self.draw(force=True)
264
265    def capture(self, title: str, filename: str | None = None) -> None:
266        """Captures the most-recently drawn buffer as `filename`.
267
268        See `pytermgui.exporters.to_svg` for more information.
269        """
270
271        with self.terminal.record() as recording:
272            self.redraw()
273
274        recording.save_svg(title=title, filename=filename)

The class used to draw pytermgui.window_managers.manager.WindowManager state.

This class handles turning a list of windows into a drawable buffer (composite), and then drawing it onto the screen.

Calling its run method will start the drawing thread, which will draw the current window states onto the screen. This routine targets framerate, though will likely not match it perfectly.

Compositor( windows: list[pytermgui.window_manager.window.Window], framerate: int)
33    def __init__(self, windows: list[Window], framerate: int) -> None:
34        """Initializes the Compositor.
35
36        Args:
37            windows: A list of the windows to be drawn.
38        """
39
40        self._windows = windows
41        self._is_running = False
42
43        self._previous: PositionedLineList = []
44        self._frametime = 0.0
45        self._should_redraw: bool = True
46        self._cache: dict[int, list[str]] = {}
47
48        self.fps = 0
49        self.framerate = framerate

Initializes the Compositor.

Args
  • windows: A list of the windows to be drawn.
framerate: int

The framerate the draw loop runs at.

Note

This will likely not be matched very accurately, mostly undershooting the given target.

Returns the current global terminal.

def clear_cache(self, window: pytermgui.window_manager.window.Window) -> None:
152    def clear_cache(self, window: Window) -> None:
153        """Clears the compositor's cache related to the given window."""
154
155        if id(window) in self._cache:
156            del self._cache[id(window)]

Clears the compositor's cache related to the given window.

def run(self) -> None:
158    def run(self) -> None:
159        """Runs the compositor draw loop as a thread."""
160
161        self._is_running = True
162        Thread(name="CompositorDrawLoop", target=self._draw_loop).start()

Runs the compositor draw loop as a thread.

def stop(self) -> None:
164    def stop(self) -> None:
165        """Stops the compositor."""
166
167        self._is_running = False

Stops the compositor.

def composite(self) -> List[Tuple[Tuple[int, int], str]]:
169    def composite(self) -> PositionedLineList:
170        """Creates a composited buffer from the assigned windows.
171
172        Note that this is currently not used."""
173
174        lines = []
175        windows = self._windows
176
177        # Don't unnecessarily print under full screen windows
178        if any(window.allow_fullscreen for window in self._windows):
179            for window in reversed(self._windows):
180                if window.allow_fullscreen:
181                    windows = [window]
182                    break
183
184        size_changes = {WidgetChange.WIDTH, WidgetChange.HEIGHT, WidgetChange.SIZE}
185        for window in reversed(windows):
186            if not window.has_focus:
187                continue
188
189            change = window.get_change()
190
191            if change is None:
192                continue
193
194            if window.is_dirty or change in size_changes:
195                for pos, line in self._iter_positioned(window):
196                    lines.append((pos, line))
197
198                window.is_dirty = False
199                continue
200
201            if change is not None:
202                remaining = window.content_dimensions[1]
203
204                for widget in window.dirty_widgets:
205                    for pos, line in self._iter_positioned(widget, until=remaining):
206                        lines.append((pos, line))
207
208                    remaining -= widget.height
209
210                window.dirty_widgets = []
211                continue
212
213            if window.allow_fullscreen:
214                break
215
216        return lines

Creates a composited buffer from the assigned windows.

Note that this is currently not used.

def set_redraw(self) -> None:
218    def set_redraw(self) -> None:
219        """Flags compositor for full redraw.
220
221        Note:
222            At the moment the compositor will always redraw the entire screen.
223        """
224
225        self._should_redraw = True

Flags compositor for full redraw.

Note

At the moment the compositor will always redraw the entire screen.

def draw(self, force: bool = False) -> None:
227    def draw(self, force: bool = False) -> None:
228        """Writes composited screen to the terminal.
229
230        At the moment this uses full-screen rewrites. There is a compositing
231        implementation in `composite`, but it is currently not performant enough to use.
232
233        Args:
234            force: When set, new composited lines will not be checked against the
235                previous ones, and everything will be redrawn.
236        """
237
238        # if self._should_redraw or force:
239        lines: PositionedLineList = []
240
241        for window in reversed(self._windows):
242            lines.extend(self._iter_positioned(window))
243
244        self._should_redraw = False
245
246        # else:
247        # lines = self.composite()
248
249        if not force and self._previous == lines:
250            return
251
252        buffer = "".join(f"\x1b[{pos[1]};{pos[0]}H{line}" for pos, line in lines)
253
254        self.terminal.clear_stream()
255        self.terminal.write(buffer)
256        self.terminal.flush()
257
258        self._previous = lines

Writes composited screen to the terminal.

At the moment this uses full-screen rewrites. There is a compositing implementation in composite, but it is currently not performant enough to use.

Args
  • force: When set, new composited lines will not be checked against the previous ones, and everything will be redrawn.
def redraw(self) -> None:
260    def redraw(self) -> None:
261        """Force-redraws the buffer."""
262
263        self.draw(force=True)

Force-redraws the buffer.

def capture(self, title: str, filename: str | None = None) -> None:
265    def capture(self, title: str, filename: str | None = None) -> None:
266        """Captures the most-recently drawn buffer as `filename`.
267
268        See `pytermgui.exporters.to_svg` for more information.
269        """
270
271        with self.terminal.record() as recording:
272            self.redraw()
273
274        recording.save_svg(title=title, filename=filename)

Captures the most-recently drawn buffer as filename.

See pytermgui.exporters.to_svg for more information.