Update api/ltx_server_refactored.py

api/ltx_server_refactored.py

@@ -459,4 +459,65 @@ class VideoService:
             try:
                 overrides["guidance_scale"] = json.loads(ltx_configs["guidance_scale_list"])
                 overrides["stg_scale"] = json.loads(ltx_configs["stg_scale_list"])
-            except (json.JSONDecodeError, KeyError
+            except (json.JSONDecodeError, KeyError) as e:
+                logging.warning(f"Failed to parse custom guidance values: {e}. Falling back to defaults.")
+        
+        if overrides:
+            logging.info(f"Applying '{preset}' guidance preset overrides.")
+        return overrides
+
+    def _save_and_log_video(self, pixel_tensor: torch.Tensor, base_filename: str) -> Path:
+        """Saves a pixel tensor to an MP4 file and returns the final path."""
+        # Work in a temporary directory to handle atomic move
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = os.path.join(temp_dir, f"{base_filename}.mp4")
+            video_encode_tool_singleton.save_video_from_tensor(
+                pixel_tensor, temp_path, fps=DEFAULT_FPS
+            )
+            final_path = RESULTS_DIR / f"{base_filename}.mp4"
+            shutil.move(temp_path, final_path)
+            logging.info(f"Video saved successfully to: {final_path}")
+            return final_path
+    
+    def _apply_precision_policy(self):
+        """Sets the autocast dtype based on the configuration file."""
+        precision = str(self.config.get("precision", "bfloat16")).lower()
+        if precision in ["float8_e4m3fn", "bfloat16"]:
+            self.runtime_autocast_dtype = torch.bfloat16
+        elif precision == "mixed_precision":
+            self.runtime_autocast_dtype = torch.float16
+        else:
+            self.runtime_autocast_dtype = torch.float32
+        logging.info(f"Runtime precision policy set for autocast: {self.runtime_autocast_dtype}")
+
+    def _align(self, dim: int, alignment: int = FRAMES_ALIGNMENT) -> int:
+        """Aligns a dimension to the nearest multiple of `alignment`."""
+        return ((dim - 1) // alignment + 1) * alignment
+    
+    def _calculate_aligned_frames(self, duration_s: float, min_frames: int = 1) -> int:
+        """Calculates the total number of frames based on duration, ensuring alignment."""
+        num_frames = int(round(duration_s * DEFAULT_FPS))
+        aligned_frames = self._align(num_frames)
+        # Ensure it's at least 1 frame longer than the alignment for some ops, and respects min_frames
+        final_frames = max(aligned_frames + 1, min_frames)
+        return final_frames
+
+    def _resolve_seed(self, seed: Optional[int]) -> int:
+        """Returns the given seed or generates a new random one."""
+        return random.randint(0, 2**32 - 1) if seed is None else int(seed)
+
+
+# ==============================================================================
+# --- SINGLETON INSTANTIATION ---
+# ==============================================================================
+# The service is instantiated once when the module is imported, ensuring a single
+# instance manages the models and GPU resources throughout the application's life.
+
+try:
+    video_generation_service = VideoService()
+    logging.info("Global VideoService instance created successfully.")
+except Exception as e:
+    logging.critical(f"Failed to initialize VideoService: {e}")
+    traceback.print_exc()
+    # Exit if the core service fails to start, as the app is non-functional
+    sys.exit(1)