def convert_image_job_pop_response_to_parameters(
api_response: ImageGenerateJobPopResponse,
model_reference_manager: ModelReferenceManager,
) -> tuple[ImageGenerationParameters, AIHordeR2DispatchParameters]:
"""Convert an API response to the parameters for a generation."""
if api_response.model is None:
raise ValueError("Model is required for generation.")
model_record = model_reference_manager.image_generation_models.get(api_response.model)
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
prompt = api_response.payload.prompt
if not prompt or prompt.isspace():
raise ValueError("Prompt is required for generation.")
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,
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,
)
img2img_params: Image2ImageGenerationParameters | None = _get_img2img_params(api_response)
remix_params: RemixGenerationParameters | None = _get_remix_params(api_response)
controlnet_params: ControlnetGenerationParameters | None = _get_controlnet_params(api_response)
hires_fix_params: HiresFixGenerationParameters | None = _get_hires_fix_params(api_response, model_baseline)
custom_workflow_params: CustomWorkflowGenerationParameters | None = _get_custom_workflow_params(api_response)
loras: list[LoRaEntry] | None = _get_lora_params(api_response)
tis: list[TIEntry] | None = _get_ti_params(api_response)
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=api_response.source_processing,
base_params=base_params,
additional_params=ImageGenerationComponentContainer(
components=additional_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 image_generation_parameters, ai_horde_dispatch_parameters