pytermgui.window_manager.compositor

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

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

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 )
View Source
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

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:
View Source
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)]

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

#   def run(self) -> None:
View Source
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()

Runs the compositor draw loop as a thread.

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

Stops the compositor.

#   def composite(self) -> List[Tuple[Tuple[int, int], str]]:
View Source
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

Creates a composited buffer from the assigned windows.

Note that this is currently not used.

#   def set_redraw(self) -> None:
View Source
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

Flags compositor for full redraw.

Note

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

#   def draw(self, force: bool = False) -> None:
View Source
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

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:
View Source
259    def redraw(self) -> None:
260        """Force-redraws the buffer."""
261
262        self.draw(force=True)

Force-redraws the buffer.

#   def capture(self, title: str, filename: str | None = None) -> None:
View Source
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)

Captures the most-recently drawn buffer as filename.

See pytermgui.exporters.to_svg for more information.