""" ComfyUI X Rodin3D(Deemos) API Nodes Rodin API docs: https://developer.hyper3d.ai/ """ import logging import math import os from inspect import cleandoc from io import BytesIO from typing import Any import aiohttp from PIL import Image from typing_extensions import override import folder_paths as comfy_paths from comfy_api.latest import IO, ComfyExtension, Types from comfy_api_nodes.apis.rodin import ( JobStatus, Rodin3DCheckStatusRequest, Rodin3DCheckStatusResponse, Rodin3DDownloadRequest, Rodin3DDownloadResponse, Rodin3DGen25Request, Rodin3DGenerateRequest, Rodin3DGenerateResponse, ) from comfy_api_nodes.util import ( ApiEndpoint, download_url_to_bytesio, download_url_to_file_3d, poll_op, sync_op, validate_string, ) COMMON_PARAMETERS = [ IO.Int.Input( "Seed", default=0, min=0, max=65535, display_mode=IO.NumberDisplay.number, optional=True, ), IO.Combo.Input("Material_Type", options=["PBR", "Shaded"], default="PBR", optional=True), IO.Combo.Input( "Polygon_count", options=["4K-Quad", "8K-Quad", "18K-Quad", "50K-Quad", "200K-Triangle"], default="18K-Quad", optional=True, ), ] _QUALITY_MESH_OPTIONS: dict[str, tuple[str, int]] = { "4K-Quad": ("Quad", 4000), "8K-Quad": ("Quad", 8000), "18K-Quad": ("Quad", 18000), "50K-Quad": ("Quad", 50000), "200K-Quad": ("Quad", 200000), "2K-Triangle": ("Raw", 2000), "20K-Triangle": ("Raw", 20000), "150K-Triangle": ("Raw", 150000), "200K-Triangle": ("Raw", 200000), "500K-Triangle": ("Raw", 500000), "1M-Triangle": ("Raw", 1000000), } def get_quality_mode(poly_count: str) -> tuple[str, int]: """Map a polygon-count preset like '18K-Quad' to (mesh_mode, quality_override). Falls back to ('Quad', 18000) for unknown labels; legacy parity. """ return _QUALITY_MESH_OPTIONS.get(poly_count, ("Quad", 18000)) def tensor_to_filelike(tensor, max_pixels: int = 2048 * 2048): """ Converts a PyTorch tensor to a file-like object. Args: - tensor (torch.Tensor): A tensor representing an image of shape (H, W, C) where C is the number of channels (3 for RGB), H is height, and W is width. Returns: - io.BytesIO: A file-like object containing the image data. """ array = tensor.cpu().numpy() array = (array * 255).astype("uint8") image = Image.fromarray(array, "RGB") original_width, original_height = image.size original_pixels = original_width * original_height if original_pixels > max_pixels: scale = math.sqrt(max_pixels / original_pixels) new_width = int(original_width * scale) new_height = int(original_height * scale) else: new_width, new_height = original_width, original_height if new_width != original_width or new_height != original_height: image = image.resize((new_width, new_height), Image.Resampling.LANCZOS) img_byte_arr = BytesIO() image.save(img_byte_arr, format="PNG") # PNG is used for lossless compression img_byte_arr.seek(0) return img_byte_arr async def create_generate_task( cls: type[IO.ComfyNode], images=None, seed=1, material="PBR", quality_override=18000, tier="Regular", mesh_mode="Quad", ta_pose: bool = False, ): if images is None: raise Exception("Rodin 3D generate requires at least 1 image.") if len(images) > 5: raise Exception("Rodin 3D generate requires up to 5 image.") response = await sync_op( cls, ApiEndpoint(path="/proxy/rodin/api/v2/rodin", method="POST"), response_model=Rodin3DGenerateResponse, data=Rodin3DGenerateRequest( seed=seed, tier=tier, material=material, quality_override=quality_override, mesh_mode=mesh_mode, TAPose=ta_pose, ), files=[ ("images", open(image, "rb") if isinstance(image, str) else tensor_to_filelike(image)) for image in images if image is not None ], content_type="multipart/form-data", ) if hasattr(response, "error"): error_message = f"Rodin3D Create 3D generate Task Failed. Message: {response.message}, error: {response.error}" logging.error(error_message) raise Exception(error_message) logging.info("[ Rodin3D API - Submit Jobs ] Submit Generate Task Success!") subscription_key = response.jobs.subscription_key task_uuid = response.uuid logging.info("[ Rodin3D API - Submit Jobs ] UUID: %s", task_uuid) return task_uuid, subscription_key def check_rodin_status(response: Rodin3DCheckStatusResponse) -> str: all_done = all(job.status == JobStatus.Done for job in response.jobs) status_list = [str(job.status) for job in response.jobs] logging.info("[ Rodin3D API - CheckStatus ] Generate Status: %s", status_list) if any(job.status == JobStatus.Failed for job in response.jobs): logging.error("[ Rodin3D API - CheckStatus ] Generate Failed: %s, Please try again.", status_list) raise Exception("[ Rodin3D API ] Generate Failed, Please Try again.") if all_done: return "DONE" return "Generating" def extract_progress(response: Rodin3DCheckStatusResponse) -> int | None: if not response.jobs: return None completed_count = sum(1 for job in response.jobs if job.status == JobStatus.Done) return int((completed_count / len(response.jobs)) * 100) async def poll_for_task_status(subscription_key: str, cls: type[IO.ComfyNode]) -> Rodin3DCheckStatusResponse: logging.info("[ Rodin3D API - CheckStatus ] Generate Start!") return await poll_op( cls, ApiEndpoint(path="/proxy/rodin/api/v2/status", method="POST"), response_model=Rodin3DCheckStatusResponse, data=Rodin3DCheckStatusRequest(subscription_key=subscription_key), status_extractor=check_rodin_status, progress_extractor=extract_progress, ) async def get_rodin_download_list(uuid: str, cls: type[IO.ComfyNode]) -> Rodin3DDownloadResponse: logging.info("[ Rodin3D API - Downloading ] Generate Successfully!") return await sync_op( cls, ApiEndpoint(path="/proxy/rodin/api/v2/download", method="POST"), response_model=Rodin3DDownloadResponse, data=Rodin3DDownloadRequest(task_uuid=uuid), monitor_progress=False, ) async def download_files(url_list, task_uuid: str) -> tuple[str | None, Types.File3D | None]: result_folder_name = f"Rodin3D_{task_uuid}" save_path = os.path.join(comfy_paths.get_output_directory(), result_folder_name) os.makedirs(save_path, exist_ok=True) model_file_path = None file_3d = None for i in url_list.items: file_path = os.path.join(save_path, i.name) if i.name.lower().endswith(".glb"): model_file_path = os.path.join(result_folder_name, i.name) file_3d = await download_url_to_file_3d(i.url, "glb") # Save to disk for backward compatibility with open(file_path, "wb") as f: f.write(file_3d.get_bytes()) else: await download_url_to_bytesio(i.url, file_path) return model_file_path, file_3d class Rodin3D_Regular(IO.ComfyNode): """Generate 3D Assets using Rodin API""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="Rodin3D_Regular", display_name="Rodin 3D Generate - Regular Generate", category="partner/3d/Rodin", description=cleandoc(cls.__doc__ or ""), inputs=[ IO.Image.Input("Images"), *COMMON_PARAMETERS, ], outputs=[ IO.String.Output(display_name="3D Model Path"), # for backward compatibility only IO.File3DGLB.Output(display_name="GLB"), ], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( expr="""{"type":"usd","usd":0.4}""", ), ) @classmethod async def execute( cls, Images, Seed, Material_Type, Polygon_count, ) -> IO.NodeOutput: tier = "Regular" num_images = Images.shape[0] m_images = [] for i in range(num_images): m_images.append(Images[i]) mesh_mode, quality_override = get_quality_mode(Polygon_count) task_uuid, subscription_key = await create_generate_task( cls, images=m_images, seed=Seed, material=Material_Type, quality_override=quality_override, tier=tier, mesh_mode=mesh_mode, ) await poll_for_task_status(subscription_key, cls) download_list = await get_rodin_download_list(task_uuid, cls) model_path, file_3d = await download_files(download_list, task_uuid) return IO.NodeOutput(model_path, file_3d) class Rodin3D_Detail(IO.ComfyNode): """Generate 3D Assets using Rodin API""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="Rodin3D_Detail", display_name="Rodin 3D Generate - Detail Generate", category="partner/3d/Rodin", description=cleandoc(cls.__doc__ or ""), inputs=[ IO.Image.Input("Images"), *COMMON_PARAMETERS, ], outputs=[ IO.String.Output(display_name="3D Model Path"), # for backward compatibility only IO.File3DGLB.Output(display_name="GLB"), ], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( expr="""{"type":"usd","usd":0.4}""", ), ) @classmethod async def execute( cls, Images, Seed, Material_Type, Polygon_count, ) -> IO.NodeOutput: tier = "Detail" num_images = Images.shape[0] m_images = [] for i in range(num_images): m_images.append(Images[i]) mesh_mode, quality_override = get_quality_mode(Polygon_count) task_uuid, subscription_key = await create_generate_task( cls, images=m_images, seed=Seed, material=Material_Type, quality_override=quality_override, tier=tier, mesh_mode=mesh_mode, ) await poll_for_task_status(subscription_key, cls) download_list = await get_rodin_download_list(task_uuid, cls) model_path, file_3d = await download_files(download_list, task_uuid) return IO.NodeOutput(model_path, file_3d) class Rodin3D_Smooth(IO.ComfyNode): """Generate 3D Assets using Rodin API""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="Rodin3D_Smooth", display_name="Rodin 3D Generate - Smooth Generate", category="partner/3d/Rodin", description=cleandoc(cls.__doc__ or ""), inputs=[ IO.Image.Input("Images"), *COMMON_PARAMETERS, ], outputs=[ IO.String.Output(display_name="3D Model Path"), # for backward compatibility only IO.File3DGLB.Output(display_name="GLB"), ], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( expr="""{"type":"usd","usd":0.4}""", ), ) @classmethod async def execute( cls, Images, Seed, Material_Type, Polygon_count, ) -> IO.NodeOutput: num_images = Images.shape[0] m_images = [] for i in range(num_images): m_images.append(Images[i]) mesh_mode, quality_override = get_quality_mode(Polygon_count) task_uuid, subscription_key = await create_generate_task( cls, images=m_images, seed=Seed, material=Material_Type, quality_override=quality_override, tier="Smooth", mesh_mode=mesh_mode, ) await poll_for_task_status(subscription_key, cls) download_list = await get_rodin_download_list(task_uuid, cls) model_path, file_3d = await download_files(download_list, task_uuid) return IO.NodeOutput(model_path, file_3d) class Rodin3D_Sketch(IO.ComfyNode): """Generate 3D Assets using Rodin API""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="Rodin3D_Sketch", display_name="Rodin 3D Generate - Sketch Generate", category="partner/3d/Rodin", description=cleandoc(cls.__doc__ or ""), inputs=[ IO.Image.Input("Images"), IO.Int.Input( "Seed", default=0, min=0, max=65535, display_mode=IO.NumberDisplay.number, optional=True, ), ], outputs=[ IO.String.Output(display_name="3D Model Path"), # for backward compatibility only IO.File3DGLB.Output(display_name="GLB"), ], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( expr="""{"type":"usd","usd":0.4}""", ), ) @classmethod async def execute( cls, Images, Seed, ) -> IO.NodeOutput: num_images = Images.shape[0] m_images = [] for i in range(num_images): m_images.append(Images[i]) task_uuid, subscription_key = await create_generate_task( cls, images=m_images, seed=Seed, material="PBR", quality_override=18000, tier="Sketch", mesh_mode="Quad", ) await poll_for_task_status(subscription_key, cls) download_list = await get_rodin_download_list(task_uuid, cls) model_path, file_3d = await download_files(download_list, task_uuid) return IO.NodeOutput(model_path, file_3d) class Rodin3D_Gen2(IO.ComfyNode): """Generate 3D Assets using Rodin API""" @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="Rodin3D_Gen2", display_name="Rodin 3D Generate - Gen-2 Generate", category="partner/3d/Rodin", description=cleandoc(cls.__doc__ or ""), inputs=[ IO.Image.Input("Images"), IO.Int.Input( "Seed", default=0, min=0, max=65535, display_mode=IO.NumberDisplay.number, optional=True, ), IO.Combo.Input("Material_Type", options=["PBR", "Shaded"], default="PBR", optional=True), IO.Combo.Input( "Polygon_count", options=[ "4K-Quad", "8K-Quad", "18K-Quad", "50K-Quad", "2K-Triangle", "20K-Triangle", "150K-Triangle", "500K-Triangle", ], default="500K-Triangle", optional=True, ), IO.Boolean.Input("TAPose", default=False, advanced=True), ], outputs=[ IO.String.Output(display_name="3D Model Path"), # for backward compatibility only IO.File3DGLB.Output(display_name="GLB"), ], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( expr="""{"type":"usd","usd":0.4}""", ), ) @classmethod async def execute( cls, Images, Seed, Material_Type, Polygon_count, TAPose, ) -> IO.NodeOutput: tier = "Gen-2" num_images = Images.shape[0] m_images = [] for i in range(num_images): m_images.append(Images[i]) mesh_mode, quality_override = get_quality_mode(Polygon_count) task_uuid, subscription_key = await create_generate_task( cls, images=m_images, seed=Seed, material=Material_Type, quality_override=quality_override, tier=tier, mesh_mode=mesh_mode, ta_pose=TAPose, ) await poll_for_task_status(subscription_key, cls) download_list = await get_rodin_download_list(task_uuid, cls) model_path, file_3d = await download_files(download_list, task_uuid) return IO.NodeOutput(model_path, file_3d) def _rodin_multipart_parser(data: dict[str, Any]) -> aiohttp.FormData: """Convert a Rodin request dict to an aiohttp form, fixing bool/list serialization. Booleans --> "true"/"false". Lists --> one field per element. """ form = aiohttp.FormData(default_to_multipart=True) for key, value in data.items(): if value is None: continue if isinstance(value, bool): form.add_field(key, "true" if value else "false") elif isinstance(value, list): for item in value: form.add_field(key, str(item)) elif isinstance(value, (bytes, bytearray)): form.add_field(key, value) else: form.add_field(key, str(value)) return form async def _create_gen25_task( cls: type[IO.ComfyNode], request: Rodin3DGen25Request, images: list | None, ) -> tuple[str, str]: """Submit a Gen-2.5 generate job; returns (task_uuid, subscription_key).""" if images is not None and len(images) > 5: raise ValueError("Rodin Gen-2.5 supports at most 5 input images.") files = None if images: files = [ ( "images", open(image, "rb") if isinstance(image, str) else tensor_to_filelike(image), ) for image in images if image is not None ] response = await sync_op( cls, ApiEndpoint(path="/proxy/rodin/api/v2/rodin", method="POST"), response_model=Rodin3DGenerateResponse, data=request, files=files, content_type="multipart/form-data", multipart_parser=_rodin_multipart_parser, ) if not response.uuid or not response.jobs or not response.jobs.subscription_key: raise RuntimeError(f"Rodin Gen-2.5 submit failed: message={response.message!r}") return response.uuid, response.jobs.subscription_key _PREVIEWABLE_3D_EXTS = {".glb", ".obj", ".fbx", ".stl", ".gltf"} async def _download_gen25_files( download_list: Rodin3DDownloadResponse, task_uuid: str, geometry_file_format: str, ) -> Types.File3D | None: """Download every file in the list; return the File3D matching the chosen format.""" folder_name = f"Rodin3D_Gen25_{task_uuid}" save_dir = os.path.join(comfy_paths.get_output_directory(), folder_name) os.makedirs(save_dir, exist_ok=True) target_ext = f".{geometry_file_format.lower().lstrip('.')}" file_3d: Types.File3D | None = None for item in download_list.items: file_path = os.path.join(save_dir, item.name) ext = os.path.splitext(item.name.lower())[1] # Prefer the file matching the user's chosen format; fall back below. if file_3d is None and ext == target_ext and ext in _PREVIEWABLE_3D_EXTS: file_3d = await download_url_to_file_3d(item.url, target_ext.lstrip(".")) with open(file_path, "wb") as f: f.write(file_3d.get_bytes()) continue await download_url_to_bytesio(item.url, file_path) # If the chosen format wasn't found, surface any model file we did get. if file_3d is None: for item in download_list.items: ext = os.path.splitext(item.name.lower())[1] if ext in _PREVIEWABLE_3D_EXTS: file_3d = await download_url_to_file_3d(item.url, ext.lstrip(".")) break return file_3d _MODE_REGULAR = "Regular" _MODE_FAST = "Fast" _MODE_EXTREME_HIGH = "Extreme-High" _REGULAR_POLY_OPTIONS = [ "Default", "4K-Quad", "8K-Quad", "18K-Quad", "50K-Quad", "2K-Triangle", "20K-Triangle", "150K-Triangle", "500K-Triangle", "1M-Triangle", ] _TEXTURE_MODE_OPTIONS = ["Default", "legacy", "extreme-low", "low", "medium", "high"] _GEOMETRY_FORMAT_OPTIONS = ["glb", "fbx", "obj", "stl"] _MATERIAL_OPTIONS = ["PBR", "Shaded", "All", "None"] def _build_mode_input(name: str = "mode") -> IO.DynamicCombo.Input: return IO.DynamicCombo.Input( name, options=[ IO.DynamicCombo.Option( _MODE_REGULAR, [ IO.Combo.Input( "tier", options=["Gen-2.5-Low", "Gen-2.5-Medium", "Gen-2.5-High"], default="Gen-2.5-High", tooltip="Quality tier. Higher tiers produce higher-fidelity geometry.", ), IO.Combo.Input( "polygon_count", options=_REGULAR_POLY_OPTIONS, default="Default", tooltip="Preset face count. 'Default' uses the server's default for the selected tier.", ), IO.Boolean.Input( "creative", default=False, tooltip="Creative mode (Medium/High only). Enhances generative robustness.", ), ], ), IO.DynamicCombo.Option( _MODE_FAST, [ IO.Combo.Input( "tier", options=[ "Gen-2.5-Extreme-Low", "Gen-2.5-Low", "Gen-2.5-Medium", "Gen-2.5-High", ], default="Gen-2.5-Low", ), IO.Int.Input( "mesh_faces", default=20000, min=1000, max=20000, display_mode=IO.NumberDisplay.number, tooltip="Mesh face count (1K-20K in Fast mode).", ), ], ), IO.DynamicCombo.Option( _MODE_EXTREME_HIGH, [ IO.Combo.Input("mesh_mode", options=["Raw", "Quad"], default="Raw"), IO.Int.Input( "mesh_faces", default=1000000, min=20000, max=2000000, display_mode=IO.NumberDisplay.number, tooltip=( "Mesh face count. Raw mode: 20K-2M. " "Quad mode: keep under 200K (upstream may reject higher values)." ), ), IO.Boolean.Input( "is_micro", default=False, tooltip="Enable micro detail (Extreme-High only).", ), IO.Boolean.Input( "creative", default=False, tooltip="Creative mode. Enhances generative robustness.", ), ], ), ], tooltip=( "Generation mode. Regular = balanced. Fast = 1K-20K faces for rapid prototyping. " "Extreme-High = 20K-2M faces with optional micro details." ), ) def _build_common_inputs(*, include_image_only: bool) -> list: inputs: list = [ IO.Combo.Input("material", options=_MATERIAL_OPTIONS, default="Shaded"), IO.Combo.Input("geometry_file_format", options=_GEOMETRY_FORMAT_OPTIONS, default="glb"), IO.Combo.Input( "texture_mode", options=_TEXTURE_MODE_OPTIONS, default="Default", optional=True, tooltip="Texture quality preset. 'Default' uses the server's default for the selected tier.", ), IO.Int.Input( "seed", default=0, min=0, max=65535, display_mode=IO.NumberDisplay.number, control_after_generate=True, optional=True, ), IO.Boolean.Input( "TAPose", default=False, optional=True, advanced=True, tooltip="T/A pose for human-like models." ), IO.Boolean.Input( "hd_texture", default=False, optional=True, advanced=True, tooltip="High-quality texture enhancement." ), IO.Boolean.Input( "texture_delight", default=False, optional=True, advanced=True, tooltip="Remove baked lighting from textures.", ), ] if include_image_only: inputs.append( IO.Boolean.Input( "use_original_alpha", default=False, optional=True, advanced=True, tooltip="Preserve image transparency.", ) ) inputs.extend( [ IO.Boolean.Input( "addon_highpack", default=False, optional=True, advanced=True, tooltip="HighPack addon: 4K textures and ~16x faces in Quad mode.", ), IO.Int.Input( "bbox_width", default=0, min=0, max=300, display_mode=IO.NumberDisplay.number, optional=True, advanced=True, tooltip="Bounding-box width (Y axis). Set to 0 with the others to skip bbox.", ), IO.Int.Input( "bbox_height", default=0, min=0, max=300, display_mode=IO.NumberDisplay.number, optional=True, advanced=True, tooltip="Bounding-box height (Z axis).", ), IO.Int.Input( "bbox_length", default=0, min=0, max=300, display_mode=IO.NumberDisplay.number, optional=True, advanced=True, tooltip="Bounding-box length (X axis).", ), IO.Int.Input( "height_cm", default=0, min=0, max=10000, display_mode=IO.NumberDisplay.number, optional=True, advanced=True, tooltip="Approximate model height in centimeters (0 to skip).", ), ] ) return inputs _PRICE_EXPR = """ ( $baseCredits := widgets.mode = "extreme-high" ? 1.0 : 0.5; $addonCredits := widgets.addon_highpack ? 1.0 : 0.0; $total := ($baseCredits * 1.5) + ($addonCredits * 0.8); {"type":"usd","usd": $total} ) """ def _resolve_mode_params(mode_input: dict) -> dict: """Translate the DynamicCombo `mode` payload into Gen-2.5 request fields. Returns a dict with: tier, quality_override, mesh_mode, geometry_instruct_mode, is_micro. Missing keys mean "do not send" (so we don't override server defaults). """ selected = mode_input["mode"] out: dict = {} if selected == _MODE_REGULAR: out["tier"] = mode_input["tier"] polygon = mode_input.get("polygon_count", "Default") if polygon != "Default": mesh_mode, faces = get_quality_mode(polygon) out["mesh_mode"] = mesh_mode out["quality_override"] = faces if mode_input.get("creative"): out["geometry_instruct_mode"] = "creative" elif selected == _MODE_FAST: out["tier"] = mode_input["tier"] out["mesh_mode"] = "Raw" out["quality_override"] = int(mode_input["mesh_faces"]) elif selected == _MODE_EXTREME_HIGH: out["tier"] = "Gen-2.5-Extreme-High" out["mesh_mode"] = mode_input["mesh_mode"] out["quality_override"] = int(mode_input["mesh_faces"]) if mode_input.get("is_micro"): out["is_micro"] = True if mode_input.get("creative"): out["geometry_instruct_mode"] = "creative" return out def _build_request( *, mode_input: dict, material: str, geometry_file_format: str, texture_mode: str, seed: int, TAPose: bool, hd_texture: bool, texture_delight: bool, addon_highpack: bool, bbox_width: int, bbox_height: int, bbox_length: int, height_cm: int, prompt: str | None = None, use_original_alpha: bool = False, ) -> Rodin3DGen25Request: mode_params = _resolve_mode_params(mode_input) bbox = None if bbox_width and bbox_height and bbox_length: bbox = [bbox_width, bbox_height, bbox_length] return Rodin3DGen25Request( tier=mode_params["tier"], prompt=prompt or None, seed=seed, material=material, geometry_file_format=geometry_file_format, texture_mode=None if texture_mode == "Default" else texture_mode, mesh_mode=mode_params.get("mesh_mode"), quality_override=mode_params.get("quality_override"), geometry_instruct_mode=mode_params.get("geometry_instruct_mode"), bbox_condition=bbox, height=height_cm or None, TAPose=TAPose or None, hd_texture=hd_texture or None, texture_delight=texture_delight or None, is_micro=mode_params.get("is_micro"), use_original_alpha=use_original_alpha or None, addons=["HighPack"] if addon_highpack else None, ) class Rodin3D_Gen25_Image(IO.ComfyNode): @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="Rodin3D_Gen25_Image", display_name="Rodin 3D Gen-2.5 - Image to 3D", category="partner/3d/Rodin", description=( "Generate a 3D model from 1-5 reference images via Rodin Gen-2.5. " "Pick a mode (Fast / Regular / Extreme-High) to tune quality vs. cost." ), inputs=[ IO.Autogrow.Input( "images", template=IO.Autogrow.TemplatePrefix(IO.Image.Input("image"), prefix="image", min=1, max=5), tooltip="1-5 images. The first image is used for materials when multi-view.", ), _build_mode_input(), *_build_common_inputs(include_image_only=True), ], outputs=[IO.File3DAny.Output(display_name="model_file")], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( depends_on=IO.PriceBadgeDepends(widgets=["mode", "addon_highpack"]), expr=_PRICE_EXPR, ), ) @classmethod async def execute( cls, images: IO.Autogrow.Type, mode: dict, material: str, geometry_file_format: str, texture_mode: str, seed: int, TAPose: bool, hd_texture: bool, texture_delight: bool, use_original_alpha: bool, addon_highpack: bool, bbox_width: int, bbox_height: int, bbox_length: int, height_cm: int, ) -> IO.NodeOutput: image_tensors = [img for img in images.values() if img is not None] if not image_tensors: raise ValueError("Rodin Gen-2.5 Image-to-3D requires at least one image.") # Flatten multi-image tensors into individual frames; the API accepts each as a separate part. flat_images: list = [] for tensor in image_tensors: if hasattr(tensor, "shape") and len(tensor.shape) == 4: for i in range(tensor.shape[0]): flat_images.append(tensor[i]) else: flat_images.append(tensor) if len(flat_images) > 5: raise ValueError(f"Rodin Gen-2.5 accepts at most 5 images; received {len(flat_images)}.") request = _build_request( mode_input=mode, material=material, geometry_file_format=geometry_file_format, texture_mode=texture_mode, seed=seed, TAPose=TAPose, hd_texture=hd_texture, texture_delight=texture_delight, addon_highpack=addon_highpack, bbox_width=bbox_width, bbox_height=bbox_height, bbox_length=bbox_length, height_cm=height_cm, prompt=None, use_original_alpha=use_original_alpha, ) task_uuid, subscription_key = await _create_gen25_task(cls, request, flat_images) await poll_for_task_status(subscription_key, cls) download_list = await get_rodin_download_list(task_uuid, cls) file_3d = await _download_gen25_files(download_list, task_uuid, geometry_file_format) return IO.NodeOutput(file_3d) class Rodin3D_Gen25_Text(IO.ComfyNode): @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id="Rodin3D_Gen25_Text", display_name="Rodin 3D Gen-2.5 - Text to 3D", category="partner/3d/Rodin", description=( "Generate a 3D model from a text prompt via Rodin Gen-2.5. " "Pick a mode (Fast / Regular / Extreme-High) to tune quality vs. cost." ), inputs=[ IO.String.Input( "prompt", multiline=True, default="", tooltip="Text prompt for the 3D model.", ), _build_mode_input(), *_build_common_inputs(include_image_only=False), ], outputs=[IO.File3DAny.Output(display_name="model_file")], hidden=[ IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id, ], is_api_node=True, price_badge=IO.PriceBadge( depends_on=IO.PriceBadgeDepends(widgets=["mode", "addon_highpack"]), expr=_PRICE_EXPR, ), ) @classmethod async def execute( cls, prompt: str, mode: dict, material: str, geometry_file_format: str, texture_mode: str, seed: int, TAPose: bool, hd_texture: bool, texture_delight: bool, addon_highpack: bool, bbox_width: int, bbox_height: int, bbox_length: int, height_cm: int, ) -> IO.NodeOutput: validate_string(prompt, field_name="prompt", min_length=1, max_length=2500) request = _build_request( mode_input=mode, material=material, geometry_file_format=geometry_file_format, texture_mode=texture_mode, seed=seed, TAPose=TAPose, hd_texture=hd_texture, texture_delight=texture_delight, addon_highpack=addon_highpack, bbox_width=bbox_width, bbox_height=bbox_height, bbox_length=bbox_length, height_cm=height_cm, prompt=prompt, ) task_uuid, subscription_key = await _create_gen25_task(cls, request, images=None) await poll_for_task_status(subscription_key, cls) download_list = await get_rodin_download_list(task_uuid, cls) file_3d = await _download_gen25_files(download_list, task_uuid, geometry_file_format) return IO.NodeOutput(file_3d) class Rodin3DExtension(ComfyExtension): @override async def get_node_list(self) -> list[type[IO.ComfyNode]]: return [ Rodin3D_Regular, Rodin3D_Detail, Rodin3D_Smooth, Rodin3D_Sketch, Rodin3D_Gen2, Rodin3D_Gen25_Image, Rodin3D_Gen25_Text, ] async def comfy_entrypoint() -> Rodin3DExtension: return Rodin3DExtension()