"""Timelapse video processing service using FFmpeg.""" import asyncio import contextlib import json import logging import tempfile from collections.abc import Callable from pathlib import Path from backend.app.services.camera import get_ffmpeg_path from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr logger = logging.getLogger(__name__) # ffprobe and a single thumbnail frame take well under a second on a healthy # file; these bound a broken or unreadable one, which runs inside a request. PROBE_TIMEOUT_SECONDS = 60 THUMBNAIL_TIMEOUT_SECONDS = 30 async def _communicate(process: asyncio.subprocess.Process, timeout: float) -> tuple[bytes, bytes]: """``process.communicate()`` that kills the process when it takes too long.""" try: return await asyncio.wait_for(process.communicate(), timeout) except (TimeoutError, asyncio.CancelledError): _kill(process) await process.wait() raise def _kill(process: asyncio.subprocess.Process) -> None: with contextlib.suppress(ProcessLookupError): process.kill() class TimelapseProcessor: """Service for processing timelapse videos with FFmpeg.""" def __init__(self, input_path: Path): self.input_path = input_path self.ffmpeg = get_ffmpeg_path() if not self.ffmpeg: raise RuntimeError("FFmpeg not found") # Derive ffprobe path from ffmpeg path self.ffprobe = self.ffmpeg.replace("ffmpeg", "ffprobe") async def get_info(self) -> dict: """Get video metadata using ffprobe.""" cmd = [ self.ffprobe, "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", str(self.input_path), ] process = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) try: stdout, stderr = await _communicate(process, PROBE_TIMEOUT_SECONDS) except TimeoutError: raise RuntimeError(f"ffprobe took longer than {PROBE_TIMEOUT_SECONDS} s") from None if process.returncode != 0: # Summarised once and used for both: the raise carried a second, # bare ``stderr.decode()`` that could itself raise UnicodeDecodeError # on the bytes ffprobe copies out of a broken file (#2968). detail = summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT logger.error("ffprobe failed: %s", detail) raise RuntimeError(f"ffprobe failed: {detail}") data = json.loads(stdout.decode()) video_stream = next( (s for s in data.get("streams", []) if s.get("codec_type") == "video"), {}, ) audio_stream = next( (s for s in data.get("streams", []) if s.get("codec_type") == "audio"), None, ) # Parse frame rate (can be "30/1" or "29.97") fps = 30.0 r_frame_rate = video_stream.get("r_frame_rate", "30/1") try: if "/" in r_frame_rate: num, den = r_frame_rate.split("/") fps = float(num) / float(den) else: fps = float(r_frame_rate) except (ValueError, ZeroDivisionError): pass # Keep default fps if frame rate string is unparseable return { "duration": float(data.get("format", {}).get("duration", 0)), "width": video_stream.get("width", 0), "height": video_stream.get("height", 0), "fps": fps, "codec": video_stream.get("codec_name", "unknown"), "file_size": int(data.get("format", {}).get("size", 0)), "has_audio": audio_stream is not None, } async def generate_thumbnails( self, count: int = 10, width: int = 160, ) -> list[tuple[float, bytes]]: """Generate evenly-spaced thumbnail frames.""" info = await self.get_info() duration = info["duration"] if duration <= 0: return [] interval = duration / max(count, 1) thumbnails = [] with tempfile.TemporaryDirectory() as tmpdir: for i in range(count): timestamp = i * interval output_path = Path(tmpdir) / f"thumb_{i:03d}.jpg" cmd = [ self.ffmpeg, "-y", "-ss", str(timestamp), "-i", str(self.input_path), "-vframes", "1", "-vf", f"scale={width}:-1", "-q:v", "5", str(output_path), ] process = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) try: await _communicate(process, THUMBNAIL_TIMEOUT_SECONDS) except TimeoutError: logger.warning("Timelapse thumbnail at %.1f s took too long; skipped", timestamp) continue if output_path.exists(): thumbnails.append((timestamp, output_path.read_bytes())) return thumbnails async def process( self, output_path: Path, trim_start: float = 0, trim_end: float | None = None, speed: float = 1.0, audio_path: Path | None = None, audio_volume: float = 1.0, on_progress: Callable[[float], None] | None = None, ) -> bool: """Process video with trim, speed, and optional audio overlay. Args: output_path: Where to save the processed video trim_start: Start time in seconds trim_end: End time in seconds (None = full duration) speed: Speed multiplier (0.25 to 4.0) audio_path: Optional music file to overlay audio_volume: Volume for audio overlay (0.0 to 1.0) on_progress: Called with the share done (0.0 to 1.0) as ffmpeg reports its position Returns: True if processing succeeded, False otherwise Cancelling the call stops ffmpeg and removes the partial output. """ # The output's length, for progress: what is kept of the input, at # the new speed. An audio overlay with -shortest may end it earlier. expected_seconds = None if on_progress is not None: end = trim_end if trim_end is not None and trim_end > trim_start else None if end is None: try: end = (await self.get_info())["duration"] except Exception as e: # Progress is a nicety: encode without it. logger.warning("Timelapse length unknown, no progress: %s", e) if end is not None: expected_seconds = max(end - trim_start, 0) / speed # Build FFmpeg command cmd = [self.ffmpeg, "-y"] # Input seeking (fast seek before input) if trim_start > 0: cmd.extend(["-ss", str(trim_start)]) cmd.extend(["-i", str(self.input_path)]) # Add audio input if provided if audio_path: cmd.extend(["-i", str(audio_path)]) # Duration limit if trim_end is not None and trim_end > trim_start: duration = trim_end - trim_start cmd.extend(["-t", str(duration)]) # Build filters - use filter_complex when we have audio overlay video_filter = "" if speed != 1.0: # setpts changes video speed: PTS/speed = faster, PTS*speed = slower setpts_value = 1.0 / speed video_filter = f"setpts={setpts_value}*PTS" if audio_path: # Use filter_complex for audio overlay (can't mix with -vf/-af) filter_parts = [] # Video filter if video_filter: filter_parts.append(f"[0:v]{video_filter}[v]") video_out = "[v]" else: video_out = "0:v" # Audio filter with volume filter_parts.append(f"[1:a]volume={audio_volume}[a]") cmd.extend(["-filter_complex", ";".join(filter_parts)]) cmd.extend(["-map", video_out, "-map", "[a]"]) cmd.extend(["-shortest"]) elif speed != 1.0: # No audio overlay - use simple -vf and -af if video_filter: cmd.extend(["-vf", video_filter]) # Adjust original audio speed with atempo atempo_chain = self._build_atempo_chain(speed) if atempo_chain: cmd.extend(["-af", atempo_chain]) # Output settings cmd.extend( [ "-c:v", "libx264", "-preset", "fast", "-crf", "23", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", # Enable streaming # Position reports on stdout, for progress. "-progress", "pipe:1", "-nostats", str(output_path), ] ) logger.info("Processing timelapse: %s", " ".join(cmd)) # Run FFmpeg process = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) try: _, stderr = await asyncio.gather( self._read_progress(process, expected_seconds, on_progress), process.stderr.read(), # type: ignore[union-attr] ) await process.wait() except asyncio.CancelledError: _kill(process) await process.wait() output_path.unlink(missing_ok=True) raise if process.returncode != 0: logger.error("FFmpeg processing failed: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT) return False return output_path.exists() @staticmethod async def _read_progress( process: asyncio.subprocess.Process, expected_seconds: float | None, on_progress: Callable[[float], None] | None, ) -> None: """Read ffmpeg's ``-progress`` lines, passing on the share done.""" assert process.stdout is not None async for raw in process.stdout: key, _, value = raw.decode(errors="replace").strip().partition("=") if on_progress is None or not expected_seconds or key != "out_time_us": continue try: on_progress(int(value) / 1_000_000 / expected_seconds) except ValueError: pass # "N/A" before the first frame def _build_atempo_chain(self, speed: float) -> str: """Build atempo filter chain. atempo filter only supports values between 0.5 and 2.0, so we chain multiple filters for extreme speeds. """ if speed == 1.0: return "" filters = [] remaining_speed = speed # Handle speeds > 2.0 by chaining atempo=2.0 while remaining_speed > 2.0: filters.append("atempo=2.0") remaining_speed /= 2.0 # Handle speeds < 0.5 by chaining atempo=0.5 while remaining_speed < 0.5: filters.append("atempo=0.5") remaining_speed *= 2.0 # Add final atempo for remaining adjustment # After the while loops above, remaining_speed is guaranteed to be in [0.5, 2.0] if remaining_speed != 1.0: filters.append(f"atempo={remaining_speed:.4f}") return ",".join(filters)