Skip to content

model_meta

ImageModelLoadResolver

Resolve meta instructions for loading models.

Source code in horde_sdk/worker/model_meta.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
 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
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
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
410
411
412
413
414
415
class ImageModelLoadResolver:
    """Resolve meta instructions for loading models."""

    # set a default timeframe for model stats
    default_timeframe = StatsModelsTimeframe.month

    _model_reference_manager: ModelReferenceManager

    async def _initialize_model_reference_manager(self) -> ModelReferenceManager:
        """Asynchronously initialize the model reference manager."""
        if ModelReferenceManager.has_instance():
            return ModelReferenceManager.get_instance()

        ModelReferenceManager(prefetch_strategy=PrefetchStrategy.ASYNC)
        handle = ModelReferenceManager.get_instance().deferred_prefetch_handle
        if not handle:
            raise Exception(
                "ModelReferenceManager instance was not properly initialized with a deferred prefetch handle."
            )

        await handle

        return ModelReferenceManager.get_instance()

    def __init__(self, model_reference_manager: ModelReferenceManager | None = None) -> None:
        """Initialise the resolver.

        Args:
            model_reference_manager: An already-initialised manager to read the reference from. Inject this
                to reuse a caller-owned manager (for example an offline, disk-only one) instead of having
                the resolver build its own. Reading the reference only needs records already on disk; left
                as ``None`` the resolver constructs a network-fetching manager and awaits its prefetch,
                which blocks the caller on a PRIMARY-API round-trip whose latency it cannot bound. A
                process that must not stall on the network (or that already owns a manager) should pass it.
        """
        # Injected manager: use it as-is and skip all event-loop / network init below.
        if model_reference_manager is not None:
            self._model_reference_manager = model_reference_manager
            return

        import asyncio

        # Check to see if there is already a running asyncio loop, and use that if possible
        try:
            loop = asyncio.get_running_loop()
            if loop.is_running():
                logger.debug("Using existing running event loop to initialize model reference manager.")
                self._model_reference_manager = loop.run_until_complete(self._initialize_model_reference_manager())
            else:
                logger.debug("No running event loop found. Initializing model reference manager with new event loop.")
                self._model_reference_manager = asyncio.run(self._initialize_model_reference_manager())
        except RuntimeError:
            logger.debug("No running event loop found. Initializing model reference manager with new event loop.")
            self._model_reference_manager = asyncio.run(self._initialize_model_reference_manager())

    def resolve_meta_instructions(
        self,
        possible_meta_instructions: list[str],
        client: AIHordeAPIManualClient,
        *,
        load_large_models: bool = True,
    ) -> set[str]:
        """Return a set of model names based on the given meta instructions.

        Args:
            possible_meta_instructions: A list of strings representing meta instructions.
            client: An AIHordeAPIManualClient object to use for making requests.
            load_large_models: A boolean representing whether to load large models.

        Returns:
            A set of strings representing the names of models to load.
        """
        # Get model stats from the API
        stats_response = client.submit_request(
            ImageStatsModelsRequest(model_state=MODEL_STATE.known),
            ImageStatsModelsResponse,
        )
        if isinstance(stats_response, RequestErrorResponse):
            raise Exception(f"Error getting stats for models: {stats_response.message}")

        return_list: list[str] = []

        found_top_n = False
        found_bottom_n = False

        # Check each possible meta instruction and return the appropriate model names
        for possible_instruction in possible_meta_instructions:
            # If the instruction is to load all models, return all model names
            if ImageModelLoadResolver.meta_instruction_regex_match(MetaInstruction.ALL_REGEX, possible_instruction):
                return self.remove_large_models(set(self.resolve_all_model_names()))

            # If the instruction is to load the top N models, add the top N model names
            top_n_matches = ImageModelLoadResolver.meta_instruction_regex_match(
                MetaInstruction.TOP_N_REGEX,
                possible_instruction,
            )
            if top_n_matches:
                return_list.extend(
                    ImageModelLoadResolver.resolve_top_n_model_names(
                        int(top_n_matches.group(1)),
                        stats_response,
                        self.default_timeframe,
                    ),
                )
                if found_top_n:
                    logger.warning("Multiple top N instructions found. This probably isn't intended.")
                found_top_n = True
                continue

            # If the instruction is to load the bottom N models, add the bottom N model names
            bottom_n_matches = ImageModelLoadResolver.meta_instruction_regex_match(
                MetaInstruction.BOTTOM_N_REGEX,
                possible_instruction,
            )
            if bottom_n_matches:
                return_list.extend(
                    ImageModelLoadResolver.resolve_bottom_n_model_names(
                        int(bottom_n_matches.group(1)),
                        stats_response,
                        self.default_timeframe,
                    ),
                )
                if found_bottom_n:
                    logger.warning("Multiple bottom N instructions found. This probably isn't intended.")
                found_bottom_n = True
                continue

            if ImageModelLoadResolver.meta_instruction_regex_match(
                MetaInstruction.ALL_SDXL_REGEX,
                possible_instruction,
            ):
                return_list.extend(self.resolve_all_models_of_baseline("stable_diffusion_xl"))

            if ImageModelLoadResolver.meta_instruction_regex_match(
                MetaInstruction.ALL_SD15_REGEX,
                possible_instruction,
            ):
                return_list.extend(self.resolve_all_models_of_baseline("stable_diffusion_1"))

            if ImageModelLoadResolver.meta_instruction_regex_match(
                MetaInstruction.ALL_SD21_REGEX,
                possible_instruction,
            ):
                return_list.extend(self.resolve_all_models_of_baseline("stable_diffusion_2_512"))
                return_list.extend(self.resolve_all_models_of_baseline("stable_diffusion_2_768"))

            if ImageModelLoadResolver.meta_instruction_regex_match(
                MetaInstruction.ALL_INPAINTING_REGEX,
                possible_instruction,
            ):
                return_list.extend(self.resolve_all_inpainting_models())

            if ImageModelLoadResolver.meta_instruction_regex_match(
                MetaInstruction.ALL_SFW_REGEX,
                possible_instruction,
            ):
                return_list.extend(self.resolve_all_sfw_model_names())

            if ImageModelLoadResolver.meta_instruction_regex_match(
                MetaInstruction.ALL_NSFW_REGEX,
                possible_instruction,
            ):
                return_list.extend(self.resolve_all_nsfw_model_names())

        if not load_large_models:
            return self.remove_large_models(set(return_list))

        return set(return_list)

    @staticmethod
    def meta_instruction_regex_match(instruction: str, target_string: str) -> re.Match[str] | None:
        """Check if a target string matches a given meta instruction regex pattern.

        Args:
            instruction: A string representing the meta instruction regex pattern.
            target_string: The string to check for a match.

        Returns:
            A Match object if the target string matches the regex pattern, otherwise None.

        """
        return re.match(instruction, target_string, re.IGNORECASE)

    def remove_large_models(
        self,
        models: set[str],
        load_large_models: bool = True,
    ) -> set[str]:
        """Remove large models from the input set of models."""
        AI_HORDE_MODEL_META_LARGE_MODELS = os.getenv("AI_HORDE_MODEL_META_LARGE_MODELS")
        if not AI_HORDE_MODEL_META_LARGE_MODELS or not load_large_models:
            cascade_models = self.resolve_all_models_of_baseline(KNOWN_IMAGE_GENERATION_BASELINE.stable_cascade)
            flux_models = self.resolve_all_models_of_baseline(KNOWN_IMAGE_GENERATION_BASELINE.flux_1)

            if not AI_HORDE_MODEL_META_LARGE_MODELS:
                logger.debug(
                    "Loading of large models is disabled with `AI_HORDE_MODEL_META_LARGE_MODELS`. "
                    f"Removing {len(cascade_models) + len(flux_models)} models. "
                    f"({cascade_models}, {flux_models})",
                )
            models = models - cascade_models - flux_models
        return models

    def resolve_all_model_names(
        self,
        ignore_large_models_env_var: bool = True,
    ) -> set[str]:
        """Get the names of all models defined in the model reference.

        Args:
            ignore_large_models_env_var (bool): A boolean representing whether to ignore the environment variable for
                large models, effectively returning all models regardless of `AI_HORDE_MODEL_META_LARGE_MODELS`.

        Returns:
            A set of strings representing the names of all models.
        """
        all_model_references = self._model_reference_manager.get_all_model_references()

        # Use .get(), not [...]: a model reference manager may omit a category entirely when its
        # reference failed to load (or simply is not present), not merely map it to None. An absent key
        # would raise KeyError here; .get() yields None, which the guards below already handle.
        sd_model_references = all_model_references.get(MODEL_REFERENCE_CATEGORY.image_generation)

        all_models = set(sd_model_references.keys()) if sd_model_references is not None else set()

        if not ignore_large_models_env_var:
            all_models = self.remove_large_models(all_models)

        if not all_models:
            logger.error("No stable diffusion models found in model reference.")
        return all_models

    def _resolve_sfw_nsfw_model_names(self, nsfw: bool) -> set[str]:
        """Get the names of all SFW or NSFW models defined in the model reference.

        Args:
            nsfw: A boolean representing whether to get SFW or NSFW models.

        Returns:
            A set of strings representing the names of all SFW or NSFW models.
        """
        all_model_references = self._model_reference_manager.get_all_model_references()

        # Use .get(), not [...]: a model reference manager may omit a category entirely when its
        # reference failed to load (or simply is not present), not merely map it to None. An absent key
        # would raise KeyError here; .get() yields None, which the guards below already handle.
        sd_model_references = all_model_references.get(MODEL_REFERENCE_CATEGORY.image_generation)

        found_models: set[str] = set()

        if sd_model_references is None:
            logger.error("No stable diffusion models found in model reference.")
            return found_models

        for model_name, model in sd_model_references.items():
            if not isinstance(model, ImageGenerationModelRecord):
                logger.error(f"Model {model_name} is not a ImageGenerationModelRecord")
                continue

            if model.nsfw == nsfw:
                found_models.add(model_name)

        return found_models

    def resolve_all_sfw_model_names(self) -> set[str]:
        """Get the names of all SFW models defined in the model reference.

        Returns:
            A set of strings representing the names of all SFW models.
        """
        return self._resolve_sfw_nsfw_model_names(nsfw=False)

    def resolve_all_nsfw_model_names(self) -> set[str]:
        """Get the names of all NSFW models defined in the model reference.

        Returns:
            A set of strings representing the names of all NSFW models.
        """
        return self._resolve_sfw_nsfw_model_names(nsfw=True)

    def resolve_all_inpainting_models(self) -> set[str]:
        """Get the names of all inpainting models defined in the model reference.

        Returns:
            A set of strings representing the names of all inpainting models.
        """
        all_model_references = self._model_reference_manager.get_all_model_references()

        # Use .get(), not [...]: a model reference manager may omit a category entirely when its
        # reference failed to load (or simply is not present), not merely map it to None. An absent key
        # would raise KeyError here; .get() yields None, which the guards below already handle.
        sd_model_references = all_model_references.get(MODEL_REFERENCE_CATEGORY.image_generation)

        found_models: set[str] = set()

        if sd_model_references is None:
            logger.error("No stable diffusion models found in model reference.")
            return found_models

        for model_name, model in sd_model_references.items():
            if not isinstance(model, ImageGenerationModelRecord):
                logger.error(f"Model {model_name} is not a ImageGenerationModelRecord")
                continue

            if model.inpainting:
                found_models.add(model_name)

        return found_models

    def resolve_all_models_of_baseline(self, baseline: str) -> set[str]:
        """Get the names of all models of a given baseline defined in the model reference.

        Args:
            baseline: A string representing the baseline to get models for.

        Returns:
            A set of strings representing the names of all models of the given baseline.
        """
        all_model_references = self._model_reference_manager.get_all_model_references()

        # Use .get(), not [...]: a model reference manager may omit a category entirely when its
        # reference failed to load (or simply is not present), not merely map it to None. An absent key
        # would raise KeyError here; .get() yields None, which the guards below already handle.
        sd_model_references = all_model_references.get(MODEL_REFERENCE_CATEGORY.image_generation)

        found_models: set[str] = set()

        if sd_model_references is None:
            logger.error("No stable diffusion models found in model reference.")
            return found_models

        for model_name, model in sd_model_references.items():
            if not isinstance(model, ImageGenerationModelRecord):
                logger.error(f"Model {model_name} is not a ImageGenerationModelRecord")
                continue

            if model.baseline == baseline:
                found_models.add(model_name)

        return found_models

    @staticmethod
    def resolve_top_n_model_names(
        number_of_top_models: int,
        response: ImageStatsModelsResponse,
        timeframe: StatsModelsTimeframe,
    ) -> list[str]:
        """Get the names of the top N models based on usage statistics.

        Args:
            number_of_top_models: An integer representing the number of top models to return.
            response: A StatsModelsResponse object containing model usage statistics.
            timeframe: A StatsModelsTimeframe object representing the timeframe to consider.

        Returns:
            A set of strings representing the names of the top N models.
        """
        models_in_timeframe = response.get_timeframe(timeframe)

        # Remove the blank or null values from the dict
        models_in_timeframe = {k: v for k, v in models_in_timeframe.items() if k and v is not None}

        # Sort by value (number of uses) ascending (highest first)
        sorted_models = sorted(models_in_timeframe.items(), key=lambda x: x[1], reverse=True)

        # Get the top n models
        top_n_models = sorted_models[:number_of_top_models]

        # Get just the model names as a list
        return [model[0] for model in top_n_models]

    @staticmethod
    def resolve_bottom_n_model_names(
        number_of_bottom_models: int,
        response: ImageStatsModelsResponse,
        timeframe: StatsModelsTimeframe,
    ) -> list[str]:
        """Get the names of the bottom N models based on usage statistics.

        Args:
            number_of_bottom_models: An integer representing the number of bottom models to return.
            response: A StatsModelsResponse object containing model usage statistics.
            timeframe: A StatsModelsTimeframe object representing the timeframe to consider.

        Returns:
            A set of strings representing the names of the bottom N models.
        """
        models_in_timeframe = response.get_timeframe(timeframe)

        # Remove the blank or null values from the dict
        models_in_timeframe = {k: v for k, v in models_in_timeframe.items() if k and v is not None}

        # Sort by value (number of uses) ascending (lowest first)
        sorted_models = sorted(models_in_timeframe.items(), key=lambda x: x[1])

        # Get the bottom n models
        bottom_n_models = sorted_models[:number_of_bottom_models]

        # Get just the model names as a list
        return [model[0] for model in bottom_n_models]

default_timeframe class-attribute instance-attribute 

default_timeframe = StatsModelsTimeframe.month

__init__ 

__init__(
    model_reference_manager: (
        ModelReferenceManager | None
    ) = None,
) -> None

Initialise the resolver.

Parameters:

  • model_reference_manager (ModelReferenceManager | None, default: None ) –

    An already-initialised manager to read the reference from. Inject this to reuse a caller-owned manager (for example an offline, disk-only one) instead of having the resolver build its own. Reading the reference only needs records already on disk; left as None the resolver constructs a network-fetching manager and awaits its prefetch, which blocks the caller on a PRIMARY-API round-trip whose latency it cannot bound. A process that must not stall on the network (or that already owns a manager) should pass it.

Source code in horde_sdk/worker/model_meta.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(self, model_reference_manager: ModelReferenceManager | None = None) -> None:
    """Initialise the resolver.

    Args:
        model_reference_manager: An already-initialised manager to read the reference from. Inject this
            to reuse a caller-owned manager (for example an offline, disk-only one) instead of having
            the resolver build its own. Reading the reference only needs records already on disk; left
            as ``None`` the resolver constructs a network-fetching manager and awaits its prefetch,
            which blocks the caller on a PRIMARY-API round-trip whose latency it cannot bound. A
            process that must not stall on the network (or that already owns a manager) should pass it.
    """
    # Injected manager: use it as-is and skip all event-loop / network init below.
    if model_reference_manager is not None:
        self._model_reference_manager = model_reference_manager
        return

    import asyncio

    # Check to see if there is already a running asyncio loop, and use that if possible
    try:
        loop = asyncio.get_running_loop()
        if loop.is_running():
            logger.debug("Using existing running event loop to initialize model reference manager.")
            self._model_reference_manager = loop.run_until_complete(self._initialize_model_reference_manager())
        else:
            logger.debug("No running event loop found. Initializing model reference manager with new event loop.")
            self._model_reference_manager = asyncio.run(self._initialize_model_reference_manager())
    except RuntimeError:
        logger.debug("No running event loop found. Initializing model reference manager with new event loop.")
        self._model_reference_manager = asyncio.run(self._initialize_model_reference_manager())

resolve_meta_instructions 

resolve_meta_instructions(
    possible_meta_instructions: list[str],
    client: AIHordeAPIManualClient,
    *,
    load_large_models: bool = True
) -> set[str]

Return a set of model names based on the given meta instructions.

Parameters:

  • possible_meta_instructions (list[str]) –

    A list of strings representing meta instructions.

  • client (AIHordeAPIManualClient) –

    An AIHordeAPIManualClient object to use for making requests.

  • load_large_models (bool, default: True ) –

    A boolean representing whether to load large models.

Returns:

  • set[str] –

    A set of strings representing the names of models to load.

Source code in horde_sdk/worker/model_meta.py 
 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
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
def resolve_meta_instructions(
    self,
    possible_meta_instructions: list[str],
    client: AIHordeAPIManualClient,
    *,
    load_large_models: bool = True,
) -> set[str]:
    """Return a set of model names based on the given meta instructions.

    Args:
        possible_meta_instructions: A list of strings representing meta instructions.
        client: An AIHordeAPIManualClient object to use for making requests.
        load_large_models: A boolean representing whether to load large models.

    Returns:
        A set of strings representing the names of models to load.
    """
    # Get model stats from the API
    stats_response = client.submit_request(
        ImageStatsModelsRequest(model_state=MODEL_STATE.known),
        ImageStatsModelsResponse,
    )
    if isinstance(stats_response, RequestErrorResponse):
        raise Exception(f"Error getting stats for models: {stats_response.message}")

    return_list: list[str] = []

    found_top_n = False
    found_bottom_n = False

    # Check each possible meta instruction and return the appropriate model names
    for possible_instruction in possible_meta_instructions:
        # If the instruction is to load all models, return all model names
        if ImageModelLoadResolver.meta_instruction_regex_match(MetaInstruction.ALL_REGEX, possible_instruction):
            return self.remove_large_models(set(self.resolve_all_model_names()))

        # If the instruction is to load the top N models, add the top N model names
        top_n_matches = ImageModelLoadResolver.meta_instruction_regex_match(
            MetaInstruction.TOP_N_REGEX,
            possible_instruction,
        )
        if top_n_matches:
            return_list.extend(
                ImageModelLoadResolver.resolve_top_n_model_names(
                    int(top_n_matches.group(1)),
                    stats_response,
                    self.default_timeframe,
                ),
            )
            if found_top_n:
                logger.warning("Multiple top N instructions found. This probably isn't intended.")
            found_top_n = True
            continue

        # If the instruction is to load the bottom N models, add the bottom N model names
        bottom_n_matches = ImageModelLoadResolver.meta_instruction_regex_match(
            MetaInstruction.BOTTOM_N_REGEX,
            possible_instruction,
        )
        if bottom_n_matches:
            return_list.extend(
                ImageModelLoadResolver.resolve_bottom_n_model_names(
                    int(bottom_n_matches.group(1)),
                    stats_response,
                    self.default_timeframe,
                ),
            )
            if found_bottom_n:
                logger.warning("Multiple bottom N instructions found. This probably isn't intended.")
            found_bottom_n = True
            continue

        if ImageModelLoadResolver.meta_instruction_regex_match(
            MetaInstruction.ALL_SDXL_REGEX,
            possible_instruction,
        ):
            return_list.extend(self.resolve_all_models_of_baseline("stable_diffusion_xl"))

        if ImageModelLoadResolver.meta_instruction_regex_match(
            MetaInstruction.ALL_SD15_REGEX,
            possible_instruction,
        ):
            return_list.extend(self.resolve_all_models_of_baseline("stable_diffusion_1"))

        if ImageModelLoadResolver.meta_instruction_regex_match(
            MetaInstruction.ALL_SD21_REGEX,
            possible_instruction,
        ):
            return_list.extend(self.resolve_all_models_of_baseline("stable_diffusion_2_512"))
            return_list.extend(self.resolve_all_models_of_baseline("stable_diffusion_2_768"))

        if ImageModelLoadResolver.meta_instruction_regex_match(
            MetaInstruction.ALL_INPAINTING_REGEX,
            possible_instruction,
        ):
            return_list.extend(self.resolve_all_inpainting_models())

        if ImageModelLoadResolver.meta_instruction_regex_match(
            MetaInstruction.ALL_SFW_REGEX,
            possible_instruction,
        ):
            return_list.extend(self.resolve_all_sfw_model_names())

        if ImageModelLoadResolver.meta_instruction_regex_match(
            MetaInstruction.ALL_NSFW_REGEX,
            possible_instruction,
        ):
            return_list.extend(self.resolve_all_nsfw_model_names())

    if not load_large_models:
        return self.remove_large_models(set(return_list))

    return set(return_list)

meta_instruction_regex_match staticmethod 

meta_instruction_regex_match(
    instruction: str, target_string: str
) -> re.Match[str] | None

Check if a target string matches a given meta instruction regex pattern.

Parameters:

  • instruction (str) –

    A string representing the meta instruction regex pattern.

  • target_string (str) –

    The string to check for a match.

Returns:

  • Match[str] | None –

    A Match object if the target string matches the regex pattern, otherwise None.

Source code in horde_sdk/worker/model_meta.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
@staticmethod
def meta_instruction_regex_match(instruction: str, target_string: str) -> re.Match[str] | None:
    """Check if a target string matches a given meta instruction regex pattern.

    Args:
        instruction: A string representing the meta instruction regex pattern.
        target_string: The string to check for a match.

    Returns:
        A Match object if the target string matches the regex pattern, otherwise None.

    """
    return re.match(instruction, target_string, re.IGNORECASE)

remove_large_models 

remove_large_models(
    models: set[str], load_large_models: bool = True
) -> set[str]

Remove large models from the input set of models.

Source code in horde_sdk/worker/model_meta.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def remove_large_models(
    self,
    models: set[str],
    load_large_models: bool = True,
) -> set[str]:
    """Remove large models from the input set of models."""
    AI_HORDE_MODEL_META_LARGE_MODELS = os.getenv("AI_HORDE_MODEL_META_LARGE_MODELS")
    if not AI_HORDE_MODEL_META_LARGE_MODELS or not load_large_models:
        cascade_models = self.resolve_all_models_of_baseline(KNOWN_IMAGE_GENERATION_BASELINE.stable_cascade)
        flux_models = self.resolve_all_models_of_baseline(KNOWN_IMAGE_GENERATION_BASELINE.flux_1)

        if not AI_HORDE_MODEL_META_LARGE_MODELS:
            logger.debug(
                "Loading of large models is disabled with `AI_HORDE_MODEL_META_LARGE_MODELS`. "
                f"Removing {len(cascade_models) + len(flux_models)} models. "
                f"({cascade_models}, {flux_models})",
            )
        models = models - cascade_models - flux_models
    return models

resolve_all_model_names 

resolve_all_model_names(
    ignore_large_models_env_var: bool = True,
) -> set[str]

Get the names of all models defined in the model reference.

Parameters:

  • ignore_large_models_env_var (bool, default: True ) –

    A boolean representing whether to ignore the environment variable for large models, effectively returning all models regardless of AI_HORDE_MODEL_META_LARGE_MODELS.

Returns:

  • set[str] –

    A set of strings representing the names of all models.

Source code in horde_sdk/worker/model_meta.py 
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
def resolve_all_model_names(
    self,
    ignore_large_models_env_var: bool = True,
) -> set[str]:
    """Get the names of all models defined in the model reference.

    Args:
        ignore_large_models_env_var (bool): A boolean representing whether to ignore the environment variable for
            large models, effectively returning all models regardless of `AI_HORDE_MODEL_META_LARGE_MODELS`.

    Returns:
        A set of strings representing the names of all models.
    """
    all_model_references = self._model_reference_manager.get_all_model_references()

    # Use .get(), not [...]: a model reference manager may omit a category entirely when its
    # reference failed to load (or simply is not present), not merely map it to None. An absent key
    # would raise KeyError here; .get() yields None, which the guards below already handle.
    sd_model_references = all_model_references.get(MODEL_REFERENCE_CATEGORY.image_generation)

    all_models = set(sd_model_references.keys()) if sd_model_references is not None else set()

    if not ignore_large_models_env_var:
        all_models = self.remove_large_models(all_models)

    if not all_models:
        logger.error("No stable diffusion models found in model reference.")
    return all_models

resolve_all_sfw_model_names 

resolve_all_sfw_model_names() -> set[str]

Get the names of all SFW models defined in the model reference.

Returns:

  • set[str] –

    A set of strings representing the names of all SFW models.

Source code in horde_sdk/worker/model_meta.py 
280
281
282
283
284
285
286
def resolve_all_sfw_model_names(self) -> set[str]:
    """Get the names of all SFW models defined in the model reference.

    Returns:
        A set of strings representing the names of all SFW models.
    """
    return self._resolve_sfw_nsfw_model_names(nsfw=False)

resolve_all_nsfw_model_names 

resolve_all_nsfw_model_names() -> set[str]

Get the names of all NSFW models defined in the model reference.

Returns:

  • set[str] –

    A set of strings representing the names of all NSFW models.

Source code in horde_sdk/worker/model_meta.py 
288
289
290
291
292
293
294
def resolve_all_nsfw_model_names(self) -> set[str]:
    """Get the names of all NSFW models defined in the model reference.

    Returns:
        A set of strings representing the names of all NSFW models.
    """
    return self._resolve_sfw_nsfw_model_names(nsfw=True)

resolve_all_inpainting_models 

resolve_all_inpainting_models() -> set[str]

Get the names of all inpainting models defined in the model reference.

Returns:

  • set[str] –

    A set of strings representing the names of all inpainting models.

Source code in horde_sdk/worker/model_meta.py 
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
def resolve_all_inpainting_models(self) -> set[str]:
    """Get the names of all inpainting models defined in the model reference.

    Returns:
        A set of strings representing the names of all inpainting models.
    """
    all_model_references = self._model_reference_manager.get_all_model_references()

    # Use .get(), not [...]: a model reference manager may omit a category entirely when its
    # reference failed to load (or simply is not present), not merely map it to None. An absent key
    # would raise KeyError here; .get() yields None, which the guards below already handle.
    sd_model_references = all_model_references.get(MODEL_REFERENCE_CATEGORY.image_generation)

    found_models: set[str] = set()

    if sd_model_references is None:
        logger.error("No stable diffusion models found in model reference.")
        return found_models

    for model_name, model in sd_model_references.items():
        if not isinstance(model, ImageGenerationModelRecord):
            logger.error(f"Model {model_name} is not a ImageGenerationModelRecord")
            continue

        if model.inpainting:
            found_models.add(model_name)

    return found_models

resolve_all_models_of_baseline 

resolve_all_models_of_baseline(baseline: str) -> set[str]

Get the names of all models of a given baseline defined in the model reference.

Parameters:

  • baseline (str) –

    A string representing the baseline to get models for.

Returns:

  • set[str] –

    A set of strings representing the names of all models of the given baseline.

Source code in horde_sdk/worker/model_meta.py 
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
def resolve_all_models_of_baseline(self, baseline: str) -> set[str]:
    """Get the names of all models of a given baseline defined in the model reference.

    Args:
        baseline: A string representing the baseline to get models for.

    Returns:
        A set of strings representing the names of all models of the given baseline.
    """
    all_model_references = self._model_reference_manager.get_all_model_references()

    # Use .get(), not [...]: a model reference manager may omit a category entirely when its
    # reference failed to load (or simply is not present), not merely map it to None. An absent key
    # would raise KeyError here; .get() yields None, which the guards below already handle.
    sd_model_references = all_model_references.get(MODEL_REFERENCE_CATEGORY.image_generation)

    found_models: set[str] = set()

    if sd_model_references is None:
        logger.error("No stable diffusion models found in model reference.")
        return found_models

    for model_name, model in sd_model_references.items():
        if not isinstance(model, ImageGenerationModelRecord):
            logger.error(f"Model {model_name} is not a ImageGenerationModelRecord")
            continue

        if model.baseline == baseline:
            found_models.add(model_name)

    return found_models

resolve_top_n_model_names staticmethod 

resolve_top_n_model_names(
    number_of_top_models: int,
    response: ImageStatsModelsResponse,
    timeframe: StatsModelsTimeframe,
) -> list[str]

Get the names of the top N models based on usage statistics.

Parameters:

  • number_of_top_models (int) –

    An integer representing the number of top models to return.

  • response (ImageStatsModelsResponse) –

    A StatsModelsResponse object containing model usage statistics.

  • timeframe (StatsModelsTimeframe) –

    A StatsModelsTimeframe object representing the timeframe to consider.

Returns:

  • list[str] –

    A set of strings representing the names of the top N models.

Source code in horde_sdk/worker/model_meta.py 
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
@staticmethod
def resolve_top_n_model_names(
    number_of_top_models: int,
    response: ImageStatsModelsResponse,
    timeframe: StatsModelsTimeframe,
) -> list[str]:
    """Get the names of the top N models based on usage statistics.

    Args:
        number_of_top_models: An integer representing the number of top models to return.
        response: A StatsModelsResponse object containing model usage statistics.
        timeframe: A StatsModelsTimeframe object representing the timeframe to consider.

    Returns:
        A set of strings representing the names of the top N models.
    """
    models_in_timeframe = response.get_timeframe(timeframe)

    # Remove the blank or null values from the dict
    models_in_timeframe = {k: v for k, v in models_in_timeframe.items() if k and v is not None}

    # Sort by value (number of uses) ascending (highest first)
    sorted_models = sorted(models_in_timeframe.items(), key=lambda x: x[1], reverse=True)

    # Get the top n models
    top_n_models = sorted_models[:number_of_top_models]

    # Get just the model names as a list
    return [model[0] for model in top_n_models]

resolve_bottom_n_model_names staticmethod 

resolve_bottom_n_model_names(
    number_of_bottom_models: int,
    response: ImageStatsModelsResponse,
    timeframe: StatsModelsTimeframe,
) -> list[str]

Get the names of the bottom N models based on usage statistics.

Parameters:

  • number_of_bottom_models (int) –

    An integer representing the number of bottom models to return.

  • response (ImageStatsModelsResponse) –

    A StatsModelsResponse object containing model usage statistics.

  • timeframe (StatsModelsTimeframe) –

    A StatsModelsTimeframe object representing the timeframe to consider.

Returns:

  • list[str] –

    A set of strings representing the names of the bottom N models.

Source code in horde_sdk/worker/model_meta.py 
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
@staticmethod
def resolve_bottom_n_model_names(
    number_of_bottom_models: int,
    response: ImageStatsModelsResponse,
    timeframe: StatsModelsTimeframe,
) -> list[str]:
    """Get the names of the bottom N models based on usage statistics.

    Args:
        number_of_bottom_models: An integer representing the number of bottom models to return.
        response: A StatsModelsResponse object containing model usage statistics.
        timeframe: A StatsModelsTimeframe object representing the timeframe to consider.

    Returns:
        A set of strings representing the names of the bottom N models.
    """
    models_in_timeframe = response.get_timeframe(timeframe)

    # Remove the blank or null values from the dict
    models_in_timeframe = {k: v for k, v in models_in_timeframe.items() if k and v is not None}

    # Sort by value (number of uses) ascending (lowest first)
    sorted_models = sorted(models_in_timeframe.items(), key=lambda x: x[1])

    # Get the bottom n models
    bottom_n_models = sorted_models[:number_of_bottom_models]

    # Get just the model names as a list
    return [model[0] for model in bottom_n_models]