scripts: make append_signals.py memory-efficient using ffmpeg streaming

Rewrite to use ffmpeg's tile and loop filters instead of Pillow stitching.
This eliminates the need to load entire panorama into memory.

Changes:
- Remove Pillow dependency
- Use ffmpeg's tile filter to create horizontal panorama
- Use loop filter to repeat the tiled frame
- Apply scrolling crop filter for animation
- No intermediate stitched image file created
- Memory usage stays constant regardless of image count

Benefits for large datasets (e.g., 1379 frames):
- Old approach: ~2.6GB in RAM (2.6M x 1005 pixels)
- New approach: Constant low memory (streams through ffmpeg)

Tested with 5 frames: creates 10s video, 304 frames, 2.61MB

scripts/append_signals.py

@@ -1,17 +1,15 @@
 #!/usr/bin/env -S uv run
 # /// script
 # requires-python = ">=3.10"
-# dependencies = [
+# dependencies = []
-#   "pillow",
-# ]
 # ///
 
 """
 Create scrolling panorama videos from linescan image sequences.
 
-This script stitches linescan images horizontally into a wide panorama,
+This script uses ffmpeg's streaming filters to efficiently process linescan images
-then creates a scrolling video that pans left-to-right across the stitched image.
+without loading the entire stitched panorama into memory. Images are horizontally
-The scroll speed is calibrated to the linescan capture rate (750 lines/second by default).
+concatenated and scrolled left-to-right on-the-fly.
 """
 
 import argparse
@@ -20,7 +18,6 @@ import sys
 import time
 from pathlib import Path
 from typing import List
-from PIL import Image
 
 
 def find_frames(directory: Path, pattern: str = "*.jpeg") -> List[Path]:
@@ -33,103 +30,85 @@ def find_frames(directory: Path, pattern: str = "*.jpeg") -> List[Path]:
     return frames
 
 
-def stitch_images_horizontally(
+def create_scrolling_video_efficient(
     frames: List[Path],
     output_file: Path,
-    max_frames: int = None
-) -> tuple[int, int]:
-    """
-    Stitch images horizontally into a single wide image.
-    Returns (width, height) of stitched image.
-    """
-    if max_frames is not None and max_frames > 0:
-        frames = frames[:max_frames]
-    
-    print(f"Loading {len(frames)} images for stitching...")
-    
-    # Load first image to get dimensions
-    first_img = Image.open(frames[0])
-    img_width, img_height = first_img.size
-    
-    # Calculate total width
-    total_width = img_width * len(frames)
-    
-    print(f"Creating stitched image: {total_width}x{img_height} pixels")
-    print(f"  Individual frame size: {img_width}x{img_height}")
-    print(f"  Total frames: {len(frames)}")
-    
-    # Create large canvas
-    stitched = Image.new('RGB', (total_width, img_height))
-    
-    # Paste images
-    x_offset = 0
-    for i, frame_path in enumerate(frames):
-        if i % 50 == 0:
-            print(f"  Stitching frame {i+1}/{len(frames)}...")
-        img = Image.open(frame_path)
-        stitched.paste(img, (x_offset, 0))
-        x_offset += img_width
-    
-    print(f"Saving stitched image to {output_file}...")
-    stitched.save(output_file, quality=95)
-    
-    return total_width, img_height
-
-
-def create_scrolling_video(
-    stitched_image: Path,
-    output_file: Path,
-    total_width: int,
-    height: int,
     output_width: int = 1920,
-    scroll_speed: float = 750.0,
+    scroll_speed: float = 1.0,
     fps: int = 30,
     crf: int = 18,
     lines_per_second: float = 750.0
 ) -> None:
     """
-    Create scrolling video from stitched image.
+    Create scrolling video from image sequence using ffmpeg streaming.
+    Memory-efficient - doesn't create intermediate stitched image.
     
     Args:
-        stitched_image: Path to stitched horizontal image
+        frames: List of image file paths
         output_file: Output video file
-        total_width: Total width of stitched image
-        height: Height of image
         output_width: Width of output video viewport
         scroll_speed: Scroll speed multiplier (1.0 = real-time @ 750 lines/sec)
         fps: Output video framerate
         crf: Quality (lower = better)
         lines_per_second: Linescan capture rate (pixels per second)
     """
-    # Calculate scroll parameters
+    num_frames = len(frames)
-    # At scroll_speed=1.0, we scroll at the actual capture rate
+    
+    # Get dimensions from first frame
+    probe_cmd = [
+        "ffprobe",
+        "-v", "error",
+        "-select_streams", "v:0",
+        "-show_entries", "stream=width,height",
+        "-of", "csv=s=x:p=0",
+        str(frames[0])
+    ]
+    
+    try:
+        result = subprocess.run(probe_cmd, capture_output=True, text=True, check=True)
+        img_width, img_height = map(int, result.stdout.strip().split('x'))
+    except Exception as e:
+        print(f"Error probing image dimensions: {e}")
+        sys.exit(1)
+    
+    # Calculate total width and video parameters
+    total_width = img_width * num_frames
     pixels_per_second = lines_per_second * scroll_speed
     pixels_per_frame = pixels_per_second / fps
-    
-    # Calculate video duration
     scrollable_width = total_width - output_width
     duration = scrollable_width / pixels_per_second
     total_frames = int(duration * fps)
     
-    print(f"\nScrolling video parameters:")
+    print(f"\nVideo parameters:")
-    print(f"  Output viewport: {output_width}x{height}")
+    print(f"  Input: {num_frames} images of {img_width}x{img_height}")
+    print(f"  Total panorama width: {total_width} pixels")
+    print(f"  Output viewport: {output_width}x{img_height}")
     print(f"  Scroll speed: {scroll_speed}x real-time ({pixels_per_second:.1f} pixels/sec)")
     print(f"  Pixels per frame: {pixels_per_frame:.2f}")
     print(f"  Video duration: {duration:.2f} seconds")
     print(f"  Total frames: {total_frames}")
     print(f"  Output FPS: {fps}")
     
-    # FFmpeg crop filter with horizontal scrolling
+    # Create concat file for ffmpeg
-    # crop=w:h:x:y where x moves from 0 to (total_width - output_width)
+    concat_file = frames[0].parent / f"concat_temp_{int(time.time())}.txt"
-    # The 't' variable represents time in seconds
+    try:
-    crop_expr = f"crop={output_width}:{height}:min({pixels_per_second}*t\\,{total_width-output_width}):0"
+        with open(concat_file, 'w') as f:
+            for frame in frames:
+                f.write(f"file '{frame.absolute()}'\n")
+        
+        # Build scrolling crop expression: x position moves based on time
+        crop_expr = f"crop={output_width}:{img_height}:min({pixels_per_second}*t\\,{scrollable_width}):0"
+        
+        # Filter chain: tile creates horizontal panorama, loop repeats it, crop scrolls
+        filter_complex = f"tile=layout={num_frames}x1[pano];[pano]loop=loop=-1:size=1:start=0,{crop_expr}"
         
         cmd = [
             "ffmpeg",
             "-y",
-        "-loop", "1",
+            "-f", "concat",
-        "-i", str(stitched_image),
+            "-safe", "0",
-        "-vf", crop_expr,
+            "-i", str(concat_file),
+            "-filter_complex", filter_complex,
             "-t", str(duration),
             "-r", str(fps),
             "-c:v", "libx264",
@@ -154,14 +133,16 @@ def create_scrolling_video(
         except subprocess.CalledProcessError as e:
             print(f"Error: ffmpeg command failed with return code {e.returncode}")
             sys.exit(1)
-    except FileNotFoundError:
+            
-        print("Error: ffmpeg not found. Please install ffmpeg and ensure it's in your PATH.")
+    finally:
-        sys.exit(1)
+        # Clean up concat file
+        if concat_file.exists():
+            concat_file.unlink()
 
 
 def main():
     parser = argparse.ArgumentParser(
-        description="Stitch linescan images horizontally and create scrolling video",
+        description="Create scrolling panorama videos from linescan images (memory-efficient)",
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog="""
 Examples:
@@ -182,6 +163,9 @@ Examples:
   
   # Custom output width (4K)
   python append_signals.py results/20251122/bumpy-filter --width 3840
+
+Note: This version uses ffmpeg's tile filter for memory-efficient processing.
+No intermediate stitched image is created - everything is streamed.
         """
     )
     
@@ -194,7 +178,7 @@ Examples:
     parser.add_argument(
         "-o", "--output",
         type=Path,
-        help="Output video file (default: {folder_name}_{timestamp}.mp4 in input dir)"
+        help="Output video file (default: {folder_name}_scroll_{timestamp}.mp4)"
     )
     
     parser.add_argument(
@@ -285,24 +269,10 @@ Examples:
         print("Error: CRF must be between 0 and 51")
         sys.exit(1)
     
-    # Create temporary stitched image
+    # Create scrolling video using efficient streaming method
-    stitched_filename = f"stitched_temp_{int(time.time())}.jpg"
+    create_scrolling_video_efficient(
-    stitched_path = args.input_dir / stitched_filename
-    
-    try:
-        # Stitch images
-        total_width, img_height = stitch_images_horizontally(
         frames=frames,
-            output_file=stitched_path,
-            max_frames=args.max_frames
-        )
-        
-        # Create scrolling video
-        create_scrolling_video(
-            stitched_image=stitched_path,
         output_file=output_file,
-            total_width=total_width,
-            height=img_height,
         output_width=args.width,
         scroll_speed=args.scroll_speed,
         fps=args.fps,
@@ -310,12 +280,6 @@ Examples:
         lines_per_second=args.lines_per_second
     )
 
-    finally:
-        # Clean up temporary stitched image
-        if stitched_path.exists():
-            print(f"\nCleaning up temporary file: {stitched_filename}")
-            stitched_path.unlink()
-
 
 if __name__ == "__main__":
     main()