horde_sdk.worker.feature_flags¶

class RESULT_RETURN_METHOD(StrEnum):
    """The method of returning results from a worker."""

    base64_post_back = auto()
    """Base64 post back in the 'job completed' message."""

    base64_post_back_with_url = auto()
    """Base64 post back to a given URL without results in the 'job completed' message."""

    byte_stream = auto()
    """Byte stream to a given URL without results in the 'job completed' message."""

    local_write_to_file = auto()
    """Can write to the local filesystem for jobs originating locally or within a closed environment."""

class WorkerFeatureFlags[ReasonTypeVar: str](ABC, BaseModel):
    """Feature flags for a worker."""

    model_config = get_default_frozen_model_config_dict()

    supported_result_return_methods: list[RESULT_RETURN_METHOD] = Field(default_factory=list)
    """The methods of returning results supported by the worker."""

    supports_threads: bool = Field(default=False)
    """Whether the worker supports threading."""

    def is_capable_of_features(self, features: GenerationFeatureFlags) -> bool:
        """Check if the worker is capable of handling the requested features.

        Args:
            features (GenerationFeatureFlags): The features to check.

        Returns:
            bool: True if the worker is capable of handling the requested features, False otherwise.
        """
        return not self.reasons_not_capable_of_features(features)

    @abstractmethod
    def get_not_capable_reason_type(self) -> type[ReasonTypeVar]:
        """Return the type of the reason for not being capable of handling the requested features.

        Returns:
            type[ReasonTypeVar]: The (python) type of the reason for not being capable of handling the requested
            features.
        """

    @abstractmethod
    def reasons_not_capable_of_features(
        self,
        features: GenerationFeatureFlags,
    ) -> list[ReasonTypeVar] | None:
        """Return a list of reasons why the worker is not capable of handling the requested features.

        Args:
            features (GenerationFeatureFlags): The features to check.

        Returns:
            list[str] | None: A list of reasons why the worker is not capable of handling the requested features,
            or None if the worker is capable.
        """

class PerBaselineFeatureFlags(BaseModel):
    """Represents exhaustive baseline-specific restrictions on a worker feature profile.

    `None` leaves the corresponding flat feature advertisement in effect. Once a map is supplied,
    a missing baseline advertises no support for that feature on the omitted baseline.
    """

    model_config = get_default_frozen_model_config_dict()

    schedulers_map: dict[KNOWN_IMAGE_GENERATION_BASELINE | str, list[KNOWN_IMAGE_SCHEDULERS | str]] | None = Field(
        default=None,
        examples=[
            {
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_1: [
                    KNOWN_IMAGE_SCHEDULERS.simple,
                    KNOWN_IMAGE_SCHEDULERS.normal,
                    KNOWN_IMAGE_SCHEDULERS.exponential,
                ],
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_xl: [
                    KNOWN_IMAGE_SCHEDULERS.simple,
                ],
            },
        ],
    )
    """If set, the supported schedulers for each baseline. If unset, it is assumed that all baselines
    support all schedulers.

    A populated map is exhaustive: a baseline absent from it advertises support for no schedulers at
    all, not for the flat `schedulers` list. A worker setting this map must therefore cover every
    baseline it serves."""

    samplers_map: dict[KNOWN_IMAGE_GENERATION_BASELINE | str, list[KNOWN_IMAGE_SAMPLERS | str]] | None = Field(
        default=None,
        examples=[
            {
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_1: [
                    KNOWN_IMAGE_SAMPLERS.k_lms,
                    KNOWN_IMAGE_SAMPLERS.k_dpm_2,
                    KNOWN_IMAGE_SAMPLERS.k_euler,
                ],
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_xl: [
                    KNOWN_IMAGE_SAMPLERS.k_lms,
                ],
            },
        ],
    )
    """If set, the supported samplers for each baseline. If unset, it is assumed that all baselines
    support all samplers.

    A populated map is exhaustive: a baseline absent from it advertises support for no samplers at
    all, not for the flat `samplers` list. A worker setting this map must therefore cover every
    baseline it serves."""

    tiling_map: dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None = Field(
        default=None,
        examples=[
            {
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_1: True,
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_xl: False,
            },
        ],
    )
    """If set, the supported tiling for each baseline. If unset, it is assumed that all baselines
    follow the flat tiling flag. A populated map is exhaustive."""

    hires_fix_map: dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None = Field(
        default=None,
        examples=[
            {
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_1: True,
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_xl: True,
                KNOWN_IMAGE_GENERATION_BASELINE.flux_1: False,
            },
        ],
    )
    """If set, the supported hires fix for each baseline. If unset, it is assumed that all baselines
    follow the flat hires-fix flag. A populated map is exhaustive."""

    transparent_map: dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None = Field(
        default=None,
        examples=[
            {
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_1: True,
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_xl: True,
                KNOWN_IMAGE_GENERATION_BASELINE.stable_cascade: False,
            },
        ],
    )
    """If set, support for transparent generation for each baseline. If unset, all advertised
    baselines follow the flat transparent flag. A populated map is exhaustive."""

    controlnet_map: dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None = Field(
        default=None,
        examples=[
            {
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_1: True,
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_xl: False,
            },
        ],
    )
    """If set, support for controlnet for each baseline. If unset, it is assumed that all baselines
    follow the flat ControlNet feature flags. A populated map is exhaustive."""

    tis_map: dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None = Field(
        default=None,
        examples=[
            {
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_1: True,
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_xl: False,
            },
        ],
    )
    """If set, support for TIs for each baseline. If unset, it is assumed that all baselines support
    the advertised TI sources. A populated map is exhaustive."""

    loras_map: dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None = Field(
        default=None,
        examples=[
            {
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_1: True,
                KNOWN_IMAGE_GENERATION_BASELINE.stable_diffusion_xl: False,
            },
        ],
    )
    """If set, support for Loras for each baseline. If unset, it is assumed that all baselines
    support the advertised LoRA sources. A populated map is exhaustive."""

class IMAGE_WORKER_NOT_CAPABLE_REASON(StrEnum):
    """Reasons why a worker is not capable of handling a request."""

    clip_skip = auto()
    """The worker does not support clip skip."""

    samplers = auto()
    """The worker does not support the requested samplers."""

    sampler_solver_knobs = auto()
    """The worker does not support one or more requested sampler solver knobs."""

    flow_shift = auto()
    """The worker does not support model-specific flow shifting."""

    transparent = auto()
    """The worker does not support transparent image generation."""

    schedulers = auto()
    """The worker does not support the requested schedulers."""

    tiling = auto()
    """The worker does not support tiling."""

    hires_fix = auto()
    """The worker does not support hires fix."""

    controlnets = auto()
    """The worker does not support controlnets."""

    tis = auto()
    """The worker does not support TIs."""

    loras = auto()
    """The worker does not support Loras."""

    extra_texts = auto()
    """The worker does not support extra texts."""

    extra_source_images = auto()
    """The worker does not support extra source images."""

    post_processing = auto()
    """The worker does not support one or more requested post-processors."""

    source_processing = auto()
    """The worker does not support the requested source-processing mode."""

    workflows = auto()
    """The worker does not support the requested workflow."""

    unsupported_baseline = auto()
    """The worker does not support the requested baseline."""

    unsupported_generation_type = auto()
    """The supplied requirements describe a different generation domain."""

class ImageWorkerFeatureFlags(WorkerFeatureFlags[IMAGE_WORKER_NOT_CAPABLE_REASON]):
    """Represents portable render features advertised by an image worker.

    This profile does not describe model residency, resource fit, queue state, worker readiness, or
    service policy. Consumers compose those constraints with `is_capable_of_features`.
    """

    image_generation_feature_flags: ImageGenerationFeatureFlags
    """The image generation feature flags for the worker."""

    per_baseline_feature_flags: PerBaselineFeatureFlags | None = None
    """The per baseline feature flags for the worker. This includes the supported schedulers and
    samplers for each baseline."""

    backend_clip_skip_representation: CLIP_SKIP_REPRESENTATION | None = None
    """The clip skip representation supported."""

    sampler_execution_contract_version: SamplerExecutionContractVersion | None = None
    """Cumulative SDK execution contract guaranteed by every backend path in this profile."""

    @override
    def get_not_capable_reason_type(self) -> type[IMAGE_WORKER_NOT_CAPABLE_REASON]:
        return IMAGE_WORKER_NOT_CAPABLE_REASON

    @override
    def reasons_not_capable_of_features(
        self,
        requested_features: GenerationFeatureFlags,
    ) -> list[IMAGE_WORKER_NOT_CAPABLE_REASON] | None:
        """Return reasons the advertised worker features do not cover a request.

        Args:
            requested_features: Features required by one image generation.

        Returns:
            The stable incompatibility reasons, or `None` when every requirement is supported.

        """
        if not isinstance(requested_features, ImageGenerationFeatureFlags):
            return [IMAGE_WORKER_NOT_CAPABLE_REASON.unsupported_generation_type]

        compatibility_checks = self._get_image_feature_compatibility_checks(requested_features)
        reasons = [check.reason for check in compatibility_checks.values() if not check.is_supported]
        return reasons or None

    def _get_image_feature_compatibility_checks(
        self,
        requested_features: ImageGenerationFeatureFlags,
    ) -> dict[str, _ImageFeatureCompatibilityCheck]:
        """Return the exhaustive compatibility registry for an image request."""
        supported_features = self.image_generation_feature_flags
        return {
            "extra_texts": _ImageFeatureCompatibilityCheck(
                not requested_features.extra_texts or supported_features.extra_texts,
                IMAGE_WORKER_NOT_CAPABLE_REASON.extra_texts,
            ),
            "extra_source_images": _ImageFeatureCompatibilityCheck(
                not requested_features.extra_source_images or supported_features.extra_source_images,
                IMAGE_WORKER_NOT_CAPABLE_REASON.extra_source_images,
            ),
            "baselines": _ImageFeatureCompatibilityCheck(
                self._worker_supports_requested_values(
                    requested_features.baselines,
                    supported_features.baselines,
                ),
                IMAGE_WORKER_NOT_CAPABLE_REASON.unsupported_baseline,
            ),
            "clip_skip": _ImageFeatureCompatibilityCheck(
                not requested_features.clip_skip or supported_features.clip_skip,
                IMAGE_WORKER_NOT_CAPABLE_REASON.clip_skip,
            ),
            "hires_fix": _ImageFeatureCompatibilityCheck(
                self.worker_supports_requested_hires_fix(requested_features),
                IMAGE_WORKER_NOT_CAPABLE_REASON.hires_fix,
            ),
            "tiling": _ImageFeatureCompatibilityCheck(
                self.worker_supports_requested_tiling(requested_features),
                IMAGE_WORKER_NOT_CAPABLE_REASON.tiling,
            ),
            "schedulers": _ImageFeatureCompatibilityCheck(
                self.worker_supports_requested_schedulers(requested_features),
                IMAGE_WORKER_NOT_CAPABLE_REASON.schedulers,
            ),
            "samplers": _ImageFeatureCompatibilityCheck(
                self.worker_supports_requested_samplers(requested_features),
                IMAGE_WORKER_NOT_CAPABLE_REASON.samplers,
            ),
            "sampler_solver_knobs": _ImageFeatureCompatibilityCheck(
                self._worker_supports_requested_values(
                    requested_features.sampler_solver_knobs,
                    supported_features.sampler_solver_knobs,
                ),
                IMAGE_WORKER_NOT_CAPABLE_REASON.sampler_solver_knobs,
            ),
            "flow_shift": _ImageFeatureCompatibilityCheck(
                not requested_features.flow_shift or supported_features.flow_shift,
                IMAGE_WORKER_NOT_CAPABLE_REASON.flow_shift,
            ),
            "transparent": _ImageFeatureCompatibilityCheck(
                self.worker_supports_requested_transparent(requested_features),
                IMAGE_WORKER_NOT_CAPABLE_REASON.transparent,
            ),
            "controlnets_feature_flags": _ImageFeatureCompatibilityCheck(
                self.worker_supports_requested_controlnets(requested_features),
                IMAGE_WORKER_NOT_CAPABLE_REASON.controlnets,
            ),
            "post_processing": _ImageFeatureCompatibilityCheck(
                self._worker_supports_requested_values(
                    requested_features.post_processing,
                    supported_features.post_processing,
                ),
                IMAGE_WORKER_NOT_CAPABLE_REASON.post_processing,
            ),
            "source_processing": _ImageFeatureCompatibilityCheck(
                self._worker_supports_requested_values(
                    requested_features.source_processing,
                    supported_features.source_processing,
                ),
                IMAGE_WORKER_NOT_CAPABLE_REASON.source_processing,
            ),
            "workflows": _ImageFeatureCompatibilityCheck(
                self._worker_supports_requested_values(
                    requested_features.workflows,
                    supported_features.workflows,
                ),
                IMAGE_WORKER_NOT_CAPABLE_REASON.workflows,
            ),
            "tis": _ImageFeatureCompatibilityCheck(
                self.worker_supports_requested_tis(requested_features),
                IMAGE_WORKER_NOT_CAPABLE_REASON.tis,
            ),
            "loras": _ImageFeatureCompatibilityCheck(
                self.worker_supports_requested_loras(requested_features),
                IMAGE_WORKER_NOT_CAPABLE_REASON.loras,
            ),
        }

    @staticmethod
    def _worker_supports_requested_values(
        requested_values: Sequence[object] | None,
        supported_values: Sequence[object] | None,
    ) -> bool:
        """Return whether all requested values appear in the advertised values."""
        if not requested_values:
            return True
        if not supported_values:
            return False
        return all(requested_value in supported_values for requested_value in requested_values)

    def _per_baseline_boolean_supports_request(
        self,
        requested_features: ImageGenerationFeatureFlags,
        per_baseline_support: dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None,
    ) -> bool:
        """Return whether an exhaustive per-baseline boolean map covers the request."""
        if per_baseline_support is None:
            return True
        return all(per_baseline_support.get(baseline, False) for baseline in requested_features.baselines)

    def worker_supports_requested_samplers(
        self,
        requested_features: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return whether the worker supports every requested sampler."""
        if not self._worker_supports_requested_values(
            requested_features.samplers,
            self.image_generation_feature_flags.samplers,
        ):
            return False
        sampler_map = self.per_baseline_feature_flags.samplers_map if self.per_baseline_feature_flags else None
        if sampler_map is None:
            return True
        return all(
            self._worker_supports_requested_values(requested_features.samplers, sampler_map.get(baseline))
            for baseline in requested_features.baselines
        )

    def worker_supports_requested_schedulers(
        self,
        requested_features: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return whether the worker supports every requested scheduler."""
        if not self._worker_supports_requested_values(
            requested_features.schedulers,
            self.image_generation_feature_flags.schedulers,
        ):
            return False
        scheduler_map = self.per_baseline_feature_flags.schedulers_map if self.per_baseline_feature_flags else None
        if scheduler_map is None:
            return True
        return all(
            self._worker_supports_requested_values(requested_features.schedulers, scheduler_map.get(baseline))
            for baseline in requested_features.baselines
        )

    def worker_supports_requested_tiling(
        self,
        requested_features: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return whether the worker supports tiling for every requested baseline."""
        if not requested_features.tiling:
            return True
        if not self.image_generation_feature_flags.tiling:
            return False
        tiling_map = self.per_baseline_feature_flags.tiling_map if self.per_baseline_feature_flags else None
        if tiling_map is not None:
            return self._per_baseline_boolean_supports_request(requested_features, tiling_map)
        return True

    def worker_supports_requested_hires_fix(
        self,
        requested_features: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return whether the worker supports hires fix for every requested baseline."""
        if not requested_features.hires_fix:
            return True
        if not self.image_generation_feature_flags.hires_fix:
            return False
        hires_fix_map = self.per_baseline_feature_flags.hires_fix_map if self.per_baseline_feature_flags else None
        if hires_fix_map is not None:
            return self._per_baseline_boolean_supports_request(requested_features, hires_fix_map)
        return True

    def worker_supports_requested_transparent(
        self,
        requested_features: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return whether the worker supports transparency for every requested baseline."""
        if not requested_features.transparent:
            return True
        if not self.image_generation_feature_flags.transparent:
            return False
        transparent_map = self.per_baseline_feature_flags.transparent_map if self.per_baseline_feature_flags else None
        if transparent_map is not None:
            return self._per_baseline_boolean_supports_request(requested_features, transparent_map)
        return True

    def worker_supports_requested_controlnets(
        self,
        requested_features: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return whether the worker supports the requested ControlNet configuration."""
        requested_controlnets = requested_features.controlnets_feature_flags
        if requested_controlnets is None:
            return True

        supported_controlnets = self.image_generation_feature_flags.controlnets_feature_flags
        if supported_controlnets is None:
            return False

        controlnet_map = self.per_baseline_feature_flags.controlnet_map if self.per_baseline_feature_flags else None
        if controlnet_map is not None and not self._per_baseline_boolean_supports_request(
            requested_features,
            controlnet_map,
        ):
            return False

        if not self._worker_supports_requested_values(
            requested_controlnets.controlnets, supported_controlnets.controlnets
        ):
            return False

        supports_control_image = not requested_controlnets.image_is_control or supported_controlnets.image_is_control
        supports_returned_map = (
            not requested_controlnets.return_control_map or supported_controlnets.return_control_map
        )
        return supports_control_image and supports_returned_map

    def worker_supports_requested_tis(
        self,
        requested_features: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return whether the worker supports every requested textual-inversion source."""
        if not self._worker_supports_requested_values(requested_features.tis, self.image_generation_feature_flags.tis):
            return False
        if not requested_features.tis:
            return True
        tis_map = self.per_baseline_feature_flags.tis_map if self.per_baseline_feature_flags else None
        return self._per_baseline_boolean_supports_request(requested_features, tis_map)

    def worker_supports_requested_loras(
        self,
        requested_features: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return whether the worker supports every requested LoRA source."""
        if not self._worker_supports_requested_values(
            requested_features.loras,
            self.image_generation_feature_flags.loras,
        ):
            return False
        if not requested_features.loras:
            return True
        loras_map = self.per_baseline_feature_flags.loras_map if self.per_baseline_feature_flags else None
        return self._per_baseline_boolean_supports_request(requested_features, loras_map)

class ALCHEMY_WORKER_NOT_CAPABLE_REASON(StrEnum):
    """Reasons why a worker is not capable of handling an alchemy request."""

    unsupported_upscaler = auto()
    """The worker does not support a requested upscaler."""

    unsupported_facefixer = auto()
    """The worker does not support a requested facefixer."""

    unsupported_interrogator = auto()
    """The worker does not support a requested interrogator."""

    unsupported_caption_model = auto()
    """The worker does not support a requested caption model."""

    unsupported_nsfw_detector = auto()
    """The worker does not support a requested NSFW detector."""

    unsupported_vectorizer = auto()
    """The worker does not support image vectorization."""

    unsupported_annotation = auto()
    """The worker does not support controlnet annotation."""

    unsupported_misc = auto()
    """The worker does not support a requested miscellaneous feature."""

    unsupported_generation_type = auto()
    """The supplied requirements describe a different generation domain."""