Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- """Advanced node variants with full parameter control and hi-res pipeline.
- Stage 1+2 run at reduced spatial resolution for speed,
- latent upscale, then Stage 3 at full resolution.
- """
- from __future__ import annotations
- import logging
- from pathlib import Path
- from typing import Any
- import comfy.model_management
- import comfy.samplers
- import comfy.utils
- import torch
- from triple_ksampler.base import TripleKSamplerBase
- from triple_ksampler.shared import alignment as core_alignment
- from triple_ksampler.shared import config as core_config
- from triple_ksampler.shared import models as core_models
- from triple_ksampler.shared import notifications as core_notifications
- _CONFIG = core_config.load_config(config_dir=Path(__file__).resolve().parent.parent.parent)
- _BASE_QUALITY_THRESHOLD = core_config.get_base_quality_threshold(_CONFIG)
- _LOG_LEVEL = core_config.get_log_level(_CONFIG)
- logger = logging.getLogger("triple_ksampler.ksampler.advanced")
- bare_logger = logging.getLogger("triple_ksampler.separator")
- STAGE3_SEED_OFFSET = 1
- UPSCALE_METHODS = ["nearest-exact", "bilinear", "area", "bislerp"]
- # WanVideo VAE spatial compression factor (pixels / latent)
- VAE_SPATIAL_FACTOR = 8
- def _downscale_latent(
- latent: dict[str, torch.Tensor], scale_factor: float
- ) -> dict[str, torch.Tensor]:
- """Downscale video latent [B, C, T, H, W] spatially by scale_factor."""
- samples = latent["samples"]
- if scale_factor <= 1.0:
- return {"samples": samples.clone()}
- b, c, t, h, w = samples.shape
- new_h = max(1, round(h / scale_factor))
- new_w = max(1, round(w / scale_factor))
- reshaped = samples.permute(0, 2, 1, 3, 4).reshape(b * t, c, h, w)
- scaled = torch.nn.functional.interpolate(
- reshaped, size=(new_h, new_w), mode="bilinear", align_corners=False
- )
- result = scaled.reshape(b, t, c, new_h, new_w).permute(0, 2, 1, 3, 4)
- return {"samples": result}
- def _upscale_latent(
- latent: dict[str, torch.Tensor],
- target_h: int,
- target_w: int,
- method: str = "bislerp",
- ) -> dict[str, torch.Tensor]:
- """Upscale video latent [B, C, T, H, W] to target spatial dimensions."""
- samples = latent["samples"]
- b, c, t, h, w = samples.shape
- if h == target_h and w == target_w:
- return {"samples": samples.clone()}
- reshaped = samples.permute(0, 2, 1, 3, 4).reshape(b * t, c, h, w)
- scaled = comfy.utils.common_upscale(reshaped, target_w, target_h, method, "disabled")
- result = scaled.reshape(b, t, c, target_h, target_w).permute(0, 2, 1, 3, 4)
- return {"samples": result}
- class TripleKSamplerAdvancedAlt(TripleKSamplerBase):
- """Advanced triple-stage sampler with hi-res pipeline. Static UI."""
- @classmethod
- def INPUT_TYPES(cls) -> dict[str, dict[str, Any]]:
- base_inputs = cls._get_base_input_types()
- return {
- "required": {
- "base_high": base_inputs["base_high"],
- "lightning_high": base_inputs["lightning_high"],
- "lightning_low": base_inputs["lightning_low"],
- "positive": base_inputs["positive"],
- "negative": base_inputs["negative"],
- "latent_image": base_inputs["latent_image"],
- "seed": base_inputs["seed"],
- "sigma_shift": base_inputs["sigma_shift"],
- "base_quality_threshold": (
- "INT",
- {
- "default": core_config.DEFAULT_BASE_QUALITY_THRESHOLD,
- "min": 1,
- "max": 100,
- "step": 1,
- "tooltip": (
- "Minimum total steps for base_steps auto-calculation "
- f"(config default: {_BASE_QUALITY_THRESHOLD}). "
- "Only applies when base_steps=-1."
- ),
- },
- ),
- "base_steps": (
- "INT",
- {
- "default": -1,
- "min": -1,
- "max": 100,
- "tooltip": (
- "Stage 1 steps for base high-noise model. "
- "Use -1 for auto-calculation."
- ),
- },
- ),
- "base_cfg": base_inputs["base_cfg"],
- "base_sampler": (
- comfy.samplers.KSampler.SAMPLERS,
- {"tooltip": "Sampler for Stage 1 (base model)."},
- ),
- "base_scheduler": (
- comfy.samplers.KSampler.SCHEDULERS,
- {"tooltip": "Scheduler for Stage 1 (base model)."},
- ),
- "lightning_start": (
- "INT",
- {
- "default": 1,
- "min": 0,
- "max": 99,
- "tooltip": (
- "Starting step within lightning schedule. "
- "Set to 0 to skip Stage 1."
- ),
- },
- ),
- "lightning_steps": base_inputs["lightning_steps"],
- "lightning_cfg": (
- "FLOAT",
- {
- "default": 1.0,
- "min": 0.0,
- "max": 100.0,
- "step": 0.1,
- "tooltip": "CFG scale for Stage 2 and Stage 3.",
- },
- ),
- "lightning_sampler": (
- comfy.samplers.KSampler.SAMPLERS,
- {"tooltip": "Sampler for Stage 2 and Stage 3."},
- ),
- "lightning_scheduler": (
- comfy.samplers.KSampler.SCHEDULERS,
- {"tooltip": "Scheduler for Stage 2 and Stage 3."},
- ),
- "switch_strategy": (
- [
- "50% of steps",
- "Manual switch step",
- "T2V boundary",
- "I2V boundary",
- "Manual boundary",
- "T2V boundary (refined)",
- "I2V boundary (refined)",
- "Manual boundary (refined)",
- ],
- {
- "default": "50% of steps",
- "tooltip": (
- "Strategy for switching between models. "
- "Refined variants auto-tune sigma_shift."
- ),
- },
- ),
- "switch_step": (
- "INT",
- {
- "default": -1,
- "min": -1,
- "max": 99,
- "tooltip": (
- "Manual switch step. Only for 'Manual switch step' strategy. "
- "-1 = auto at 50%."
- ),
- },
- ),
- "switch_boundary": (
- "FLOAT",
- {
- "default": 0.875,
- "min": 0.0,
- "max": 1.0,
- "step": 0.001,
- "tooltip": "Sigma boundary. Only for 'Manual boundary' strategy.",
- },
- ),
- "hr_scale_factor": (
- "FLOAT",
- {
- "default": 2.0,
- "min": 1.0,
- "max": 8.0,
- "step": 0.25,
- "tooltip": (
- "Spatial downscale factor for Stage 1+2. "
- "2.0 = half pixel resolution. 1.0 = disabled."
- ),
- },
- ),
- "hr_upscale_method": (
- UPSCALE_METHODS,
- {
- "default": "bislerp",
- "tooltip": "Latent upscale method between Stage 2 and Stage 3.",
- },
- ),
- "dry_run": (
- "BOOLEAN",
- {
- "default": False,
- "tooltip": "Test stage calculations without sampling.",
- },
- ),
- },
- }
- RETURN_TYPES = ("LATENT",)
- RETURN_NAMES = ("samples",)
- FUNCTION = "sample"
- CATEGORY = "TripleKSampler"
- DESCRIPTION = (
- "Advanced triple-stage cascade sampler with hi-res pipeline. "
- "Stage 1+2 at reduced resolution, upscale, Stage 3 at full resolution."
- )
- def sample(
- self,
- base_high: Any,
- lightning_high: Any,
- lightning_low: Any,
- positive: Any,
- negative: Any,
- latent_image: dict[str, torch.Tensor],
- seed: int,
- sigma_shift: float,
- base_quality_threshold: int,
- base_steps: int,
- base_cfg: float,
- base_sampler: str,
- base_scheduler: str,
- lightning_start: int,
- lightning_steps: int,
- lightning_cfg: float,
- lightning_sampler: str,
- lightning_scheduler: str,
- switch_strategy: str,
- switch_step: int = -1,
- switch_boundary: float = 0.875,
- hr_scale_factor: float = 2.0,
- hr_upscale_method: str = "bislerp",
- dry_run: bool = False,
- ) -> tuple[dict[str, torch.Tensor], ...]:
- """Execute 4-phase pipeline with hi-res upscale."""
- if str(_LOG_LEVEL).upper() == "DEBUG":
- bare_logger.info("")
- logger.debug("=== TripleKSampler Advanced - Input Parameters ===")
- logger.debug("seed=%d, sigma_shift=%.3f", seed, sigma_shift)
- logger.debug(
- "base: threshold=%d, steps=%d, cfg=%.1f, sampler=%s, scheduler=%s",
- base_quality_threshold, base_steps, base_cfg, base_sampler, base_scheduler,
- )
- logger.debug(
- "lightning: start=%d, steps=%d, cfg=%.1f, sampler=%s, scheduler=%s",
- lightning_start, lightning_steps, lightning_cfg,
- lightning_sampler, lightning_scheduler,
- )
- logger.debug(
- "strategy=%s, switch_step=%d, switch_boundary=%.3f",
- switch_strategy, switch_step, switch_boundary,
- )
- logger.debug(
- "hr: scale=%.2f, method=%s, dry_run=%s",
- hr_scale_factor, hr_upscale_method, dry_run,
- )
- base_calculation_info = ""
- model_switching_info = ""
- hr_enabled = hr_scale_factor > 1.0
- # === Validation ===
- self._validate_basic_parameters(
- lightning_steps, lightning_start, switch_strategy, switch_step
- )
- bare_logger.info("")
- # === Calculate base_steps alignment ===
- optimal_total_base_steps = None
- if base_steps == -1:
- base_steps, optimal_total_base_steps, method = self._calculate_perfect_alignment(
- base_quality_threshold, lightning_start, lightning_steps
- )
- if lightning_start > 0:
- if method in ("mathematical_search", "simple_math"):
- base_calculation_info = (
- f"Auto-calculated base_steps = {base_steps}, "
- f"total_base_steps = {optimal_total_base_steps} "
- f"({method.replace('_', ' ')})"
- )
- else:
- base_calculation_info = (
- f"Auto-calculated base_steps = {base_steps} "
- f"(fallback - no perfect alignment found)"
- )
- logger.info(base_calculation_info)
- else:
- optimal_total_base_steps = core_alignment.calculate_manual_base_steps_alignment(
- base_steps, lightning_start, lightning_steps
- )
- base_calculation_info = (
- f"Auto-calculated total_base_steps = {optimal_total_base_steps} "
- f"for manual base_steps = {base_steps}"
- )
- logger.info(base_calculation_info)
- if lightning_start > 0 and base_steps > 0 and optimal_total_base_steps > 0:
- stage1_end_pct = base_steps / optimal_total_base_steps
- stage2_start_pct = lightning_start / lightning_steps
- if stage1_end_pct > stage2_start_pct:
- overlap_pct = (stage1_end_pct - stage2_start_pct) * 100.0
- logger.warning(
- "Stage 1 and 2 overlap (%.1f%%)!", overlap_pct,
- )
- core_notifications.send_overlap_warning(overlap_pct)
- total_base_steps = optimal_total_base_steps or base_steps
- # === Post-calculation validation ===
- self._validate_resolved_parameters(lightning_start, base_steps)
- self._validate_special_modes(
- lightning_start, lightning_steps, base_steps, switch_strategy, switch_step
- )
- # === Patch models ===
- patched_base_high, patched_lightning_high, patched_lightning_low = (
- self._patch_models_for_sampling(base_high, lightning_high, lightning_low, sigma_shift)
- )
- # === Calculate switch point ===
- switch_step_final, _, model_switching_info = self._calculate_switch_step_and_strategy(
- switch_strategy, switch_step, switch_boundary,
- lightning_steps, patched_lightning_high, lightning_scheduler,
- )
- # === Sigma shift refinement ===
- sigma_shift_final = sigma_shift
- if self._is_refined_strategy(switch_strategy):
- try:
- target_boundary = self._get_target_boundary_from_strategy(
- switch_strategy, switch_boundary
- )
- refined_shift, refine_msg = core_models.calculate_perfect_shift_for_step(
- model=lightning_high,
- scheduler=lightning_scheduler,
- total_steps=lightning_steps,
- target_step=switch_step_final,
- target_sigma=target_boundary,
- initial_shift=sigma_shift,
- is_wanvideo=False,
- )
- logger.info(
- "Refined sigma_shift: %.2f → %.2f at step %d",
- sigma_shift, refined_shift, switch_step_final,
- )
- sigma_shift_final = refined_shift
- model_switching_info += (
- f" [Refined shift: {sigma_shift:.2f}→{refined_shift:.2f}]"
- )
- except Exception as e:
- logger.warning("Shift refinement failed: %s. Using %.2f", e, sigma_shift)
- if sigma_shift_final != sigma_shift:
- patched_base_high, patched_lightning_high, patched_lightning_low = (
- self._patch_models_for_sampling(
- base_high, lightning_high, lightning_low, sigma_shift_final
- )
- )
- logger.info(model_switching_info)
- # === Resolve spatial dimensions ===
- orig_samples = latent_image["samples"] # [B, C, T, H, W]
- orig_h, orig_w = orig_samples.shape[3], orig_samples.shape[4]
- orig_px_h = orig_h * VAE_SPATIAL_FACTOR
- orig_px_w = orig_w * VAE_SPATIAL_FACTOR
- if hr_enabled:
- lr_h = max(1, round(orig_h / hr_scale_factor))
- lr_w = max(1, round(orig_w / hr_scale_factor))
- lr_px_h = lr_h * VAE_SPATIAL_FACTOR
- lr_px_w = lr_w * VAE_SPATIAL_FACTOR
- logger.info(
- "Hi-res pipeline: Stage 1+2 at %dx%d px (latent %dx%d) "
- "→ upscale (%s) → Stage 3 at %dx%d px (latent %dx%d)",
- lr_px_w, lr_px_h, lr_w, lr_h,
- hr_upscale_method,
- orig_px_w, orig_px_h, orig_w, orig_h,
- )
- else:
- lr_h, lr_w = orig_h, orig_w
- lr_px_h, lr_px_w = orig_px_h, orig_px_w
- logger.info("Hi-res pipeline: disabled (scale_factor=1.0)")
- # === Execution flags ===
- skip_stage1 = lightning_start == 0
- skip_stage2 = lightning_start == switch_step_final
- if not skip_stage1 and lightning_start > switch_step_final:
- raise ValueError("lightning_start cannot be greater than switch_step.")
- # === Prepare low-res latent ===
- if hr_enabled:
- lr_latent = _downscale_latent(latent_image, hr_scale_factor)
- else:
- lr_latent = latent_image
- # ================================================================
- # STAGE 1: Base high-noise model (low resolution)
- # ================================================================
- if skip_stage1:
- bare_logger.info("")
- logger.info("Stage 1: Skipped (Lightning-only mode)")
- stage1_info = "Skipped (Lightning-only mode)"
- stage1_output = lr_latent
- else:
- lr_tag = f" [{lr_px_w}×{lr_px_h} px]" if hr_enabled else ""
- stage1_info = (
- self._format_stage_range(0, base_steps, total_base_steps) + lr_tag
- )
- stage1_result = self._run_sampling_stage(
- model=patched_base_high,
- positive=positive,
- negative=negative,
- latent=lr_latent,
- seed=seed,
- steps=total_base_steps,
- cfg=base_cfg,
- sampler_name=base_sampler,
- scheduler=base_scheduler,
- start_at_step=0,
- end_at_step=base_steps,
- add_noise=True,
- return_with_leftover_noise=True,
- dry_run=dry_run,
- stage_name="Stage 1",
- stage_info=stage1_info,
- )
- stage1_output = stage1_result[0]
- # ================================================================
- # STAGE 2: Lightning high-noise model (low resolution)
- # When hr_enabled: return clean signal (no leftover noise)
- # so we upscale clean data, Stage 3 adds fresh noise.
- # ================================================================
- if skip_stage2:
- bare_logger.info("")
- logger.info("Stage 2: Skipped (lightning_start == switch point)")
- stage2_info = "Skipped (lightning_start == switch point)"
- stage2_output = stage1_output
- else:
- lr_tag = f" [{lr_px_w}×{lr_px_h} px]" if hr_enabled else ""
- stage2_info = (
- self._format_stage_range(
- lightning_start, switch_step_final, lightning_steps
- ) + lr_tag
- )
- stage2_result = self._run_sampling_stage(
- model=patched_lightning_high,
- positive=positive,
- negative=negative,
- latent=stage1_output,
- seed=seed,
- steps=lightning_steps,
- cfg=lightning_cfg,
- sampler_name=lightning_sampler,
- scheduler=lightning_scheduler,
- start_at_step=lightning_start,
- end_at_step=switch_step_final,
- add_noise=skip_stage1,
- return_with_leftover_noise=not hr_enabled,
- dry_run=dry_run,
- stage_name="Stage 2",
- stage_info=stage2_info,
- )
- stage2_output = stage2_result[0]
- # ================================================================
- # LATENT UPSCALE
- # ================================================================
- if hr_enabled:
- bare_logger.info("")
- if dry_run:
- logger.info(
- "Latent Upscale: %dx%d px → %dx%d px (%s) [DRY RUN]",
- lr_px_w, lr_px_h, orig_px_w, orig_px_h, hr_upscale_method,
- )
- stage3_input = stage2_output
- else:
- stage3_input = _upscale_latent(
- stage2_output, orig_h, orig_w, hr_upscale_method
- )
- logger.info(
- "Latent Upscale: %dx%d px → %dx%d px (%s)",
- lr_px_w, lr_px_h, orig_px_w, orig_px_h, hr_upscale_method,
- )
- else:
- stage3_input = stage2_output
- upscale_info = ""
- if hr_enabled:
- upscale_info = (
- f"Latent Upscale: {lr_px_w}×{lr_px_h} px → "
- f"{orig_px_w}×{orig_px_h} px ({hr_upscale_method})"
- )
- # ================================================================
- # STAGE 3: Lightning low-noise model (full resolution)
- # When hr_enabled: add_noise=True because Stage 2 returned
- # fully denoised signal. Stage 3 injects fresh noise at the
- # correct sigma level for full resolution.
- # ================================================================
- stage3_start = max(lightning_start, switch_step_final)
- if hr_enabled:
- stage3_add_noise = True
- else:
- stage3_add_noise = skip_stage1 and skip_stage2
- hr_tag = f" [{orig_px_w}×{orig_px_h} px]" if hr_enabled else ""
- stage3_info = (
- self._format_stage_range(stage3_start, lightning_steps, lightning_steps)
- + hr_tag
- )
- stage3_result = self._run_sampling_stage(
- model=patched_lightning_low,
- positive=positive,
- negative=negative,
- latent=stage3_input,
- seed=seed + STAGE3_SEED_OFFSET,
- steps=lightning_steps,
- cfg=lightning_cfg,
- sampler_name=lightning_sampler,
- scheduler=lightning_scheduler,
- start_at_step=stage3_start,
- end_at_step=lightning_steps,
- add_noise=stage3_add_noise,
- return_with_leftover_noise=False,
- dry_run=dry_run,
- stage_name="Stage 3",
- stage_info=stage3_info,
- )
- bare_logger.info("")
- # === Dry run notification ===
- if dry_run:
- summary_lines = []
- has_base_calc = not skip_stage1 and base_calculation_info
- if has_base_calc or model_switching_info:
- summary_lines.append("Calculations:")
- if has_base_calc:
- formatted = self._format_base_calculation_compact(base_calculation_info)
- summary_lines.append(f"• {formatted}")
- if model_switching_info:
- formatted = self._format_switch_info_compact(model_switching_info)
- summary_lines.append(f"• {formatted}")
- summary_lines.append("")
- summary_lines.append("Stage Configuration:")
- summary_lines.append(f"• Stage 1: {stage1_info}")
- summary_lines.append(f"• Stage 2: {stage2_info}")
- if upscale_info:
- summary_lines.append(f"• {upscale_info}")
- summary_lines.append(f"• Stage 3: {stage3_info}")
- if hr_enabled:
- summary_lines.append("")
- summary_lines.append(
- "Note: Stage 3 uses add_noise=True (fresh noise at full resolution)"
- )
- detail_text = "\n".join(summary_lines)
- core_notifications.send_dry_run_notification(detail_text)
- logger.info("[DRY RUN] Complete - interrupting workflow")
- bare_logger.info("")
- raise comfy.model_management.InterruptProcessingException()
- return stage3_result
- class TripleKSamplerAdvanced(TripleKSamplerAdvancedAlt):
- """Advanced triple-stage sampler with dynamic UI."""
- @classmethod
- def INPUT_TYPES(cls) -> dict[str, dict[str, Any]]:
- base_inputs = cls._get_base_input_types()
- return {
- "required": {
- "base_high": base_inputs["base_high"],
- "lightning_high": base_inputs["lightning_high"],
- "lightning_low": base_inputs["lightning_low"],
- "positive": base_inputs["positive"],
- "negative": base_inputs["negative"],
- "latent_image": base_inputs["latent_image"],
- "seed": base_inputs["seed"],
- "sigma_shift": base_inputs["sigma_shift"],
- "base_quality_threshold": (
- "INT",
- {
- "default": core_config.DEFAULT_BASE_QUALITY_THRESHOLD,
- "min": 1,
- "max": 100,
- "step": 1,
- "tooltip": (
- "Minimum total steps for base_steps auto-calculation "
- f"(config default: {_BASE_QUALITY_THRESHOLD}). "
- "Only applies when base_steps=-1."
- ),
- },
- ),
- "base_steps": (
- "INT",
- {
- "default": -1,
- "min": -1,
- "max": 100,
- "tooltip": "Stage 1 steps. -1 = auto.",
- },
- ),
- "base_cfg": base_inputs["base_cfg"],
- "base_sampler": (
- comfy.samplers.KSampler.SAMPLERS,
- {"tooltip": "Sampler for Stage 1."},
- ),
- "base_scheduler": (
- comfy.samplers.KSampler.SCHEDULERS,
- {"tooltip": "Scheduler for Stage 1."},
- ),
- "lightning_start": (
- "INT",
- {
- "default": 1,
- "min": 0,
- "max": 99,
- "tooltip": "Starting step in lightning schedule. 0 = skip Stage 1.",
- },
- ),
- "lightning_steps": base_inputs["lightning_steps"],
- "lightning_cfg": (
- "FLOAT",
- {
- "default": 1.0,
- "min": 0.0,
- "max": 100.0,
- "step": 0.1,
- "tooltip": "CFG for Stage 2 and Stage 3.",
- },
- ),
- "lightning_sampler": (
- comfy.samplers.KSampler.SAMPLERS,
- {"tooltip": "Sampler for Stage 2 and Stage 3."},
- ),
- "lightning_scheduler": (
- comfy.samplers.KSampler.SCHEDULERS,
- {"tooltip": "Scheduler for Stage 2 and Stage 3."},
- ),
- "switch_strategy": (
- [
- "50% of steps",
- "Manual switch step",
- "T2V boundary",
- "I2V boundary",
- "Manual boundary",
- "T2V boundary (refined)",
- "I2V boundary (refined)",
- "Manual boundary (refined)",
- ],
- {"default": "50% of steps", "tooltip": "Model switching strategy."},
- ),
- "hr_scale_factor": (
- "FLOAT",
- {
- "default": 2.0,
- "min": 1.0,
- "max": 8.0,
- "step": 0.25,
- "tooltip": "Spatial downscale for Stage 1+2. 1.0 = disabled.",
- },
- ),
- "hr_upscale_method": (
- UPSCALE_METHODS,
- {
- "default": "bislerp",
- "tooltip": "Latent upscale method between Stage 2 and Stage 3.",
- },
- ),
- },
- "optional": {
- "switch_step": (
- "INT",
- {
- "default": -1,
- "min": -1,
- "max": 99,
- "tooltip": "Manual switch step. -1 = 50%.",
- },
- ),
- "switch_boundary": (
- "FLOAT",
- {
- "default": 0.875,
- "min": 0.0,
- "max": 1.0,
- "step": 0.001,
- "tooltip": "Sigma boundary for 'Manual boundary'.",
- },
- ),
- "dry_run": (
- "BOOLEAN",
- {
- "default": False,
- "tooltip": "Test calculations without sampling.",
- },
- ),
- },
- }
- DESCRIPTION = (
- "Advanced triple-stage cascade sampler with hi-res pipeline and dynamic UI."
- )
- # sample() inherited from TripleKSamplerAdvancedAlt
Advertisement
Add Comment
Please, Sign In to add comment