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)
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.
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.
The framerate the draw loop runs at.
Note
This will likely not be matched very accurately, mostly undershooting the given target.
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.
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.
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.
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.
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.
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.