Spaces:
Sleeping
Sleeping
| 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 | |
| class RenderInput: | |
| asset_id: str | |
| path: Path | |
| kind: str | |
| source_start_ms: int | |
| duration_ms: int | |
| input_index: int | |
| 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) | |