Spaces:
Build error
Build error
Download embodied_gen/scripts/imageto3d.py from Tim-ops/EmbodiedGen-Text-to-3D_2: direct link, hf CLI and curl.
- Browser
- Download file 15.9 kB
-
https://huggingface.co/spaces/Tim-ops/EmbodiedGen-Text-to-3D_2/resolve/main/embodied_gen/scripts/imageto3d.py
- Command line
-
hf download hf://spaces/Tim-ops/EmbodiedGen-Text-to-3D_2/embodied_gen/scripts/imageto3d.py
-
curl -L -o imageto3d.py https://huggingface.co/spaces/Tim-ops/EmbodiedGen-Text-to-3D_2/resolve/main/embodied_gen/scripts/imageto3d.py
15.9 kB
| # Project EmbodiedGen | |
| # | |
| # Copyright (c) 2025 Horizon Robotics. All Rights Reserved. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or | |
| # implied. See the License for the specific language governing | |
| # permissions and limitations under the License. | |
| import argparse | |
| import os | |
| import random | |
| from glob import glob | |
| from shutil import copy, copytree, rmtree | |
| import numpy as np | |
| import trimesh | |
| from PIL import Image | |
| from embodied_gen.data.backproject_v3 import entrypoint as backproject_api | |
| from embodied_gen.data.utils import delete_dir | |
| # from embodied_gen.models.sr_model import ImageRealESRGAN | |
| # from embodied_gen.models.delight_model import DelightingModel | |
| from embodied_gen.models.gs_model import GaussianOperator | |
| from embodied_gen.models.hunyuan3d import ( | |
| HunyuanConfig, | |
| load_credentials, | |
| process_image, | |
| ) | |
| from embodied_gen.models.segment_model import RembgRemover | |
| from embodied_gen.scripts.render_gs import entrypoint as render_gs_api | |
| from embodied_gen.utils.gpt_clients import GPT_CLIENT | |
| from embodied_gen.utils.inference import image3d_model_infer | |
| from embodied_gen.utils.log import logger | |
| from embodied_gen.utils.process_media import ( | |
| combine_images_to_grid, | |
| merge_images_video, | |
| ) | |
| from embodied_gen.utils.tags import VERSION | |
| from embodied_gen.utils.trender import render_video | |
| from embodied_gen.validators.quality_checkers import ( | |
| BaseChecker, | |
| ImageAestheticChecker, | |
| ImageSegChecker, | |
| MeshGeoChecker, | |
| ) | |
| from embodied_gen.validators.urdf_convertor import URDFGenerator | |
| # random.seed(0) | |
| IMAGE3D_MODEL = "SAM3D" # default backend; SAM3D, TRELLIS, or HUNYUAN3D | |
| SUPPORTED_IMAGE3D_MODELS = ("SAM3D", "TRELLIS", "HUNYUAN3D") | |
| _PIPELINE_CACHE: dict = {} | |
| def _build_image3d_pipeline(name: str): | |
| """Lazily instantiate (and cache) the local image-to-3D pipeline. | |
| The cache preserves the pre-refactor invariant that the local backend | |
| is loaded once per process: ``textto3d.py`` calls ``entrypoint`` in a | |
| per-node loop, and re-loading weights each call would regress runtime. | |
| Returns ``None`` for backends that have no local model (HUNYUAN3D). | |
| """ | |
| if name == "HUNYUAN3D": | |
| return None | |
| if name in _PIPELINE_CACHE: | |
| return _PIPELINE_CACHE[name] | |
| if name == "TRELLIS": | |
| logger.info("Loading TRELLIS as Image3D Models...") | |
| from thirdparty.TRELLIS.trellis.pipelines import ( | |
| TrellisImageTo3DPipeline, | |
| ) | |
| pipeline = TrellisImageTo3DPipeline.from_pretrained( | |
| "microsoft/TRELLIS-image-large" | |
| ) | |
| elif name == "SAM3D": | |
| logger.info("Loading SAM3D as Image3D Models...") | |
| from embodied_gen.models.sam3d import Sam3dInference | |
| pipeline = Sam3dInference() | |
| else: | |
| raise ValueError( | |
| f"Unsupported image3d backend {name!r}; " | |
| f"expected one of {SUPPORTED_IMAGE3D_MODELS}." | |
| ) | |
| _PIPELINE_CACHE[name] = pipeline | |
| return pipeline | |
| # DELIGHT = DelightingModel() | |
| # IMAGESR_MODEL = ImageRealESRGAN(outscale=4) | |
| RBG_REMOVER = RembgRemover() | |
| SEG_CHECKER = ImageSegChecker(GPT_CLIENT) | |
| GEO_CHECKER = MeshGeoChecker(GPT_CLIENT) | |
| AESTHETIC_CHECKER = ImageAestheticChecker() | |
| CHECKERS = [GEO_CHECKER, SEG_CHECKER, AESTHETIC_CHECKER] | |
| def parse_args(): | |
| parser = argparse.ArgumentParser(description="Image to 3D pipeline args.") | |
| parser.add_argument( | |
| "--image_path", type=str, nargs="+", help="Path to the input images." | |
| ) | |
| parser.add_argument( | |
| "--image_root", type=str, help="Path to the input images folder." | |
| ) | |
| parser.add_argument( | |
| "--output_root", | |
| type=str, | |
| help="Root directory for saving outputs.", | |
| ) | |
| parser.add_argument( | |
| "--height_range", | |
| type=str, | |
| default=None, | |
| help="The hight in meter to restore the mesh real size.", | |
| ) | |
| parser.add_argument( | |
| "--mass_range", | |
| type=str, | |
| default=None, | |
| help="The mass in kg to restore the mesh real weight.", | |
| ) | |
| parser.add_argument("--asset_type", type=str, nargs="+", default=None) | |
| parser.add_argument("--skip_exists", action="store_true") | |
| parser.add_argument("--version", type=str, default=VERSION) | |
| parser.add_argument("--keep_intermediate", action="store_true") | |
| parser.add_argument("--seed", type=int, default=0) | |
| parser.add_argument( | |
| "--n_retry", | |
| type=int, | |
| default=3, | |
| ) | |
| parser.add_argument("--disable_decompose_convex", action="store_true") | |
| parser.add_argument("--texture_size", type=int, default=2048) | |
| parser.add_argument( | |
| "--image3d_model", | |
| type=str, | |
| default=IMAGE3D_MODEL, | |
| help=( | |
| "Image-to-3D backend. One of " | |
| f"{', '.join(SUPPORTED_IMAGE3D_MODELS)} (case-insensitive). " | |
| "HUNYUAN3D calls Tencent Hunyuan3D Pro API and requires " | |
| "TENCENT_SECRET_ID/TENCENT_SECRET_KEY in the environment." | |
| ), | |
| ) | |
| args, unknown = parser.parse_known_args() | |
| return args | |
| def entrypoint(**kwargs): | |
| args = parse_args() | |
| for k, v in kwargs.items(): | |
| if hasattr(args, k) and v is not None: | |
| setattr(args, k, v) | |
| args.image3d_model = str(args.image3d_model).strip().upper() | |
| if args.image3d_model not in SUPPORTED_IMAGE3D_MODELS: | |
| raise ValueError( | |
| f"Unsupported --image3d_model {args.image3d_model!r}; " | |
| f"expected one of {SUPPORTED_IMAGE3D_MODELS}." | |
| ) | |
| hunyuan_config = None | |
| hunyuan_credentials = None | |
| if args.image3d_model == "HUNYUAN3D": | |
| # Fail fast on missing creds before any local model load or network I/O. | |
| hunyuan_credentials = load_credentials() | |
| hunyuan_config = HunyuanConfig() | |
| logger.info( | |
| "HUNYUAN3D backend: action=%s host=%s result_format=%s", | |
| hunyuan_config.image_action, | |
| hunyuan_config.host, | |
| hunyuan_config.result_format, | |
| ) | |
| pipeline = _build_image3d_pipeline(args.image3d_model) | |
| assert args.image_path or args.image_root, ( | |
| "Please provide either --image_path or --image_root." | |
| ) | |
| if not args.image_path: | |
| args.image_path = glob(os.path.join(args.image_root, "*.png")) | |
| args.image_path += glob(os.path.join(args.image_root, "*.jpg")) | |
| args.image_path += glob(os.path.join(args.image_root, "*.jpeg")) | |
| for idx, image_path in enumerate(args.image_path): | |
| try: | |
| filename = os.path.basename(image_path).split(".")[0] | |
| output_root = args.output_root | |
| if args.image_root is not None or len(args.image_path) > 1: | |
| output_root = os.path.join(output_root, filename) | |
| os.makedirs(output_root, exist_ok=True) | |
| mesh_out = f"{output_root}/{filename}.obj" | |
| if args.skip_exists and os.path.exists(mesh_out): | |
| logger.warning( | |
| f"Skip {image_path}, already processed in {mesh_out}" | |
| ) | |
| continue | |
| image = Image.open(image_path) | |
| image.save(f"{output_root}/{filename}_raw.png") | |
| # Segmentation: Get segmented image using Rembg. | |
| seg_path = f"{output_root}/{filename}_cond.png" | |
| seg_image = RBG_REMOVER(image) if image.mode != "RGBA" else image | |
| seg_image.save(seg_path) | |
| if args.image3d_model == "HUNYUAN3D": | |
| process_image( | |
| args=args, | |
| idx=idx, | |
| image_path=image_path, | |
| output_root=output_root, | |
| filename=filename, | |
| hunyuan_config=hunyuan_config, | |
| hunyuan_credentials=hunyuan_credentials, | |
| checkers=CHECKERS, | |
| ) | |
| continue | |
| seed = args.seed | |
| asset_node = "unknown" | |
| gs_model = None | |
| if isinstance(args.asset_type, list) and args.asset_type[idx]: | |
| asset_node = args.asset_type[idx] | |
| for try_idx in range(args.n_retry): | |
| logger.info( | |
| f"Try: {try_idx + 1}/{args.n_retry}, Seed: {seed}, Prompt: {seg_path}" | |
| ) | |
| try: | |
| outputs = image3d_model_infer(pipeline, seg_image, seed) | |
| except Exception as e: | |
| logger.error( | |
| f"[Image3D Failed] process {image_path}: {e}, retry: {try_idx + 1}/{args.n_retry}" | |
| ) | |
| seed = ( | |
| random.randint(0, 100000) if seed is not None else None | |
| ) | |
| continue | |
| gs_model = outputs["gaussian"][0] | |
| mesh_model = outputs["mesh"][0] | |
| # Save the raw Gaussian model | |
| gs_path = mesh_out.replace(".obj", "_gs.ply") | |
| gs_model.save_ply(gs_path) | |
| # Rotate mesh and GS by 90 degrees around Z-axis. | |
| rot_matrix = [[0, 0, -1], [0, 1, 0], [1, 0, 0]] | |
| gs_add_rot = [[1, 0, 0], [0, -1, 0], [0, 0, -1]] | |
| mesh_add_rot = [[1, 0, 0], [0, 0, -1], [0, 1, 0]] | |
| # Addtional rotation for GS to align mesh. | |
| gs_rot = np.array(gs_add_rot) @ np.array(rot_matrix) | |
| pose = GaussianOperator.trans_to_quatpose(gs_rot) | |
| aligned_gs_path = gs_path.replace(".ply", "_aligned.ply") | |
| GaussianOperator.resave_ply( | |
| in_ply=gs_path, | |
| out_ply=aligned_gs_path, | |
| instance_pose=pose, | |
| device="cpu", | |
| ) | |
| color_path = os.path.join(output_root, "color.png") | |
| render_gs_api( | |
| input_gs=aligned_gs_path, | |
| output_path=color_path, | |
| elevation=[30, -30], | |
| num_images=4, | |
| ) | |
| color_img = Image.open(color_path) | |
| geo_flag, geo_result = GEO_CHECKER( | |
| [color_img], text=asset_node | |
| ) | |
| logger.warning( | |
| f"{GEO_CHECKER.__class__.__name__}: {geo_result} for {seg_path}" | |
| ) | |
| if geo_flag is True or geo_flag is None: | |
| break | |
| seed = random.randint(0, 100000) if seed is not None else None | |
| if gs_model is None: | |
| logger.error(f"Exceed image3d retry num, skip {image_path}.") | |
| continue | |
| # Render the video for generated 3D asset. | |
| color_images = render_video(gs_model, r=1.85)["color"] | |
| normal_images = render_video(mesh_model, r=1.85)["normal"] | |
| video_path = os.path.join(output_root, "gs_mesh.mp4") | |
| merge_images_video(color_images, normal_images, video_path) | |
| mesh = trimesh.Trimesh( | |
| vertices=mesh_model.vertices.cpu().numpy(), | |
| faces=mesh_model.faces.cpu().numpy(), | |
| ) | |
| mesh.vertices = mesh.vertices @ np.array(mesh_add_rot) | |
| mesh.vertices = mesh.vertices @ np.array(rot_matrix) | |
| mesh_obj_path = os.path.join(output_root, f"{filename}.obj") | |
| mesh.export(mesh_obj_path) | |
| mesh = backproject_api( | |
| # delight_model=DELIGHT, | |
| # imagesr_model=IMAGESR_MODEL, | |
| gs_path=aligned_gs_path, | |
| mesh_path=mesh_obj_path, | |
| output_path=mesh_obj_path, | |
| skip_fix_mesh=False, | |
| texture_size=args.texture_size, | |
| delight=False, | |
| ) | |
| mesh_glb_path = os.path.join(output_root, f"{filename}.glb") | |
| mesh.export(mesh_glb_path) | |
| urdf_convertor = URDFGenerator( | |
| GPT_CLIENT, | |
| render_view_num=4, | |
| decompose_convex=not args.disable_decompose_convex, | |
| ) | |
| asset_attrs = { | |
| "version": VERSION, | |
| "gs_model": f"{urdf_convertor.output_mesh_dir}/{filename}_gs.ply", | |
| } | |
| if args.height_range: | |
| min_height, max_height = map( | |
| float, args.height_range.split("-") | |
| ) | |
| asset_attrs["min_height"] = min_height | |
| asset_attrs["max_height"] = max_height | |
| if args.mass_range: | |
| min_mass, max_mass = map(float, args.mass_range.split("-")) | |
| asset_attrs["min_mass"] = min_mass | |
| asset_attrs["max_mass"] = max_mass | |
| if isinstance(args.asset_type, list) and args.asset_type[idx]: | |
| asset_attrs["category"] = args.asset_type[idx] | |
| if args.version: | |
| asset_attrs["version"] = args.version | |
| urdf_root = f"{output_root}/URDF_{filename}" | |
| urdf_path = urdf_convertor( | |
| mesh_path=mesh_obj_path, | |
| output_root=urdf_root, | |
| **asset_attrs, | |
| ) | |
| # Rescale GS and save to URDF/mesh folder. | |
| real_height = urdf_convertor.get_attr_from_urdf( | |
| urdf_path, attr_name="real_height" | |
| ) | |
| out_gs = f"{urdf_root}/{urdf_convertor.output_mesh_dir}/{filename}_gs.ply" # noqa | |
| GaussianOperator.resave_ply( | |
| in_ply=aligned_gs_path, | |
| out_ply=out_gs, | |
| real_height=real_height, | |
| device="cpu", | |
| ) | |
| # Quality check and update .urdf file. | |
| mesh_out = ( | |
| f"{urdf_root}/{urdf_convertor.output_mesh_dir}/{filename}.obj" # noqa | |
| ) | |
| trimesh.load(mesh_out).export(mesh_out.replace(".obj", ".glb")) | |
| image_dir = ( | |
| f"{urdf_root}/{urdf_convertor.output_render_dir}/image_color" # noqa | |
| ) | |
| image_paths = glob(f"{image_dir}/*.png") | |
| images_list = [] | |
| for checker in CHECKERS: | |
| images = combine_images_to_grid(image_paths) | |
| if isinstance(checker, ImageSegChecker): | |
| images = [ | |
| f"{output_root}/{filename}_raw.png", | |
| f"{output_root}/{filename}_cond.png", | |
| ] | |
| images_list.append(images) | |
| qa_results = BaseChecker.validate(CHECKERS, images_list) | |
| urdf_convertor.add_quality_tag(urdf_path, qa_results) | |
| # Organize the final result files | |
| result_dir = f"{output_root}/result" | |
| if os.path.exists(result_dir): | |
| rmtree(result_dir, ignore_errors=True) | |
| os.makedirs(result_dir, exist_ok=True) | |
| copy(urdf_path, f"{result_dir}/{os.path.basename(urdf_path)}") | |
| copytree( | |
| f"{urdf_root}/{urdf_convertor.output_mesh_dir}", | |
| f"{result_dir}/{urdf_convertor.output_mesh_dir}", | |
| ) | |
| copy(video_path, f"{result_dir}/video.mp4") | |
| if not args.keep_intermediate: | |
| delete_dir(output_root, keep_subs=["result"]) | |
| logger.info(f"Saved results for {image_path} in {result_dir}") | |
| except Exception as e: | |
| logger.error(f"Failed to process {image_path}: {e}, skip.") | |
| continue | |
| logger.info(f"Processing complete. Outputs saved to {args.output_root}") | |
| if __name__ == "__main__": | |
| entrypoint() | |