perf: Optimize asciinema cast processing with line-by-line streaming and early exit

This commit is contained in:
Paul Gauthier (aider) 2025-03-13 16:08:53 -07:00
parent 03733516cc
commit 520eb4a932

View file

@ -1,7 +1,6 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import json import json
import os import os
import re
import sys import sys
import pyte import pyte
@ -15,45 +14,48 @@ def main():
input_file = sys.argv[1] input_file = sys.argv[1]
output_file = sys.argv[2] output_file = sys.argv[2]
# Count total lines for progress bar
total_lines = sum(1 for _ in open(input_file, 'r'))
# Read and parse the cast file with open(input_file, "r") as fin, open(output_file, "w") as fout:
with open(input_file, "r") as f: # Process header
# First line is header header = fin.readline().strip()
header = f.readline().strip() fout.write(header + "\n")
# Parse header to extract terminal dimensions # Parse header for terminal dimensions
header_data = json.loads(header) header_data = json.loads(header)
width = header_data.get("width", 80) width = header_data.get("width", 80)
height = header_data.get("height", 24) height = header_data.get("height", 24)
print(f"Terminal dimensions: {width}x{height}") print(f"Terminal dimensions: {width}x{height}")
# Initialize pyte screen and stream with dimensions from header # Initialize terminal emulator
screen = pyte.Screen(width, height) screen = pyte.Screen(width, height)
stream = pyte.Stream(screen) stream = pyte.Stream(screen)
# Read the events # Process events line by line
events = [json.loads(line) for line in f if line.strip()] for line in tqdm(fin, desc="Processing events", total=total_lines-1):
if not line.strip():
# Write the header to the output file continue
with open(output_file, "w") as f:
f.write(header + "\n") event = json.loads(line)
# Process each event through the terminal emulator and stream to file # Only run terminal emulation for output events
for event in tqdm(events, desc="Processing events"): if len(event) >= 3 and event[1] == "o":
# Process the event in the terminal
if len(event) >= 3 and event[1] == "o": # Output event
stream.feed(event[2]) stream.feed(event[2])
# Check if "Atuin" is visible on screen # Check if "Atuin" is visible on screen - exit early if found
if any( atuin_visible = False
("Atuin" in line) for display_line in screen.display:
for line in screen.display if "Atuin" in display_line:
): atuin_visible = True
break
if atuin_visible:
continue # Skip this event continue # Skip this event
# Write this event directly to the output file # Write event to output file
f.write(json.dumps(event) + "\n") fout.write(json.dumps(event) + "\n")
if __name__ == "__main__": if __name__ == "__main__":