Update api/ltx_server.py

api/ltx_server.py

@@ -246,7 +246,12 @@ def log_tensor_info(tensor, name="Tensor"):
     print("------------------------------------------\n")
 
 
-
+@dataclass
+class LatentConditioningItem:
+    """Item de dados para condicionamento da pipeline LTX."""
+    latent_tensor: torch.Tensor
+    media_frame_number: int
+    conditioning_strength: float
 
 
 # --- 5. CLASSE PRINCIPAL DO SERVIÇO ---
@@ -289,73 +294,72 @@ class VideoService:
         print(f"[DEBUG] VideoService pronto. boot_time={time.perf_counter()-t0:.3f}s")
 
 
-def _aduc_prepare_conditioning_patch(
-    self: "LTXVideoPipeline",
-    conditioning_items: Optional[List[Union["ConditioningItem", "LatentConditioningItem"]]],
-    init_latents: torch.Tensor,
-    num_frames: int,
-    height: int,
-    width: int,
-    vae_per_channel_normalize: bool = False,
-    generator=None,
-) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor, int]:
-    if not conditioning_items:
+    def _aduc_prepare_conditioning_patch(
+        self: "LTXVideoPipeline",
+        conditioning_items: Optional[List[Union["ConditioningItem", "LatentConditioningItem"]]],
+        init_latents: torch.Tensor,
+        num_frames: int,
+        height: int,
+        width: int,
+        vae_per_channel_normalize: bool = False,
+        generator=None,
+    ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor, int]:
+        if not conditioning_items:
+            init_latents, init_latent_coords = self.patchifier.patchify(latents=init_latents)
+            init_pixel_coords = latent_to_pixel_coords(init_latent_coords, self.vae, causal_fix=self.transformer.config.causal_temporal_positioning)
+            return init_latents, init_pixel_coords, None, 0
+        
+        init_conditioning_mask = torch.zeros_like(init_latents[:, 0, ...], dtype=torch.float32, device=init_latents.device)
+        extra_conditioning_latents, extra_conditioning_pixel_coords, extra_conditioning_mask = [], [], []
+        extra_conditioning_num_latents = 0
+        
+        for item in conditioning_items:
+            if not isinstance(item, LatentConditioningItem):
+                logger.warning("Patch ADUC: Item de condicionamento não é um LatentConditioningItem e será ignorado.")
+                continue
+            
+            media_item_latents = item.latent_tensor.to(dtype=init_latents.dtype, device=init_latents.device)
+            media_frame_number, strength = item.media_frame_number, item.conditioning_strength
+            
+            if media_frame_number == 0:
+                f_l, h_l, w_l = media_item_latents.shape[-3:]
+                init_latents[..., :f_l, :h_l, :w_l] = torch.lerp(init_latents[..., :f_l, :h_l, :w_l], media_item_latents, strength)
+                init_conditioning_mask[..., :f_l, :h_l, :w_l] = strength
+            else:
+                noise = randn_tensor(media_item_latents.shape, generator=generator, device=media_item_latents.device, dtype=media_item_latents.dtype)
+                media_item_latents = torch.lerp(noise, media_item_latents, strength)
+                patched_latents, latent_coords = self.patchifier.patchify(latents=media_item_latents)
+                pixel_coords = latent_to_pixel_coords(latent_coords, self.vae, causal_fix=self.transformer.config.causal_temporal_positioning)
+                pixel_coords[:, 0] += media_frame_number
+                extra_conditioning_num_latents += patched_latents.shape[1]
+                new_mask = torch.full(patched_latents.shape[:2], strength, dtype=torch.float32, device=init_latents.device)
+                extra_conditioning_latents.append(patched_latents)
+                extra_conditioning_pixel_coords.append(pixel_coords)
+                extra_conditioning_mask.append(new_mask)
+    
         init_latents, init_latent_coords = self.patchifier.patchify(latents=init_latents)
         init_pixel_coords = latent_to_pixel_coords(init_latent_coords, self.vae, causal_fix=self.transformer.config.causal_temporal_positioning)
-        return init_latents, init_pixel_coords, None, 0
-    
-    init_conditioning_mask = torch.zeros_like(init_latents[:, 0, ...], dtype=torch.float32, device=init_latents.device)
-    extra_conditioning_latents, extra_conditioning_pixel_coords, extra_conditioning_mask = [], [], []
-    extra_conditioning_num_latents = 0
-    
-    for item in conditioning_items:
-        if not isinstance(item, LatentConditioningItem):
-            logger.warning("Patch ADUC: Item de condicionamento não é um LatentConditioningItem e será ignorado.")
-            continue
+        init_conditioning_mask, _ = self.patchifier.patchify(latents=init_conditioning_mask.unsqueeze(1))
+        init_conditioning_mask = init_conditioning_mask.squeeze(-1)
         
-        media_item_latents = item.latent_tensor.to(dtype=init_latents.dtype, device=init_latents.device)
-        media_frame_number, strength = item.media_frame_number, item.conditioning_strength
+        if extra_conditioning_latents:
+            init_latents = torch.cat([*extra_conditioning_latents, init_latents], dim=1)
+            init_pixel_coords = torch.cat([*extra_conditioning_pixel_coords, init_pixel_coords], dim=2)
+            init_conditioning_mask = torch.cat([*extra_conditioning_mask, init_conditioning_mask], dim=1)
         
-        if media_frame_number == 0:
-            f_l, h_l, w_l = media_item_latents.shape[-3:]
-            init_latents[..., :f_l, :h_l, :w_l] = torch.lerp(init_latents[..., :f_l, :h_l, :w_l], media_item_latents, strength)
-            init_conditioning_mask[..., :f_l, :h_l, :w_l] = strength
-        else:
-            noise = randn_tensor(media_item_latents.shape, generator=generator, device=media_item_latents.device, dtype=media_item_latents.dtype)
-            media_item_latents = torch.lerp(noise, media_item_latents, strength)
-            patched_latents, latent_coords = self.patchifier.patchify(latents=media_item_latents)
-            pixel_coords = latent_to_pixel_coords(latent_coords, self.vae, causal_fix=self.transformer.config.causal_temporal_positioning)
-            pixel_coords[:, 0] += media_frame_number
-            extra_conditioning_num_latents += patched_latents.shape[1]
-            new_mask = torch.full(patched_latents.shape[:2], strength, dtype=torch.float32, device=init_latents.device)
-            extra_conditioning_latents.append(patched_latents)
-            extra_conditioning_pixel_coords.append(pixel_coords)
-            extra_conditioning_mask.append(new_mask)
-
-    init_latents, init_latent_coords = self.patchifier.patchify(latents=init_latents)
-    init_pixel_coords = latent_to_pixel_coords(init_latent_coords, self.vae, causal_fix=self.transformer.config.causal_temporal_positioning)
-    init_conditioning_mask, _ = self.patchifier.patchify(latents=init_conditioning_mask.unsqueeze(1))
-    init_conditioning_mask = init_conditioning_mask.squeeze(-1)
+        return init_latents, init_pixel_coords, init_conditioning_mask, extra_conditioning_num_latents
     
-    if extra_conditioning_latents:
-        init_latents = torch.cat([*extra_conditioning_latents, init_latents], dim=1)
-        init_pixel_coords = torch.cat([*extra_conditioning_pixel_coords, init_pixel_coords], dim=2)
-        init_conditioning_mask = torch.cat([*extra_conditioning_mask, init_conditioning_mask], dim=1)
+    def _apply_ltx_pipeline_patches(self):
+            """Aplica patches em tempo de execução na pipeline LTX para compatibilidade com ADUC-SDR."""
+            logger.info("LTX POOL MANAGER: Aplicando patches ADUC-SDR na pipeline LTX...")
+            for worker in self.workers:
+                worker.pipeline.prepare_conditioning = _aduc_prepare_conditioning_patch.__get__(worker.pipeline, LTXVideoPipeline)
+            logger.info("LTX POOL MANAGER: Todas as instâncias da pipeline foram corrigidas com sucesso.")
     
-    return init_latents, init_pixel_coords, init_conditioning_mask, extra_conditioning_num_latents
-
-
-def _apply_ltx_pipeline_patches(self):
-        """Aplica patches em tempo de execução na pipeline LTX para compatibilidade com ADUC-SDR."""
-        logger.info("LTX POOL MANAGER: Aplicando patches ADUC-SDR na pipeline LTX...")
-        for worker in self.workers:
-            worker.pipeline.prepare_conditioning = _aduc_prepare_conditioning_patch.__get__(worker.pipeline, LTXVideoPipeline)
-        logger.info("LTX POOL MANAGER: Todas as instâncias da pipeline foram corrigidas com sucesso.")
-
-
-self._apply_ltx_pipeline_patches()
     
+    self._apply_ltx_pipeline_patches()
     
+        
     
     
     def _log_gpu_memory(self, stage_name: str):