Update deformes4D_engine.py

deformes4D_engine.py

@@ -2,7 +2,7 @@
 #
 # Copyright (C) August 4, 2025  Carlos Rodrigues dos Santos
 #
-# Version: 2.0.0
+# Version: 2.0.1
 #
 # 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 (Ψ)
@@ -58,6 +58,7 @@ class Deformes4DEngine:
     @property
     def vae(self):
         if self._vae is None:
+            # Assumes the VAE from the first LTX worker is representative
             self._vae = self.ltx_manager.workers[0].pipeline.vae
         self._vae.to(self.device); self._vae.eval()
         return self._vae
@@ -66,16 +67,19 @@ class Deformes4DEngine:
 
     @torch.no_grad()
     def pixels_to_latents(self, tensor: torch.Tensor) -> torch.Tensor:
+        """Encodes a pixel-space tensor to the latent space using the VAE."""
         tensor = tensor.to(self.device, dtype=self.vae.dtype)
         return vae_encode(tensor, self.vae, vae_per_channel_normalize=True)
 
     @torch.no_grad()
     def latents_to_pixels(self, latent_tensor: torch.Tensor, decode_timestep: float = 0.05) -> torch.Tensor:
+        """Decodes a latent-space tensor to pixels using the VAE."""
         latent_tensor = latent_tensor.to(self.device, dtype=self.vae.dtype)
         timestep_tensor = torch.tensor([decode_timestep] * latent_tensor.shape[0], device=self.device, dtype=latent_tensor.dtype)
         return vae_decode(latent_tensor, self.vae, is_video=True, timestep=timestep_tensor, vae_per_channel_normalize=True)
 
     def save_video_from_tensor(self, video_tensor: torch.Tensor, path: str, fps: int = 24):
+        """Saves a pixel-space tensor as an MP4 video file."""
         if video_tensor is None or video_tensor.ndim != 5 or video_tensor.shape[2] == 0: return
         video_tensor = video_tensor.squeeze(0).permute(1, 2, 3, 0)
         video_tensor = (video_tensor.clamp(-1, 1) + 1) / 2.0
@@ -84,17 +88,20 @@ class Deformes4DEngine:
             for frame in video_np: writer.append_data(frame)
 
     def _preprocess_image_for_latent_conversion(self, image: Image.Image, target_resolution: tuple) -> Image.Image:
+        """Resizes and fits an image to the target resolution for VAE encoding."""
         if image.size != target_resolution:
             return ImageOps.fit(image, target_resolution, Image.Resampling.LANCZOS)
         return image
 
     def pil_to_latent(self, pil_image: Image.Image) -> torch.Tensor:
+        """Converts a PIL Image to a latent tensor."""
         image_np = np.array(pil_image).astype(np.float32) / 255.0
         tensor = torch.from_numpy(image_np).permute(2, 0, 1).unsqueeze(0).unsqueeze(2)
         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:
@@ -124,7 +131,6 @@ class Deformes4DEngine:
                                 progress: gr.Progress = gr.Progress()):
         """
         Step 3: Production. Generates the original master video from keyframes.
-        This involves generating latent tensors for each segment and then decoding them into a video file.
         """
         FPS = 24
         FRAMES_PER_LATENT_CHUNK = 8
@@ -269,15 +275,10 @@ class Deformes4DEngine:
         yield {"final_path": final_video_path}
 
     def master_video_hd(self, source_video_path: str, model_version: str, steps: int, prompt: str, progress: gr.Progress):
-        """
-        Post-Production Step 4B: Applies SeedVR super-resolution to an existing video file.
-        """
         logger.info(f"--- STARTING POST-PRODUCTION: HD Mastering with SeedVR {model_version} ---")
         progress(0.1, desc=f"Preparing for HD Mastering with SeedVR {model_version}...")
-        
         run_timestamp = int(time.time())
-        output_path = os.path.join(self.workspace_dir, f"hd_mastered_movie_{run_timestamp}.mp4")
-        
+        output_path = os.path.join(self.workspace_dir, f"hd_mastered_movie_{model_version}_{run_timestamp}.mp4")
         try:
             final_path = hd_specialist_singleton.process_video(
                 input_video_path=source_video_path,
@@ -292,49 +293,41 @@ class Deformes4DEngine:
         except Exception as e:
             logger.error(f"HD Mastering failed: {e}", exc_info=True)
             raise gr.Error(f"HD Mastering failed. Details: {e}")
-
+    
     def generate_audio_for_final_video(self, source_video_path: str, audio_prompt: str, progress: gr.Progress):
-        """
-        Post-Production Step 4C: Generates audio for a final video file and muxes it in.
-        """
         logger.info(f"--- STARTING POST-PRODUCTION: Audio Generation ---")
         progress(0.1, desc="Preparing for audio generation...")
-        
+        run_timestamp = int(time.time())
         try:
-            # Get video duration using ffprobe
             result = subprocess.run(
                 ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", source_video_path],
                 capture_output=True, text=True, check=True)
             duration = float(result.stdout.strip())
             logger.info(f"Source video duration: {duration:.2f} seconds.")
-
             progress(0.5, desc="Generating audio track...")
-            # The audio specialist handles file naming and muxing internally
+            output_path = os.path.join(self.workspace_dir, f"final_movie_with_audio_{run_timestamp}.mp4")
             final_path = audio_specialist_singleton.generate_audio_for_video(
                 video_path=source_video_path,
                 prompt=audio_prompt,
                 duration_seconds=duration,
+                output_path_override=output_path
             )
-            
             logger.info(f"Audio generation complete! Final video with audio at: {final_path}")
             progress(1.0, desc="Audio generation complete!")
             yield {"final_path": final_path}
-
         except Exception as e:
             logger.error(f"Audio generation failed: {e}", exc_info=True)
             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
-        }
+        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
+        return m if n > 0 and quantized == 0 else quantized```
+
+Com este arquivo finalizado, toda a lógica do backend para o ADUC 2.0 está completa. Todas as funções estão prontas para serem chamadas pela UI através do orquestrador.
+
+Agora estamos prontos para testar a aplicação completa.