Blame for ComfyUI-ChebyFlow/chebyflow.py

import math
import torch

import comfy.model_sampling
import comfy.samplers
from comfy.utils import model_trange as trange

_EPS = 1.0e-6


def _get_model_sampling(model):
    try:
        return model.inner_model.model_patcher.get_model_object("model_sampling")
    except Exception as exc:
        raise RuntimeError(
            "ChebyFlow could not access ComfyUI model_sampling. "
            "Use the normal SamplerCustomAdvanced / KSAMPLER path."
        ) from exc



def _validate_flow_model(model):
    ms = _get_model_sampling(model)
    if not isinstance(ms, comfy.model_sampling.CONST):
        raise ValueError(
            "ChebyFlow requires a CONST/rectified-flow model. "
            f"Got {type(ms).__name__}."
        )
    return ms



def _sigma_list(sigmas):
    vals = [float(v) for v in sigmas.detach().float().cpu().tolist()]
    for i, s in enumerate(vals):
        if s < -1.0e-7 or s > 1.0 + 1.0e-5:
            raise ValueError(f"Flow sigma at index {i} is {s:.8g}, outside [0, 1].")
    for i in range(max(0, len(vals) - 1)):
        if vals[i + 1] > vals[i] + 1.0e-7:
            raise ValueError("Sigmas must be monotonically non-increasing.")
    return vals



def _flow_lambda(sigma):
    s = min(max(float(sigma), _EPS), 1.0 - _EPS)
    return math.log1p(-s) - math.log(s)



def _flow_sigma_from_lambda(lam):
    lam = float(lam)
    if lam >= 0.0:
        e = math.exp(-lam)
        return e / (1.0 + e)
    e = math.exp(lam)
    return 1.0 / (1.0 + e)



def _coordinate_value(sigma, coordinate):
    if coordinate == "sigma":
        return float(sigma)
    if coordinate == "flow_lambda":
        return _flow_lambda(sigma)
    raise ValueError(f"Unknown coordinate: {coordinate}")



def _sigma_from_coordinate(q, coordinate):
    if coordinate == "sigma":
        return min(max(float(q), 0.0), 1.0)
    if coordinate == "flow_lambda":
        return min(max(_flow_sigma_from_lambda(q), 0.0), 1.0)
    raise ValueError(f"Unknown coordinate: {coordinate}")



def _flow_velocity(x, denoised, sigma):
    if float(sigma) <= _EPS:
        return torch.zeros_like(x)
    return (x - denoised) / float(sigma)



def _flow_rhs(x, denoised, sigma, coordinate):
    v = _flow_velocity(x, denoised, sigma)
    if coordinate == "sigma":
        return v
    if coordinate == "flow_lambda":
        s = float(sigma)
        return (-s * (1.0 - s)) * v
    raise ValueError(f"Unknown coordinate: {coordinate}")



def _rms_scalar(t):
    return torch.sqrt(torch.mean(t.float() * t.float()) + 1.0e-20)



def _channel_reduce_dims(t):
    return tuple(i for i in range(t.ndim) if i != 1)



def _metric_components_global(a, b):
    af = a.float()
    bf = b.float()
    diff = bf - af

    rms_a = _rms_scalar(af)
    rms_b = _rms_scalar(bf)
    residual = _rms_scalar(diff) / torch.clamp(rms_a, min=1.0e-12)

    dot = torch.mean(af * bf)
    denom = torch.clamp(rms_a * rms_b, min=1.0e-12)
    cosine = torch.clamp(dot / denom, -1.0, 1.0)
    angle = 1.0 - cosine

    magnitude = torch.abs(torch.log(torch.clamp(rms_b, min=1.0e-12) / torch.clamp(rms_a, min=1.0e-12)))
    return residual, angle, magnitude



def _metric_components_channel(a, b):
    af = a.float()
    bf = b.float()
    dims = _channel_reduce_dims(af)
    diff = bf - af

    rms_a = torch.sqrt(torch.mean(af * af, dim=dims) + 1.0e-20)
    rms_b = torch.sqrt(torch.mean(bf * bf, dim=dims) + 1.0e-20)
    rms_d = torch.sqrt(torch.mean(diff * diff, dim=dims) + 1.0e-20)
    residual = rms_d / torch.clamp(rms_a, min=1.0e-12)

    dot = torch.mean(af * bf, dim=dims)
    denom = torch.clamp(rms_a * rms_b, min=1.0e-12)
    cosine = torch.clamp(dot / denom, -1.0, 1.0)
    angle = 1.0 - cosine

    magnitude = torch.abs(torch.log(torch.clamp(rms_b, min=1.0e-12) / torch.clamp(rms_a, min=1.0e-12)))
    return residual, angle, magnitude



def _combine_metric(residual, angle, magnitude, metric, residual_weight, angle_weight, magnitude_weight):
    if metric == "rms_delta":
        return residual
    if metric == "angle":
        return angle
    if metric == "magnitude":
        return magnitude
    if metric == "combined":
        wr = float(max(residual_weight, 0.0))
        wa = float(max(angle_weight, 0.0))
        wm = float(max(magnitude_weight, 0.0))
        denom = max(wr + wa + wm, 1.0e-12)
        return (wr * residual + wa * angle + wm * magnitude) / denom
    raise ValueError(f"Unknown curvature metric: {metric}")



def _metric_global(a, b, metric, rw, aw, mw):
    parts = _metric_components_global(a, b)
    return _combine_metric(*parts, metric, rw, aw, mw)



def _metric_channel(a, b, metric, rw, aw, mw):
    parts = _metric_components_channel(a, b)
    return _combine_metric(*parts, metric, rw, aw, mw)



def _score_from_metric(metric_value, scale):
    scale = float(max(scale, 1.0e-6))
    return metric_value / (metric_value + scale)



def _broadcast_channel(values, like):
    shape = [1] * like.ndim
    shape[1] = int(values.numel())
    return values.to(device=like.device, dtype=like.dtype).reshape(shape)



def _observer_init(value):
    v = value.detach().float()
    base_dev = torch.clamp(torch.abs(v) * 0.25 + 0.05, min=0.05)
    return {
        "slow": v.clone(),
        "fast": v.clone(),
        "dev": base_dev.clone(),
    }



def _observer_update(obs, value, slow_decay, fast_decay, dev_decay):
    v = value.detach().float()
    if obs is None:
        return _observer_init(v)
    slow = slow_decay * obs["slow"] + (1.0 - slow_decay) * v
    fast = fast_decay * obs["fast"] + (1.0 - fast_decay) * v
    dev = dev_decay * obs["dev"] + (1.0 - dev_decay) * torch.abs(v - slow)
    dev = torch.clamp(dev, min=0.01)
    return {"slow": slow, "fast": fast, "dev": dev}



def _observer_score(obs, value, current_weight, trend_weight, temperature):
    if obs is None:
        return torch.full_like(value.detach().float(), 0.5)
    v = value.detach().float()
    slow = obs["slow"]
    fast = obs["fast"]
    dev = torch.clamp(obs["dev"], min=0.01)
    z_current = (v - slow) / dev
    z_trend = (fast - slow) / dev
    z = current_weight * z_current + trend_weight * z_trend
    return torch.sigmoid(z / float(max(temperature, 1.0e-4)))



def _weighted_score(scores, weights):
    num = None
    denom = 0.0
    for score, weight in zip(scores, weights):
        w = float(max(weight, 0.0))
        if w == 0.0:
            continue
        term = score * w
        num = term if num is None else num + term
        denom += w
    if num is None or denom <= 0.0:
        ref = scores[0]
        return torch.full_like(ref, 0.5)
    return num / denom



def _centered_response(score):
    return 2.0 * torch.clamp(score, 0.0, 1.0) - 1.0



def _stage_from_difficulty(base_c, min_c, max_c, difficulty, adaptivity, stage_gain=1.0):
    shift = float(torch.clamp(_centered_response(difficulty) * float(stage_gain), -1.0, 1.0).item())
    if shift >= 0.0:
        target = base_c + shift * (max_c - base_c)
    else:
        target = base_c + shift * (base_c - min_c)
    c = (1.0 - adaptivity) * base_c + adaptivity * target
    return min(max(float(c), min_c), max_c)



def _correction_strength_from_difficulty(difficulty, low, high):
    low = float(low)
    high = float(high)
    shift = torch.clamp(_centered_response(difficulty), -1.0, 1.0)
    hard = torch.clamp(shift, min=0.0)
    easy = torch.clamp(-shift, min=0.0)
    out = torch.ones_like(shift)
    out = out + hard * (low - 1.0)
    out = out + easy * (high - 1.0)
    return out



def _auto_profile(name):
    name = str(name)
    if name == "detail":
        return {
            "weights": (0.50, 0.35, 0.15),
            "slow_decay": 0.88,
            "fast_decay": 0.42,
            "dev_decay": 0.84,
            "current_weight": 0.70,
            "trend_weight": 0.30,
            "temperature": 1.05,
            "stage_gain": 1.10,
            "corr_low": 0.92,
            "corr_high": 1.10,
            "warmup_steps": 2,
            "soft_start_steps": 0,
        }
    if name == "smooth":
        return {
            "weights": (0.40, 0.45, 0.15),
            "slow_decay": 0.93,
            "fast_decay": 0.34,
            "dev_decay": 0.89,
            "current_weight": 0.62,
            "trend_weight": 0.38,
            "temperature": 1.35,
            "stage_gain": 0.88,
            "corr_low": 0.94,
            "corr_high": 1.04,
            "warmup_steps": 2,
            "soft_start_steps": 0,
        }
    if name == "stable":
        return {
            "weights": (0.42, 0.43, 0.15),
            "slow_decay": 0.94,
            "fast_decay": 0.32,
            "dev_decay": 0.90,
            "current_weight": 0.60,
            "trend_weight": 0.40,
            "temperature": 1.45,
            "stage_gain": 0.78,
            "corr_low": 0.95,
            "corr_high": 1.03,
            "warmup_steps": 2,
            "soft_start_steps": 0,
        }
    if name == "dynamic":
        return {
            "weights": (0.35, 0.50, 0.15),
            "slow_decay": 0.87,
            "fast_decay": 0.38,
            "dev_decay": 0.83,
            "current_weight": 0.67,
            "trend_weight": 0.33,
            "temperature": 1.00,
            "stage_gain": 1.18,
            "corr_low": 0.91,
            "corr_high": 1.09,
            "warmup_steps": 2,
            "soft_start_steps": 0,
        }
    # balanced
    return {
        "weights": (0.45, 0.40, 0.15),
        "slow_decay": 0.90,
        "fast_decay": 0.40,
        "dev_decay": 0.86,
        "current_weight": 0.68,
        "trend_weight": 0.32,
        "temperature": 1.18,
        "stage_gain": 1.00,
        "corr_low": 0.90,
        "corr_high": 1.08,
        "warmup_steps": 2,
        "soft_start_steps": 0,
    }


def _apply_turbo_profile(profile):
    """Return a low-step controller variant without changing the preset's character."""
    p = dict(profile)
    # Faster online calibration: fewer historical steps exist, so both baselines must
    # move sooner. Keep slow > fast so trend information remains meaningful.
    p["slow_decay"] = max(0.68, float(p["slow_decay"]) - 0.15)
    p["fast_decay"] = max(0.16, float(p["fast_decay"]) - 0.15)
    p["dev_decay"] = max(0.68, float(p["dev_decay"]) - 0.12)

    # A little more weight on the current interval and a lower temperature make the
    # normalized controller useful after only one baseline interval.
    p["current_weight"] = min(0.78, float(p["current_weight"]) + 0.05)
    p["trend_weight"] = max(0.0, 1.0 - p["current_weight"])
    p["temperature"] = max(0.75, float(p["temperature"]) * 0.78)
    p["stage_gain"] = float(p["stage_gain"]) * 1.08

    # Low-step intervals are large: move the probe decisively, but keep algebraic
    # correction closer to 1.0 so one interval cannot redirect the whole trajectory.
    p["corr_low"] = 1.0 + (float(p["corr_low"]) - 1.0) * 0.50
    p["corr_high"] = 1.0 + (float(p["corr_high"]) - 1.0) * 0.50

    p["warmup_steps"] = 1
    p["soft_start_steps"] = 1
    p["stage_base_shift"] = 0.05
    p["stage_min_shift"] = 0.07
    p["stage_max_shift"] = 0.01
    return p


def _controller_confidence(step_index, warmup_steps, soft_start_steps):
    """0 during warmup, then a short confidence ramp before full adaptation."""
    i = int(step_index)
    warm = int(max(warmup_steps, 0))
    soft = int(max(soft_start_steps, 0))
    if i < warm:
        return 0.0
    if soft <= 0:
        return 1.0
    j = i - warm
    if j < soft:
        return float(j + 1) / float(soft + 1)
    return 1.0


@torch.no_grad()
def sample_h3_adaptive_chebyshev_advanced(
    model,
    x,
    sigmas,
    extra_args=None,
    callback=None,
    disable=None,
    coordinate="flow_lambda",
    base_stage_position=0.70,
    min_stage_position=0.55,
    max_stage_position=0.95,
    adapt_strength=1.0,
    curvature_smoothing=0.50,
    stage_curvature_scale=0.12,
    curvature_metric="combined",
    residual_weight=1.0,
    angle_weight=1.0,
    magnitude_weight=0.5,
    correction_mode="curvature_down",
    correction_granularity="per_channel",
    correction_min=0.88,
    correction_max=1.08,
    correction_curvature_scale=0.12,
):
    """
    Advanced adaptive 2-stage Chebyshev/GS sampler for H3 CONST flow models.

    This is the manually tunable version. Defaults are intentionally set to a strong
    "all three improvements enabled" configuration:
      - combined curvature metric,
      - curvature-dependent correction damping,
      - per-channel correction strengths.
    """
    extra_args = {} if extra_args is None else extra_args
    _validate_flow_model(model)
    sigma_vals = _sigma_list(sigmas)
    if len(sigma_vals) <= 1:
        return x

    base_c = min(max(float(base_stage_position), 0.05), 0.99)
    min_c = min(max(float(min_stage_position), 0.05), 0.99)
    max_c = min(max(float(max_stage_position), min_c + 1.0e-4), 0.99)
    adapt = float(max(adapt_strength, 0.0))
    smooth = float(min(max(curvature_smoothing, 0.0), 0.99))
    stage_scale = float(max(stage_curvature_scale, 1.0e-6))

    corr_min = float(min(correction_min, correction_max))
    corr_max = float(max(correction_min, correction_max))
    corr_scale = float(max(correction_curvature_scale, 1.0e-6))

    c_state = base_c
    global_metric_ema = None
    channel_metric_ema = None
    s_in = x.new_ones([x.shape[0]])

    for i in trange(len(sigmas) - 1, disable=disable):
        sigma_s = sigma_vals[i]
        sigma_t = sigma_vals[i + 1]
        denoised_s = model(x, sigmas[i] * s_in, **extra_args)

        if callback is not None:
            callback({"x": x, "i": i, "sigma": sigmas[i], "sigma_hat": sigmas[i], "denoised": denoised_s})

        if sigma_t <= _EPS:
            x = denoised_s
            break

        c = min(max(float(c_state), min_c), max_c)
        b_stage = 1.0 / (2.0 * c)
        b_start = 1.0 - b_stage

        q_s = _coordinate_value(sigma_s, coordinate)
        q_t = _coordinate_value(sigma_t, coordinate)
        h = q_t - q_s
        if abs(h) < 1.0e-12:
            continue

        rhs_s = _flow_rhs(x, denoised_s, sigma_s, coordinate)
        q_c = q_s + c * h
        sigma_c = _sigma_from_coordinate(q_c, coordinate)
        x_c = x + (c * h) * rhs_s
        denoised_c = model(x_c, x.new_full((x.shape[0],), sigma_c), **extra_args)
        rhs_c = _flow_rhs(x_c, denoised_c, sigma_c, coordinate)

        x_first = x + h * rhs_s
        x_cheb = x + h * (b_start * rhs_s + b_stage * rhs_c)
        correction = x_cheb - x_first

        g_metric = _metric_global(rhs_s, rhs_c, curvature_metric, residual_weight, angle_weight, magnitude_weight)
        global_metric_ema = g_metric if global_metric_ema is None else (
            smooth * global_metric_ema + (1.0 - smooth) * g_metric
        )
        g_score = _score_from_metric(global_metric_ema, stage_scale)
        target_c = min_c + (max_c - min_c) * float(g_score.item())
        c_state = (1.0 - adapt) * base_c + adapt * target_c
        c_state = min(max(c_state, min_c), max_c)

        if correction_mode == "constant":
            strength = corr_max
        else:
            if correction_granularity == "per_channel":
                c_metric = _metric_channel(rhs_s, rhs_c, curvature_metric, residual_weight, angle_weight, magnitude_weight)
                channel_metric_ema = c_metric if channel_metric_ema is None else (
                    smooth * channel_metric_ema + (1.0 - smooth) * c_metric
                )
                score = _score_from_metric(channel_metric_ema, corr_scale)
                if correction_mode == "curvature_down":
                    strength = corr_max - (corr_max - corr_min) * score
                elif correction_mode == "curvature_up":
                    strength = corr_min + (corr_max - corr_min) * score
                else:
                    raise ValueError(f"Unknown correction mode: {correction_mode}")
                strength = _broadcast_channel(strength, correction)
            else:
                score = _score_from_metric(global_metric_ema, corr_scale)
                if correction_mode == "curvature_down":
                    strength = corr_max - (corr_max - corr_min) * score
                elif correction_mode == "curvature_up":
                    strength = corr_min + (corr_max - corr_min) * score
                else:
                    raise ValueError(f"Unknown correction mode: {correction_mode}")
                strength = float(strength.item())

        x = x_first + correction * strength

    return x


@torch.no_grad()
def sample_h3_adaptive_chebyshev_auto(
    model,
    x,
    sigmas,
    extra_args=None,
    callback=None,
    disable=None,
    coordinate="flow_lambda",
    base_stage_position=0.72,
    min_stage_position=0.58,
    max_stage_position=0.96,
    adaptivity=1.0,
    quality_bias="balanced",
    turbo_mode=False,
    correction_granularity="per_channel",
):
    """
    Self-calibrating adaptive Chebyshev sampler.

    No extra model evaluations are added. The sampler self-calibrates online during the
    current generation using only the two evaluations it already performs per interval.

    turbo_mode is an orthogonal numerical regime, not a quality preset. It keeps the
    selected quality_bias character but accelerates observer dynamics, shortens startup,
    biases the internal stage later, and narrows correction excursions for large low-step
    intervals such as ~6-10 step distilled/Turbo-LoRA workflows.
    """
    extra_args = {} if extra_args is None else extra_args
    _validate_flow_model(model)
    sigma_vals = _sigma_list(sigmas)
    if len(sigma_vals) <= 1:
        return x

    profile = _auto_profile(quality_bias)
    if bool(turbo_mode):
        profile = _apply_turbo_profile(profile)

    # User values remain the baseline. Turbo mode shifts the operating envelope later
    # without requiring separate turbo presets. Keep the internal stage strictly below 1.
    base_c = float(base_stage_position)
    min_c = float(min_stage_position)
    max_c = float(max_stage_position)
    if bool(turbo_mode):
        base_c += float(profile.get("stage_base_shift", 0.0))
        min_c += float(profile.get("stage_min_shift", 0.0))
        max_c += float(profile.get("stage_max_shift", 0.0))

    base_c = min(max(base_c, 0.05), 0.99)
    min_c = min(max(min_c, 0.05), 0.99)
    max_c = min(max(max_c, min_c + 1.0e-4), 0.99)
    base_c = min(max(base_c, min_c), max_c)

    adaptivity = float(max(adaptivity, 0.0))
    weights = profile["weights"]
    warmup_steps = int(profile["warmup_steps"])
    soft_start_steps = int(profile.get("soft_start_steps", 0))

    obs_g_res = obs_g_ang = obs_g_mag = None
    obs_c_res = obs_c_ang = obs_c_mag = None
    c_state = base_c
    s_in = x.new_ones([x.shape[0]])

    for i in trange(len(sigmas) - 1, disable=disable):
        sigma_s = sigma_vals[i]
        sigma_t = sigma_vals[i + 1]
        denoised_s = model(x, sigmas[i] * s_in, **extra_args)

        if callback is not None:
            callback({"x": x, "i": i, "sigma": sigmas[i], "sigma_hat": sigmas[i], "denoised": denoised_s})

        if sigma_t <= _EPS:
            x = denoised_s
            break

        confidence = _controller_confidence(i, warmup_steps, soft_start_steps)
        c = base_c if confidence <= 0.0 else c_state
        c = min(max(float(c), min_c), max_c)
        b_stage = 1.0 / (2.0 * c)
        b_start = 1.0 - b_stage

        q_s = _coordinate_value(sigma_s, coordinate)
        q_t = _coordinate_value(sigma_t, coordinate)
        h = q_t - q_s
        if abs(h) < 1.0e-12:
            continue

        rhs_s = _flow_rhs(x, denoised_s, sigma_s, coordinate)
        q_c = q_s + c * h
        sigma_c = _sigma_from_coordinate(q_c, coordinate)
        x_c = x + (c * h) * rhs_s
        denoised_c = model(x_c, x.new_full((x.shape[0],), sigma_c), **extra_args)
        rhs_c = _flow_rhs(x_c, denoised_c, sigma_c, coordinate)

        x_first = x + h * rhs_s
        x_cheb = x + h * (b_start * rhs_s + b_stage * rhs_c)
        correction = x_cheb - x_first

        # Current raw local difficulty components.
        g_res, g_ang, g_mag = _metric_components_global(rhs_s, rhs_c)

        if obs_g_res is None:
            g_difficulty = torch.full_like(g_res, 0.5)
        else:
            g_res_s = _observer_score(obs_g_res, g_res, profile["current_weight"], profile["trend_weight"], profile["temperature"])
            g_ang_s = _observer_score(obs_g_ang, g_ang, profile["current_weight"], profile["trend_weight"], profile["temperature"])
            g_mag_s = _observer_score(obs_g_mag, g_mag, profile["current_weight"], profile["trend_weight"], profile["temperature"])
            g_difficulty = _weighted_score((g_res_s, g_ang_s, g_mag_s), weights)

        # Global stage controller for NEXT interval. During turbo soft-start, the
        # confidence factor ramps this from fixed-base behavior to full adaptation.
        c_state = _stage_from_difficulty(
            base_c,
            min_c,
            max_c,
            g_difficulty,
            adaptivity * confidence,
            profile["stage_gain"],
        )

        c_res = c_ang = c_mag = None
        if correction_granularity == "per_channel":
            c_res, c_ang, c_mag = _metric_components_channel(rhs_s, rhs_c)

        # Correction controller for CURRENT interval. The automatic strength is blended
        # toward 1.0 by controller confidence, giving Turbo mode a 50% soft-start step.
        if confidence <= 0.0:
            strength = 1.0
        else:
            if correction_granularity == "per_channel":
                if obs_c_res is None:
                    c_difficulty = torch.full_like(c_res, 0.5)
                else:
                    c_res_s = _observer_score(obs_c_res, c_res, profile["current_weight"], profile["trend_weight"], profile["temperature"])
                    c_ang_s = _observer_score(obs_c_ang, c_ang, profile["current_weight"], profile["trend_weight"], profile["temperature"])
                    c_mag_s = _observer_score(obs_c_mag, c_mag, profile["current_weight"], profile["trend_weight"], profile["temperature"])
                    c_difficulty = _weighted_score((c_res_s, c_ang_s, c_mag_s), weights)
                auto_strength = _correction_strength_from_difficulty(
                    c_difficulty, profile["corr_low"], profile["corr_high"]
                )
                auto_strength = 1.0 + confidence * (auto_strength - 1.0)
                strength = _broadcast_channel(auto_strength, correction)
            else:
                auto_strength = _correction_strength_from_difficulty(
                    g_difficulty, profile["corr_low"], profile["corr_high"]
                )
                auto_strength = 1.0 + confidence * (auto_strength - 1.0)
                strength = float(auto_strength.item())

        x = x_first + correction * strength

        # Update observers AFTER using them: current difficulty is judged against the
        # previously learned baseline, rather than being normalized away immediately.
        obs_g_res = _observer_update(obs_g_res, g_res, profile["slow_decay"], profile["fast_decay"], profile["dev_decay"])
        obs_g_ang = _observer_update(obs_g_ang, g_ang, profile["slow_decay"], profile["fast_decay"], profile["dev_decay"])
        obs_g_mag = _observer_update(obs_g_mag, g_mag, profile["slow_decay"], profile["fast_decay"], profile["dev_decay"])

        if correction_granularity == "per_channel":
            obs_c_res = _observer_update(obs_c_res, c_res, profile["slow_decay"], profile["fast_decay"], profile["dev_decay"])
            obs_c_ang = _observer_update(obs_c_ang, c_ang, profile["slow_decay"], profile["fast_decay"], profile["dev_decay"])
            obs_c_mag = _observer_update(obs_c_mag, c_mag, profile["slow_decay"], profile["fast_decay"], profile["dev_decay"])

    return x


class H3AdaptiveChebyshevAdvancedSampler:
    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "coordinate": (["flow_lambda", "sigma"], {"default": "flow_lambda"}),
                "base_stage_position": ("FLOAT", {"default": 0.70, "min": 0.05, "max": 0.99, "step": 0.01}),
                "min_stage_position": ("FLOAT", {"default": 0.55, "min": 0.05, "max": 0.99, "step": 0.01}),
                "max_stage_position": ("FLOAT", {"default": 0.95, "min": 0.05, "max": 0.99, "step": 0.01}),
                "adapt_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.5, "step": 0.05}),
                "curvature_smoothing": ("FLOAT", {"default": 0.50, "min": 0.0, "max": 0.95, "step": 0.05}),
                "stage_curvature_scale": ("FLOAT", {"default": 0.12, "min": 0.01, "max": 2.0, "step": 0.01}),
                "curvature_metric": (["rms_delta", "angle", "magnitude", "combined"], {"default": "combined"}),
                "residual_weight": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 4.0, "step": 0.05}),
                "angle_weight": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 4.0, "step": 0.05}),
                "magnitude_weight": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 4.0, "step": 0.05}),
                "correction_mode": (["constant", "curvature_down", "curvature_up"], {"default": "curvature_down"}),
                "correction_granularity": (["global", "per_channel"], {"default": "per_channel"}),
                "correction_min": ("FLOAT", {"default": 0.88, "min": 0.0, "max": 1.5, "step": 0.01}),
                "correction_max": ("FLOAT", {"default": 1.08, "min": 0.0, "max": 1.5, "step": 0.01}),
                "correction_curvature_scale": ("FLOAT", {"default": 0.12, "min": 0.01, "max": 2.0, "step": 0.01}),
            }
        }

    RETURN_TYPES = ("SAMPLER",)
    RETURN_NAMES = ("sampler",)
    FUNCTION = "get_sampler"
    CATEGORY = "sampling/custom_sampling/samplers"

    def get_sampler(
        self,
        coordinate,
        base_stage_position,
        min_stage_position,
        max_stage_position,
        adapt_strength,
        curvature_smoothing,
        stage_curvature_scale,
        curvature_metric,
        residual_weight,
        angle_weight,
        magnitude_weight,
        correction_mode,
        correction_granularity,
        correction_min,
        correction_max,
        correction_curvature_scale,
    ):
        sampler = comfy.samplers.KSAMPLER(
            sample_h3_adaptive_chebyshev_advanced,
            extra_options={
                "coordinate": coordinate,
                "base_stage_position": base_stage_position,
                "min_stage_position": min_stage_position,
                "max_stage_position": max_stage_position,
                "adapt_strength": adapt_strength,
                "curvature_smoothing": curvature_smoothing,
                "stage_curvature_scale": stage_curvature_scale,
                "curvature_metric": curvature_metric,
                "residual_weight": residual_weight,
                "angle_weight": angle_weight,
                "magnitude_weight": magnitude_weight,
                "correction_mode": correction_mode,
                "correction_granularity": correction_granularity,
                "correction_min": correction_min,
                "correction_max": correction_max,
                "correction_curvature_scale": correction_curvature_scale,
            },
        )
        return (sampler,)


class H3AdaptiveChebyshevAutoSampler:
    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "coordinate": (["flow_lambda", "sigma"], {"default": "flow_lambda"}),
                "base_stage_position": ("FLOAT", {"default": 0.72, "min": 0.05, "max": 0.99, "step": 0.01}),
                "min_stage_position": ("FLOAT", {"default": 0.58, "min": 0.05, "max": 0.99, "step": 0.01}),
                "max_stage_position": ("FLOAT", {"default": 0.96, "min": 0.05, "max": 0.99, "step": 0.01}),
                "adaptivity": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.5, "step": 0.05}),
                "quality_bias": (["balanced", "detail", "smooth", "stable", "dynamic"], {"default": "balanced"}),
                "turbo_mode": ("BOOLEAN", {"default": False}),
                "correction_granularity": (["global", "per_channel"], {"default": "per_channel"}),
            }
        }

    RETURN_TYPES = ("SAMPLER",)
    RETURN_NAMES = ("sampler",)
    FUNCTION = "get_sampler"
    CATEGORY = "sampling/custom_sampling/samplers"

    def get_sampler(self, coordinate, base_stage_position, min_stage_position, max_stage_position, adaptivity, quality_bias, turbo_mode, correction_granularity):
        sampler = comfy.samplers.KSAMPLER(
            sample_h3_adaptive_chebyshev_auto,
            extra_options={
                "coordinate": coordinate,
                "base_stage_position": base_stage_position,
                "min_stage_position": min_stage_position,
                "max_stage_position": max_stage_position,
                "adaptivity": adaptivity,
                "quality_bias": quality_bias,
                "turbo_mode": turbo_mode,
                "correction_granularity": correction_granularity,
            },
        )
        return (sampler,)


# Backward-compatible alias for earlier pack versions.
H3AdaptiveChebyshevProSampler = H3AdaptiveChebyshevAdvancedSampler


NODE_CLASS_MAPPINGS = {
    "H3AdaptiveChebyshevAdvancedSampler": H3AdaptiveChebyshevAdvancedSampler,
    "H3AdaptiveChebyshevAutoSampler": H3AdaptiveChebyshevAutoSampler,
    "H3AdaptiveChebyshevProSampler": H3AdaptiveChebyshevAdvancedSampler,
}

NODE_DISPLAY_NAME_MAPPINGS = {
    "H3AdaptiveChebyshevAdvancedSampler": "ChebyFlow Advanced Sampler",
    "H3AdaptiveChebyshevAutoSampler": "ChebyFlow Auto Sampler",
    "H3AdaptiveChebyshevProSampler": "ChebyFlow Advanced Sampler",
}