Skip to content

bridge_data

The bridge data file definitions and functions for any horde worker type.

ALL_DATA_FILES_TYPES module-attribute 

ALL_DATA_FILES_TYPES: list[type[BaseModel]] = [
    SharedHordeBridgeData,
    ImageWorkerBridgeData,
    ScribeBridgeData,
    CombinedHordeBridgeData,
]

MetaInstruction

Bases: StrEnum

Model load instructions which requiring further processing to resolve.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class MetaInstruction(StrEnum):
    """Model load instructions which requiring further processing to resolve."""

    ALL_REGEX = r"all$|all models?$"

    ALL_SDXL_REGEX = r"all sdxl$|all sdxl models?$"
    ALL_SD15_REGEX = r"all sd15$|all sd15 models?$"
    ALL_SD21_REGEX = r"all sd21$|all sd21 models?$"

    ALL_SFW_REGEX = r"all sfw$|all sfw models?$"
    ALL_NSFW_REGEX = r"all nsfw$|all nsfw models?$"

    ALL_INPAINTING_REGEX = r"all inpainting$|all inpainting models?$"

    TOP_N_REGEX = r"TOP (\d+)"
    """The regex to use to match the top N models. The number is in a capture group on its own."""

    BOTTOM_N_REGEX = r"BOTTOM (\d+)"

ALL_REGEX class-attribute instance-attribute 

ALL_REGEX = 'all$|all models?$'

ALL_SDXL_REGEX class-attribute instance-attribute 

ALL_SDXL_REGEX = 'all sdxl$|all sdxl models?$'

ALL_SD15_REGEX class-attribute instance-attribute 

ALL_SD15_REGEX = 'all sd15$|all sd15 models?$'

ALL_SD21_REGEX class-attribute instance-attribute 

ALL_SD21_REGEX = 'all sd21$|all sd21 models?$'

ALL_SFW_REGEX class-attribute instance-attribute 

ALL_SFW_REGEX = 'all sfw$|all sfw models?$'

ALL_NSFW_REGEX class-attribute instance-attribute 

ALL_NSFW_REGEX = 'all nsfw$|all nsfw models?$'

ALL_INPAINTING_REGEX class-attribute instance-attribute 

ALL_INPAINTING_REGEX = (
    "all inpainting$|all inpainting models?$"
)

TOP_N_REGEX class-attribute instance-attribute 

TOP_N_REGEX = 'TOP (\\d+)'

The regex to use to match the top N models. The number is in a capture group on its own.

BOTTOM_N_REGEX class-attribute instance-attribute 

BOTTOM_N_REGEX = 'BOTTOM (\\d+)'

BaseHordeBridgeData

Bases: BaseModel

The base bridge data file for all worker types.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
class BaseHordeBridgeData(BaseModel):
    """The base bridge data file for all worker types."""

    model_config = ConfigDict(extra="allow")

    @model_validator(mode="after")
    def validate_extra_params_warning(self) -> BaseHordeBridgeData:
        """Warn on extra parameters being passed."""
        if self.model_extra is not None:
            for key in self.model_extra:
                logger.warning(f"Unknown parameter {key} in bridge data file.")

        return self

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='allow')

validate_extra_params_warning 

validate_extra_params_warning() -> BaseHordeBridgeData

Warn on extra parameters being passed.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
46
47
48
49
50
51
52
53
@model_validator(mode="after")
def validate_extra_params_warning(self) -> BaseHordeBridgeData:
    """Warn on extra parameters being passed."""
    if self.model_extra is not None:
        for key in self.model_extra:
            logger.warning(f"Unknown parameter {key} in bridge data file.")

    return self

SharedHordeBridgeData

Bases: BaseHordeBridgeData

The base bridge data file for all worker types.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
 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
 91
 92
 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
class SharedHordeBridgeData(BaseHordeBridgeData):
    """The base bridge data file for all worker types."""

    #
    # Shared Worker-type Settings
    #

    api_key: str = Field(
        default=ANON_API_KEY,
        title="API Key",
    )
    """The API key to use to authenticate with the Horde API. Defaults to the anonymous API key."""

    default_worker_name: str = Field("An Awesome AI Horde Worker", alias="worker_name")
    """The default name for any worker type. This is used to easily identify the worker.

    Note that if the user switches to a different worker type after having run a different worker type, the name will
    be rejected by the Horde API. This is because the name is unique to the worker type.
    """

    models_folder_parent: str = Field(
        default="./",
        alias="cache_home",
    )
    """The directories to use to store the model folders. The spirit of this derives from `XDG_CACHE_HOME`, but it is
    not a *volatile* cache. It is the ordinary location for the models. Defaults to the current working directory."""

    temp_dir: str = "./tmp/"
    """The directory to use for temporary files such as model caches. Defaults to `./tmp/`."""

    allow_unsafe_ip: bool = True
    """Whether to allow img2img from unsafe IP addresses. This is only relevant if img2img is enabled."""

    priority_usernames: list[str] = Field(default_factory=list)
    """A list of usernames to prioritize. The API always considers this worker's user to be on this list."""

    stats_output_frequency: int = Field(default=30, json_schema_extra={"unit": "seconds"})
    """The frequency at which to output stats to the console, in seconds. Set to 0 to disable."""
    disable_terminal_ui: bool = False
    """Whether to disable the terminal UI. This is useful for running the worker in a container."""
    always_download: bool = True
    """Whether to always download models, without prompting for permission for each one."""
    suppress_speed_warnings: bool = False
    """Whether to suppress warnings about slower than ideal generations."""
    horde_url: str = ""  # TODO: consts.DEFAULT_HORDE_API_URL
    """The URL of the Horde API to use. Defaults to the official API."""

    @field_validator("horde_url")
    def validate_url(cls, v: str) -> str:
        """Validate the URL is valid."""
        parsed_url = urllib.parse.urlparse(v)
        if not parsed_url.scheme or not parsed_url.netloc:
            raise ValueError(f"Invalid URL: {v}")
        return v

    @field_validator("api_key")
    def validate_api_key_length(cls, v: str) -> str:
        """Validate that the API key is the correct length."""
        if v == ANON_API_KEY:
            return v
        if len(v) != 22:
            raise ValueError("API key must be 22 characters long")
        return v

    @field_validator("models_folder_parent", "temp_dir")
    def validate_is_dir(cls, v: str) -> str:
        """Validate that the path is a directory."""
        paths_list = []
        # Does it specific multiple directories?
        if ";" in v:
            for sub_v in v.split(";"):
                paths_list.append(sub_v)
        else:
            paths_list.append(v)
        for sub_v in paths_list:
            # See if it's a well-formed path, but don't check if it exists
            pathlib.Path(sub_v).absolute()

        return v

api_key class-attribute instance-attribute 

api_key: str = Field(default=ANON_API_KEY, title='API Key')

The API key to use to authenticate with the Horde API. Defaults to the anonymous API key.

default_worker_name class-attribute instance-attribute 

default_worker_name: str = Field(
    "An Awesome AI Horde Worker", alias="worker_name"
)

The default name for any worker type. This is used to easily identify the worker.

Note that if the user switches to a different worker type after having run a different worker type, the name will be rejected by the Horde API. This is because the name is unique to the worker type.

models_folder_parent class-attribute instance-attribute 

models_folder_parent: str = Field(
    default="./", alias="cache_home"
)

The directories to use to store the model folders. The spirit of this derives from XDG_CACHE_HOME, but it is not a volatile cache. It is the ordinary location for the models. Defaults to the current working directory.

temp_dir class-attribute instance-attribute 

temp_dir: str = './tmp/'

The directory to use for temporary files such as model caches. Defaults to ./tmp/.

allow_unsafe_ip class-attribute instance-attribute 

allow_unsafe_ip: bool = True

Whether to allow img2img from unsafe IP addresses. This is only relevant if img2img is enabled.

priority_usernames class-attribute instance-attribute 

priority_usernames: list[str] = Field(default_factory=list)

A list of usernames to prioritize. The API always considers this worker's user to be on this list.

stats_output_frequency class-attribute instance-attribute 

stats_output_frequency: int = Field(
    default=30, json_schema_extra={"unit": "seconds"}
)

The frequency at which to output stats to the console, in seconds. Set to 0 to disable.

disable_terminal_ui class-attribute instance-attribute 

disable_terminal_ui: bool = False

Whether to disable the terminal UI. This is useful for running the worker in a container.

always_download class-attribute instance-attribute 

always_download: bool = True

Whether to always download models, without prompting for permission for each one.

suppress_speed_warnings class-attribute instance-attribute 

suppress_speed_warnings: bool = False

Whether to suppress warnings about slower than ideal generations.

horde_url class-attribute instance-attribute 

horde_url: str = ''

The URL of the Horde API to use. Defaults to the official API.

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='allow')

validate_url 

validate_url(v: str) -> str

Validate the URL is valid.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
103
104
105
106
107
108
109
@field_validator("horde_url")
def validate_url(cls, v: str) -> str:
    """Validate the URL is valid."""
    parsed_url = urllib.parse.urlparse(v)
    if not parsed_url.scheme or not parsed_url.netloc:
        raise ValueError(f"Invalid URL: {v}")
    return v

validate_api_key_length 

validate_api_key_length(v: str) -> str

Validate that the API key is the correct length.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
111
112
113
114
115
116
117
118
@field_validator("api_key")
def validate_api_key_length(cls, v: str) -> str:
    """Validate that the API key is the correct length."""
    if v == ANON_API_KEY:
        return v
    if len(v) != 22:
        raise ValueError("API key must be 22 characters long")
    return v

validate_is_dir 

validate_is_dir(v: str) -> str

Validate that the path is a directory.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@field_validator("models_folder_parent", "temp_dir")
def validate_is_dir(cls, v: str) -> str:
    """Validate that the path is a directory."""
    paths_list = []
    # Does it specific multiple directories?
    if ";" in v:
        for sub_v in v.split(";"):
            paths_list.append(sub_v)
    else:
        paths_list.append(v)
    for sub_v in paths_list:
        # See if it's a well-formed path, but don't check if it exists
        pathlib.Path(sub_v).absolute()

    return v

validate_extra_params_warning 

validate_extra_params_warning() -> BaseHordeBridgeData

Warn on extra parameters being passed.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
46
47
48
49
50
51
52
53
@model_validator(mode="after")
def validate_extra_params_warning(self) -> BaseHordeBridgeData:
    """Warn on extra parameters being passed."""
    if self.model_extra is not None:
        for key in self.model_extra:
            logger.warning(f"Unknown parameter {key} in bridge data file.")

    return self

ImageWorkerBridgeData

Bases: SharedHordeBridgeData

The bridge data file for a Dreamer or Alchemist worker.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
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
194
195
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
231
232
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
class ImageWorkerBridgeData(SharedHordeBridgeData):
    """The bridge data file for a Dreamer or Alchemist worker."""

    extra_stable_diffusion_models_folders: list[str] = Field(default_factory=list)
    """A list of extra folders to search for stable diffusion models."""

    allow_controlnet: bool = False
    """Whether to allow the use of ControlNet. This requires img2img to be enabled."""

    allow_sdxl_controlnet: bool = False
    """Whether to allow the use of SDXL ControlNet. This requires controlnet to be enabled."""

    allow_img2img: bool = True
    """Whether to allow the use of img2img."""

    allow_lora: bool = False
    """Whether to allow the use of LoRA."""

    extra_slow_worker: bool = False
    """Marks the worker as extra slow."""

    limit_max_steps: bool = False
    """Prevents the worker picking up jobs with more steps than the model average."""

    max_lora_cache_size: int = Field(
        default=10,
        ge=10,
        le=2048,  # I do not foresee anyone having a 2TB lora cache
        json_schema_extra={"unit": "GB"},
    )
    """The maximum size of the LoRA cache, in gigabytes."""

    allow_inpainting: bool = Field(
        default=False,
        alias="allow_painting",
    )
    """Whether to allow the use of inpainting."""

    allow_post_processing: bool = False
    """Whether to allow the use of post-processing."""

    blacklist: list[str] = Field(
        default_factory=list,
    )
    """A list of terms or phrases to blacklist. Jobs containing these terms will be not be assigned to this worker."""

    censor_nsfw: bool = False
    """Whether to censor NSFW content."""

    censorlist: list[str] = Field(
        default_factory=list,
    )
    """A list of terms to allow through the censor. Jobs containing these terms will not be censored for those words
    alone."""

    disable_disk_cache: bool = False
    """Whether to disable the disk cache."""

    dreamer_worker_name: str = Field(
        default="An Awesome Dreamer",
        alias="dreamer_name",
    )

    dynamic_models: bool = False
    """Whether to dynamically switch the models loaded to correspond to the models with the longest queues."""

    max_dynamic_models_to_download: int = Field(
        default=10,
        ge=0,
        le=_UNREASONABLE_NUMBER_OF_MODELS,
        alias="max_models_to_download",
    )
    """The maximum number of models to download."""

    number_of_dynamic_models_to_load: int = Field(
        default=1,
        ge=0,
        le=_UNREASONABLE_NUMBER_OF_MODELS,
        alias="number_of_dynamic_models",
    )
    """The number of dynamic models to load at a time."""

    max_power: int = Field(
        default=8,
        ge=1,
        le=512,
        json_schema_extra={"unit": "* (8 * 64 * 64) pixels"},
    )
    """The factor in the equation `max_power * (8 * 64 * 64)`. This will be the maximum number of pixels that can be
    generated (with inference) in a single job by this worker."""

    @property
    def max_pixels(self) -> int:
        """Calculate the maximum number of pixels that can be generated in a single job by this worker."""
        return self.max_power * (8 * 64 * 64)

    image_models_to_load: list[str] = Field(
        default_factory=list,
        alias="models_to_load",
    )

    """A list of models to load. This can be a list of model names, or a list of model loading instructions, such as
    `top 100` or `all models`."""

    image_models_to_skip: list[str] = Field(
        default_factory=list,
        alias="models_to_skip",
    )
    """When a meta model loading instruction is given, like `top 100`, models in this list will be skipped."""

    nsfw: bool = True
    """Whether to allow NSFW content."""

    queue_size: int = Field(default=1, ge=0, le=4)
    """The number of jobs to queue."""

    max_threads: int = Field(
        default=1,
        ge=1,
        le=16,  # 8 already seems incredibly out of reach for most people, but I have set it to 16 just in case
    )
    """The number of threads to use for inference."""

    max_batch: int = Field(
        default=1,
        ge=1,
        le=20,  # 20 is the max per request on the horde
    )
    """The number of threads to use for inference."""

    require_upfront_kudos: bool = False
    """Whether to require upfront kudos for jobs. This effectively makes the worker reject jobs from
    the anonymous API key, or from keys with no kudos."""

    ram_to_leave_free: str = Field(
        default="80%",
        pattern=r"^\d+%$|^\d+$",
        json_schema_extra={"unit": "Percent or MB"},
    )
    """The amount of RAM to leave free for the system to use. This is a percentage or a number in megabytes."""

    vram_to_leave_free: str | None = Field(
        default="80%",
        pattern=r"^\d+%$|^\d+$",
        json_schema_extra={"unit": "Percent or MB"},
    )
    """The amount of VRAM to leave free for the system to use. This is a percentage or a number in megabytes."""

    #
    # Alchemist Settings
    #

    alchemist_name: str = "An Awesome Alchemist"
    """The name of the worker if it is an alchemist. This is used to easily identify the worker."""

    forms: list[str] = Field(
        default_factory=list,
    )
    """The type of services or processing an alchemist worker will provide."""

    @field_validator("forms", mode="before")
    def default_forms(cls, v: list[str]) -> list[str]:
        """Set the default forms if none are specified."""
        if v is None or len(v) == 0:
            logger.info("Using the default alchemy forms as none were specified.")
            return ["caption", "nsfw", "interrogation", "post-process"]

        return v

    @model_validator(mode="after")
    def validate_model(self) -> ImageWorkerBridgeData:
        """Validate that the parameters are not conflicting and make any fixed adjustments."""
        if not self.allow_img2img and self.allow_controlnet:
            logger.warning(
                (
                    "allow_controlnet is set to True, but allow_img2img is set to False. "
                    "ControlNet requires img2img to be enabled. Setting allow_controlnet to false."
                ),
            )
            self.allow_controlnet = False

        if not self.allow_controlnet and self.allow_sdxl_controlnet:
            logger.warning(
                (
                    "allow_sdxl_controlnet is set to True, but allow_controlnet is set to False. "
                    "SDXL ControlNet requires allow_controlnet to be enabled. Setting allow_sdxl_controlnet to false."
                ),
            )
            self.allow_sdxl_controlnet = False

        self.image_models_to_skip.append("SDXL_beta::stability.ai#6901")  # FIXME: no magic strings

        return self

    _meta_load_instructions: list[str] | None = None
    _meta_skip_instructions: list[str] | None = None

    @property
    def meta_load_instructions(self) -> list[str] | None:
        """The meta load instructions."""
        return self._meta_load_instructions

    @model_validator(mode="after")
    def handle_meta_instructions(self) -> ImageWorkerBridgeData:
        """Handle the meta instructions by resolving and applying them."""
        # See if any entries are meta instructions, and if so, remove them and place them in _meta_load_instructions
        for instruction_regex in MetaInstruction.__members__.values():
            for i, model in enumerate(self.image_models_to_load):
                if re.match(instruction_regex, model, re.IGNORECASE):
                    if self._meta_load_instructions is None:
                        self._meta_load_instructions = []
                    self._meta_load_instructions.append(model)
                    self.image_models_to_load.pop(i)

        return self

    @property
    def meta_skip_instructions(self) -> list[str] | None:
        """The meta skip instructions."""
        return self._meta_skip_instructions

    @model_validator(mode="after")
    def handle_meta_skip_instructions(self) -> ImageWorkerBridgeData:
        """Handle the meta skip instructions by resolving and applying them."""
        # See if any entries are meta instructions, and if so, remove them and place them in _meta_skip_instructions
        for instruction_regex in MetaInstruction.__members__.values():
            for i, model in enumerate(self.image_models_to_skip):
                if re.match(instruction_regex, model, re.IGNORECASE):
                    if self._meta_skip_instructions is None:
                        self._meta_skip_instructions = []
                    self._meta_skip_instructions.append(model)
                    self.image_models_to_skip.pop(i)

        return self

    @field_validator("image_models_to_load")
    def validate_models_to_load(cls, v: list[str]) -> list[str]:
        """Validate and parse the models to load."""
        if not isinstance(v, list):
            v = [v]
        if not v:
            v = ["top 2"]
            logger.warning("No models specified in bridgeData.yaml. Defaulting to top 2 models.")
        return v

    @field_validator("ram_to_leave_free", "vram_to_leave_free")
    def validate_ram_to_leave_free(cls, v: str | int | float) -> str | int | float:
        """Validate VRAM/RAM to leave free."""
        if isinstance(v, str):
            if v.isdigit():
                v = int(v)
            elif v.endswith("%"):
                _ = float(v[:-1])
                if _ < 0 or _ > 100:
                    raise ValueError("RAM to leave free must be between 0 and 100%")
        else:
            if v < 0:
                raise ValueError("RAM to leave free must be greater than or equal to 0")
        return v

    @field_validator("forms")
    def validate_alchemy_forms(cls, v: list[str]) -> list[str | KNOWN_ALCHEMY_FORMS]:
        """Validate the alchemy forms (services offered)."""
        if not isinstance(v, list):
            raise ValueError("forms must be a list")
        validated_forms = []
        for form in v:
            form = str(form).lower()
            form = form.replace("-", "_")
            if form not in KNOWN_ALCHEMY_FORMS.__members__:
                raise ValueError(f"Invalid form: {form}")
            validated_forms.append(form)
        return validated_forms

extra_stable_diffusion_models_folders class-attribute instance-attribute 

extra_stable_diffusion_models_folders: list[str] = Field(
    default_factory=list
)

A list of extra folders to search for stable diffusion models.

allow_controlnet class-attribute instance-attribute 

allow_controlnet: bool = False

Whether to allow the use of ControlNet. This requires img2img to be enabled.

allow_sdxl_controlnet class-attribute instance-attribute 

allow_sdxl_controlnet: bool = False

Whether to allow the use of SDXL ControlNet. This requires controlnet to be enabled.

allow_img2img class-attribute instance-attribute 

allow_img2img: bool = True

Whether to allow the use of img2img.

allow_lora class-attribute instance-attribute 

allow_lora: bool = False

Whether to allow the use of LoRA.

extra_slow_worker class-attribute instance-attribute 

extra_slow_worker: bool = False

Marks the worker as extra slow.

limit_max_steps class-attribute instance-attribute 

limit_max_steps: bool = False

Prevents the worker picking up jobs with more steps than the model average.

max_lora_cache_size class-attribute instance-attribute 

max_lora_cache_size: int = Field(
    default=10,
    ge=10,
    le=2048,
    json_schema_extra={"unit": "GB"},
)

The maximum size of the LoRA cache, in gigabytes.

allow_inpainting class-attribute instance-attribute 

allow_inpainting: bool = Field(
    default=False, alias="allow_painting"
)

Whether to allow the use of inpainting.

allow_post_processing class-attribute instance-attribute 

allow_post_processing: bool = False

Whether to allow the use of post-processing.

blacklist class-attribute instance-attribute 

blacklist: list[str] = Field(default_factory=list)

A list of terms or phrases to blacklist. Jobs containing these terms will be not be assigned to this worker.

censor_nsfw class-attribute instance-attribute 

censor_nsfw: bool = False

Whether to censor NSFW content.

censorlist class-attribute instance-attribute 

censorlist: list[str] = Field(default_factory=list)

A list of terms to allow through the censor. Jobs containing these terms will not be censored for those words alone.

disable_disk_cache class-attribute instance-attribute 

disable_disk_cache: bool = False

Whether to disable the disk cache.

dreamer_worker_name class-attribute instance-attribute 

dreamer_worker_name: str = Field(
    default="An Awesome Dreamer", alias="dreamer_name"
)

dynamic_models class-attribute instance-attribute 

dynamic_models: bool = False

Whether to dynamically switch the models loaded to correspond to the models with the longest queues.

max_dynamic_models_to_download class-attribute instance-attribute 

max_dynamic_models_to_download: int = Field(
    default=10,
    ge=0,
    le=_UNREASONABLE_NUMBER_OF_MODELS,
    alias="max_models_to_download",
)

The maximum number of models to download.

number_of_dynamic_models_to_load class-attribute instance-attribute 

number_of_dynamic_models_to_load: int = Field(
    default=1,
    ge=0,
    le=_UNREASONABLE_NUMBER_OF_MODELS,
    alias="number_of_dynamic_models",
)

The number of dynamic models to load at a time.

max_power class-attribute instance-attribute 

max_power: int = Field(
    default=8,
    ge=1,
    le=512,
    json_schema_extra={"unit": "* (8 * 64 * 64) pixels"},
)

The factor in the equation max_power * (8 * 64 * 64). This will be the maximum number of pixels that can be generated (with inference) in a single job by this worker.

max_pixels property 

max_pixels: int

Calculate the maximum number of pixels that can be generated in a single job by this worker.

image_models_to_load class-attribute instance-attribute 

image_models_to_load: list[str] = Field(
    default_factory=list, alias="models_to_load"
)

A list of models to load. This can be a list of model names, or a list of model loading instructions, such as top 100 or all models.

image_models_to_skip class-attribute instance-attribute 

image_models_to_skip: list[str] = Field(
    default_factory=list, alias="models_to_skip"
)

When a meta model loading instruction is given, like top 100, models in this list will be skipped.

nsfw class-attribute instance-attribute 

nsfw: bool = True

Whether to allow NSFW content.

queue_size class-attribute instance-attribute 

queue_size: int = Field(default=1, ge=0, le=4)

The number of jobs to queue.

max_threads class-attribute instance-attribute 

max_threads: int = Field(default=1, ge=1, le=16)

The number of threads to use for inference.

max_batch class-attribute instance-attribute 

max_batch: int = Field(default=1, ge=1, le=20)

The number of threads to use for inference.

require_upfront_kudos class-attribute instance-attribute 

require_upfront_kudos: bool = False

Whether to require upfront kudos for jobs. This effectively makes the worker reject jobs from the anonymous API key, or from keys with no kudos.

ram_to_leave_free class-attribute instance-attribute 

ram_to_leave_free: str = Field(
    default="80%",
    pattern="^\\d+%$|^\\d+$",
    json_schema_extra={"unit": "Percent or MB"},
)

The amount of RAM to leave free for the system to use. This is a percentage or a number in megabytes.

vram_to_leave_free class-attribute instance-attribute 

vram_to_leave_free: str | None = Field(
    default="80%",
    pattern="^\\d+%$|^\\d+$",
    json_schema_extra={"unit": "Percent or MB"},
)

The amount of VRAM to leave free for the system to use. This is a percentage or a number in megabytes.

alchemist_name class-attribute instance-attribute 

alchemist_name: str = 'An Awesome Alchemist'

The name of the worker if it is an alchemist. This is used to easily identify the worker.

forms class-attribute instance-attribute 

forms: list[str] = Field(default_factory=list)

The type of services or processing an alchemist worker will provide.

meta_load_instructions property 

meta_load_instructions: list[str] | None

The meta load instructions.

meta_skip_instructions property 

meta_skip_instructions: list[str] | None

The meta skip instructions.

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='allow')

api_key class-attribute instance-attribute 

api_key: str = Field(default=ANON_API_KEY, title='API Key')

The API key to use to authenticate with the Horde API. Defaults to the anonymous API key.

default_worker_name class-attribute instance-attribute 

default_worker_name: str = Field(
    "An Awesome AI Horde Worker", alias="worker_name"
)

The default name for any worker type. This is used to easily identify the worker.

Note that if the user switches to a different worker type after having run a different worker type, the name will be rejected by the Horde API. This is because the name is unique to the worker type.

models_folder_parent class-attribute instance-attribute 

models_folder_parent: str = Field(
    default="./", alias="cache_home"
)

The directories to use to store the model folders. The spirit of this derives from XDG_CACHE_HOME, but it is not a volatile cache. It is the ordinary location for the models. Defaults to the current working directory.

temp_dir class-attribute instance-attribute 

temp_dir: str = './tmp/'

The directory to use for temporary files such as model caches. Defaults to ./tmp/.

allow_unsafe_ip class-attribute instance-attribute 

allow_unsafe_ip: bool = True

Whether to allow img2img from unsafe IP addresses. This is only relevant if img2img is enabled.

priority_usernames class-attribute instance-attribute 

priority_usernames: list[str] = Field(default_factory=list)

A list of usernames to prioritize. The API always considers this worker's user to be on this list.

stats_output_frequency class-attribute instance-attribute 

stats_output_frequency: int = Field(
    default=30, json_schema_extra={"unit": "seconds"}
)

The frequency at which to output stats to the console, in seconds. Set to 0 to disable.

disable_terminal_ui class-attribute instance-attribute 

disable_terminal_ui: bool = False

Whether to disable the terminal UI. This is useful for running the worker in a container.

always_download class-attribute instance-attribute 

always_download: bool = True

Whether to always download models, without prompting for permission for each one.

suppress_speed_warnings class-attribute instance-attribute 

suppress_speed_warnings: bool = False

Whether to suppress warnings about slower than ideal generations.

horde_url class-attribute instance-attribute 

horde_url: str = ''

The URL of the Horde API to use. Defaults to the official API.

default_forms 

default_forms(v: list[str]) -> list[str]

Set the default forms if none are specified.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
297
298
299
300
301
302
303
304
@field_validator("forms", mode="before")
def default_forms(cls, v: list[str]) -> list[str]:
    """Set the default forms if none are specified."""
    if v is None or len(v) == 0:
        logger.info("Using the default alchemy forms as none were specified.")
        return ["caption", "nsfw", "interrogation", "post-process"]

    return v

validate_model 

validate_model() -> ImageWorkerBridgeData

Validate that the parameters are not conflicting and make any fixed adjustments.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
@model_validator(mode="after")
def validate_model(self) -> ImageWorkerBridgeData:
    """Validate that the parameters are not conflicting and make any fixed adjustments."""
    if not self.allow_img2img and self.allow_controlnet:
        logger.warning(
            (
                "allow_controlnet is set to True, but allow_img2img is set to False. "
                "ControlNet requires img2img to be enabled. Setting allow_controlnet to false."
            ),
        )
        self.allow_controlnet = False

    if not self.allow_controlnet and self.allow_sdxl_controlnet:
        logger.warning(
            (
                "allow_sdxl_controlnet is set to True, but allow_controlnet is set to False. "
                "SDXL ControlNet requires allow_controlnet to be enabled. Setting allow_sdxl_controlnet to false."
            ),
        )
        self.allow_sdxl_controlnet = False

    self.image_models_to_skip.append("SDXL_beta::stability.ai#6901")  # FIXME: no magic strings

    return self

handle_meta_instructions 

handle_meta_instructions() -> ImageWorkerBridgeData

Handle the meta instructions by resolving and applying them.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
@model_validator(mode="after")
def handle_meta_instructions(self) -> ImageWorkerBridgeData:
    """Handle the meta instructions by resolving and applying them."""
    # See if any entries are meta instructions, and if so, remove them and place them in _meta_load_instructions
    for instruction_regex in MetaInstruction.__members__.values():
        for i, model in enumerate(self.image_models_to_load):
            if re.match(instruction_regex, model, re.IGNORECASE):
                if self._meta_load_instructions is None:
                    self._meta_load_instructions = []
                self._meta_load_instructions.append(model)
                self.image_models_to_load.pop(i)

    return self

handle_meta_skip_instructions 

handle_meta_skip_instructions() -> ImageWorkerBridgeData

Handle the meta skip instructions by resolving and applying them.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
@model_validator(mode="after")
def handle_meta_skip_instructions(self) -> ImageWorkerBridgeData:
    """Handle the meta skip instructions by resolving and applying them."""
    # See if any entries are meta instructions, and if so, remove them and place them in _meta_skip_instructions
    for instruction_regex in MetaInstruction.__members__.values():
        for i, model in enumerate(self.image_models_to_skip):
            if re.match(instruction_regex, model, re.IGNORECASE):
                if self._meta_skip_instructions is None:
                    self._meta_skip_instructions = []
                self._meta_skip_instructions.append(model)
                self.image_models_to_skip.pop(i)

    return self

validate_models_to_load 

validate_models_to_load(v: list[str]) -> list[str]

Validate and parse the models to load.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
372
373
374
375
376
377
378
379
380
@field_validator("image_models_to_load")
def validate_models_to_load(cls, v: list[str]) -> list[str]:
    """Validate and parse the models to load."""
    if not isinstance(v, list):
        v = [v]
    if not v:
        v = ["top 2"]
        logger.warning("No models specified in bridgeData.yaml. Defaulting to top 2 models.")
    return v

validate_ram_to_leave_free 

validate_ram_to_leave_free(
    v: str | int | float,
) -> str | int | float

Validate VRAM/RAM to leave free.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
@field_validator("ram_to_leave_free", "vram_to_leave_free")
def validate_ram_to_leave_free(cls, v: str | int | float) -> str | int | float:
    """Validate VRAM/RAM to leave free."""
    if isinstance(v, str):
        if v.isdigit():
            v = int(v)
        elif v.endswith("%"):
            _ = float(v[:-1])
            if _ < 0 or _ > 100:
                raise ValueError("RAM to leave free must be between 0 and 100%")
    else:
        if v < 0:
            raise ValueError("RAM to leave free must be greater than or equal to 0")
    return v

validate_alchemy_forms 

validate_alchemy_forms(
    v: list[str],
) -> list[str | KNOWN_ALCHEMY_FORMS]

Validate the alchemy forms (services offered).

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
397
398
399
400
401
402
403
404
405
406
407
408
409
@field_validator("forms")
def validate_alchemy_forms(cls, v: list[str]) -> list[str | KNOWN_ALCHEMY_FORMS]:
    """Validate the alchemy forms (services offered)."""
    if not isinstance(v, list):
        raise ValueError("forms must be a list")
    validated_forms = []
    for form in v:
        form = str(form).lower()
        form = form.replace("-", "_")
        if form not in KNOWN_ALCHEMY_FORMS.__members__:
            raise ValueError(f"Invalid form: {form}")
        validated_forms.append(form)
    return validated_forms

validate_extra_params_warning 

validate_extra_params_warning() -> BaseHordeBridgeData

Warn on extra parameters being passed.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
46
47
48
49
50
51
52
53
@model_validator(mode="after")
def validate_extra_params_warning(self) -> BaseHordeBridgeData:
    """Warn on extra parameters being passed."""
    if self.model_extra is not None:
        for key in self.model_extra:
            logger.warning(f"Unknown parameter {key} in bridge data file.")

    return self

validate_url 

validate_url(v: str) -> str

Validate the URL is valid.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
103
104
105
106
107
108
109
@field_validator("horde_url")
def validate_url(cls, v: str) -> str:
    """Validate the URL is valid."""
    parsed_url = urllib.parse.urlparse(v)
    if not parsed_url.scheme or not parsed_url.netloc:
        raise ValueError(f"Invalid URL: {v}")
    return v

validate_api_key_length 

validate_api_key_length(v: str) -> str

Validate that the API key is the correct length.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
111
112
113
114
115
116
117
118
@field_validator("api_key")
def validate_api_key_length(cls, v: str) -> str:
    """Validate that the API key is the correct length."""
    if v == ANON_API_KEY:
        return v
    if len(v) != 22:
        raise ValueError("API key must be 22 characters long")
    return v

validate_is_dir 

validate_is_dir(v: str) -> str

Validate that the path is a directory.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@field_validator("models_folder_parent", "temp_dir")
def validate_is_dir(cls, v: str) -> str:
    """Validate that the path is a directory."""
    paths_list = []
    # Does it specific multiple directories?
    if ";" in v:
        for sub_v in v.split(";"):
            paths_list.append(sub_v)
    else:
        paths_list.append(v)
    for sub_v in paths_list:
        # See if it's a well-formed path, but don't check if it exists
        pathlib.Path(sub_v).absolute()

    return v

ScribeBridgeData

Bases: SharedHordeBridgeData

The bridge data file subset for a Scribe worker.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
class ScribeBridgeData(SharedHordeBridgeData):
    """The bridge data file subset for a Scribe worker."""

    # Scribe Settings
    scribe_name: str = "An Awesome Scribe"
    kai_url: str = "http://localhost:5000"
    """The URL of the KoboldAI API to use. Defaults to the official API."""

    max_context_length: int = Field(
        default=1024,
        ge=1,
        le=16384,  # This is just a sanity check, I doubt anyone would want to go this high
        json_schema_extra={"unit": "tokens"},
    )
    """The maximum number of tokens to use for context with a scribe job."""

    max_length: int = Field(
        default=80,
        ge=1,
        le=16384,  # This is just a sanity check, I doubt anyone would want to go this high
        json_schema_extra={"unit": "tokens"},
    )
    """The maximum number of tokens to generate for a scribe job."""

    branded_model: bool = True
    """If true, this is a custom scribe model, and the horde API will identify uniquely to this worker's API key."""

scribe_name class-attribute instance-attribute 

scribe_name: str = 'An Awesome Scribe'

kai_url class-attribute instance-attribute 

kai_url: str = 'http://localhost:5000'

The URL of the KoboldAI API to use. Defaults to the official API.

max_context_length class-attribute instance-attribute 

max_context_length: int = Field(
    default=1024,
    ge=1,
    le=16384,
    json_schema_extra={"unit": "tokens"},
)

The maximum number of tokens to use for context with a scribe job.

max_length class-attribute instance-attribute 

max_length: int = Field(
    default=80,
    ge=1,
    le=16384,
    json_schema_extra={"unit": "tokens"},
)

The maximum number of tokens to generate for a scribe job.

branded_model class-attribute instance-attribute 

branded_model: bool = True

If true, this is a custom scribe model, and the horde API will identify uniquely to this worker's API key.

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='allow')

api_key class-attribute instance-attribute 

api_key: str = Field(default=ANON_API_KEY, title='API Key')

The API key to use to authenticate with the Horde API. Defaults to the anonymous API key.

default_worker_name class-attribute instance-attribute 

default_worker_name: str = Field(
    "An Awesome AI Horde Worker", alias="worker_name"
)

The default name for any worker type. This is used to easily identify the worker.

Note that if the user switches to a different worker type after having run a different worker type, the name will be rejected by the Horde API. This is because the name is unique to the worker type.

models_folder_parent class-attribute instance-attribute 

models_folder_parent: str = Field(
    default="./", alias="cache_home"
)

The directories to use to store the model folders. The spirit of this derives from XDG_CACHE_HOME, but it is not a volatile cache. It is the ordinary location for the models. Defaults to the current working directory.

temp_dir class-attribute instance-attribute 

temp_dir: str = './tmp/'

The directory to use for temporary files such as model caches. Defaults to ./tmp/.

allow_unsafe_ip class-attribute instance-attribute 

allow_unsafe_ip: bool = True

Whether to allow img2img from unsafe IP addresses. This is only relevant if img2img is enabled.

priority_usernames class-attribute instance-attribute 

priority_usernames: list[str] = Field(default_factory=list)

A list of usernames to prioritize. The API always considers this worker's user to be on this list.

stats_output_frequency class-attribute instance-attribute 

stats_output_frequency: int = Field(
    default=30, json_schema_extra={"unit": "seconds"}
)

The frequency at which to output stats to the console, in seconds. Set to 0 to disable.

disable_terminal_ui class-attribute instance-attribute 

disable_terminal_ui: bool = False

Whether to disable the terminal UI. This is useful for running the worker in a container.

always_download class-attribute instance-attribute 

always_download: bool = True

Whether to always download models, without prompting for permission for each one.

suppress_speed_warnings class-attribute instance-attribute 

suppress_speed_warnings: bool = False

Whether to suppress warnings about slower than ideal generations.

horde_url class-attribute instance-attribute 

horde_url: str = ''

The URL of the Horde API to use. Defaults to the official API.

validate_extra_params_warning 

validate_extra_params_warning() -> BaseHordeBridgeData

Warn on extra parameters being passed.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
46
47
48
49
50
51
52
53
@model_validator(mode="after")
def validate_extra_params_warning(self) -> BaseHordeBridgeData:
    """Warn on extra parameters being passed."""
    if self.model_extra is not None:
        for key in self.model_extra:
            logger.warning(f"Unknown parameter {key} in bridge data file.")

    return self

validate_url 

validate_url(v: str) -> str

Validate the URL is valid.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
103
104
105
106
107
108
109
@field_validator("horde_url")
def validate_url(cls, v: str) -> str:
    """Validate the URL is valid."""
    parsed_url = urllib.parse.urlparse(v)
    if not parsed_url.scheme or not parsed_url.netloc:
        raise ValueError(f"Invalid URL: {v}")
    return v

validate_api_key_length 

validate_api_key_length(v: str) -> str

Validate that the API key is the correct length.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
111
112
113
114
115
116
117
118
@field_validator("api_key")
def validate_api_key_length(cls, v: str) -> str:
    """Validate that the API key is the correct length."""
    if v == ANON_API_KEY:
        return v
    if len(v) != 22:
        raise ValueError("API key must be 22 characters long")
    return v

validate_is_dir 

validate_is_dir(v: str) -> str

Validate that the path is a directory.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@field_validator("models_folder_parent", "temp_dir")
def validate_is_dir(cls, v: str) -> str:
    """Validate that the path is a directory."""
    paths_list = []
    # Does it specific multiple directories?
    if ";" in v:
        for sub_v in v.split(";"):
            paths_list.append(sub_v)
    else:
        paths_list.append(v)
    for sub_v in paths_list:
        # See if it's a well-formed path, but don't check if it exists
        pathlib.Path(sub_v).absolute()

    return v

CombinedHordeBridgeData

Bases: ImageWorkerBridgeData, ScribeBridgeData

The bridge data file for a worker bridge that supports all worker types.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
440
441
class CombinedHordeBridgeData(ImageWorkerBridgeData, ScribeBridgeData):
    """The bridge data file for a worker bridge that supports all worker types."""

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='allow')

api_key class-attribute instance-attribute 

api_key: str = Field(default=ANON_API_KEY, title='API Key')

The API key to use to authenticate with the Horde API. Defaults to the anonymous API key.

default_worker_name class-attribute instance-attribute 

default_worker_name: str = Field(
    "An Awesome AI Horde Worker", alias="worker_name"
)

The default name for any worker type. This is used to easily identify the worker.

Note that if the user switches to a different worker type after having run a different worker type, the name will be rejected by the Horde API. This is because the name is unique to the worker type.

models_folder_parent class-attribute instance-attribute 

models_folder_parent: str = Field(
    default="./", alias="cache_home"
)

The directories to use to store the model folders. The spirit of this derives from XDG_CACHE_HOME, but it is not a volatile cache. It is the ordinary location for the models. Defaults to the current working directory.

temp_dir class-attribute instance-attribute 

temp_dir: str = './tmp/'

The directory to use for temporary files such as model caches. Defaults to ./tmp/.

allow_unsafe_ip class-attribute instance-attribute 

allow_unsafe_ip: bool = True

Whether to allow img2img from unsafe IP addresses. This is only relevant if img2img is enabled.

priority_usernames class-attribute instance-attribute 

priority_usernames: list[str] = Field(default_factory=list)

A list of usernames to prioritize. The API always considers this worker's user to be on this list.

stats_output_frequency class-attribute instance-attribute 

stats_output_frequency: int = Field(
    default=30, json_schema_extra={"unit": "seconds"}
)

The frequency at which to output stats to the console, in seconds. Set to 0 to disable.

disable_terminal_ui class-attribute instance-attribute 

disable_terminal_ui: bool = False

Whether to disable the terminal UI. This is useful for running the worker in a container.

always_download class-attribute instance-attribute 

always_download: bool = True

Whether to always download models, without prompting for permission for each one.

suppress_speed_warnings class-attribute instance-attribute 

suppress_speed_warnings: bool = False

Whether to suppress warnings about slower than ideal generations.

horde_url class-attribute instance-attribute 

horde_url: str = ''

The URL of the Horde API to use. Defaults to the official API.

scribe_name class-attribute instance-attribute 

scribe_name: str = 'An Awesome Scribe'

kai_url class-attribute instance-attribute 

kai_url: str = 'http://localhost:5000'

The URL of the KoboldAI API to use. Defaults to the official API.

max_context_length class-attribute instance-attribute 

max_context_length: int = Field(
    default=1024,
    ge=1,
    le=16384,
    json_schema_extra={"unit": "tokens"},
)

The maximum number of tokens to use for context with a scribe job.

max_length class-attribute instance-attribute 

max_length: int = Field(
    default=80,
    ge=1,
    le=16384,
    json_schema_extra={"unit": "tokens"},
)

The maximum number of tokens to generate for a scribe job.

branded_model class-attribute instance-attribute 

branded_model: bool = True

If true, this is a custom scribe model, and the horde API will identify uniquely to this worker's API key.

extra_stable_diffusion_models_folders class-attribute instance-attribute 

extra_stable_diffusion_models_folders: list[str] = Field(
    default_factory=list
)

A list of extra folders to search for stable diffusion models.

allow_controlnet class-attribute instance-attribute 

allow_controlnet: bool = False

Whether to allow the use of ControlNet. This requires img2img to be enabled.

allow_sdxl_controlnet class-attribute instance-attribute 

allow_sdxl_controlnet: bool = False

Whether to allow the use of SDXL ControlNet. This requires controlnet to be enabled.

allow_img2img class-attribute instance-attribute 

allow_img2img: bool = True

Whether to allow the use of img2img.

allow_lora class-attribute instance-attribute 

allow_lora: bool = False

Whether to allow the use of LoRA.

extra_slow_worker class-attribute instance-attribute 

extra_slow_worker: bool = False

Marks the worker as extra slow.

limit_max_steps class-attribute instance-attribute 

limit_max_steps: bool = False

Prevents the worker picking up jobs with more steps than the model average.

max_lora_cache_size class-attribute instance-attribute 

max_lora_cache_size: int = Field(
    default=10,
    ge=10,
    le=2048,
    json_schema_extra={"unit": "GB"},
)

The maximum size of the LoRA cache, in gigabytes.

allow_inpainting class-attribute instance-attribute 

allow_inpainting: bool = Field(
    default=False, alias="allow_painting"
)

Whether to allow the use of inpainting.

allow_post_processing class-attribute instance-attribute 

allow_post_processing: bool = False

Whether to allow the use of post-processing.

blacklist class-attribute instance-attribute 

blacklist: list[str] = Field(default_factory=list)

A list of terms or phrases to blacklist. Jobs containing these terms will be not be assigned to this worker.

censor_nsfw class-attribute instance-attribute 

censor_nsfw: bool = False

Whether to censor NSFW content.

censorlist class-attribute instance-attribute 

censorlist: list[str] = Field(default_factory=list)

A list of terms to allow through the censor. Jobs containing these terms will not be censored for those words alone.

disable_disk_cache class-attribute instance-attribute 

disable_disk_cache: bool = False

Whether to disable the disk cache.

dreamer_worker_name class-attribute instance-attribute 

dreamer_worker_name: str = Field(
    default="An Awesome Dreamer", alias="dreamer_name"
)

dynamic_models class-attribute instance-attribute 

dynamic_models: bool = False

Whether to dynamically switch the models loaded to correspond to the models with the longest queues.

max_dynamic_models_to_download class-attribute instance-attribute 

max_dynamic_models_to_download: int = Field(
    default=10,
    ge=0,
    le=_UNREASONABLE_NUMBER_OF_MODELS,
    alias="max_models_to_download",
)

The maximum number of models to download.

number_of_dynamic_models_to_load class-attribute instance-attribute 

number_of_dynamic_models_to_load: int = Field(
    default=1,
    ge=0,
    le=_UNREASONABLE_NUMBER_OF_MODELS,
    alias="number_of_dynamic_models",
)

The number of dynamic models to load at a time.

max_power class-attribute instance-attribute 

max_power: int = Field(
    default=8,
    ge=1,
    le=512,
    json_schema_extra={"unit": "* (8 * 64 * 64) pixels"},
)

The factor in the equation max_power * (8 * 64 * 64). This will be the maximum number of pixels that can be generated (with inference) in a single job by this worker.

max_pixels property 

max_pixels: int

Calculate the maximum number of pixels that can be generated in a single job by this worker.

image_models_to_load class-attribute instance-attribute 

image_models_to_load: list[str] = Field(
    default_factory=list, alias="models_to_load"
)

A list of models to load. This can be a list of model names, or a list of model loading instructions, such as top 100 or all models.

image_models_to_skip class-attribute instance-attribute 

image_models_to_skip: list[str] = Field(
    default_factory=list, alias="models_to_skip"
)

When a meta model loading instruction is given, like top 100, models in this list will be skipped.

nsfw class-attribute instance-attribute 

nsfw: bool = True

Whether to allow NSFW content.

queue_size class-attribute instance-attribute 

queue_size: int = Field(default=1, ge=0, le=4)

The number of jobs to queue.

max_threads class-attribute instance-attribute 

max_threads: int = Field(default=1, ge=1, le=16)

The number of threads to use for inference.

max_batch class-attribute instance-attribute 

max_batch: int = Field(default=1, ge=1, le=20)

The number of threads to use for inference.

require_upfront_kudos class-attribute instance-attribute 

require_upfront_kudos: bool = False

Whether to require upfront kudos for jobs. This effectively makes the worker reject jobs from the anonymous API key, or from keys with no kudos.

ram_to_leave_free class-attribute instance-attribute 

ram_to_leave_free: str = Field(
    default="80%",
    pattern="^\\d+%$|^\\d+$",
    json_schema_extra={"unit": "Percent or MB"},
)

The amount of RAM to leave free for the system to use. This is a percentage or a number in megabytes.

vram_to_leave_free class-attribute instance-attribute 

vram_to_leave_free: str | None = Field(
    default="80%",
    pattern="^\\d+%$|^\\d+$",
    json_schema_extra={"unit": "Percent or MB"},
)

The amount of VRAM to leave free for the system to use. This is a percentage or a number in megabytes.

alchemist_name class-attribute instance-attribute 

alchemist_name: str = 'An Awesome Alchemist'

The name of the worker if it is an alchemist. This is used to easily identify the worker.

forms class-attribute instance-attribute 

forms: list[str] = Field(default_factory=list)

The type of services or processing an alchemist worker will provide.

meta_load_instructions property 

meta_load_instructions: list[str] | None

The meta load instructions.

meta_skip_instructions property 

meta_skip_instructions: list[str] | None

The meta skip instructions.

validate_extra_params_warning 

validate_extra_params_warning() -> BaseHordeBridgeData

Warn on extra parameters being passed.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
46
47
48
49
50
51
52
53
@model_validator(mode="after")
def validate_extra_params_warning(self) -> BaseHordeBridgeData:
    """Warn on extra parameters being passed."""
    if self.model_extra is not None:
        for key in self.model_extra:
            logger.warning(f"Unknown parameter {key} in bridge data file.")

    return self

validate_url 

validate_url(v: str) -> str

Validate the URL is valid.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
103
104
105
106
107
108
109
@field_validator("horde_url")
def validate_url(cls, v: str) -> str:
    """Validate the URL is valid."""
    parsed_url = urllib.parse.urlparse(v)
    if not parsed_url.scheme or not parsed_url.netloc:
        raise ValueError(f"Invalid URL: {v}")
    return v

validate_api_key_length 

validate_api_key_length(v: str) -> str

Validate that the API key is the correct length.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
111
112
113
114
115
116
117
118
@field_validator("api_key")
def validate_api_key_length(cls, v: str) -> str:
    """Validate that the API key is the correct length."""
    if v == ANON_API_KEY:
        return v
    if len(v) != 22:
        raise ValueError("API key must be 22 characters long")
    return v

validate_is_dir 

validate_is_dir(v: str) -> str

Validate that the path is a directory.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@field_validator("models_folder_parent", "temp_dir")
def validate_is_dir(cls, v: str) -> str:
    """Validate that the path is a directory."""
    paths_list = []
    # Does it specific multiple directories?
    if ";" in v:
        for sub_v in v.split(";"):
            paths_list.append(sub_v)
    else:
        paths_list.append(v)
    for sub_v in paths_list:
        # See if it's a well-formed path, but don't check if it exists
        pathlib.Path(sub_v).absolute()

    return v

default_forms 

default_forms(v: list[str]) -> list[str]

Set the default forms if none are specified.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
297
298
299
300
301
302
303
304
@field_validator("forms", mode="before")
def default_forms(cls, v: list[str]) -> list[str]:
    """Set the default forms if none are specified."""
    if v is None or len(v) == 0:
        logger.info("Using the default alchemy forms as none were specified.")
        return ["caption", "nsfw", "interrogation", "post-process"]

    return v

validate_model 

validate_model() -> ImageWorkerBridgeData

Validate that the parameters are not conflicting and make any fixed adjustments.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
@model_validator(mode="after")
def validate_model(self) -> ImageWorkerBridgeData:
    """Validate that the parameters are not conflicting and make any fixed adjustments."""
    if not self.allow_img2img and self.allow_controlnet:
        logger.warning(
            (
                "allow_controlnet is set to True, but allow_img2img is set to False. "
                "ControlNet requires img2img to be enabled. Setting allow_controlnet to false."
            ),
        )
        self.allow_controlnet = False

    if not self.allow_controlnet and self.allow_sdxl_controlnet:
        logger.warning(
            (
                "allow_sdxl_controlnet is set to True, but allow_controlnet is set to False. "
                "SDXL ControlNet requires allow_controlnet to be enabled. Setting allow_sdxl_controlnet to false."
            ),
        )
        self.allow_sdxl_controlnet = False

    self.image_models_to_skip.append("SDXL_beta::stability.ai#6901")  # FIXME: no magic strings

    return self

handle_meta_instructions 

handle_meta_instructions() -> ImageWorkerBridgeData

Handle the meta instructions by resolving and applying them.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
@model_validator(mode="after")
def handle_meta_instructions(self) -> ImageWorkerBridgeData:
    """Handle the meta instructions by resolving and applying them."""
    # See if any entries are meta instructions, and if so, remove them and place them in _meta_load_instructions
    for instruction_regex in MetaInstruction.__members__.values():
        for i, model in enumerate(self.image_models_to_load):
            if re.match(instruction_regex, model, re.IGNORECASE):
                if self._meta_load_instructions is None:
                    self._meta_load_instructions = []
                self._meta_load_instructions.append(model)
                self.image_models_to_load.pop(i)

    return self

handle_meta_skip_instructions 

handle_meta_skip_instructions() -> ImageWorkerBridgeData

Handle the meta skip instructions by resolving and applying them.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
@model_validator(mode="after")
def handle_meta_skip_instructions(self) -> ImageWorkerBridgeData:
    """Handle the meta skip instructions by resolving and applying them."""
    # See if any entries are meta instructions, and if so, remove them and place them in _meta_skip_instructions
    for instruction_regex in MetaInstruction.__members__.values():
        for i, model in enumerate(self.image_models_to_skip):
            if re.match(instruction_regex, model, re.IGNORECASE):
                if self._meta_skip_instructions is None:
                    self._meta_skip_instructions = []
                self._meta_skip_instructions.append(model)
                self.image_models_to_skip.pop(i)

    return self

validate_models_to_load 

validate_models_to_load(v: list[str]) -> list[str]

Validate and parse the models to load.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
372
373
374
375
376
377
378
379
380
@field_validator("image_models_to_load")
def validate_models_to_load(cls, v: list[str]) -> list[str]:
    """Validate and parse the models to load."""
    if not isinstance(v, list):
        v = [v]
    if not v:
        v = ["top 2"]
        logger.warning("No models specified in bridgeData.yaml. Defaulting to top 2 models.")
    return v

validate_ram_to_leave_free 

validate_ram_to_leave_free(
    v: str | int | float,
) -> str | int | float

Validate VRAM/RAM to leave free.

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
@field_validator("ram_to_leave_free", "vram_to_leave_free")
def validate_ram_to_leave_free(cls, v: str | int | float) -> str | int | float:
    """Validate VRAM/RAM to leave free."""
    if isinstance(v, str):
        if v.isdigit():
            v = int(v)
        elif v.endswith("%"):
            _ = float(v[:-1])
            if _ < 0 or _ > 100:
                raise ValueError("RAM to leave free must be between 0 and 100%")
    else:
        if v < 0:
            raise ValueError("RAM to leave free must be greater than or equal to 0")
    return v

validate_alchemy_forms 

validate_alchemy_forms(
    v: list[str],
) -> list[str | KNOWN_ALCHEMY_FORMS]

Validate the alchemy forms (services offered).

Source code in horde_sdk/worker/dispatch/ai_horde/bridge_data.py 
397
398
399
400
401
402
403
404
405
406
407
408
409
@field_validator("forms")
def validate_alchemy_forms(cls, v: list[str]) -> list[str | KNOWN_ALCHEMY_FORMS]:
    """Validate the alchemy forms (services offered)."""
    if not isinstance(v, list):
        raise ValueError("forms must be a list")
    validated_forms = []
    for form in v:
        form = str(form).lower()
        form = form.replace("-", "_")
        if form not in KNOWN_ALCHEMY_FORMS.__members__:
            raise ValueError(f"Invalid form: {form}")
        validated_forms.append(form)
    return validated_forms