docs: add comments to explain MarkdownStream class and methods

This commit is contained in:
Paul Gauthier (aider) 2025-01-07 05:57:35 -08:00
parent 37ad4758a1
commit be82b6bed9

View file

@ -41,72 +41,97 @@ The end.
class MarkdownStream:
live = None
when = 0
min_delay = 0.050
live_window = 6
"""Streaming markdown renderer that progressively displays content with a live updating window.
Uses rich.console and rich.live to render markdown content with smooth scrolling and partial updates.
Maintains a sliding window of visible content while streaming in new markdown text.
"""
live = None # Rich Live display instance
when = 0 # Timestamp of last update
min_delay = 0.050 # Minimum time between updates (20fps)
live_window = 6 # Number of lines to keep visible at bottom during streaming
def __init__(self, mdargs=None):
self.printed = []
"""Initialize the markdown stream.
Args:
mdargs (dict, optional): Additional arguments to pass to rich Markdown renderer
"""
self.printed = [] # Stores lines that have already been printed
if mdargs:
self.mdargs = mdargs
else:
self.mdargs = dict()
# Initialize rich Live display with empty text
self.live = Live(Text(""), refresh_per_second=1.0 / self.min_delay)
self.live.start()
def __del__(self):
"""Destructor to ensure Live display is properly cleaned up."""
if self.live:
try:
self.live.stop()
except Exception:
pass
pass # Ignore any errors during cleanup
def update(self, text, final=False):
"""Update the displayed markdown content.
Args:
text (str): New markdown text to append
final (bool): If True, this is the final update and we should clean up
"""
now = time.time()
# Throttle updates to maintain smooth rendering
if not final and now - self.when < self.min_delay:
return
self.when = now
# Render the markdown to a string buffer
string_io = io.StringIO()
console = Console(file=string_io, force_terminal=True)
markdown = Markdown(text, **self.mdargs)
console.print(markdown)
output = string_io.getvalue()
# Split rendered output into lines
lines = output.splitlines(keepends=True)
num_lines = len(lines)
# During streaming, keep a window of lines at the bottom visible
if not final:
num_lines -= self.live_window
# If we have new content to display...
if final or num_lines > 0:
num_printed = len(self.printed)
show = num_lines - num_printed
# Skip if no new lines to show
if show <= 0:
return
# Get the new lines and display them
show = lines[num_printed:num_lines]
show = "".join(show)
show = Text.from_ansi(show)
self.live.console.print(show)
# Update our record of printed lines
self.printed = lines[:num_lines]
# Handle final update cleanup
if final:
self.live.update(Text(""))
self.live.stop()
self.live = None
else:
# Update the live window with remaining lines
rest = lines[num_lines:]
rest = "".join(rest)
# rest = '...\n' + rest
rest = Text.from_ansi(rest)
self.live.update(rest)