def convert_image_job_pop_response_to_parameters(
api_response: ImageGenerateJobPopResponse,
model_reference_manager: ModelReferenceManager,
) -> ImageConversionResult:
"""Convert an API response to the parameters for a generation.
Conversion is fault-tolerant for source images: when an img2img-family or remix request has
no usable source image, the generation degrades to txt2img and a fault is recorded, matching
the `TXT2IMG_FALLBACK` choice declared in the dispatch parameters.
"""
if api_response.model is None:
raise ValueError("Model is required for generation.")
faults: list[GenMetadataEntry] = []
model_record = (
model_reference_manager.query(MODEL_REFERENCE_CATEGORY.image_generation).where(name=api_response.model).first()
)
model_baseline: KNOWN_IMAGE_GENERATION_BASELINE | None = None
if model_record is not None:
try:
model_baseline = KNOWN_IMAGE_GENERATION_BASELINE(model_record.baseline)
except ValueError:
logger.debug(
f"Invalid baseline {model_record.baseline} for model {api_response.model}. Using None instead.",
)
model_baseline = None
# model_version = model_record.version # TODO
combined_prompt = api_response.payload.prompt
if not combined_prompt or combined_prompt.isspace():
raise ValueError("Prompt is required for generation.")
prompt, negative_prompt = _split_ai_horde_prompt(combined_prompt)
base_params = BasicImageGenerationParameters(
model=api_response.model,
model_baseline=model_baseline,
# model_version=model_version,
# model_filename=None, # TODO
# model_hash=None, # TODO
prompt=prompt,
negative_prompt=negative_prompt,
seed=api_response.payload.seed,
height=api_response.payload.height,
width=api_response.payload.width,
steps=api_response.payload.ddim_steps,
cfg_scale=api_response.payload.cfg_scale,
sampler_name=api_response.payload.sampler_name,
scheduler=KNOWN_IMAGE_SCHEDULERS.karras if api_response.payload.karras else KNOWN_IMAGE_SCHEDULERS.normal,
clip_skip=api_response.payload.clip_skip,
denoising_strength=api_response.payload.denoising_strength,
tiling=api_response.payload.tiling,
transparent=api_response.payload.transparent,
)
img2img_params: Image2ImageGenerationParameters | None = _get_img2img_params(api_response, faults)
remix_params: RemixGenerationParameters | None = _get_remix_params(api_response, faults)
controlnet_params: ControlnetGenerationParameters | None = _get_controlnet_params(api_response, faults)
hires_fix_params: HiresFixGenerationParameters | None = _get_hires_fix_params(
api_response,
model_baseline,
prompt,
negative_prompt,
)
custom_workflow_params: CustomWorkflowGenerationParameters | None = _get_custom_workflow_params(api_response)
alchemy_params: AlchemyParameters | None = _get_alchemy_params(api_response)
loras: list[LoRaEntry] | None = _get_lora_params(api_response)
tis: list[TIEntry] | None = _get_ti_params(api_response)
source_processing = api_response.source_processing
if source_processing in SOURCE_IMAGE_REQUIRING_PROCESSING and not _source_image_is_usable(api_response):
# The source image is unusable; degrade to txt2img rather than abort, matching the
# TXT2IMG_FALLBACK choice declared below.
logger.warning(
f"Source processing {source_processing} requested without a usable source image; falling back to txt2img.",
)
if not any(fault.type_ == METADATA_TYPE.source_image for fault in faults):
faults.append(
GenMetadataEntry(type=METADATA_TYPE.source_image, value=METADATA_VALUE.parse_failed),
)
source_processing = KNOWN_IMAGE_SOURCE_PROCESSING.txt2img
raw_uuids = [id_.root for id_ in api_response.ids]
additional_params: list[
Image2ImageGenerationParameters
| RemixGenerationParameters
| ControlnetGenerationParameters
| HiresFixGenerationParameters
| LoRaEntry
| TIEntry
| CustomWorkflowGenerationParameters
] = []
if img2img_params is not None:
additional_params.append(img2img_params)
if remix_params is not None:
additional_params.append(remix_params)
if controlnet_params is not None:
additional_params.append(controlnet_params)
if hires_fix_params is not None:
additional_params.append(hires_fix_params)
if loras is not None:
additional_params.extend(loras)
if tis is not None:
additional_params.extend(tis)
if custom_workflow_params is not None:
additional_params.append(custom_workflow_params)
image_generation_parameters = ImageGenerationParameters(
result_ids=raw_uuids,
batch_size=api_response.payload.n_iter,
source_processing=source_processing,
base_params=base_params,
additional_params=ImageGenerationComponentContainer(
components=additional_params,
),
alchemy_params=alchemy_params,
)
r2_upload_url_map = {}
if api_response.r2_upload is not None:
r2_upload_url_map[api_response.id_] = api_response.r2_upload
elif api_response.r2_uploads is not None:
r2_upload_url_map = dict(zip(api_response.ids, api_response.r2_uploads, strict=True))
else:
raise ValueError("No R2 upload URL found in the API response.")
ai_horde_dispatch_parameters = AIHordeR2DispatchParameters(
generation_ids=[GenerationID(root=uuid_) for uuid_ in raw_uuids],
dispatch_source=KNOWN_DISPATCH_SOURCE.AI_HORDE_API_OFFICIAL,
ttl=api_response.ttl,
inference_backend=KNOWN_INFERENCE_BACKEND.COMFYUI,
requested_backend_constraints=REQUESTED_BACKEND_CONSTRAINTS.SPECIFIED,
no_valid_request_found_reasons=api_response.skipped,
source_image_fallback_choice=REQUESTED_SOURCE_IMAGE_FALLBACK_CHOICE.TXT2IMG_FALLBACK,
r2_upload_url_map=r2_upload_url_map,
)
return ImageConversionResult(
generation_parameters=image_generation_parameters,
dispatch_parameters=ai_horde_dispatch_parameters,
faults=faults,
)