def hyper3d_run_api()

in tools.py [0:0]

• 63 lines of code
• 11 McCabe index (conditional complexity)

def hyper3d_run_api(api_key: str, object_description: str):
    import os
    import tempfile
    import time

    import requests

    files = [
        ("mesh_mode", (None, "Raw")),
        ("prompt", (None, object_description)),
    ]

    response = requests.post(
        "https://hyperhuman.deemos.com/api/v2/rodin",
        headers={"Authorization": f"Bearer {api_key}"},
        files=files,
    )

    data = response.json()

    succeed = data.get("submit_time", False)
    if not succeed:
        raise RuntimeError("Failed to create generate job")

    task_uuid = data["uuid"]
    subscription_key = data["jobs"]["subscription_key"]

    start_time = time.time()
    max_wait_time = 300  # 5 minutes

    print(f"Generation started. Task UUID: {task_uuid}")
    print(f"Waiting up to {max_wait_time} seconds for generation to complete...")

    while True:
        if time.time() - start_time > max_wait_time:
            raise RuntimeError(f"Generation timed out after {max_wait_time} seconds")

        response = requests.post(
            "https://hyperhuman.deemos.com/api/v2/status",
            headers={"Authorization": f"Bearer {api_key}"},
            json={"subscription_key": subscription_key},
        )

        data = response.json()
        status_list = [i["status"] for i in data["jobs"]]
        if all(status == "Done" for status in status_list):
            break

        time.sleep(2)

    print("Generation completed. Downloading result...")

    response = requests.post(
        "https://hyperhuman.deemos.com/api/v2/download",
        headers={"Authorization": f"Bearer {api_key}"},
        json={"task_uuid": task_uuid},
    )

    data = response.json()
    temp_file = None

    for i in data["list"]:
        if i["name"].endswith(".glb"):
            temp_file = tempfile.NamedTemporaryFile(
                delete=False,
                prefix=task_uuid,
                suffix=".glb",
            )

            try:
                response = requests.get(i["url"], stream=True)
                response.raise_for_status()

                for chunk in response.iter_content(chunk_size=8192):
                    temp_file.write(chunk)

                temp_file.close()
            except Exception as e:
                temp_file.close()
                os.unlink(temp_file.name)
                raise e

    return temp_file.name