def convert_alchemy_job_pop_response_to_parameters(
api_response: AlchemyJobPopResponse,
) -> tuple[AlchemyParameters, AIHordeR2DispatchParameters]:
"""Convert an API response to the parameters for a generation."""
if api_response.forms is None or len(api_response.forms) == 0:
raise ValueError("The API response did not contain any forms. Was this a skipped response?")
parsed_upscalers: list[UpscaleAlchemyParameters] = []
parsed_facefixers: list[FacefixAlchemyParameters] = []
parsed_interrogators: list[InterrogateAlchemyParameters] = []
parsed_captions: list[CaptionAlchemyParameters] = []
parsed_nsfw_detectors: list[NSFWAlchemyParameters] = []
parsed_unknown_forms: list[SingleAlchemyParameters] = []
r2_upload_url_map = {}
for form in api_response.forms:
if form.source_image is None:
raise ValueError("The API response did not contain a source image for a form.")
r2_upload_url_map[str(form.id_)] = form.r2_upload
if is_upscaler_form(form.form):
parsed_upscalers.append(
UpscaleAlchemyParameters(
result_id=str(form.id_),
form=KNOWN_ALCHEMY_FORMS.post_process,
source_image=base64_str_to_bytes(form.source_image),
upscaler=form.form,
),
)
elif is_facefixer_form(form.form):
parsed_facefixers.append(
FacefixAlchemyParameters(
result_id=str(form.id_),
form=KNOWN_ALCHEMY_FORMS.post_process,
source_image=base64_str_to_bytes(form.source_image),
facefixer=form.form,
),
)
elif is_interrogator_form(form.form):
parsed_interrogators.append(
InterrogateAlchemyParameters(
result_id=str(form.id_),
form=KNOWN_ALCHEMY_FORMS.interrogation,
source_image=base64_str_to_bytes(form.source_image),
interrogator=KNOWN_INTERROGATORS.vit_l_14,
),
)
elif is_caption_form(form.form):
parsed_captions.append(
CaptionAlchemyParameters(
result_id=str(form.id_),
form=KNOWN_ALCHEMY_FORMS.caption,
source_image=base64_str_to_bytes(form.source_image),
caption_model=KNOWN_CAPTION_MODELS.BLIP_BASE_SALESFORCE,
),
)
elif is_nsfw_detector_form(form.form):
parsed_nsfw_detectors.append(
NSFWAlchemyParameters(
result_id=str(form.id_),
form=KNOWN_ALCHEMY_FORMS.nsfw,
source_image=base64_str_to_bytes(form.source_image),
nsfw_detector=KNOWN_NSFW_DETECTOR.HORDE_SAFETY,
),
)
else:
parsed_unknown_forms.append(
SingleAlchemyParameters(
result_id=str(form.id_),
form=form.form,
source_image=base64_str_to_bytes(form.source_image),
),
)
alchemy_parameters = AlchemyParameters(
upscalers=parsed_upscalers or None,
facefixers=parsed_facefixers or None,
interrogators=parsed_interrogators or None,
captions=parsed_captions or None,
nsfw_detectors=parsed_nsfw_detectors or None,
misc_post_processors=parsed_unknown_forms or None,
)
dispatch_parameters = AIHordeR2DispatchParameters(
generation_ids=[str(form.id_) for form in api_response.forms],
dispatch_source=KNOWN_DISPATCH_SOURCE.AI_HORDE_API_OFFICIAL,
inference_backend=KNOWN_ALCHEMY_BACKEND.HORDE_ALCHEMIST,
requested_backend_constraints=REQUESTED_BACKEND_CONSTRAINTS.SPECIFIED,
no_valid_request_found_reasons=api_response.skipped or NoValidAlchemyFound(),
r2_upload_url_map=r2_upload_url_map,
source_image_fallback_choice=REQUESTED_SOURCE_IMAGE_FALLBACK_CHOICE.ABANDON,
)
return alchemy_parameters, dispatch_parameters