horde_sdk.ai_horde_api.apimodels.alchemy.status¶

class AlchemyUpscaleResult(HordeAPIData):
    """Represents the result of an upscale job."""

    upscaler_used: KNOWN_UPSCALERS | str
    """The upscaler used."""
    url: str
    """The URL of the upscaled image."""

class AlchemyCaptionResult(HordeAPIData):
    """Represents the result of a caption job."""

    caption: str
    """The resulting caption of the image."""

class AlchemyNSFWResult(HordeAPIData):
    """Represents the result of an NSFW evaluation."""

    nsfw: bool
    """Whether the image is likely to be NSFW."""

class AlchemyVectorizeResult(HordeAPIData):
    """Represents the result of a vectorize (raster -> SVG) job."""

    vectorize: str
    """The resulting SVG markup of the image."""

class AlchemyAnnotationResult(HordeAPIData):
    """Represents the result of a controlnet annotation (raster -> control map) job."""

    annotation: str
    """The URL of the annotated control-map image."""

class AlchemyInterrogationResultItem(HordeAPIData):
    """Represents an item in the result of an interrogation job."""

    text: str
    """The text of the item."""
    confidence: float
    """The confidence of the item."""

class AlchemyInterrogationDetails(HordeAPIData):
    """The details of an interrogation job."""

    tags: list[AlchemyInterrogationResultItem]
    """The resulting similar tags of the image."""
    sites: list[AlchemyInterrogationResultItem]
    """The resulting similar sites of the image."""
    artists: list[AlchemyInterrogationResultItem]
    """The resulting similar artists of the image."""
    flavors: list[AlchemyInterrogationResultItem]
    """The resulting similar flavors of the image."""
    mediums: list[AlchemyInterrogationResultItem]
    """The resulting similar mediums of the image."""
    movements: list[AlchemyInterrogationResultItem]
    """The resulting similar movements of the image."""
    techniques: list[AlchemyInterrogationResultItem]
    """The resulting similar techniques of the image."""

class AlchemyInterrogationResult(HordeAPIData):
    """Represents the result of an interrogation job. Use the `interrogation` field for the details."""

    interrogation: AlchemyInterrogationDetails
    """The details of the interrogation."""

class AlchemyFormStatus(HordeAPIData):
    """Represents the status of a form in an interrogation job."""

    form: KNOWN_ALCHEMY_TYPES | str
    """The form type."""
    state: GENERATION_STATE
    """The state of the form."""
    result: (
        AlchemyInterrogationDetails
        | AlchemyNSFWResult
        | AlchemyCaptionResult
        | AlchemyUpscaleResult
        | AlchemyVectorizeResult
        | AlchemyAnnotationResult
        | None
    ) = None
    """The result of the form."""

    @field_validator("form", mode="before")
    def validate_form(cls, v: str | KNOWN_ALCHEMY_TYPES) -> KNOWN_ALCHEMY_TYPES | str:
        """Ensure that the form is a known alchemy type."""
        if isinstance(v, KNOWN_ALCHEMY_TYPES):
            return v
        if str(v) not in KNOWN_ALCHEMY_TYPES.__members__:
            logger.warning(f"Unknown form type {v}. Is your SDK out of date or did the API change?")
        return v

    @property
    def done(self) -> bool:
        """Return whether the form is done."""
        return self.state == "done"

    @field_validator("result", mode="before")
    def validate_result(
        cls,
        v: dict[str, object],
    ) -> dict[str, object] | None:
        """Ensure that the result is valid and convert it to the correct type, if possible."""
        if not isinstance(v, dict):
            return v

        if "additionalProp1" in v:
            logger.debug("Found additionalProp1 in result, this is a dummy result. Ignoring.")
            return None

        # The API wraps interrogation results in an extra key ({"interrogation": {...}});
        # unwrap so the union resolves to AlchemyInterrogationDetails (as
        # `all_interrogation_results` expects).
        interrogation_details = v.get("interrogation")
        if isinstance(interrogation_details, dict):
            return interrogation_details

        for key in list(v.keys()):
            if key in KNOWN_UPSCALERS.__members__:
                v["upscaler_used"] = KNOWN_UPSCALERS(key)
                v["url"] = v[key]
                del v[key]

        return v

class AlchemyStatusResponse(HordeResponseBaseModel, ResponseWithProgressMixin):
    """Contains the status of an alchemy job and any completed work (if any).

    Represents the data returned from the following endpoints and http status codes:
        - /v2/interrogate/status/{id} | AlchemyStatusRequest [GET] -> 200
        - /v2/interrogate/status/{id} | AlchemyDeleteRequest [DELETE] -> 200

    v2 API Model: `InterrogationStatus`
    """

    state: GENERATION_STATE
    """The state of the job. See `GENERATION_STATE` for possible values."""
    forms: list[AlchemyFormStatus]
    """The status of each form in the job."""

    @property
    def all_interrogation_results(self) -> list[AlchemyInterrogationDetails]:
        """Return all completed interrogation results."""
        return [
            form.result for form in self.forms if form.done and isinstance(form.result, AlchemyInterrogationDetails)
        ]

    @property
    def all_nsfw_results(self) -> list[AlchemyNSFWResult]:
        """Return all completed nsfw results."""
        return [form.result for form in self.forms if form.done and isinstance(form.result, AlchemyNSFWResult)]

    @property
    def all_caption_results(self) -> list[AlchemyCaptionResult]:
        """Return all completed caption results."""
        return [form.result for form in self.forms if form.done and isinstance(form.result, AlchemyCaptionResult)]

    @property
    def all_upscale_results(self) -> list[AlchemyUpscaleResult]:
        """Return all completed upscale results."""
        return [form.result for form in self.forms if form.done and isinstance(form.result, AlchemyUpscaleResult)]

    @property
    def all_vectorize_results(self) -> list[AlchemyVectorizeResult]:
        """Return all completed vectorize results."""
        return [form.result for form in self.forms if form.done and isinstance(form.result, AlchemyVectorizeResult)]

    @property
    def all_annotation_results(self) -> list[AlchemyAnnotationResult]:
        """Return all completed controlnet annotation results."""
        return [form.result for form in self.forms if form.done and isinstance(form.result, AlchemyAnnotationResult)]

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

    @override
    def is_job_complete(self, number_of_result_expected: int) -> bool:
        found_results = 0
        if not self.forms:
            return False
        for form in self.forms:
            if form.state != "done":
                return False
            found_results += 1
        return found_results == number_of_result_expected

    @override
    def is_job_possible(self) -> bool:
        return True  # FIXME

    @override
    @classmethod
    def is_final_follow_up(cls) -> bool:
        return True

    @override
    @classmethod
    def get_finalize_success_request_type(cls) -> None:
        return None

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, AlchemyStatusResponse):
            return False
        return self.state == other.state and all(form in other.forms for form in self.forms)

    def __hash__(self) -> int:
        return hash((self.state, tuple(self.forms)))

class AlchemyStatusRequest(
    BaseAIHordeRequest,
    JobRequestMixin,
    APIKeyAllowedInRequestMixin,
):
    """Poll for the status of an interrogation job, and retrieve any completed work.

    Represents a GET request to the /v2/interrogate/status/{id} endpoint.
    """

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

    @override
    @classmethod
    def get_http_method(cls) -> HTTPMethod:
        return HTTPMethod.GET

    @override
    @classmethod
    def get_api_endpoint_subpath(cls) -> AI_HORDE_API_ENDPOINT_SUBPATH:
        return AI_HORDE_API_ENDPOINT_SUBPATH.v2_interrogate_status

    @override
    @classmethod
    def get_default_success_response_type(cls) -> type[AlchemyStatusResponse]:
        return AlchemyStatusResponse

class AlchemyDeleteRequest(
    BaseAIHordeRequest,
    JobRequestMixin,
):
    """Cancel an in-progress interrogation job.

    Represents a DELETE request to the /v2/interrogate/status/{id} endpoint.
    """

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

    @override
    @classmethod
    def get_http_method(cls) -> HTTPMethod:
        return HTTPMethod.DELETE

    @override
    @classmethod
    def get_api_endpoint_subpath(cls) -> AI_HORDE_API_ENDPOINT_SUBPATH:
        return AI_HORDE_API_ENDPOINT_SUBPATH.v2_interrogate_status

    @override
    @classmethod
    def get_default_success_response_type(cls) -> type[AlchemyStatusResponse]:
        return AlchemyStatusResponse