horde_sdk.ai_horde_api.apimodels.base¶

class BaseAIHordeRequest(HordeRequest):
    """Base class for all AI Horde API requests."""

    @override
    @classmethod
    def get_api_url(cls) -> str:
        return AI_HORDE_BASE_URL

class JobRequestMixin(HordeAPIData):
    """Mix-in class for data relating to any generation jobs."""

    id_: GenerationID = Field(alias="id")
    """The UUID for this job. Use this to post the results in the future."""

    @field_validator("id_", mode="before")
    def validate_id(cls, v: str | GenerationID) -> GenerationID | str:
        """Ensure that the job ID is not empty."""
        if isinstance(v, str) and v == "":
            logger.warning("Job ID is empty")
            return GenerationID(root=uuid.uuid4())

        return v

    def __eq__(self, __value: object) -> bool:
        if isinstance(__value, JobRequestMixin):
            return self.id_ == __value.id_
        return False

    def __hash__(self) -> int:
        return hash(JobRequestMixin.__name__) + hash(self.id_)

class JobResponseMixin(HordeAPIData):
    """Mix-in class for data relating to any generation jobs."""

    id_: GenerationID = Field(alias="id")
    """The UUID for this job."""

    @field_validator("id_", mode="before")
    def validate_id(cls, v: str | GenerationID) -> GenerationID | str:
        """Ensure that the job ID is not empty."""
        if isinstance(v, str) and v == "":
            logger.warning("Job ID is empty")
            return GenerationID(root=uuid.uuid4())

        return v

class WorkerRequestMixin(HordeAPIData):
    """Mix-in class for data relating to worker requests."""

    worker_id: str | WorkerID
    """The UUID of the worker in question for this request."""

class WorkerRequestNameMixin(HordeAPIData):
    """Mix-in class for data relating to worker requests."""

    worker_name: str
    """The name of the worker in question for this request."""

class LorasPayloadEntry(HordeAPIObjectBaseModel):
    """Represents a single lora parameter.

    v2 API Model: `ModelPayloadLorasStable`
    """

    name: str = Field(min_length=1, max_length=255)
    """The name of the LoRa model to use."""
    model: float = Field(default=1, ge=-5, le=5)
    """The strength of the LoRa against the stable diffusion model."""
    clip: float = Field(default=1, ge=-5, le=5)
    """The strength of the LoRa against the clip model."""
    inject_trigger: str | None = Field(default=None, min_length=1, max_length=30)
    """Any trigger required to activate the LoRa model."""
    is_version: bool = Field(default=False)
    """If true, will treat the lora name as a version ID."""

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "ModelPayloadLorasStable"

class TIPayloadEntry(HordeAPIObjectBaseModel):
    """Represents a single textual inversion (embedding) parameter.

    v2 API Model: `ModelPayloadTextualInversionsStable`
    """

    name: str = Field(min_length=1, max_length=255)
    """The name or ID of the textual inversion model to use."""
    inject_ti: str | None = None
    """Whether to automatically insert the TI into the prompt or negprompt."""
    strength: float = Field(default=1, ge=-5, le=5)
    """The strength to apply the textual inversion model."""

    @field_validator("inject_ti")
    def validate_inject_ti(cls, v: str | None) -> str | None:
        """Ensure that the inject_ti is either 'prompt' or 'negprompt'."""
        if v is None:
            return None
        if v not in ["prompt", "negprompt"]:
            raise ValueError("inject_ti must be either 'prompt' or 'negprompt'")
        return v

    @field_validator("strength")
    def validate_strength(cls, v: float) -> float:
        """Ensure that the strength is non-zero."""
        if v == 0:
            logger.debug("strength should be non-zero")

        return v

    @model_validator(mode="after")
    def strength_only_if_inject_ti(self) -> TIPayloadEntry:
        """Ensure that the strength is only set if the inject_ti is set."""
        if self.strength and self.inject_ti is None:
            logger.debug("strength is only valid when inject_ti is set")
        return self

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "ModelPayloadTextualInversionsStable"

class ExtraSourceImageEntry(HordeAPIObjectBaseModel):
    """Represents a single extra source image.

    v2 API Model: `ExtraSourceImage`
    """

    original_url: str | None = None
    """The URL of the original image after it was downloaded."""

    image: str = Field(min_length=1)
    """The URL of the image to download, or the base64 string once downloaded."""
    strength: float = Field(default=1, ge=-5, le=5)
    """The strength to apply to this image on various operations."""

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "ExtraSourceImage"

class ExtraTextEntry(HordeAPIObjectBaseModel):
    """Represents a single extra text.

    v2 API Model: `ExtraText`
    """

    text: str = Field(min_length=1)
    """Extra text required for this generation."""
    reference: str = Field(min_length=3)
    """Reference pointing to how this text is to be used."""

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "ExtraText"

class SingleWarningEntry(HordeAPIObjectBaseModel):
    """Represents a single warning.

    v2 API Model: `RequestSingleWarning`
    """

    code: WarningCode | str = Field(min_length=1)
    """The code uniquely identifying this warning."""
    message: str = Field(min_length=1)
    """The human-readable description of this warning"""

    @field_validator("code")
    def code_must_be_known(cls, v: str | WarningCode) -> str | WarningCode:
        """Ensure that the warning code is in this list of supported warning codes."""
        if isinstance(v, WarningCode):
            return v
        if v not in WarningCode.__members__ or v not in WarningCode.__members__.values():
            logger.warning(f"Unknown warning code {v}. Is your SDK out of date or did the API change?")

        return v

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "RequestSingleWarning"

class ImageGenerateParamMixin(_BaseImageGenerateParamMixin):
    """Contains basic and api-specific parameters for image generation.

    v2 API Model: `ModelPayloadRootStable`
    """

    model_config = (
        ConfigDict(frozen=True, extra="allow")
        if not os.getenv("TESTS_ONGOING")
        else ConfigDict(frozen=True, extra="forbid")
    )

    seed: str | None = None
    """The seed to use for this generation. If not provided, a random seed will be used."""
    seed_variation: int | None = Field(default=None, ge=1, le=1000)
    """Deprecated."""

    control_type: str | KNOWN_IMAGE_CONTROLNETS | None = None
    """The type of control net type to use."""
    image_is_control: bool | None = None
    """Set to True if the image is a control image."""
    return_control_map: bool | None = None
    """Set to True if you want the ControlNet map returned instead of a generated image."""

    extra_texts: list[ExtraTextEntry] | None = None
    """A list of extra texts and prompts to use in the comfyUI workflow."""
    use_nsfw_censor: bool = False
    """If the request is SFW, and the worker accidentally generates NSFW, it will send back a censored image."""

    # @model_validator(mode="after")
    # def validate_hires_fix(self) -> ImageGenerateParamMixin:
    #     if self.hires_fix and (self.width < 512 or self.height < 512):
    #         raise ValueError("hires_fix is only valid when width and height are both >= 512")
    #     return self

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "ModelPayloadRootStable"

    @field_validator("seed")
    def random_seed_if_none(cls, v: str | None) -> str | None:
        """If the seed is None, generate a random seed."""
        if v is None:
            random_seed = str(random.randint(1, 1000000000))
            logger.debug(f"Using random seed ({random_seed})")
            return random_seed

        return v

    @field_validator("control_type")
    def control_type_must_be_known(
        cls,
        v: str | KNOWN_IMAGE_CONTROLNETS | None,
    ) -> str | KNOWN_IMAGE_CONTROLNETS | None:
        """Ensure that the control type is in this list of supported control types."""
        if v is None:
            return None
        if isinstance(v, KNOWN_IMAGE_CONTROLNETS):
            return v

        try:
            KNOWN_IMAGE_CONTROLNETS(v)
        except ValueError:
            logger.warning(f"Unknown control type {v}. Is your SDK out of date or did the API change?")

        return v

class JobSubmitResponse(HordeResponseBaseModel):
    """Indicates that a generation job was successfully submitted and the amount of kudos gained.

    Represents the data returned from the following endpoints and http status codes:
        - /v2/generate/text/submit | TextGenerationJobSubmitRequest [POST] -> 200
        - /v2/generate/submit | ImageGenerationJobSubmitRequest [POST] -> 200

    v2 API Model: `GenerationSubmitted`
    """

    reward: float
    """The amount of kudos gained for submitting this request."""

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "GenerationSubmitted"

class GenMetadataEntry(HordeAPIObjectBaseModel):
    """Represents a single generation metadata entry.

    v2 API Model: `GenerationMetadataStable`
    """

    type_: METADATA_TYPE | str = Field(alias="type")
    """The relevance of the metadata field."""
    value: METADATA_VALUE | str = Field()
    """The value of the metadata field."""
    ref: str | None = Field(default=None, max_length=255)
    """Optionally a reference for the metadata (e.g. a lora ID)"""

    @field_validator("type_")
    def validate_type(cls, v: str | METADATA_TYPE) -> str | METADATA_TYPE:
        """Ensure that the type is in this list of supported types."""
        if isinstance(v, METADATA_TYPE):
            return v
        if isinstance(v, str) and v not in METADATA_TYPE.__members__:
            logger.warning(f"Unknown metadata type {v}. Is your SDK out of date or did the API change?")
        return v

    @field_validator("value")
    def validate_value(cls, v: str | METADATA_VALUE) -> str | METADATA_VALUE:
        """Ensure that the value is in this list of supported values."""
        if isinstance(v, METADATA_VALUE):
            return v
        if isinstance(v, str) and v not in METADATA_VALUE.__members__:
            logger.warning(f"Unknown metadata value {v}. Is your SDK out of date or did the API change?")
        return v

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "GenerationMetadataStable"

class MessageSpecifiesSharedKeyMixin(HordeAPIData):
    """Mix-in class to describe an endpoint for which you can specify a shared key."""

    sharedkey_id: SharedKeyID = Field(alias="id")
    """The shared key ID to use for this request."""

    @field_validator("sharedkey_id", mode="before")
    def validate_sharedkey_id(cls, v: str | SharedKeyID) -> SharedKeyID | str:
        """Ensure that the shared key ID is not empty."""
        if isinstance(v, str) and v == "":
            logger.warning("Shared key ID is empty")

        return v

class ActiveModelLite(HordeAPIObjectBaseModel):
    """Represents a single active model.

    v2 API Model: `ActiveModelLite`
    """

    count: int | None = Field(
        default=None,
    )
    """How many of workers in this horde are running this model."""
    name: str | None = Field(
        default=None,
    )
    """The Name of a model available by workers in this horde."""

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "ActiveModelLite"

class ActiveModel(ActiveModelLite):
    """Represents a single active model.

    v2 API Model: `ActiveModel`
    """

    eta: int | None = Field(
        default=None,
    )
    """Estimated time in seconds for this model's queue to be cleared."""
    jobs: float | None = Field(
        default=None,
    )
    """The job count waiting to be generated by this model."""
    performance: float | None = Field(
        default=None,
    )
    """The average speed of generation for this model."""
    queued: float | None = Field(
        default=None,
    )
    """The amount waiting to be generated by this model."""
    type_: MODEL_TYPE | None = Field(
        examples=[MODEL_TYPE.image, MODEL_TYPE.text],
        alias="type",
    )
    """The model type (text or image)."""

    @override
    @classmethod
    def get_api_model_name(cls) -> str | None:
        return "ActiveModel"