Update deformes4D_engine.py

deformes4D_engine.py

@@ -2,7 +2,7 @@
 #
 # Copyright (C) August 4, 2025  Carlos Rodrigues dos Santos
 #
-# Version: 2.0.1
+# Version: 2.0.2
 #
 # This file contains the Deformes4D Engine, which acts as the primary "Editor" or
 # "Film Crew" specialist within the ADUC-SDR architecture. It implements the Camera (Ψ)
@@ -24,6 +24,7 @@ import subprocess
 import gc
 import shutil
 from pathlib import Path
+from typing import List, Tuple, Generator, Dict, Any, Optional
 
 from ltx_manager_helpers import ltx_manager_singleton
 from gemini_helpers import gemini_singleton
@@ -31,6 +32,7 @@ from latent_enhancer_specialist import latent_enhancer_specialist_singleton
 from hd_specialist import hd_specialist_singleton
 from ltx_video.models.autoencoders.vae_encode import vae_encode, vae_decode
 from audio_specialist import audio_specialist_singleton
+from tools.video_encode_tool import video_encode_tool_singleton
 
 logger = logging.getLogger(__name__)
 
@@ -100,27 +102,6 @@ class Deformes4DEngine:
         tensor = (tensor * 2.0) - 1.0
         return self.pixels_to_latents(tensor)
 
-    def concatenate_videos_ffmpeg(self, video_paths: list[str], output_path: str):
-        """Concatenates multiple video clips into a single file using FFmpeg."""
-        if not video_paths: raise gr.Error("No video fragments to assemble.")
-        list_file_path = os.path.join(self.workspace_dir, "concat_list.txt")
-        with open(list_file_path, 'w', encoding='utf-8') as f:
-            for path in video_paths: f.write(f"file '{os.path.abspath(path)}'\n")
-
-        cmd_list = ['ffmpeg', '-y', '-hwaccel', 'auto', '-f', 'concat', '-safe', '0', '-i', list_file_path, '-c', 'copy', output_path]
-        logger.info(f"Concatenating {len(video_paths)} video clips into {output_path}...")
-        try:
-            subprocess.run(cmd_list, check=True, capture_output=True, text=True)
-        except subprocess.CalledProcessError as e:
-            logger.error(f"FFmpeg error: {e.stderr}")
-            logger.info("Attempting concatenation again without hardware acceleration...")
-            cmd_list = ['ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', list_file_path, '-c', 'copy', output_path]
-            try:
-                subprocess.run(cmd_list, check=True, capture_output=True, text=True)
-            except subprocess.CalledProcessError as e_fallback:
-                 logger.error(f"FFmpeg error (fallback): {e_fallback.stderr}")
-                 raise gr.Error(f"Failed to assemble the final video. Details: {e_fallback.stderr}")
-
     # --- CORE ADUC-SDR LOGIC ---
 
     def generate_original_movie(self, keyframes: list, global_prompt: str, storyboard: list,
@@ -222,7 +203,7 @@ class Deformes4DEngine:
 
         progress(0.98, desc="Final assembly of clips...")
         final_video_path = os.path.join(self.workspace_dir, f"original_movie_{run_timestamp}.mp4")
-        self.concatenate_videos_ffmpeg(final_video_clip_paths, final_video_path)
+        video_encode_tool_singleton.concatenate_videos(video_paths=final_video_clip_paths, output_path=final_video_path, workspace_dir=self.workspace_dir)
         logger.info("Cleaning up temporary clip files...")
         try:
             shutil.rmtree(temp_video_clips_dir)
@@ -263,12 +244,10 @@ class Deformes4DEngine:
             logger.info(f"Saved upscaled clip: {Path(current_clip_path).name}")
         progress(0.98, desc="Assembling upscaled clips...")
         final_video_path = os.path.join(self.workspace_dir, f"upscaled_movie_{run_timestamp}.mp4")
-        self.concatenate_videos_ffmpeg(final_upscaled_clip_paths, final_video_path)
+        video_encode_tool_singleton.concatenate_videos(video_paths=final_upscaled_clip_paths, output_path=final_video_path, workspace_dir=self.workspace_dir)
         logger.info("Cleaning up temporary upscaled clip files...")
         try:
             shutil.rmtree(temp_upscaled_clips_dir)
-            concat_list_path = os.path.join(self.workspace_dir, "concat_list.txt")
-            if os.path.exists(concat_list_path): os.remove(concat_list_path)
         except OSError as e:
             logger.warning(f"Could not remove temporary upscaled clip directory: {e}")
         logger.info(f"Latent upscaling complete! Final video at: {final_video_path}")
@@ -320,10 +299,12 @@ class Deformes4DEngine:
             raise gr.Error(f"Audio generation failed. Details: {e}")
 
     def _generate_latent_tensor_internal(self, conditioning_items, ltx_params, target_resolution, total_frames_to_generate):
+        """Internal helper to call the LTX manager."""
         final_ltx_params = {**ltx_params, 'width': target_resolution[0], 'height': target_resolution[1], 'video_total_frames': total_frames_to_generate, 'video_fps': 24, 'current_fragment_index': int(time.time()), 'conditioning_items_data': conditioning_items}
         return self.ltx_manager.generate_latent_fragment(**final_ltx_params)
 
     def _quantize_to_multiple(self, n, m):
+        """Helper to round n to the nearest multiple of m."""
         if m == 0: return n
         quantized = int(round(n / m) * m)
         return m if n > 0 and quantized == 0 else quantized