horde_sdk.generation_parameters.image.sampler_work¶

def get_sampler_work_profile(sampler: KNOWN_IMAGE_SAMPLERS) -> SamplerWorkProfile:
    """Return the work profile for ``sampler``.

    Args:
        sampler: Sampler whose marginal work profile is required.

    Returns:
        The sampler's fixed-rate or adaptive work profile.

    Raises:
        KeyError: If the profile registry has drifted from the sampler enum.
    """
    return SAMPLER_WORK_PROFILES[sampler]

def get_sampler_execution_contract(
    version: SamplerExecutionContractVersion,
) -> SamplerExecutionContract:
    """Return the sampler execution contract identified by ``version``.

    Args:
        version: Stable execution contract version advertised by a worker.

    Returns:
        The cumulative contract associated with the version.

    Raises:
        KeyError: If the SDK's version enum and execution-contract registry have drifted.
    """
    return SAMPLER_EXECUTION_CONTRACTS[version]

def minimum_common_sampler_execution_contract_version(
    versions: Collection[SamplerExecutionContractVersion | None],
) -> SamplerExecutionContractVersion | None:
    """Return the newest contract guaranteed by every represented backend path.

    Contract versions are cumulative. A missing version therefore makes the combined profile legacy,
    while combining newer and older contracts yields the older shared contract.

    Args:
        versions: Contract versions for every backend path represented by a worker profile.

    Returns:
        The newest shared contract, or ``None`` when any path is legacy.

    Raises:
        ValueError: If ``versions`` is empty.
    """
    if not versions:
        raise ValueError("At least one sampler execution contract version is required.")
    if any(version is None for version in versions):
        return None

    contract_ranks = {version: rank for rank, version in enumerate(_SAMPLER_EXECUTION_CONTRACT_ORDER)}
    concrete_versions = [version for version in versions if version is not None]
    return min(concrete_versions, key=contract_ranks.__getitem__)

def estimate_sampler_work(
    *,
    sampler: KNOWN_IMAGE_SAMPLERS,
    trajectory_steps: TrajectoryStepCount,
    estimation_policy: SamplerWorkEstimationPolicy,
) -> SamplerWorkEstimate:
    """Return estimated service-accounting work without claiming an execution ceiling.

    Args:
        sampler: Sampler selected by the request.
        trajectory_steps: Requested denoising-schedule length.
        estimation_policy: Service-owned adaptive work estimates.

    Returns:
        The request's typed trajectory and estimated work values.

    Raises:
        ValueError: If an adaptive sampler has no service estimate.
    """
    profile = get_sampler_work_profile(sampler)
    if isinstance(profile, FixedRateSamplerWorkProfile):
        work_units = trajectory_steps.value * profile.marginal_work_units_per_trajectory_step
    else:
        try:
            work_units = estimation_policy.adaptive_sampler_work_units[sampler].value
        except KeyError as error:
            raise ValueError(f"No adaptive work estimate is configured for {sampler!s}.") from error
    return SamplerWorkEstimate(sampler, trajectory_steps, SamplerWorkUnitCount(work_units))

def maximum_adaptive_solver_iterations(
    *,
    trajectory_steps: TrajectoryStepCount,
    execution_policy: AdaptiveSamplerExecutionPolicy = BOUNDED_DPM_ADAPTIVE_V1,
) -> int:
    """Return the exact integer iteration ceiling for an adaptive execution policy.

    Args:
        trajectory_steps: Requested denoising-schedule length.
        execution_policy: Adaptive policy whose rational ceiling is enforced.

    Returns:
        Maximum permitted solver iterations, rounded upward.
    """
    numerator = trajectory_steps.value * execution_policy.iteration_multiplier_numerator
    denominator = execution_policy.iteration_multiplier_denominator
    return max(1, (numerator + denominator - 1) // denominator)

def maximum_sampler_work(
    *,
    sampler: KNOWN_IMAGE_SAMPLERS,
    trajectory_steps: TrajectoryStepCount,
    execution_contract_version: SamplerExecutionContractVersion | None,
    adaptive_work_units_per_iteration: int | None = None,
) -> SamplerWorkCeiling | None:
    """Return a finite work ceiling when the backend contract proves one.

    Args:
        sampler: Sampler selected by the request.
        trajectory_steps: Requested denoising-schedule length.
        execution_contract_version: Cumulative execution contract advertised by the backend.
        adaptive_work_units_per_iteration: Work units consumed by each adaptive solver iteration.

    Returns:
        A typed finite ceiling, or ``None`` when no finite ceiling is guaranteed.

    Raises:
        ValueError: If adaptive work per iteration is missing or invalid for a bounded contract.
    """
    profile = get_sampler_work_profile(sampler)
    if isinstance(profile, FixedRateSamplerWorkProfile):
        return SamplerWorkCeiling(
            sampler,
            trajectory_steps,
            SamplerWorkUnitCount(trajectory_steps.value * profile.marginal_work_units_per_trajectory_step),
            None,
        )

    if execution_contract_version is None:
        return None
    execution_contract = get_sampler_execution_contract(execution_contract_version)
    guarantee = SamplerExecutionGuarantee.BOUNDED_DPM_ADAPTIVE_V1
    if sampler is not BOUNDED_DPM_ADAPTIVE_V1.sampler or guarantee not in execution_contract.guarantees:
        return None
    if (
        not isinstance(adaptive_work_units_per_iteration, int)
        or isinstance(adaptive_work_units_per_iteration, bool)
        or adaptive_work_units_per_iteration < 1
    ):
        raise ValueError("Adaptive work ceilings require positive work units per solver iteration.")
    iterations = maximum_adaptive_solver_iterations(trajectory_steps=trajectory_steps)
    return SamplerWorkCeiling(
        sampler,
        trajectory_steps,
        SamplerWorkUnitCount(iterations * adaptive_work_units_per_iteration),
        guarantee,
    )

def maximum_trajectory_steps_for_work_budget(
    *,
    sampler: KNOWN_IMAGE_SAMPLERS,
    requested_trajectory_steps: TrajectoryStepCount,
    work_budget: SamplerWorkUnitCount,
    estimation_policy: SamplerWorkEstimationPolicy,
) -> TrajectoryStepCount | None:
    """Return the largest requested trajectory length within an estimated-work budget.

    An adaptive service estimate is request-independent. Reducing requested steps therefore cannot
    make an over-budget adaptive request fit; callers must reject it without changing sampler identity.

    Args:
        sampler: Sampler selected by the request.
        requested_trajectory_steps: Requested denoising-schedule length.
        work_budget: Maximum estimated work the service permits.
        estimation_policy: Service-owned adaptive work estimates.

    Returns:
        The largest permitted trajectory length, or ``None`` when step reduction cannot fit the budget.
    """
    profile = get_sampler_work_profile(sampler)
    if isinstance(profile, AdaptiveSamplerWorkProfile):
        estimate = estimate_sampler_work(
            sampler=sampler,
            trajectory_steps=requested_trajectory_steps,
            estimation_policy=estimation_policy,
        )
        return requested_trajectory_steps if estimate.work_units <= work_budget else None

    maximum_steps = min(
        requested_trajectory_steps.value,
        work_budget.value // profile.marginal_work_units_per_trajectory_step,
    )
    return TrajectoryStepCount(maximum_steps) if maximum_steps >= 1 else None