Release NaraMotion v0.26.0: Structure of Arrays (SoA), Path Arena Allocator, Uniform Grid Indexing, Dual-Buffer Rendering & Centroid Spring Pop-in
b516e00 verified | """ | |
| Distributed Temporal Composition and Sequence Engine for NaraMotion. | |
| Enables declarative, modular video creation (similar to Remotion Sequence and Composition architecture). | |
| """ | |
| from __future__ import annotations | |
| import math | |
| from typing import List, Dict, Any, Optional, Callable, Generator, Union | |
| from naramotion.image import Image, Format | |
| from naramotion.context import Context | |
| from naramotion.color import parse_color, Rgba32 | |
| from naramotion.video import VideoWriter | |
| class Sequence: | |
| """ | |
| Temporal video clip or animation block active within a specific frame range [from_frame, from_frame + duration_in_frames). | |
| """ | |
| def __init__( | |
| self, | |
| from_frame: int = 0, | |
| duration_in_frames: Optional[int] = None, | |
| render_fn: Optional[Callable[[Context, float, int], None]] = None, | |
| name: str = "Sequence" | |
| ): | |
| self.from_frame = int(from_frame) | |
| self.duration_in_frames = int(duration_in_frames) if duration_in_frames is not None else None | |
| self.render_fn = render_fn | |
| self.name = name | |
| def is_active(self, frame_index: int) -> bool: | |
| """Check if sequence is active at absolute frame_index.""" | |
| if frame_index < self.from_frame: | |
| return False | |
| if self.duration_in_frames is not None and frame_index >= (self.from_frame + self.duration_in_frames): | |
| return False | |
| return True | |
| def get_progress(self, frame_index: int) -> float: | |
| """Get normalized local progress t in [0.0, 1.0] within this sequence.""" | |
| if frame_index < self.from_frame: | |
| return 0.0 | |
| dur = self.duration_in_frames or 1 | |
| rel = (frame_index - self.from_frame) / max(1, dur - 1 if dur > 1 else 1) | |
| return max(0.0, min(1.0, float(rel))) | |
| def render(self, ctx: Context, frame_index: int) -> None: | |
| """Render this sequence to context if active.""" | |
| if self.is_active(frame_index) and self.render_fn: | |
| progress = self.get_progress(frame_index) | |
| local_frame = frame_index - self.from_frame | |
| self.render_fn(ctx, progress, local_frame) | |
| class Composition: | |
| """ | |
| High-level declarative video animation builder. | |
| Eliminates repetitive video boilerplates and renders directly to hardware-accelerated MP4. | |
| """ | |
| def __init__( | |
| self, | |
| width: int = 1920, | |
| height: int = 1080, | |
| fps: float = 60.0, | |
| duration_in_frames: int = 180, | |
| bg_color: Union[str, Rgba32] = "#050711" | |
| ): | |
| self.width = int(width) | |
| self.height = int(height) | |
| self.fps = float(fps) | |
| self.duration_in_frames = int(duration_in_frames) | |
| self.bg_color = parse_color(bg_color) | |
| self.sequences: List[Sequence] = [] | |
| self._custom_frame_fn: Optional[Callable[[Context, float, int], None]] = None | |
| self._post_fx_fn: Optional[Callable[[Image, float, int], None]] = None | |
| def add_sequence( | |
| self, | |
| from_frame: int = 0, | |
| duration_in_frames: Optional[int] = None, | |
| render_fn: Optional[Callable[[Context, float, int], None]] = None, | |
| name: str = "Sequence" | |
| ) -> "Composition": | |
| """Add a temporal sequence block.""" | |
| seq = Sequence(from_frame=from_frame, duration_in_frames=duration_in_frames, render_fn=render_fn, name=name) | |
| self.sequences.append(seq) | |
| return self | |
| def sequence(self, from_frame: int = 0, duration_in_frames: Optional[int] = None, name: str = "Sequence"): | |
| """Decorator for adding a sequence.""" | |
| def decorator(fn: Callable[[Context, float, int], None]): | |
| self.add_sequence(from_frame=from_frame, duration_in_frames=duration_in_frames, render_fn=fn, name=name) | |
| return fn | |
| return decorator | |
| def on_frame(self, fn: Callable[[Context, float, int], None]) -> "Composition": | |
| """Set a global per-frame callback function (ctx, progress, frame_index).""" | |
| self._custom_frame_fn = fn | |
| return self | |
| def on_post_fx(self, fn: Callable[[Image, float, int], None]) -> "Composition": | |
| """Set a post-processing filter callback (img, progress, frame_index).""" | |
| self._post_fx_fn = fn | |
| return self | |
| def render_video( | |
| self, | |
| output_filename: str, | |
| codec: str = "h264", | |
| hwaccel: str = "auto", | |
| bitrate: Optional[str] = None, | |
| show_progress: bool = True | |
| ) -> str: | |
| """ | |
| Render all sequences directly to high-speed hardware-encoded video. | |
| """ | |
| frame_img = Image(self.width, self.height, Format.PRGB32) | |
| with VideoWriter( | |
| output_filename, | |
| width=self.width, | |
| height=self.height, | |
| fps=self.fps, | |
| codec=codec, | |
| hwaccel=hwaccel, | |
| bitrate=bitrate | |
| ) as vw: | |
| for f in range(self.duration_in_frames): | |
| t = f / max(1, self.duration_in_frames - 1) | |
| with Context(frame_img) as ctx: | |
| ctx.fill_all(self.bg_color) | |
| # 1. Global frame function | |
| if self._custom_frame_fn: | |
| self._custom_frame_fn(ctx, t, f) | |
| # 2. Render all active sequence layers | |
| for seq in self.sequences: | |
| seq.render(ctx, f) | |
| # 3. Post-processing FX | |
| if self._post_fx_fn: | |
| self._post_fx_fn(frame_img, t, f) | |
| # 4. Stream frame | |
| vw.write_frame(frame_img) | |
| return output_filename | |
| def delay_render(callback: Callable[[], Any]) -> Any: | |
| """Execute pre-render tasks or asset downloads with exception wrapping.""" | |
| try: | |
| return callback() | |
| except Exception as e: | |
| raise RuntimeError(f"delay_render precomputation failed: {e}") from e | |