Skip to content

feature_flags

ReasonTypeVar module-attribute 

ReasonTypeVar = TypeVar('ReasonTypeVar', bound=str)

RESULT_RETURN_METHOD

Bases: StrEnum

The method of returning results from a worker.

Source code in horde_sdk/worker/feature_flags.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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."""

base64_post_back class-attribute instance-attribute 

base64_post_back = auto()

Base64 post back in the 'job completed' message.

base64_post_back_with_url class-attribute instance-attribute 

base64_post_back_with_url = auto()

Base64 post back to a given URL without results in the 'job completed' message.

byte_stream class-attribute instance-attribute 

byte_stream = auto()

Byte stream to a given URL without results in the 'job completed' message.

local_write_to_file class-attribute instance-attribute 

local_write_to_file = auto()

Can write to the local filesystem for jobs originating locally or within a closed environment.

WorkerFeatureFlags

Bases: ABC, BaseModel

Feature flags for a worker.

Source code in horde_sdk/worker/feature_flags.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
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.
        """

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

supported_result_return_methods class-attribute instance-attribute 

supported_result_return_methods: list[
    RESULT_RETURN_METHOD
] = Field(default_factory=list)

The methods of returning results supported by the worker.

supports_threads class-attribute instance-attribute 

supports_threads: bool = Field(default=False)

Whether the worker supports threading.

is_capable_of_features 

is_capable_of_features(
    features: GenerationFeatureFlags,
) -> bool

Check if the worker is capable of handling the requested features.

Parameters:

Returns:

  • bool ( bool ) –

    True if the worker is capable of handling the requested features, False otherwise.

Source code in horde_sdk/worker/feature_flags.py 
57
58
59
60
61
62
63
64
65
66
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)

get_not_capable_reason_type abstractmethod 

get_not_capable_reason_type() -> type[ReasonTypeVar]

Return the type of the reason for not being capable of handling the requested features.

Returns:

  • type[ReasonTypeVar] –

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

  • type[ReasonTypeVar] –

    features.

Source code in horde_sdk/worker/feature_flags.py 
68
69
70
71
72
73
74
75
@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.
    """

reasons_not_capable_of_features abstractmethod 

reasons_not_capable_of_features(
    features: GenerationFeatureFlags,
) -> list[ReasonTypeVar] | None

Return a list of reasons why the worker is not capable of handling the requested features.

Parameters:

Returns:

  • list[ReasonTypeVar] | None –

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

  • list[ReasonTypeVar] | None –

    or None if the worker is capable.

Source code in horde_sdk/worker/feature_flags.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@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.
    """

PerBaselineFeatureFlags

Bases: BaseModel

Feature flags for a worker per baseline.

Source code in horde_sdk/worker/feature_flags.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
class PerBaselineFeatureFlags(BaseModel):
    """Feature flags for a worker per 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."""

    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."""

    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
    support tiling."""

    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
    support hires fix."""

    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
    support controlnets."""

    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
    TIs."""

    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 Loras."""

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

schedulers_map class-attribute instance-attribute 

schedulers_map: (
    dict[
        KNOWN_IMAGE_GENERATION_BASELINE | str,
        list[KNOWN_IMAGE_SCHEDULERS | str],
    ]
    | None
) = Field(
    default=None,
    examples=[
        {
            stable_diffusion_1: [
                simple,
                normal,
                exponential,
            ],
            stable_diffusion_xl: [simple],
        }
    ],
)

If set, the supported schedulers for each baseline. If unset, it is assumed that all baselines support all schedulers.

samplers_map class-attribute instance-attribute 

samplers_map: (
    dict[
        KNOWN_IMAGE_GENERATION_BASELINE | str,
        list[KNOWN_IMAGE_SAMPLERS | str],
    ]
    | None
) = Field(
    default=None,
    examples=[
        {
            stable_diffusion_1: [k_lms, k_dpm_2, k_euler],
            stable_diffusion_xl: [k_lms],
        }
    ],
)

If set, the supported samplers for each baseline. If unset, it is assumed that all baselines support all samplers.

tiling_map class-attribute instance-attribute 

tiling_map: (
    dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None
) = Field(
    default=None,
    examples=[
        {
            stable_diffusion_1: True,
            stable_diffusion_xl: False,
        }
    ],
)

If set, the supported tiling for each baseline. If unset, it is assumed that all baselines support tiling.

hires_fix_map class-attribute instance-attribute 

hires_fix_map: (
    dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None
) = Field(
    default=None,
    examples=[
        {
            stable_diffusion_1: True,
            stable_diffusion_xl: True,
            flux_1: False,
        }
    ],
)

If set, the supported hires fix for each baseline. If unset, it is assumed that all baselines support hires fix.

controlnet_map class-attribute instance-attribute 

controlnet_map: (
    dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None
) = Field(
    default=None,
    examples=[
        {
            stable_diffusion_1: True,
            stable_diffusion_xl: False,
        }
    ],
)

If set, support for controlnet for each baseline. If unset, it is assumed that all baselines support controlnets.

tis_map class-attribute instance-attribute 

tis_map: (
    dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None
) = Field(
    default=None,
    examples=[
        {
            stable_diffusion_1: True,
            stable_diffusion_xl: False,
        }
    ],
)

If set, support for TIs for each baseline. If unset, it is assumed that all baselines support TIs.

loras_map class-attribute instance-attribute 

loras_map: (
    dict[KNOWN_IMAGE_GENERATION_BASELINE | str, bool] | None
) = Field(
    default=None,
    examples=[
        {
            stable_diffusion_1: True,
            stable_diffusion_xl: False,
        }
    ],
)

If set, support for Loras for each baseline. If unset, it is assumed that all baselines support Loras.

IMAGE_WORKER_NOT_CAPABLE_REASON

Bases: StrEnum

Reasons why a worker is not capable of handling a request.

Source code in horde_sdk/worker/feature_flags.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
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."""

    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."""

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

clip_skip class-attribute instance-attribute 

clip_skip = auto()

The worker does not support clip skip.

samplers class-attribute instance-attribute 

samplers = auto()

The worker does not support the requested samplers.

schedulers class-attribute instance-attribute 

schedulers = auto()

The worker does not support the requested schedulers.

tiling class-attribute instance-attribute 

tiling = auto()

The worker does not support tiling.

hires_fix class-attribute instance-attribute 

hires_fix = auto()

The worker does not support hires fix.

controlnets class-attribute instance-attribute 

controlnets = auto()

The worker does not support controlnets.

tis class-attribute instance-attribute 

tis = auto()

The worker does not support TIs.

loras class-attribute instance-attribute 

loras = auto()

The worker does not support Loras.

extra_texts class-attribute instance-attribute 

extra_texts = auto()

The worker does not support extra texts.

extra_source_images class-attribute instance-attribute 

extra_source_images = auto()

The worker does not support extra source images.

unsupported_baseline class-attribute instance-attribute 

unsupported_baseline = auto()

The worker does not support the requested baseline.

ImageWorkerFeatureFlags

Bases: WorkerFeatureFlags[IMAGE_WORKER_NOT_CAPABLE_REASON]

Feature flags for an image worker.

Source code in horde_sdk/worker/feature_flags.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
class ImageWorkerFeatureFlags(WorkerFeatureFlags[IMAGE_WORKER_NOT_CAPABLE_REASON]):
    """Feature flags for an image worker."""

    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."""

    @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,
        request: GenerationFeatureFlags,
    ) -> list[IMAGE_WORKER_NOT_CAPABLE_REASON] | None:
        """Return a list of reasons why a worker is not capable of handling a request."""
        if not isinstance(request, ImageGenerationFeatureFlags):
            logger.debug(f"Request is not an ImageGenerationFeatureFlags instance. Request type: {type(request)}")
            return None

        reasons = []

        if request.clip_skip and not self.image_generation_feature_flags.clip_skip:
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.clip_skip)

        if not self.worker_supports_requested_samplers(request):
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.samplers)

        if not self.worker_supports_requested_schedulers(request):
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.schedulers)

        if not self.worker_supports_requested_tiling(request):
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.tiling)

        if not self.worker_supports_requested_hires_fix(request):
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.hires_fix)

        if not self.worker_supports_requested_controlnets(request):
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.controlnets)

        if not self.worker_supports_requested_tis(request):
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.tis)

        if not self.worker_supports_requested_loras(request):
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.loras)

        if request.extra_texts and not self.image_generation_feature_flags.extra_texts:
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.extra_texts)

        if request.extra_source_images and not self.image_generation_feature_flags.extra_source_images:
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.extra_source_images)

        if (
            request.baselines
            and self.image_generation_feature_flags.baselines
            and (not any(baseline in self.image_generation_feature_flags.baselines for baseline in request.baselines))
        ):
            reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.unsupported_baseline)

        return reasons if reasons else None

    def worker_supports_requested_samplers(
        self,
        request: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return True if the worker supports the samplers requested."""
        if self.per_baseline_feature_flags and self.per_baseline_feature_flags.samplers_map:
            for baseline in request.baselines or []:
                if any(
                    sampler not in self.per_baseline_feature_flags.samplers_map.get(baseline, [])
                    for sampler in request.samplers
                ):
                    return False
        else:
            if any(sampler not in self.image_generation_feature_flags.samplers for sampler in request.samplers):
                return False
        return True

    def worker_supports_requested_schedulers(
        self,
        request: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return True if the worker supports the schedulers requested."""
        if self.per_baseline_feature_flags and self.per_baseline_feature_flags.schedulers_map:
            for baseline in request.baselines or []:
                if any(
                    scheduler not in self.per_baseline_feature_flags.schedulers_map.get(baseline, [])
                    for scheduler in request.schedulers
                ):
                    return False
        else:
            if any(
                scheduler not in self.image_generation_feature_flags.schedulers for scheduler in request.schedulers
            ):
                return False
        return True

    def worker_supports_requested_tiling(
        self,
        request: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return True if the worker supports tiling."""
        if self.per_baseline_feature_flags and self.per_baseline_feature_flags.tiling_map:
            for baseline in request.baselines or []:
                if not self.per_baseline_feature_flags.tiling_map.get(baseline, True):
                    return False
        else:
            if request.tiling and not self.image_generation_feature_flags.tiling:
                return False
        return True

    def worker_supports_requested_hires_fix(
        self,
        request: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return True if the worker supports hires fix."""
        if self.per_baseline_feature_flags and self.per_baseline_feature_flags.hires_fix_map:
            for baseline in request.baselines or []:
                if not self.per_baseline_feature_flags.hires_fix_map.get(baseline, True):
                    return False
        else:
            if request.hires_fix and not self.image_generation_feature_flags.hires_fix:
                return False
        return True

    def worker_supports_requested_controlnets(
        self,
        request: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return True if the worker supports controlnets."""
        if not request.controlnets_feature_flags:
            return True

        if not self.image_generation_feature_flags.controlnets_feature_flags:
            return False

        if self.per_baseline_feature_flags and self.per_baseline_feature_flags.controlnet_map:
            for baseline in request.baselines or []:
                if not self.per_baseline_feature_flags.controlnet_map.get(baseline, True):
                    return False

        if (
            request.controlnets_feature_flags.image_is_control
            and not self.image_generation_feature_flags.controlnets_feature_flags.image_is_control
        ):
            return False

        if (  # noqa SIM103: For readability, we return this False directly
            request.controlnets_feature_flags.return_control_map
            and not self.image_generation_feature_flags.controlnets_feature_flags.return_control_map
        ):
            return False

        return True

    def worker_supports_requested_tis(
        self,
        request: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return True if the worker supports TIs."""
        if self.per_baseline_feature_flags and self.per_baseline_feature_flags.tis_map:
            for baseline in request.baselines or []:
                if not self.per_baseline_feature_flags.tis_map.get(baseline, True):
                    return False
        else:
            if request.tis and not self.image_generation_feature_flags.tis:
                return False
        return True

    def worker_supports_requested_loras(
        self,
        request: ImageGenerationFeatureFlags,
    ) -> bool:
        """Return True if the worker supports Loras."""
        if self.per_baseline_feature_flags and self.per_baseline_feature_flags.loras_map:
            for baseline in request.baselines or []:
                if not self.per_baseline_feature_flags.loras_map.get(baseline, True):
                    return False
        else:
            if request.loras and not self.image_generation_feature_flags.loras:
                return False
        return True

image_generation_feature_flags instance-attribute 

image_generation_feature_flags: ImageGenerationFeatureFlags

The image generation feature flags for the worker.

per_baseline_feature_flags class-attribute instance-attribute 

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 class-attribute instance-attribute 

backend_clip_skip_representation: (
    CLIP_SKIP_REPRESENTATION | None
) = None

The clip skip representation supported.

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

supported_result_return_methods class-attribute instance-attribute 

supported_result_return_methods: list[
    RESULT_RETURN_METHOD
] = Field(default_factory=list)

The methods of returning results supported by the worker.

supports_threads class-attribute instance-attribute 

supports_threads: bool = Field(default=False)

Whether the worker supports threading.

get_not_capable_reason_type 

get_not_capable_reason_type() -> (
    type[IMAGE_WORKER_NOT_CAPABLE_REASON]
)
Source code in horde_sdk/worker/feature_flags.py 
246
247
248
@override
def get_not_capable_reason_type(self) -> type[IMAGE_WORKER_NOT_CAPABLE_REASON]:
    return IMAGE_WORKER_NOT_CAPABLE_REASON

reasons_not_capable_of_features 

reasons_not_capable_of_features(
    request: GenerationFeatureFlags,
) -> list[IMAGE_WORKER_NOT_CAPABLE_REASON] | None

Return a list of reasons why a worker is not capable of handling a request.

Source code in horde_sdk/worker/feature_flags.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
@override
def reasons_not_capable_of_features(
    self,
    request: GenerationFeatureFlags,
) -> list[IMAGE_WORKER_NOT_CAPABLE_REASON] | None:
    """Return a list of reasons why a worker is not capable of handling a request."""
    if not isinstance(request, ImageGenerationFeatureFlags):
        logger.debug(f"Request is not an ImageGenerationFeatureFlags instance. Request type: {type(request)}")
        return None

    reasons = []

    if request.clip_skip and not self.image_generation_feature_flags.clip_skip:
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.clip_skip)

    if not self.worker_supports_requested_samplers(request):
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.samplers)

    if not self.worker_supports_requested_schedulers(request):
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.schedulers)

    if not self.worker_supports_requested_tiling(request):
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.tiling)

    if not self.worker_supports_requested_hires_fix(request):
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.hires_fix)

    if not self.worker_supports_requested_controlnets(request):
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.controlnets)

    if not self.worker_supports_requested_tis(request):
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.tis)

    if not self.worker_supports_requested_loras(request):
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.loras)

    if request.extra_texts and not self.image_generation_feature_flags.extra_texts:
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.extra_texts)

    if request.extra_source_images and not self.image_generation_feature_flags.extra_source_images:
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.extra_source_images)

    if (
        request.baselines
        and self.image_generation_feature_flags.baselines
        and (not any(baseline in self.image_generation_feature_flags.baselines for baseline in request.baselines))
    ):
        reasons.append(IMAGE_WORKER_NOT_CAPABLE_REASON.unsupported_baseline)

    return reasons if reasons else None

worker_supports_requested_samplers 

worker_supports_requested_samplers(
    request: ImageGenerationFeatureFlags,
) -> bool

Return True if the worker supports the samplers requested.

Source code in horde_sdk/worker/feature_flags.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def worker_supports_requested_samplers(
    self,
    request: ImageGenerationFeatureFlags,
) -> bool:
    """Return True if the worker supports the samplers requested."""
    if self.per_baseline_feature_flags and self.per_baseline_feature_flags.samplers_map:
        for baseline in request.baselines or []:
            if any(
                sampler not in self.per_baseline_feature_flags.samplers_map.get(baseline, [])
                for sampler in request.samplers
            ):
                return False
    else:
        if any(sampler not in self.image_generation_feature_flags.samplers for sampler in request.samplers):
            return False
    return True

worker_supports_requested_schedulers 

worker_supports_requested_schedulers(
    request: ImageGenerationFeatureFlags,
) -> bool

Return True if the worker supports the schedulers requested.

Source code in horde_sdk/worker/feature_flags.py 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def worker_supports_requested_schedulers(
    self,
    request: ImageGenerationFeatureFlags,
) -> bool:
    """Return True if the worker supports the schedulers requested."""
    if self.per_baseline_feature_flags and self.per_baseline_feature_flags.schedulers_map:
        for baseline in request.baselines or []:
            if any(
                scheduler not in self.per_baseline_feature_flags.schedulers_map.get(baseline, [])
                for scheduler in request.schedulers
            ):
                return False
    else:
        if any(
            scheduler not in self.image_generation_feature_flags.schedulers for scheduler in request.schedulers
        ):
            return False
    return True

worker_supports_requested_tiling 

worker_supports_requested_tiling(
    request: ImageGenerationFeatureFlags,
) -> bool

Return True if the worker supports tiling.

Source code in horde_sdk/worker/feature_flags.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
def worker_supports_requested_tiling(
    self,
    request: ImageGenerationFeatureFlags,
) -> bool:
    """Return True if the worker supports tiling."""
    if self.per_baseline_feature_flags and self.per_baseline_feature_flags.tiling_map:
        for baseline in request.baselines or []:
            if not self.per_baseline_feature_flags.tiling_map.get(baseline, True):
                return False
    else:
        if request.tiling and not self.image_generation_feature_flags.tiling:
            return False
    return True

worker_supports_requested_hires_fix 

worker_supports_requested_hires_fix(
    request: ImageGenerationFeatureFlags,
) -> bool

Return True if the worker supports hires fix.

Source code in horde_sdk/worker/feature_flags.py 
351
352
353
354
355
356
357
358
359
360
361
362
363
def worker_supports_requested_hires_fix(
    self,
    request: ImageGenerationFeatureFlags,
) -> bool:
    """Return True if the worker supports hires fix."""
    if self.per_baseline_feature_flags and self.per_baseline_feature_flags.hires_fix_map:
        for baseline in request.baselines or []:
            if not self.per_baseline_feature_flags.hires_fix_map.get(baseline, True):
                return False
    else:
        if request.hires_fix and not self.image_generation_feature_flags.hires_fix:
            return False
    return True

worker_supports_requested_controlnets 

worker_supports_requested_controlnets(
    request: ImageGenerationFeatureFlags,
) -> bool

Return True if the worker supports controlnets.

Source code in horde_sdk/worker/feature_flags.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def worker_supports_requested_controlnets(
    self,
    request: ImageGenerationFeatureFlags,
) -> bool:
    """Return True if the worker supports controlnets."""
    if not request.controlnets_feature_flags:
        return True

    if not self.image_generation_feature_flags.controlnets_feature_flags:
        return False

    if self.per_baseline_feature_flags and self.per_baseline_feature_flags.controlnet_map:
        for baseline in request.baselines or []:
            if not self.per_baseline_feature_flags.controlnet_map.get(baseline, True):
                return False

    if (
        request.controlnets_feature_flags.image_is_control
        and not self.image_generation_feature_flags.controlnets_feature_flags.image_is_control
    ):
        return False

    if (  # noqa SIM103: For readability, we return this False directly
        request.controlnets_feature_flags.return_control_map
        and not self.image_generation_feature_flags.controlnets_feature_flags.return_control_map
    ):
        return False

    return True

worker_supports_requested_tis 

worker_supports_requested_tis(
    request: ImageGenerationFeatureFlags,
) -> bool

Return True if the worker supports TIs.

Source code in horde_sdk/worker/feature_flags.py 
395
396
397
398
399
400
401
402
403
404
405
406
407
def worker_supports_requested_tis(
    self,
    request: ImageGenerationFeatureFlags,
) -> bool:
    """Return True if the worker supports TIs."""
    if self.per_baseline_feature_flags and self.per_baseline_feature_flags.tis_map:
        for baseline in request.baselines or []:
            if not self.per_baseline_feature_flags.tis_map.get(baseline, True):
                return False
    else:
        if request.tis and not self.image_generation_feature_flags.tis:
            return False
    return True

worker_supports_requested_loras 

worker_supports_requested_loras(
    request: ImageGenerationFeatureFlags,
) -> bool

Return True if the worker supports Loras.

Source code in horde_sdk/worker/feature_flags.py 
409
410
411
412
413
414
415
416
417
418
419
420
421
def worker_supports_requested_loras(
    self,
    request: ImageGenerationFeatureFlags,
) -> bool:
    """Return True if the worker supports Loras."""
    if self.per_baseline_feature_flags and self.per_baseline_feature_flags.loras_map:
        for baseline in request.baselines or []:
            if not self.per_baseline_feature_flags.loras_map.get(baseline, True):
                return False
    else:
        if request.loras and not self.image_generation_feature_flags.loras:
            return False
    return True

is_capable_of_features 

is_capable_of_features(
    features: GenerationFeatureFlags,
) -> bool

Check if the worker is capable of handling the requested features.

Parameters:

Returns:

  • bool ( bool ) –

    True if the worker is capable of handling the requested features, False otherwise.

Source code in horde_sdk/worker/feature_flags.py 
57
58
59
60
61
62
63
64
65
66
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)

ALCHEMY_WORKER_NOT_CAPABLE_REASON

Bases: StrEnum

Reasons why a worker is not capable of handling an alchemy request.

Source code in horde_sdk/worker/feature_flags.py 
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
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_misc = auto()
    """The worker does not support a requested miscellaneous feature."""

unsupported_upscaler class-attribute instance-attribute 

unsupported_upscaler = auto()

The worker does not support a requested upscaler.

unsupported_facefixer class-attribute instance-attribute 

unsupported_facefixer = auto()

The worker does not support a requested facefixer.

unsupported_interrogator class-attribute instance-attribute 

unsupported_interrogator = auto()

The worker does not support a requested interrogator.

unsupported_caption_model class-attribute instance-attribute 

unsupported_caption_model = auto()

The worker does not support a requested caption model.

unsupported_nsfw_detector class-attribute instance-attribute 

unsupported_nsfw_detector = auto()

The worker does not support a requested NSFW detector.

unsupported_misc class-attribute instance-attribute 

unsupported_misc = auto()

The worker does not support a requested miscellaneous feature.

AlchemyWorkerFeatureFlags

Bases: WorkerFeatureFlags[ALCHEMY_WORKER_NOT_CAPABLE_REASON]

Feature flags for an alchemy worker.

Source code in horde_sdk/worker/feature_flags.py 
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
class AlchemyWorkerFeatureFlags(WorkerFeatureFlags[ALCHEMY_WORKER_NOT_CAPABLE_REASON]):
    """Feature flags for an alchemy worker."""

    alchemy_feature_flags: AlchemyFeatureFlags

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

    @override
    def reasons_not_capable_of_features(
        self,
        request: GenerationFeatureFlags,
    ) -> list[ALCHEMY_WORKER_NOT_CAPABLE_REASON] | None:
        """Return a list of reasons why a worker is not capable of handling an alchemy request."""
        if not isinstance(request, AlchemyFeatureFlags):
            logger.debug(f"Request is not an AlchemyFeatureFlags instance. Request type: {type(request)}")
            return None

        if not self.alchemy_feature_flags:
            logger.debug("Worker does not have alchemy feature flags.")
            return None

        if not request.alchemy_types:
            logger.debug("Request does not have alchemy types.")
            return None

        reasons = []

        for alchemy_type in request.alchemy_types:
            if alchemy_type not in self.alchemy_feature_flags.alchemy_types:
                if is_upscaler_form(alchemy_type):
                    reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_upscaler)
                elif is_facefixer_form(alchemy_type):
                    reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_facefixer)
                elif is_interrogator_form(alchemy_type):
                    reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_interrogator)
                elif is_caption_form(alchemy_type):
                    reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_caption_model)
                elif is_nsfw_detector_form(alchemy_type):
                    reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_nsfw_detector)
                else:
                    reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_misc)
        return reasons if reasons else None

alchemy_feature_flags instance-attribute 

alchemy_feature_flags: AlchemyFeatureFlags

model_config class-attribute instance-attribute 

model_config = get_default_frozen_model_config_dict()

supported_result_return_methods class-attribute instance-attribute 

supported_result_return_methods: list[
    RESULT_RETURN_METHOD
] = Field(default_factory=list)

The methods of returning results supported by the worker.

supports_threads class-attribute instance-attribute 

supports_threads: bool = Field(default=False)

Whether the worker supports threading.

get_not_capable_reason_type 

get_not_capable_reason_type() -> (
    type[ALCHEMY_WORKER_NOT_CAPABLE_REASON]
)
Source code in horde_sdk/worker/feature_flags.py 
455
456
457
@override
def get_not_capable_reason_type(self) -> type[ALCHEMY_WORKER_NOT_CAPABLE_REASON]:
    return ALCHEMY_WORKER_NOT_CAPABLE_REASON

reasons_not_capable_of_features 

reasons_not_capable_of_features(
    request: GenerationFeatureFlags,
) -> list[ALCHEMY_WORKER_NOT_CAPABLE_REASON] | None

Return a list of reasons why a worker is not capable of handling an alchemy request.

Source code in horde_sdk/worker/feature_flags.py 
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
@override
def reasons_not_capable_of_features(
    self,
    request: GenerationFeatureFlags,
) -> list[ALCHEMY_WORKER_NOT_CAPABLE_REASON] | None:
    """Return a list of reasons why a worker is not capable of handling an alchemy request."""
    if not isinstance(request, AlchemyFeatureFlags):
        logger.debug(f"Request is not an AlchemyFeatureFlags instance. Request type: {type(request)}")
        return None

    if not self.alchemy_feature_flags:
        logger.debug("Worker does not have alchemy feature flags.")
        return None

    if not request.alchemy_types:
        logger.debug("Request does not have alchemy types.")
        return None

    reasons = []

    for alchemy_type in request.alchemy_types:
        if alchemy_type not in self.alchemy_feature_flags.alchemy_types:
            if is_upscaler_form(alchemy_type):
                reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_upscaler)
            elif is_facefixer_form(alchemy_type):
                reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_facefixer)
            elif is_interrogator_form(alchemy_type):
                reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_interrogator)
            elif is_caption_form(alchemy_type):
                reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_caption_model)
            elif is_nsfw_detector_form(alchemy_type):
                reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_nsfw_detector)
            else:
                reasons.append(ALCHEMY_WORKER_NOT_CAPABLE_REASON.unsupported_misc)
    return reasons if reasons else None

is_capable_of_features 

is_capable_of_features(
    features: GenerationFeatureFlags,
) -> bool

Check if the worker is capable of handling the requested features.

Parameters:

Returns:

  • bool ( bool ) –

    True if the worker is capable of handling the requested features, False otherwise.

Source code in horde_sdk/worker/feature_flags.py 
57
58
59
60
61
62
63
64
65
66
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)