from __future__ import annotations from dataclasses import dataclass from pathlib import Path from app.projects.editor_schemas import AudioClip, EditorDocument, MediaClip from app.projects.errors import ProjectRenderInvalidError, ProjectRenderUnsupportedError @dataclass(frozen=True, slots=True) class RenderInput: asset_id: str path: Path kind: str source_start_ms: int duration_ms: int input_index: int @dataclass(frozen=True, slots=True) class RenderPlan: args: tuple[str | Path, ...] inputs: tuple[RenderInput, ...] duration_ms: int output_extension: str def validate_renderable_document(document: EditorDocument) -> None: if document.timeline.transitions: raise ProjectRenderUnsupportedError( "Timeline transitions are not supported by the render compiler yet." ) for track in document.timeline.tracks: if not track.visible: continue for clip in track.clips: if clip.visible and clip.kind in {"caption", "effect"}: raise ProjectRenderUnsupportedError( f"The render compiler does not support {clip.kind} clips yet." ) def compile_render( document: EditorDocument, *, asset_paths: dict[str, tuple[Path, str]], width: int, height: int, frame_rate: float, output_format: str, quality: str, preset: str, ) -> RenderPlan: """Compile the currently supported editor primitives into FFmpeg args. The compiler is deterministic: all inputs, offsets, trims, canvas settings, and codecs are derived solely from the immutable editor snapshot/settings. Captions, effects, transitions, and unsupported clip types fail closed. """ if output_format not in {"mp4", "webm"}: raise ProjectRenderInvalidError("Unsupported render output format.") if width % 2 or height % 2 or width < 2 or height < 2: raise ProjectRenderInvalidError("Render resolution must be positive and even.") validate_renderable_document(document) inputs: list[RenderInput] = [] video_clips: list[tuple[MediaClip, RenderInput]] = [] audio_clips: list[tuple[AudioClip, RenderInput]] = [] for track in sorted(document.timeline.tracks, key=lambda item: item.order): if not track.visible: continue for clip in track.clips: if not clip.visible: continue if clip.kind in {"caption", "effect"}: raise ProjectRenderUnsupportedError( f"The render compiler does not support {clip.kind} clips yet." ) if clip.kind == "audio" and track.muted: continue if clip.asset_id not in asset_paths: raise ProjectRenderInvalidError("A referenced media asset is unavailable.") path, mime_type = asset_paths[clip.asset_id] if not path.is_file(): raise ProjectRenderInvalidError("A referenced media asset file is unavailable.") input_index = len(inputs) + 1 # 0 is the generated black canvas. item = RenderInput( asset_id=clip.asset_id, path=path, kind=clip.kind, source_start_ms=clip.source_start_ms, duration_ms=clip.duration_ms, input_index=input_index, ) inputs.append(item) if clip.kind == "media": if clip.media_type == "image" and not mime_type.startswith("image/"): raise ProjectRenderInvalidError("An image clip references a non-image asset.") if clip.media_type == "video" and not mime_type.startswith("video/"): raise ProjectRenderInvalidError("A video clip references a non-video asset.") video_clips.append((clip, item)) elif clip.kind == "audio": if not mime_type.startswith("audio/"): raise ProjectRenderInvalidError("An audio clip references a non-audio asset.") audio_clips.append((clip, item)) if not video_clips and not audio_clips: raise ProjectRenderInvalidError("The editor timeline contains no renderable media.") # Hidden tracks/clips are non-rendering state and must not extend the # generated canvas with a long black tail. duration_ms = max( (clip.start_ms + clip.duration_ms for clip, _ in [*video_clips, *audio_clips]), default=0, ) args: list[str | Path] = [ "-f", "lavfi", "-t", f"{duration_ms / 1000:.6f}", "-i", f"color=c=black:s={width}x{height}:r={frame_rate:.6g}", ] for item in inputs: if item.kind == "media" and any( clip.media_type == "image" and ref is item for clip, ref in video_clips ): args.extend(["-loop", "1"]) if item.source_start_ms: args.extend(["-ss", f"{item.source_start_ms / 1000:.6f}"]) args.extend(["-t", f"{item.duration_ms / 1000:.6f}", "-i", item.path]) filters: list[str] = [] current_video = "base0" filters.append(f"[0:v]setpts=PTS-STARTPTS[{current_video}]") for number, (clip, item) in enumerate(video_clips, start=1): label = f"clipv{number}" scale_x = f"{clip.transform.scale_x:.6g}" scale_y = f"{clip.transform.scale_y:.6g}" chain = f"setpts=PTS-STARTPTS,scale=trunc(iw*{scale_x}/2)*2:trunc(ih*{scale_y}/2)*2" if abs(clip.transform.rotation) > 1e-9: chain += f",rotate={clip.transform.rotation:.6g}*PI/180:c=none:ow=rotw(iw):oh=roth(ih)" if clip.opacity < 1: chain += f",format=rgba,colorchannelmixer=aa={clip.opacity:.6g}" chain += f",setpts=PTS+{clip.start_ms / 1000:.6f}/TB" filters.append(f"[{item.input_index}:v]{chain}[{label}]") next_video = f"mixv{number}" x = f"(W-w)/2+{clip.transform.x:.6g}" y = f"(H-h)/2+{clip.transform.y:.6g}" filters.append( f"[{current_video}][{label}]overlay=x={x}:y={y}:eof_action=pass:shortest=0[{next_video}]" ) current_video = next_video audio_labels: list[str] = [] for number, (clip, item) in enumerate(audio_clips, start=1): label = f"clipa{number}" chain = f"atrim=duration={clip.duration_ms / 1000:.6f},asetpts=PTS-STARTPTS,volume={clip.volume:.6g}" if clip.fade_in_ms: chain += f",afade=t=in:st=0:d={clip.fade_in_ms / 1000:.6f}" if clip.fade_out_ms: chain += f",afade=t=out:st={(clip.duration_ms - clip.fade_out_ms) / 1000:.6f}:d={clip.fade_out_ms / 1000:.6f}" chain += f",adelay={clip.start_ms}|{clip.start_ms}" filters.append(f"[{item.input_index}:a]{chain}[{label}]") audio_labels.append(label) if audio_labels: filters.append( "".join(f"[{label}]" for label in audio_labels) + f"amix=inputs={len(audio_labels)}:duration=longest:dropout_transition=0[aout]" ) args.extend(["-filter_complex", ";".join(filters), "-map", f"[{current_video}]"]) if audio_labels: args.extend(["-map", "[aout]"]) else: args.append("-an") if output_format == "webm": args.extend( [ "-c:v", "libvpx-vp9", "-crf", {"draft": "30", "standard": "24", "high": "18"}[quality], "-b:v", "0", "-deadline", "good", "-cpu-used", {"fast": "5", "balanced": "2", "quality": "0"}[preset], "-c:a", "libopus", ] ) else: crf = {"draft": "28", "standard": "23", "high": "18"}[quality] args.extend( [ "-c:v", "libx264", "-pix_fmt", "yuv420p", "-preset", {"fast": "veryfast", "balanced": "medium", "quality": "slow"}[preset], "-crf", crf, "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", ] ) args.extend(["-r", f"{frame_rate:.6g}", "-t", f"{duration_ms / 1000:.6f}"]) return RenderPlan(tuple(args), tuple(inputs), duration_ms, output_format)