# # app/core/pipeline.py # import os # import shutil # import subprocess # import sys # import tempfile # import time # from pathlib import Path # from typing import Dict, Optional, Tuple # from core.vectorizer import vectorize_image # # Import animation modules (each file inside core/) # import core.text_animations as text_animations # import core.logo_animations as logo_animations # import core.shape_animations as shape_animations # import core.fx_animations as fx_animations # import core.transitions as transitions # import core.infographic_animations as infographic_animations # import core.character_animations as character_animations # import core.background_animations as background_animations # import core.overlay_animations as overlay_animations # # Global outputs dir (stable) # GLOBAL_OUTPUTS_DIR = Path(tempfile.gettempdir()) / "manim_render_service" / "outputs" # GLOBAL_OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) # # Manim installation directory under tmp (pip --target here) # MANIM_TARGET_DIR = Path(tempfile.gettempdir()) / "manim_render_service" / "manim_env" # MANIM_TARGET_DIR.mkdir(parents=True, exist_ok=True) # # Map style strings to animation functions across modules. # STYLE_MAP = { # # Text basics # "fade-in": text_animations.fade_in, # "slide-in-left": text_animations.slide_in_left, # "pop-bounce": text_animations.pop_bounce, # "zoom-in": text_animations.zoom_in, # "typewriter": text_animations.typewriter, # "wipe": text_animations.wipe_mask, # "flip": text_animations.flip_rotate, # "blur-in": text_animations.blur_in, # "scale-up": text_animations.scale_up, # # Text premium # "neon-glow": text_animations.neon_glow, # "gradient-fill": text_animations.gradient_fill, # "wave-ripple": text_animations.wave_ripple, # "split-text": text_animations.split_text, # # Logo # "logo-build": logo_animations.logo_build_lines, # "logo-particle": logo_animations.logo_fade_particle, # "logo-spin": logo_animations.logo_spin_scale, # "logo-stroke": logo_animations.logo_stroke_draw, # "logo-glitch": logo_animations.logo_glitch, # # Shapes # "line-draw": shape_animations.line_draw, # "shape-morph": shape_animations.shape_morph, # "grow-center": shape_animations.grow_center, # "floating": shape_animations.floating_bounce, # # FX # "particle-burst": fx_animations.particle_burst, # "smoke": fx_animations.smoke_effect, # "light-rays": fx_animations.light_rays, # # Transitions & infographic & character etc (examples) # "fade-transition": transitions.fade_transition, # "bar-chart": infographic_animations.bar_chart_grow, # "count-number": infographic_animations.number_count, # "walk-cycle": character_animations.walk_cycle, # "bobbing": character_animations.bobbing, # "animated-gradient": background_animations.animated_gradient, # "particle-motion": background_animations.particle_motion, # "confetti": overlay_animations.confetti, # "checkmark": overlay_animations.checkmark_tick, # } # # Helper: detect manim in PATH or in current python env # def _manim_executable_available() -> bool: # # 1) check for manim binary in PATH # if shutil.which("manim"): # return True # # 2) check if importable in current python env # try: # import manim # noqa: F401 # return True # except Exception: # return False # def _ensure_manim_installed(log_file: Path) -> tuple[bool, str]: # """ # Ensure Manim is available. Strategy: # - If system has manim in PATH or importable, OK. # - Otherwise, run pip install --target MANIM_TARGET_DIR manim # and set PYTHONPATH to include MANIM_TARGET_DIR when invoking manim via `python -m manim`. # Returns (success, message). Writes pip logs to log_file. # NOTE: pip install requires network and may take time; call this once at startup or first run. # """ # if _manim_executable_available(): # return True, "manim-available" # try: # log_file.parent.mkdir(parents=True, exist_ok=True) # with log_file.open("ab") as lf: # # pip install into target dir # cmd = [sys.executable, "-m", "pip", "install", "--upgrade", "--target", str(MANIM_TARGET_DIR), "manim"] # proc = subprocess.run(cmd, stdout=lf, stderr=lf, check=False, timeout=1800) # if proc.returncode != 0: # return False, f"pip-install-failed:{proc.returncode}" # except Exception as e: # return False, f"pip-install-exception:{e}" # # final check # if _manim_executable_available(): # return True, "manim-installed-system" # # Even if not on PATH, we can still run via python -m manim with PYTHONPATH pointing to MANIM_TARGET_DIR # return True, "manim-installed-target" # def _run_manim(scene_py_path: Path, scene_class: str, work_dir: Path, log_file: Path, quality_flag: str = "-ql", timeout: int = 300) -> Optional[Path]: # """ # Run manim to render the scene class from scene_py_path inside work_dir. # - Uses python -m manim with PYTHONPATH including MANIM_TARGET_DIR to pick up target install. # - quality_flag example: "-ql" quick low, "-qm" medium, "-qh" high. # - Always uses "-t" for transparent background. # Returns Path to produced mp4 or None on failure. # """ # # Build base command. prefer system manim if available (shutil.which) # if shutil.which("manim"): # base = ["manim"] # else: # base = [sys.executable, "-m", "manim"] # # Ensure output name deterministic: use scene_py stem + scene_class # out_name = f"{scene_py_path.stem}_{scene_class}" # cmd = [*base, quality_flag, "-t", str(scene_py_path), scene_class, "-o", out_name] # env = os.environ.copy() # # include MANIM_TARGET_DIR in PYTHONPATH so python -m manim can import if installed with --target # env_py = env.get("PYTHONPATH", "") # new_py = str(MANIM_TARGET_DIR) # if env_py: # new_py = new_py + os.pathsep + env_py # env["PYTHONPATH"] = new_py # try: # with log_file.open("ab") as lf: # start = time.time() # proc = subprocess.run(cmd, cwd=str(work_dir), env=env, stdout=lf, stderr=lf, timeout=timeout) # elapsed = time.time() - start # if proc.returncode != 0: # return None # except Exception: # return None # # search for produced mp4 under work_dir (manim writes into media/videos/... usually) # mp4s = list(work_dir.rglob("*.mp4")) # if not mp4s: # return None # mp4s.sort(key=lambda p: p.stat().st_mtime, reverse=True) # return mp4s[0] # def process_image_pipeline(task_meta: Dict) -> Dict: # """ # Synchronous pipeline entry. Steps: # - set up task_dir/working/outputs/logs # - vectorize input image -> svg # - call animation placeholder (get scene_code) # - write scene file & copy svg into working dir # - ensure manim installed (pip --target if needed) # - run manim with -t -> get mp4 # - copy mp4 to GLOBAL_OUTPUTS_DIR and return path # """ # task_id = task_meta.get("task_id") # task_dir = Path(task_meta.get("task_dir", "")) # if not task_id or not task_dir.exists(): # return {"success": False, "error": "invalid-task"} # working = task_dir / "working" # outputs = task_dir / "outputs" # logs = task_dir / "logs" # working.mkdir(parents=True, exist_ok=True) # outputs.mkdir(parents=True, exist_ok=True) # logs.mkdir(parents=True, exist_ok=True) # log_file = logs / f"{task_id}.log" # try: # # 1) copy input to working folder # input_image = Path(task_meta.get("input_image")) # if not input_image.exists(): # return {"success": False, "error": "input-missing"} # safe_input = working / input_image.name # shutil.copy2(input_image, safe_input) # # 2) vectorize -> svg # svg_path = working / f"{task_id}.svg" # vec_res = vectorize_image(safe_input, svg_path, options={"quality": task_meta.get("quality")}) # if not vec_res.get("success"): # return {"success": False, "error": f"vectorize-failed:{vec_res.get('error')}"} # # 3) pick animation function # style = task_meta.get("style", "fade-in") # anim_fn = STYLE_MAP.get(style) # if anim_fn is None: # # fallback to fade-in # anim_fn = text_animations.fade_in # # 4) animation function returns scene info (name + code) # scene_info = anim_fn(svg_path, working, task_meta) # if not scene_info or "scene_name" not in scene_info or "scene_code" not in scene_info: # return {"success": False, "error": "animation-fn-invalid"} # scene_name = scene_info["scene_name"] # scene_code = scene_info["scene_code"] # # 5) write scene python file and ensure svg is in same folder # scene_py = working / f"scene_{task_id}.py" # scene_py.write_text(scene_code, encoding="utf-8") # # ensure svg copied into working dir # try: # shutil.copy2(svg_path, working / svg_path.name) # except Exception: # pass # # 6) ensure manim is available (installs into MANIM_TARGET_DIR if not) # ok, msg = _ensure_manim_installed(log_file) # if not ok: # return {"success": False, "error": f"manim-install-failed:{msg}"} # # 7) run manim to render scene with transparent background # # quality: map task_meta quality -> flag # q = task_meta.get("quality", "preview") # quality_flag = "-ql" if q == "preview" else "-qh" if q == "final" else "-qm" # rendered_mp4 = _run_manim(scene_py, scene_name, working, log_file, quality_flag=quality_flag, timeout=900) # if not rendered_mp4: # # attach log path for debugging # return {"success": False, "error": "manim-render-failed", "log": str(log_file)} # # 8) copy to GLOBAL_OUTPUTS_DIR # dest = GLOBAL_OUTPUTS_DIR / f"{task_id}_{rendered_mp4.name}" # try: # if dest.exists(): # dest.unlink() # shutil.copy2(rendered_mp4, dest) # except Exception as e: # return {"success": False, "error": f"copy-failed:{e}", "log": str(log_file)} # # optional: cleanup working to save disk # try: # shutil.rmtree(working) # except Exception: # pass # return {"success": True, "output_path": str(dest), "log": str(log_file)} # except Exception as e: # # keep task_dir for debug # return {"success": False, "error": str(e), "log": str(log_file)} # # app/core/pipeline.py # import os # import shutil # import subprocess # import sys # import tempfile # import time # from pathlib import Path # from typing import Dict, Optional, Tuple # from core.vectorizer import vectorize_image # # Import animation modules (each file inside core/) # import core.text_animations as text_animations # import core.logo_animations as logo_animations # import core.shape_animations as shape_animations # import core.fx_animations as fx_animations # import core.transitions as transitions # import core.infographic_animations as infographic_animations # import core.character_animations as character_animations # import core.background_animations as background_animations # import core.overlay_animations as overlay_animations # # Global outputs dir (stable) # GLOBAL_OUTPUTS_DIR = Path(tempfile.gettempdir()) / "manim_render_service" / "outputs" # GLOBAL_OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) # # Map style strings to animation functions across modules. # STYLE_MAP = { # # Text basics # "fade-in": text_animations.fade_in, # "slide-in-left": text_animations.slide_in_left, # "pop-bounce": text_animations.pop_bounce, # "zoom-in": text_animations.zoom_in, # "typewriter": text_animations.typewriter, # "wipe": text_animations.wipe_mask, # "flip": text_animations.flip_rotate, # "blur-in": text_animations.blur_in, # "scale-up": text_animations.scale_up, # # Text premium # "neon-glow": text_animations.neon_glow, # "gradient-fill": text_animations.gradient_fill, # "wave-ripple": text_animations.wave_ripple, # "split-text": text_animations.split_text, # # Logo # "logo-build": logo_animations.logo_build_lines, # "logo-particle": logo_animations.logo_fade_particle, # "logo-spin": logo_animations.logo_spin_scale, # "logo-stroke": logo_animations.logo_stroke_draw, # "logo-glitch": logo_animations.logo_glitch, # # Shapes # "line-draw": shape_animations.line_draw, # "shape-morph": shape_animations.shape_morph, # "grow-center": shape_animations.grow_center, # "floating": shape_animations.floating_bounce, # # FX # "particle-burst": fx_animations.particle_burst, # "smoke": fx_animations.smoke_effect, # "light-rays": fx_animations.light_rays, # # Transitions & infographic & character etc (examples) # "fade-transition": transitions.fade_transition, # "bar-chart": infographic_animations.bar_chart_grow, # "count-number": infographic_animations.number_count, # "walk-cycle": character_animations.walk_cycle, # "bobbing": character_animations.bobbing, # "animated-gradient": background_animations.animated_gradient, # "particle-motion": background_animations.particle_motion, # "confetti": overlay_animations.confetti, # "checkmark": overlay_animations.checkmark_tick, # } # def _run_manim(scene_py_path: Path, scene_class: str, work_dir: Path, log_file: Path, quality_flag: str = "-ql", timeout: int = 300) -> Optional[Path]: # """ # Run manim to render the scene class from scene_py_path inside work_dir. # - Uses python -m manim command. # - Always uses "-t" for transparent background. # Returns Path to produced mp4 or None on failure. # """ # base = [sys.executable, "-m", "manim"] # out_name = f"{scene_py_path.stem}_{scene_class}" # cmd = [*base, quality_flag, "-t", str(scene_py_path), scene_class, "-o", out_name] # try: # with log_file.open("ab") as lf: # start = time.time() # proc = subprocess.run(cmd, cwd=str(work_dir), stdout=lf, stderr=lf, timeout=timeout) # elapsed = time.time() - start # if proc.returncode != 0: # return None # except Exception: # return None # mp4s = list(work_dir.rglob("*.mp4")) # if not mp4s: # return None # mp4s.sort(key=lambda p: p.stat().st_mtime, reverse=True) # return mp4s[0] # def process_image_pipeline(task_meta: Dict) -> Dict: # """ # Synchronous pipeline entry. Steps: # - set up task_dir/working/outputs/logs # - vectorize input image -> svg # - call animation placeholder (get scene_code) # - write scene file & copy svg into working dir # - run manim with -t -> get mp4 # - copy mp4 to GLOBAL_OUTPUTS_DIR and return path # """ # task_id = task_meta.get("task_id") # task_dir = Path(task_meta.get("task_dir", "")) # if not task_id or not task_dir.exists(): # return {"success": False, "error": "invalid-task"} # working = task_dir / "working" # outputs = task_dir / "outputs" # logs = task_dir / "logs" # working.mkdir(parents=True, exist_ok=True) # outputs.mkdir(parents=True, exist_ok=True) # logs.mkdir(parents=True, exist_ok=True) # log_file = logs / f"{task_id}.log" # try: # input_image = Path(task_meta.get("input_image")) # if not input_image.exists(): # return {"success": False, "error": "input-missing"} # safe_input = working / input_image.name # shutil.copy2(input_image, safe_input) # # Step 1: Prepare vectorization # svg_path = working / f"{task_id}.svg" # # Run vectorization — returns {"success": True, "svg": "..."} # vec_res = vectorize_image(safe_input, options={"quality": task_meta.get("quality")}) # if not vec_res.get("success"): # return {"success": False, "error": f"vectorize-failed:{vec_res.get('error')}"} # # Step 2: Write SVG to file (since later parts of pipeline expect a file path) # svg_content = vec_res.get("svg") # if not svg_content: # return {"success": False, "error": "vectorizer-did-not-return-svg"} # svg_path.write_text(svg_content, encoding="utf-8") # style = task_meta.get("style", "fade-in") # anim_fn = STYLE_MAP.get(style, text_animations.fade_in) # scene_info = anim_fn(svg_path, working, task_meta) # if not scene_info or "scene_name" not in scene_info or "scene_code" not in scene_info: # return {"success": False, "error": "animation-fn-invalid"} # scene_name = scene_info["scene_name"] # scene_code = scene_info["scene_code"] # scene_py = working / f"scene_{task_id}.py" # scene_py.write_text(scene_code, encoding="utf-8") # try: # shutil.copy2(svg_path, working / svg_path.name) # except Exception: # pass # q = task_meta.get("quality", "preview") # quality_flag = "-ql" if q == "preview" else "-qh" if q == "final" else "-qm" # rendered_mp4 = _run_manim(scene_py, scene_name, working, log_file, quality_flag=quality_flag, timeout=900) # if not rendered_mp4: # return {"success": False, "error": "manim-render-failed", "log": str(log_file)} # dest = GLOBAL_OUTPUTS_DIR / f"{task_id}_{rendered_mp4.name}" # try: # if dest.exists(): # dest.unlink() # shutil.copy2(rendered_mp4, dest) # except Exception as e: # return {"success": False, "error": f"copy-failed:{e}", "log": str(log_file)} # try: # shutil.rmtree(working) # except Exception: # pass # return {"success": True, "output_path": str(dest), "log": str(log_file)} # except Exception as e: # return {"success": False, "error": str(e), "log": str(log_file)} # # app/core/pipeline.py # import os # import shutil # import subprocess # import sys # import tempfile # import time # from pathlib import Path # from typing import Dict, Optional, Tuple # from core.vectorizer import vectorize_image # # Import animation modules (each file inside core/) # import core.text_animations as text_animations # import core.logo_animations as logo_animations # import core.shape_animations as shape_animations # import core.fx_animations as fx_animations # import core.transitions as transitions # import core.infographic_animations as infographic_animations # import core.character_animations as character_animations # import core.background_animations as background_animations # import core.overlay_animations as overlay_animations # # Global outputs dir (stable) # GLOBAL_OUTPUTS_DIR = Path("tmp")/ "manim_render_service" / "outputs" # GLOBAL_OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) # # Map style strings to animation functions across modules. # STYLE_MAP = { # # Text basics # "fade-in": text_animations.fade_in, # "slide-in-left": text_animations.slide_in_left, # "pop-bounce": text_animations.pop_bounce, # "zoom-in": text_animations.zoom_in, # "typewriter": text_animations.typewriter, # "wipe": text_animations.wipe_mask, # "flip": text_animations.flip_rotate, # "blur-in": text_animations.blur_in, # "scale-up": text_animations.scale_up, # # Text premium # "neon-glow": text_animations.neon_glow, # "gradient-fill": text_animations.gradient_fill, # "wave-ripple": text_animations.wave_ripple, # "split-text": text_animations.split_text, # # Logo # "logo-build": logo_animations.logo_build_lines, # "logo-particle": logo_animations.logo_fade_particle, # "logo-spin": logo_animations.logo_spin_scale, # "logo-stroke": logo_animations.logo_stroke_draw, # "logo-glitch": logo_animations.logo_glitch, # # Shapes # "line-draw": shape_animations.line_draw, # "shape-morph": shape_animations.shape_morph, # "grow-center": shape_animations.grow_center, # "floating": shape_animations.floating_bounce, # # FX # "particle-burst": fx_animations.particle_burst, # "smoke": fx_animations.smoke_effect, # "light-rays": fx_animations.light_rays, # # Transitions & infographic & character etc (examples) # "fade-transition": transitions.fade_transition, # "bar-chart": infographic_animations.bar_chart_grow, # "count-number": infographic_animations.number_count, # "walk-cycle": character_animations.walk_cycle, # "bobbing": character_animations.bobbing, # "animated-gradient": background_animations.animated_gradient, # "particle-motion": background_animations.particle_motion, # "confetti": overlay_animations.confetti, # "checkmark": overlay_animations.checkmark_tick, # } # # def _run_manim(scene_py_path: Path, scene_class: str, work_dir: Path, log_file: Path, quality_flag: str = "-ql", timeout: int = 300) -> Optional[Path]: # # print(f"[DEBUG] Running manim render for scene '{scene_class}' in '{work_dir}'...") # # base = [sys.executable, "-m", "manim"] # # out_name = f"{scene_py_path.stem}_{scene_class}" # # cmd = [*base, quality_flag, "-t", str(scene_py_path), scene_class, "-o", out_name] # # print(f"[DEBUG] Command: {' '.join(cmd)}") # # try: # # with log_file.open("ab") as lf: # # start = time.time() # # proc = subprocess.run(cmd, cwd=str(work_dir), stdout=lf, stderr=lf, timeout=timeout) # # elapsed = time.time() - start # # print(f"[DEBUG] Manim completed in {elapsed:.2f}s with return code {proc.returncode}") # # if proc.returncode != 0: # # print("[ERROR] Manim render failed.") # # return None # # except Exception as e: # # print(f"[ERROR] Exception during manim run: {e}") # # return None # # mp4s = list(work_dir.rglob("*.mp4")) # # if not mp4s: # # print("[ERROR] No MP4 files found after rendering.") # # return None # # mp4s.sort(key=lambda p: p.stat().st_mtime, reverse=True) # # print(f"[DEBUG] Rendered MP4: {mp4s[0]}") # # return mp4s[0] # def _run_manim( # scene_py_path: Path, # scene_class: str, # work_dir: Path, # log_file: Path, # quality_flag: str = "-ql", # timeout: int = 300 # ) -> Optional[Path]: # """ # Run a Manim render inside a temp working directory and return the final video path (.mp4 or .mov). # Logs full debug details and ensures the output is stored inside `work_dir`. # """ # import sys, subprocess, shutil, time # print(f"\n[DEBUG] 🎬 Starting Manim render for scene '{scene_class}'...") # print(f"[DEBUG] Working directory: {work_dir}") # print(f"[DEBUG] Scene file: {scene_py_path}") # base = [sys.executable, "-m", "manim"] # out_name = f"{scene_py_path.stem}_{scene_class}" # # ✅ Expected output path (any format) # expected_output = work_dir / f"{out_name}" # print(f"[DEBUG] Expected output file prefix: {expected_output}") # # ✅ Manim command (saves media inside tmp/media/) # scene_filename = Path(scene_py_path).name # only file name, not full path # cmd = [*base, quality_flag, "-t", scene_filename, scene_class, "-o", out_name] # print(f"[DEBUG] Command: {' '.join(cmd)}") # # ✅ Run the process # try: # with log_file.open("ab") as lf: # start = time.time() # proc = subprocess.run(cmd, cwd=str(work_dir), stdout=lf, stderr=lf, timeout=timeout) # elapsed = time.time() - start # print(f"[DEBUG] Manim completed in {elapsed:.2f}s with code {proc.returncode}") # if proc.returncode != 0: # print("[ERROR] ❌ Manim render failed.") # return None # except Exception as e: # print(f"[ERROR] ❌ Exception during manim run: {e}") # return None # # ✅ Search for rendered video files (.mp4 or .mov) # print("[DEBUG] Scanning for video outputs...") # search_dirs = [ # work_dir, # work_dir / "media", # work_dir / "media" / "videos" # ] # videos = [] # for folder in search_dirs: # if folder.exists(): # found = list(folder.rglob("*.mp4")) + list(folder.rglob("*.mov")) # videos.extend(found) # print(f"[DEBUG] → {folder}: {len(found)} video(s) found") # if not videos: # print("[ERROR] ❌ No video file found after scanning all media directories.") # return None # videos.sort(key=lambda p: p.stat().st_mtime, reverse=True) # final_video = videos[0] # print(f"[SUCCESS] ✅ Rendered video detected at: {final_video}") # # ✅ Copy file back to work_dir (for consistent access) # dest = work_dir / final_video.name # if final_video != dest: # try: # shutil.copy2(final_video, dest) # print(f"[DEBUG] Copied render result → {dest}") # except Exception as e: # print(f"[WARN] ⚠️ Could not copy final video: {e}") # print(f"[INFO] 🎉 Final render saved successfully at:\n {dest}") # print("-" * 70) # return dest # def process_image_pipeline(task_meta: Dict) -> Dict: # print("\n================= PIPELINE START =================") # print(f"[INFO] Task metadata received: {task_meta}") # task_id = task_meta.get("task_id") # task_dir = Path(task_meta.get("task_dir", "")) # if not task_id or not task_dir.exists(): # print("[ERROR] Invalid task: Missing task_id or task_dir.") # return {"success": False, "error": "invalid-task"} # working = task_dir / "working" # outputs = task_dir / "outputs" # logs = task_dir / "logs" # working.mkdir(parents=True, exist_ok=True) # outputs.mkdir(parents=True, exist_ok=True) # logs.mkdir(parents=True, exist_ok=True) # print(f"[DEBUG] Working directories prepared under {task_dir}") # log_file = logs / f"{task_id}.log" # try: # input_image = Path(task_meta.get("input_image")) # if not input_image.exists(): # print(f"[ERROR] Input image missing: {input_image}") # return {"success": False, "error": "input-missing"} # safe_input = working / input_image.name # shutil.copy2(input_image, safe_input) # print(f"[DEBUG] Input image copied to working directory: {safe_input}") # svg_path = working / f"{task_id}.svg" # print("[DEBUG] Starting vectorization...") # vec_res = vectorize_image(safe_input, options={"quality": task_meta.get("quality")}) # print(f"[DEBUG] Vectorization result: {vec_res.keys()}") # if not vec_res.get("success"): # print(f"[ERROR] Vectorization failed: {vec_res.get('error')}") # return {"success": False, "error": f"vectorize-failed:{vec_res.get('error')}"} # svg_content = vec_res.get("svg") # if not svg_content: # print("[ERROR] Vectorizer returned empty SVG content.") # return {"success": False, "error": "vectorizer-did-not-return-svg"} # svg_path.write_text(svg_content, encoding="utf-8") # print(f"[DEBUG] SVG written to: {svg_path}") # style = task_meta.get("style", "fade-in") # anim_fn = STYLE_MAP.get(style, text_animations.fade_in) # print(f"[DEBUG] Selected animation style: {style}") # scene_info = anim_fn(svg_path, working, task_meta) # if not scene_info or "scene_name" not in scene_info or "scene_code" not in scene_info: # print("[ERROR] Invalid animation function result.") # return {"success": False, "error": "animation-fn-invalid"} # scene_name = scene_info["scene_name"] # scene_code = scene_info["scene_code"] # scene_py = working / f"scene_{task_id}.py" # scene_py.write_text(scene_code, encoding="utf-8") # print(f"[DEBUG] Scene file written: {scene_py}") # try: # shutil.copy2(svg_path, working / svg_path.name) # except Exception as e: # print(f"[WARN] Could not copy SVG: {e}") # q = task_meta.get("quality", "preview") # quality_flag = "-ql" if q == "preview" else "-qh" if q == "final" else "-qm" # print(f"[DEBUG] Render quality flag: {quality_flag}") # rendered_mp4 = _run_manim(scene_py, scene_name, working, log_file, quality_flag=quality_flag, timeout=900) # if not rendered_mp4: # print("[ERROR] Manim render failed.") # return {"success": False, "error": "manim-render-failed", "log": str(log_file)} # dest = GLOBAL_OUTPUTS_DIR / f"{task_id}_{rendered_mp4.name}" # try: # if dest.exists(): # dest.unlink() # shutil.copy2(rendered_mp4, dest) # print(f"[DEBUG] Final output copied to global outputs dir: {dest}") # except Exception as e: # print(f"[ERROR] Failed to copy output: {e}") # return {"success": False, "error": f"copy-failed:{e}", "log": str(log_file)} # try: # shutil.rmtree(working) # print("[DEBUG] Cleaned up working directory.") # except Exception as e: # print(f"[WARN] Could not remove working directory: {e}") # print(f"[SUCCESS] Task {task_id} completed successfully.") # print("================= PIPELINE END =================\n") # return {"success": True, "output_path": str(dest), "log": str(log_file),"output_bytes": dest.read_bytes(),} # except Exception as e: # print(f"[ERROR] Unexpected exception in pipeline: {e}") # print("================= PIPELINE FAILED =================\n") # return {"success": False, "error": str(e), "log": str(log_file)} # app/core/pipeline.py import sys import shutil import subprocess import time from pathlib import Path from typing import Dict, Optional from core.vectorizer import vectorize_image # type: ignore from moviepy.video.io.VideoFileClip import VideoFileClip # Animation modules import core.text_animations as text_animations import core.logo_animations as logo_animations import core.shape_animations as shape_animations import core.fx_animations as fx_animations import core.transitions as transitions import core.infographic_animations as infographic_animations import core.character_animations as character_animations import core.background_animations as background_animations import core.overlay_animations as overlay_animations # === GLOBAL OUTPUT DIR === TMP_DIR = Path("tmp") TMP_DIR.mkdir(exist_ok=True) GLOBAL_OUTPUTS_DIR = TMP_DIR / "outputs" GLOBAL_OUTPUTS_DIR.mkdir(parents=True, exist_ok=True) # === STYLE MAP === STYLE_MAP = { "fade-in": text_animations.fade_in, "slide-in-left": text_animations.slide_in_left, "pop-bounce": text_animations.pop_bounce, "zoom-in": text_animations.zoom_in, "typewriter": text_animations.typewriter, "wipe": text_animations.wipe_mask, "flip": text_animations.flip_rotate, "blur-in": text_animations.blur_in, "scale-up": text_animations.scale_up, "neon-glow": text_animations.neon_glow, "gradient-fill": text_animations.gradient_fill, "wave-ripple": text_animations.wave_ripple, "split-text": text_animations.split_text, "logo-build": logo_animations.logo_build_lines, "logo-particle": logo_animations.logo_fade_particle, "logo-spin": logo_animations.logo_spin_scale, "logo-stroke": logo_animations.logo_stroke_draw, "logo-glitch": logo_animations.logo_glitch, "line-draw": shape_animations.line_draw, "shape-morph": shape_animations.shape_morph, "grow-center": shape_animations.grow_center, "floating": shape_animations.floating_bounce, "particle-burst": fx_animations.particle_burst, "smoke": fx_animations.smoke_effect, "light-rays": fx_animations.light_rays, "fade-transition": transitions.fade_transition, "bar-chart": infographic_animations.bar_chart_grow, "count-number": infographic_animations.number_count, "walk-cycle": character_animations.walk_cycle, "bobbing": character_animations.bobbing, "animated-gradient": background_animations.animated_gradient, "particle-motion": background_animations.particle_motion, "confetti": overlay_animations.confetti, "checkmark": overlay_animations.checkmark_tick, } # # === MANIM RUNNER === # def _run_manim(scene_py_path: Path, scene_class: str, quality_flag: str = "-ql", timeout: int = 300) -> Optional[Path]: # """Run Manim and return the rendered video path.""" # print(f"\n🎬 Running Manim for scene: {scene_class}") # cmd = [ # sys.executable, # "-m", # "manim", # quality_flag, # "-t", # scene_py_path.name, # scene_class, # "-o", # f"{scene_class}_output", # ] # print(f"→ Command: {' '.join(cmd)}") # try: # subprocess.run(cmd, cwd=scene_py_path.parent, check=True, timeout=timeout) # except subprocess.CalledProcessError: # print("❌ Manim render failed.") # return None # except Exception as e: # print(f"❌ Exception: {e}") # return None # # Search output # rendered = list(scene_py_path.parent.rglob("*.mp4")) + list(scene_py_path.parent.rglob("*.mov")) # if not rendered: # print("❌ No output file found after render.") # return None # rendered.sort(key=lambda p: p.stat().st_mtime, reverse=True) # output_file = rendered[0] # print(f"✅ Rendered file: {output_file}") # return output_file # # === PIPELINE === # def process_image_pipeline(task_meta: Dict) -> Dict: # """Simplified image → SVG → Manim video pipeline.""" # print("\n================= PIPELINE START =================") # print(f"Metadata: {task_meta}") # try: # task_id = task_meta.get("task_id", "no_id") # input_image = Path(task_meta.get("input_image", "")) # if not input_image.exists(): # print(f"❌ Input image not found: {input_image}") # return {"success": False, "error": "input-not-found"} # style = task_meta.get("style", "fade-in") # quality = task_meta.get("quality", "preview") # # === Work directly inside tmp/ === # work_dir = TMP_DIR / f"{task_id}" # work_dir.mkdir(exist_ok=True) # safe_input = work_dir / input_image.name # shutil.copy2(input_image, safe_input) # print(f"🖼️ Copied input → {safe_input}") # # === Vectorize === # vec_res = vectorize_image(safe_input, options={"quality": quality}) # if not vec_res.get("success"): # print("❌ Vectorization failed.") # return {"success": False, "error": vec_res.get("error")} # svg_path = work_dir / f"{task_id}.svg" # svg_path.write_text(vec_res.get("svg", ""), encoding="utf-8") # print(f"🧩 SVG written → {svg_path}") # # === Animation === # anim_fn = STYLE_MAP.get(style, text_animations.fade_in) # scene_info = anim_fn(svg_path, work_dir, task_meta) # scene_name = scene_info["scene_name"] # scene_code = scene_info["scene_code"] # scene_py = work_dir / f"{scene_name}.py" # scene_py.write_text(scene_code, encoding="utf-8") # print(f"🎞️ Scene file written → {scene_py}") # quality_flag = "-ql" if quality == "preview" else "-qh" if quality == "final" else "-qm" # output_file = _run_manim(scene_py, scene_name, quality_flag=quality_flag) # if not output_file: # return {"success": False, "error": "render-failed"} # final_output = GLOBAL_OUTPUTS_DIR / f"{task_id}.mp4" # shutil.copy2(output_file, final_output) # print(f"📦 Final output copied → {final_output}") # print("================= PIPELINE END =================\n") # return { # "success": True, # "output_path": str(final_output), # "output_bytes": final_output.read_bytes(), # } # except Exception as e: # print(f"❌ Exception in pipeline: {e}") # print("================= PIPELINE FAILED =================\n") # return {"success": False, "error": str(e)} # === MANIM RUNNER === def _run_manim(scene_py_path: Path, scene_class: str, quality_flag: str = "-ql", timeout: int = 300) -> Optional[Path]: """Run Manim and return the rendered video path.""" print(f"\n🎬 Running Manim for scene: {scene_class}") cmd = [ sys.executable, "-m", "manim", quality_flag, "--transparent", "-t", scene_py_path.name, scene_class, "-o", f"{scene_class}_output", ] print(f"→ Command: {' '.join(cmd)}") try: subprocess.run(cmd, cwd=scene_py_path.parent, check=True, timeout=timeout) except subprocess.CalledProcessError: print("❌ Manim render failed.") return None except Exception as e: print(f"❌ Exception: {e}") return None # Search for rendered outputs rendered = list(scene_py_path.parent.rglob("*.mp4")) + list(scene_py_path.parent.rglob("*.mov")) if not rendered: print("❌ No output file found after render.") return None rendered.sort(key=lambda p: p.stat().st_mtime, reverse=True) output_file = rendered[0] print(f"✅ Rendered file: {output_file}") # 🔄 Convert MOV → MP4 if needed (safe cross-platform) # if output_file.suffix.lower() == ".mov": # try: # converted = output_file.with_suffix(".mp4") # print("🎞️ Converting .mov → .mp4 for compatibility...") # clip = VideoFileClip(str(output_file)) # clip.write_videofile(str(converted), codec="libx264", audio_codec="aac") # clip.close() # output_file = converted # except Exception as e: # print(f"⚠️ MOV→MP4 conversion failed: {e}") # return None # if output_file.suffix.lower() == ".mov": # try: # converted = output_file.with_suffix(".webm") # print("🎞️ Converting .mov → .webm (keeping transparency)...") # cmd = [ # "ffmpeg", # "-y", # "-i", str(output_file), # "-c:v", "libvpx-vp9", # "-pix_fmt", "yuva420p", # ✅ keep alpha channel # "-b:v", "4M", # "-auto-alt-ref", "0", # str(converted) # ] # subprocess.run(cmd, check=True) # output_file1 = converted # print(f"✅ Converted successfully → {converted}") # except Exception as e: # print(f"⚠️ MOV→WEBM conversion failed: {e}") # return None return output_file # === PIPELINE === def process_image_pipeline(task_meta: Dict) -> Dict: """Simplified image → SVG → Manim video pipeline.""" print("\n================= PIPELINE START =================") print(f"Metadata: {task_meta}") try: task_id = task_meta.get("task_id", "no_id") input_image = Path(task_meta.get("input_image", "")) if not input_image.exists(): print(f"❌ Input image not found: {input_image}") return {"success": False, "error": "input-not-found"} style = task_meta.get("style", "fade-in") quality = task_meta.get("quality", "final") # === Work directly inside tmp/ === work_dir = TMP_DIR / f"{task_id}" work_dir.mkdir(exist_ok=True) safe_input = work_dir / input_image.name shutil.copy2(input_image, safe_input) print(f"🖼️ Copied input → {safe_input}") # === Vectorize === vec_res = vectorize_image(safe_input, options={"quality": quality}) if not vec_res.get("success"): print("❌ Vectorization failed.") return {"success": False, "error": vec_res.get("error")} svg_path = work_dir / f"{task_id}.svg" svg_path.write_text(vec_res.get("svg", ""), encoding="utf-8") print(f"🧩 SVG written → {svg_path}") # === Animation === anim_fn = STYLE_MAP.get(style, text_animations.fade_in) scene_info = anim_fn(svg_path, work_dir, task_meta) scene_name = scene_info["scene_name"] scene_code = scene_info["scene_code"] scene_py = work_dir / f"{scene_name}.py" scene_py.write_text(scene_code, encoding="utf-8") print(f"🎞️ Scene file written → {scene_py}") # === Run Manim === quality_flag = "-ql" if quality == "preview" else "-qh" if quality == "final" else "-qm" output_file = _run_manim(scene_py, scene_name, quality_flag=quality_flag) if not output_file: print("❌ Render failed, no output returned.") return {"success": False, "error": "render-failed"} # === Copy Final Output === final_output = GLOBAL_OUTPUTS_DIR / f"{task_id}.mov" shutil.copy2(output_file, final_output) print(f"📦 Final output copied → {final_output}") print("================= PIPELINE END =================\n") # === Return result in-memory === return { "success": True, "output_path": str(final_output), # "output_bytes": final_output.read_bytes(), } except Exception as e: print(f"❌ Exception in pipeline: {e}") print("================= PIPELINE FAILED =================\n") return {"success": False, "error": str(e)}