From b70b22253f34fe27fdea2c921b4804130b1f5535 Mon Sep 17 00:00:00 2001 From: "Abe Diaz (@abe238)" Date: Tue, 26 May 2026 02:05:30 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20timelapse=20stitching=20=E2=80=94=20bam?= =?UTF-8?q?bu=20timelapse=20stitch=20(#15)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bambu timelapse stitch --out OUT.mp4 [--fps N] [--pattern GLOB] Uses ffmpeg concat demuxer to stitch frames captured by 'bambu stream'. Frames are ordered by filename (timestamp prefix works correctly). Default fps=24, output is H.264/yuv420p in MP4. Bypasses Bambu Studio's much-complained-about 'dramatic zoom-out only' A1 timelapse entirely. Workflow: bambu stream --interval 30 --out-dir ./tl & # ... print runs ... bambu timelapse stitch ./tl --out my-print.mp4 6 unit tests in tests/test_timelapse.py (ffmpeg mocked) + 1 integration test (gated, runs ffmpeg for real locally). 134 total passing. Closes #15. --- completions/_bambu | 1 + completions/bambu.bash | 2 +- src/bambu_ai/cli.py | 26 ++++++++++ src/bambu_ai/timelapse.py | 87 +++++++++++++++++++++++++++++++ tests/test_timelapse.py | 106 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 221 insertions(+), 1 deletion(-) create mode 100644 src/bambu_ai/timelapse.py create mode 100644 tests/test_timelapse.py diff --git a/completions/_bambu b/completions/_bambu index e281329..bbbd75c 100644 --- a/completions/_bambu +++ b/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)' ) diff --git a/completions/bambu.bash b/completions/bambu.bash index c800b22..42e2731 100644 --- a/completions/bambu.bash +++ b/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 diff --git a/src/bambu_ai/cli.py b/src/bambu_ai/cli.py index 28ad3b6..6509a99 100644 --- a/src/bambu_ai/cli.py +++ b/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)") diff --git a/src/bambu_ai/timelapse.py b/src/bambu_ai/timelapse.py new file mode 100644 index 0000000..24969bd --- /dev/null +++ b/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)" diff --git a/tests/test_timelapse.py b/tests/test_timelapse.py new file mode 100644 index 0000000..c4405ec --- /dev/null +++ b/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