Skip to content

compositor

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

Compositor

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.

Source code in pytermgui/window_manager/compositor.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
class Compositor:
    """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.
    """

    def __init__(self, windows: list[Window], framerate: int) -> None:
        """Initializes the Compositor.

        Args:
            windows: A list of the windows to be drawn.
        """

        self._windows = windows
        self._is_running = False

        self._previous: PositionedLineList = []
        self._frametime = 0.0
        self._should_redraw: bool = True
        self._cache: dict[int, list[str]] = {}

        self.fps = 0
        self.framerate = framerate

    @property
    def terminal(self) -> Terminal:
        """Returns the current global terminal."""

        return get_terminal()

    def _draw_loop(self) -> None:
        """A loop that draws at regular intervals."""

        framecount = 0
        last_frame = fps_start_time = time.perf_counter()

        while self._is_running:
            elapsed = time.perf_counter() - last_frame

            if elapsed < self._frametime:
                time.sleep(self._frametime - elapsed)
                continue

            animator.step(elapsed)

            last_frame = time.perf_counter()
            self.draw()

            framecount += 1

            if last_frame - fps_start_time >= 1:
                self.fps = framecount
                fps_start_time = last_frame
                framecount = 0

    # NOTE: This is not needed at the moment, but might be at some point soon.
    # def _get_lines(self, window: Window) -> list[str]:
    #     """Gets lines from the window, caching when possible.

    #     This also applies the blurred style of the window, if it has no focus.
    #     """

    #     if window.allow_fullscreen:
    #         window.pos = self.terminal.origin
    #         window.width = self.terminal.width
    #         window.height = self.terminal.height

    #     return window.get_lines()

    #     if window.has_focus or window.is_noblur:
    #         return window.get_lines()

    #     _id = id(window)
    #     if not window.is_dirty and _id in self._cache:
    #         return self._cache[_id]

    #     lines: list[str] = []
    #     for line in window.get_lines():
    #         if not window.has_focus:
    #             line = tim.parse("[239]" + strip_ansi(line).replace("[", r"\["))

    #         lines.append(line)

    #     self._cache[_id] = lines
    #     return lines

    def _iter_positioned(
        self, widget: Widget, until: int | None = None
    ) -> Iterator[tuple[tuple[int, int], str]]:
        """Iterates through (pos, line) tuples from widget.get_lines()."""

        # get_lines = widget.get_lines
        # if isinstance(widget, Window):
        #     get_lines = lambda *_: self._get_lines(widget)  # type: ignore
        width, height = self.terminal.size

        if until is None:
            until = widget.height

        for i, line in enumerate(widget.get_lines()[:until]):
            if i >= until:
                break

            pos = (widget.pos[0], widget.pos[1] + i)

            yield (pos, line)

        for item in widget.positioned_line_buffer.copy():
            pos, line = item

            if 0 <= pos[0] <= width and 0 <= pos[1] <= height:
                yield item

            widget.positioned_line_buffer.remove(item)

    @property
    def framerate(self) -> int:
        """The framerate the draw loop runs at.

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

        return self._framerate

    @framerate.setter
    def framerate(self, new: int) -> None:
        """Updates the framerate."""

        self._frametime = 1 / new
        self._framerate = new

    def clear_cache(self, window: Window) -> None:
        """Clears the compositor's cache related to the given window."""

        if id(window) in self._cache:
            del self._cache[id(window)]

    def run(self) -> None:
        """Runs the compositor draw loop as a thread."""

        self._is_running = True
        Thread(name="CompositorDrawLoop", target=self._draw_loop, daemon=True).start()

    def stop(self) -> None:
        """Stops the compositor."""

        self._is_running = False

    def composite(self) -> PositionedLineList:
        """Creates a composited buffer from the assigned windows.

        Note that this is currently not used."""

        lines = []
        windows = self._windows

        # Don't unnecessarily print under full screen windows
        if any(window.allow_fullscreen for window in self._windows):
            for window in reversed(self._windows):
                if window.allow_fullscreen:
                    windows = [window]
                    break

        size_changes = {WidgetChange.WIDTH, WidgetChange.HEIGHT, WidgetChange.SIZE}
        for window in reversed(windows):
            if not window.has_focus:
                continue

            change = window.get_change()

            if change is None:
                continue

            if window.is_dirty or change in size_changes:
                for pos, line in self._iter_positioned(window):
                    lines.append((pos, line))

                window.is_dirty = False
                continue

            if change is not None:
                remaining = window.content_dimensions[1]

                for widget in window.dirty_widgets:
                    for pos, line in self._iter_positioned(widget, until=remaining):
                        lines.append((pos, line))

                    remaining -= widget.height

                window.dirty_widgets = []
                continue

            if window.allow_fullscreen:
                break

        return lines

    def set_redraw(self) -> None:
        """Flags compositor for full redraw.

        Note:
            At the moment the compositor will always redraw the entire screen.
        """

        self._should_redraw = True

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

        # if self._should_redraw or force:
        lines: PositionedLineList = []

        for window in reversed(self._windows):
            lines.extend(self._iter_positioned(window))

        self._should_redraw = False

        # else:
        # lines = self.composite()

        if not force and self._previous == lines:
            return

        self.terminal.clear_stream()
        with self.terminal.frame() as frame:
            frame_write = frame.write

            for pos, line in lines:
                frame_write(f"\x1b[{pos[1]};{pos[0]}H{line}")

        self._previous = lines

    def redraw(self) -> None:
        """Force-redraws the buffer."""

        self.draw(force=True)

    def capture(self, title: str, filename: str | None = None) -> None:
        """Captures the most-recently drawn buffer as `filename`.

        See `pytermgui.exporters.to_svg` for more information.
        """

        with self.terminal.record() as recording:
            self.redraw()

        recording.save_svg(title=title, filename=filename)

framerate: int property writable

The framerate the draw loop runs at.

Note

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

terminal: Terminal property

Returns the current global terminal.

__init__(windows, framerate)

Initializes the Compositor.

Parameters:

Name Type Description Default
windows list[Window]

A list of the windows to be drawn.

 required
Source code in pytermgui/window_manager/compositor.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(self, windows: list[Window], framerate: int) -> None:
    """Initializes the Compositor.

    Args:
        windows: A list of the windows to be drawn.
    """

    self._windows = windows
    self._is_running = False

    self._previous: PositionedLineList = []
    self._frametime = 0.0
    self._should_redraw: bool = True
    self._cache: dict[int, list[str]] = {}

    self.fps = 0
    self.framerate = framerate

capture(title, filename=None)

Captures the most-recently drawn buffer as filename.

See pytermgui.exporters.to_svg for more information.

Source code in pytermgui/window_manager/compositor.py 
272
273
274
275
276
277
278
279
280
281
def capture(self, title: str, filename: str | None = None) -> None:
    """Captures the most-recently drawn buffer as `filename`.

    See `pytermgui.exporters.to_svg` for more information.
    """

    with self.terminal.record() as recording:
        self.redraw()

    recording.save_svg(title=title, filename=filename)

clear_cache(window)

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

Source code in pytermgui/window_manager/compositor.py 
158
159
160
161
162
def clear_cache(self, window: Window) -> None:
    """Clears the compositor's cache related to the given window."""

    if id(window) in self._cache:
        del self._cache[id(window)]

composite()

Creates a composited buffer from the assigned windows.

Note that this is currently not used.

Source code in pytermgui/window_manager/compositor.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def composite(self) -> PositionedLineList:
    """Creates a composited buffer from the assigned windows.

    Note that this is currently not used."""

    lines = []
    windows = self._windows

    # Don't unnecessarily print under full screen windows
    if any(window.allow_fullscreen for window in self._windows):
        for window in reversed(self._windows):
            if window.allow_fullscreen:
                windows = [window]
                break

    size_changes = {WidgetChange.WIDTH, WidgetChange.HEIGHT, WidgetChange.SIZE}
    for window in reversed(windows):
        if not window.has_focus:
            continue

        change = window.get_change()

        if change is None:
            continue

        if window.is_dirty or change in size_changes:
            for pos, line in self._iter_positioned(window):
                lines.append((pos, line))

            window.is_dirty = False
            continue

        if change is not None:
            remaining = window.content_dimensions[1]

            for widget in window.dirty_widgets:
                for pos, line in self._iter_positioned(widget, until=remaining):
                    lines.append((pos, line))

                remaining -= widget.height

            window.dirty_widgets = []
            continue

        if window.allow_fullscreen:
            break

    return lines

draw(force=False)

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.

Parameters:

Name Type Description Default
force bool

When set, new composited lines will not be checked against the previous ones, and everything will be redrawn.

 False
Source code in pytermgui/window_manager/compositor.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def draw(self, force: bool = False) -> None:
    """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.
    """

    # if self._should_redraw or force:
    lines: PositionedLineList = []

    for window in reversed(self._windows):
        lines.extend(self._iter_positioned(window))

    self._should_redraw = False

    # else:
    # lines = self.composite()

    if not force and self._previous == lines:
        return

    self.terminal.clear_stream()
    with self.terminal.frame() as frame:
        frame_write = frame.write

        for pos, line in lines:
            frame_write(f"\x1b[{pos[1]};{pos[0]}H{line}")

    self._previous = lines

redraw()

Force-redraws the buffer.

Source code in pytermgui/window_manager/compositor.py 
267
268
269
270
def redraw(self) -> None:
    """Force-redraws the buffer."""

    self.draw(force=True)

run()

Runs the compositor draw loop as a thread.

Source code in pytermgui/window_manager/compositor.py 
164
165
166
167
168
def run(self) -> None:
    """Runs the compositor draw loop as a thread."""

    self._is_running = True
    Thread(name="CompositorDrawLoop", target=self._draw_loop, daemon=True).start()

set_redraw()

Flags compositor for full redraw.

Note

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

Source code in pytermgui/window_manager/compositor.py 
224
225
226
227
228
229
230
231
def set_redraw(self) -> None:
    """Flags compositor for full redraw.

    Note:
        At the moment the compositor will always redraw the entire screen.
    """

    self._should_redraw = True

stop()

Stops the compositor.

Source code in pytermgui/window_manager/compositor.py 
170
171
172
173
def stop(self) -> None:
    """Stops the compositor."""

    self._is_running = False