feat: timelapse stitching — bambu timelapse stitch (#15)

completions/_bambu

@@ -19,6 +19,7 @@ _bambu() {
         'queue:persistent print queue (add/list/clear/start)'
         'snap:capture one frame from the chamber camera'
         'stream:capture frames periodically into a folder'
+        'timelapse:stitch captured frames into MP4'
         'vision:AI failure detection (watch / classify)'
     )
 

completions/bambu.bash

@@ -12,7 +12,7 @@ _bambu() {
     }
 
     if [ "${cword}" -eq 1 ]; then
-        COMPREPLY=( $(compgen -W "status pause resume cancel home light print queue schedule quiet auto-off snap stream vision" -- "${cur}") )
+        COMPREPLY=( $(compgen -W "status pause resume cancel home light print queue schedule quiet auto-off snap stream timelapse vision" -- "${cur}") )
         return
     fi
 

src/bambu_ai/cli.py

@@ -87,6 +87,23 @@ def _cmd_light(p, args):
 # ----------------------------------------------------------------- new subcommands
 
 
+def _cmd_timelapse_stitch(args) -> None:
+    """Stitch a folder of JPEG frames into an MP4 (issue #15)."""
+    from pathlib import Path
+
+    from .timelapse import stitch_frames
+
+    ok, msg = stitch_frames(
+        Path(args.frames_dir),
+        Path(args.out),
+        fps=args.fps,
+        pattern=args.pattern,
+    )
+    print(f"[timelapse] {msg}")
+    if not ok:
+        sys.exit(1)
+
+
 def _cmd_queue_add(args) -> None:
     from pathlib import Path
 
@@ -493,6 +510,15 @@ def main() -> None:
     snap.add_argument("out", help="output path (e.g. frame.jpg)")
     snap.set_defaults(func=_cmd_snap)
 
+    tl = sub.add_parser("timelapse", help="stitch captured frames into an MP4 (issue #15)")
+    tlsub = tl.add_subparsers(dest="timelapse_cmd", required=True)
+    tlstitch = tlsub.add_parser("stitch", help="ffmpeg-stitch JPEGs in a folder into an MP4")
+    tlstitch.add_argument("frames_dir", help="folder of JPEG frames (use `bambu stream` to capture)")
+    tlstitch.add_argument("--out", required=True, help="output MP4 path")
+    tlstitch.add_argument("--fps", type=int, default=24, help="output frame rate (default: 24)")
+    tlstitch.add_argument("--pattern", default="*.jpg", help="glob pattern for frame files (default: *.jpg)")
+    tlstitch.set_defaults(func=_cmd_timelapse_stitch)
+
     stream = sub.add_parser("stream", help="capture frames periodically into a folder")
     stream.add_argument("--out-dir", default="./frames", help="output directory (default: ./frames)")
     stream.add_argument("--interval", type=float, default=30.0, help="seconds between frames (default: 30)")

src/bambu_ai/timelapse.py

@@ -0,0 +1,87 @@
+"""Timelapse stitching (issue #15).
+
+The A1 mini's built-in timelapse is the much-complained-about 'dramatic
+zoom-out only' variant. This module ignores Bambu's pause-between-layers
+approach entirely — it just stitches frames captured by ``bambu stream``
+(see :mod:`bambu_ai.camera`) with ffmpeg.
+
+Requires the ``ffmpeg`` binary on PATH.
+"""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+from pathlib import Path
+
+
+def _sorted_frame_paths(directory: Path, pattern: str = "*.jpg") -> list[Path]:
+    """Return frames in deterministic order. Sorted by filename — which works
+    because :func:`bambu_ai.camera.stream_frames` names frames with a Unix
+    timestamp prefix."""
+    return sorted(directory.glob(pattern))
+
+
+def stitch_frames(
+    frames_dir: Path,
+    out_path: Path,
+    *,
+    fps: int = 24,
+    pattern: str = "*.jpg",
+    ffmpeg_path: str | None = None,
+    runner=subprocess.run,
+) -> tuple[bool, str]:
+    """ffmpeg-stitch every JPEG in ``frames_dir`` into a video at ``out_path``.
+
+    Returns ``(ok, message)``. ``runner`` is injected for tests.
+    """
+    if ffmpeg_path is None:
+        ffmpeg_path = shutil.which("ffmpeg") or "ffmpeg"
+
+    frames = _sorted_frame_paths(frames_dir, pattern)
+    if len(frames) < 2:
+        return False, f"need at least 2 frames in {frames_dir}, found {len(frames)}"
+
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+
+    # Build a concat-friendly file list (handles non-sequential timestamps and
+    # arbitrary file names — more robust than ffmpeg's image2 glob/sequence).
+    listfile = out_path.with_suffix(".txt")
+    listfile.write_text(
+        "\n".join(f"file '{f.absolute()}'\nduration {1.0 / fps:.6f}" for f in frames)
+        + f"\nfile '{frames[-1].absolute()}'\n"
+    )
+
+    cmd = [
+        ffmpeg_path,
+        "-y",
+        "-f",
+        "concat",
+        "-safe",
+        "0",
+        "-i",
+        str(listfile),
+        "-vsync",
+        "vfr",
+        "-pix_fmt",
+        "yuv420p",
+        "-c:v",
+        "libx264",
+        "-crf",
+        "23",
+        str(out_path),
+    ]
+    try:
+        result = runner(cmd, capture_output=True, text=True, timeout=600)
+    except FileNotFoundError:
+        return False, f"ffmpeg not found at {ffmpeg_path}"
+    except subprocess.TimeoutExpired:
+        return False, "ffmpeg timed out after 10 minutes"
+    finally:
+        listfile.unlink(missing_ok=True)
+
+    if result.returncode != 0:
+        return False, f"ffmpeg rc={result.returncode}: {result.stderr[-500:]}"
+    if not out_path.is_file() or out_path.stat().st_size == 0:
+        return False, "ffmpeg succeeded but produced no output file"
+    return True, f"stitched {len(frames)} frames -> {out_path} ({out_path.stat().st_size} bytes)"

tests/test_timelapse.py

@@ -0,0 +1,106 @@
+"""Tests for the timelapse stitcher (issue #15)."""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from bambu_ai.timelapse import _sorted_frame_paths, stitch_frames
+
+
+def _make_frame(path: Path) -> None:
+    """Drop a tiny but valid JPEG at ``path``."""
+    from PIL import Image
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.new("RGB", (8, 8), color=(0, 0, 0)).save(path, format="JPEG")
+
+
+class TestSortedFramePaths:
+    def test_returns_in_filename_order(self, tmp_path) -> None:
+        for name in ("frame-1010.jpg", "frame-1001.jpg", "frame-1100.jpg"):
+            _make_frame(tmp_path / name)
+        order = [p.name for p in _sorted_frame_paths(tmp_path)]
+        assert order == ["frame-1001.jpg", "frame-1010.jpg", "frame-1100.jpg"]
+
+
+class TestStitchFrames:
+    def test_rejects_too_few_frames(self, tmp_path) -> None:
+        _make_frame(tmp_path / "f1.jpg")
+        ok, msg = stitch_frames(
+            tmp_path,
+            tmp_path / "out.mp4",
+            runner=MagicMock(side_effect=AssertionError("should not run")),
+        )
+        assert not ok
+        assert "at least 2" in msg
+
+    def test_calls_ffmpeg_with_expected_args(self, tmp_path) -> None:
+        for n in (1, 2, 3):
+            _make_frame(tmp_path / f"frame-{n}.jpg")
+        out = tmp_path / "out.mp4"
+
+        # Mock ffmpeg: write an output file so the size check passes.
+        def fake_runner(cmd, **_kw):
+            out.write_bytes(b"FAKEMP4")
+            return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+
+        ok, msg = stitch_frames(tmp_path, out, fps=12, runner=fake_runner, ffmpeg_path="ffmpeg")
+        assert ok, msg
+        assert out.read_bytes() == b"FAKEMP4"
+
+    def test_propagates_ffmpeg_failure(self, tmp_path) -> None:
+        for n in (1, 2):
+            _make_frame(tmp_path / f"f{n}.jpg")
+        out = tmp_path / "out.mp4"
+
+        def fake_runner(cmd, **_kw):
+            return subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="boom")
+
+        ok, msg = stitch_frames(tmp_path, out, runner=fake_runner, ffmpeg_path="ffmpeg")
+        assert not ok
+        assert "rc=1" in msg
+        assert "boom" in msg
+
+    def test_handles_missing_ffmpeg(self, tmp_path) -> None:
+        for n in (1, 2):
+            _make_frame(tmp_path / f"f{n}.jpg")
+        out = tmp_path / "out.mp4"
+
+        def fake_runner(*_a, **_kw):
+            raise FileNotFoundError("ffmpeg")
+
+        ok, msg = stitch_frames(tmp_path, out, runner=fake_runner, ffmpeg_path="/nonexistent/ffmpeg")
+        assert not ok
+        assert "not found" in msg
+
+    def test_cleans_up_listfile(self, tmp_path) -> None:
+        for n in (1, 2):
+            _make_frame(tmp_path / f"f{n}.jpg")
+        out = tmp_path / "out.mp4"
+
+        def fake_runner(cmd, **_kw):
+            out.write_bytes(b"x")
+            return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+
+        stitch_frames(tmp_path, out, runner=fake_runner, ffmpeg_path="ffmpeg")
+        listfile = out.with_suffix(".txt")
+        assert not listfile.exists()
+
+
+@pytest.mark.integration
+def test_real_ffmpeg_stitches_two_frames(tmp_path) -> None:
+    """Marked integration — only runs locally via `pytest -m integration`. Hits real ffmpeg."""
+    import shutil
+
+    if not shutil.which("ffmpeg"):
+        pytest.skip("ffmpeg not installed")
+    for n in (1, 2, 3):
+        _make_frame(tmp_path / f"frame-{n}.jpg")
+    out = tmp_path / "out.mp4"
+    ok, msg = stitch_frames(tmp_path, out)
+    assert ok, msg
+    assert out.stat().st_size > 0