init

.gitattributes

@@ -33,3 +33,6 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+data/demo_glb/*.glb filter=lfs diff=lfs merge=lfs -text
+assets/*.jpg filter=lfs diff=lfs merge=lfs -text
+data/basic_shapes_norm_pc10000/*.ply filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,4 @@
+**/__pycache__/
+ckpt
+gradio_cached_examples
+results

app.py

@@ -0,0 +1,390 @@
+import os
+import time
+import glob
+import json
+import yaml
+import torch
+import trimesh
+import argparse
+import mesh2sdf.core
+import numpy as np
+import skimage.measure
+import seaborn as sns
+from scipy.spatial.transform import Rotation
+from mesh_to_sdf import get_surface_point_cloud
+from accelerate.utils import set_seed
+from accelerate import Accelerator
+from huggingface_hub.file_download import hf_hub_download
+from huggingface_hub import list_repo_files
+
+from primitive_anything.utils import path_mkdir, count_parameters
+from primitive_anything.utils.logger import print_log
+
+os.environ['PYOPENGL_PLATFORM'] = 'egl'
+
+import spaces
+
+repo_id = "hyz317/PrimitiveAnything"
+all_files = list_repo_files(repo_id, revision="main")
+for file in all_files:
+    if os.path.exists(file):
+        continue
+    hf_hub_download(repo_id, file, local_dir="./ckpt")
+hf_hub_download("Maikou/Michelangelo", "checkpoints/aligned_shape_latents/shapevae-256.ckpt", local_dir="./ckpt")
+
+def parse_args():
+    parser = argparse.ArgumentParser(description='Process 3D model files')
+    
+    parser.add_argument(
+        '--input',
+        type=str,
+        default='./data/demo_glb/',
+        help='Input file or directory path (default: ./data/demo_glb/)'
+    )
+    
+    parser.add_argument(
+        '--log_path',
+        type=str,
+        default='./results/demo',
+        help='Output directory path (default: results/demo)'
+    )
+    
+    return parser.parse_args()
+
+def get_input_files(input_path):
+    if os.path.isfile(input_path):
+        return [input_path]
+    elif os.path.isdir(input_path):
+        return glob.glob(os.path.join(input_path, '*'))
+    else:
+        raise ValueError(f"Input path {input_path} is neither a file nor a directory")
+
+args = parse_args()
+
+# Create output directory (keeping your original variable name)
+LOG_PATH = args.log_path
+os.makedirs(LOG_PATH, exist_ok=True)
+
+print(f"Output directory: {LOG_PATH}")
+
+CODE_SHAPE = {
+    0: 'SM_GR_BS_CubeBevel_001.ply',
+    1: 'SM_GR_BS_SphereSharp_001.ply',
+    2: 'SM_GR_BS_CylinderSharp_001.ply',
+}
+
+shapename_map = {
+    'SM_GR_BS_CubeBevel_001.ply': 1101002001034001,
+    'SM_GR_BS_SphereSharp_001.ply': 1101002001034010,
+    'SM_GR_BS_CylinderSharp_001.ply': 1101002001034002,
+}
+
+#### config
+bs_dir = 'data/basic_shapes_norm'
+config_path = './configs/infer.yml'
+AR_checkpoint_path = './ckpt/mesh-transformer.ckpt.60.pt'
+temperature= 0.0
+#### init model
+mesh_bs = {}
+for bs_path in glob.glob(os.path.join(bs_dir, '*.ply')):
+    bs_name = os.path.basename(bs_path)
+    bs = trimesh.load(bs_path)
+    bs.visual.uv = np.clip(bs.visual.uv, 0, 1)
+    bs.visual = bs.visual.to_color()
+    mesh_bs[bs_name] = bs
+
+def create_model(cfg_model):
+    kwargs = cfg_model
+    name = kwargs.pop('name')
+    model = get_model(name)(**kwargs)
+    print_log("Model '{}' init: nb_params={:,}, kwargs={}".format(name, count_parameters(model), kwargs))
+    return model
+
+from primitive_anything.primitive_transformer import PrimitiveTransformerDiscrete
+def get_model(name):
+    return {
+        'discrete': PrimitiveTransformerDiscrete,
+    }[name]
+
+with open(config_path, mode='r') as fp:
+    AR_train_cfg = yaml.load(fp, Loader=yaml.FullLoader)
+
+AR_checkpoint = torch.load(AR_checkpoint_path)
+
+transformer = create_model(AR_train_cfg['model'])
+transformer.load_state_dict(AR_checkpoint)
+
+device = torch.device('cuda')
+accelerator = Accelerator(
+    mixed_precision='fp16',
+)
+transformer = accelerator.prepare(transformer)
+transformer.eval()
+transformer.bs_pc = transformer.bs_pc.cuda()
+transformer.rotation_matrix_align_coord = transformer.rotation_matrix_align_coord.cuda()
+print('model loaded to device')
+
+
+def sample_surface_points(mesh, number_of_points=500000, surface_point_method='scan', sign_method='normal',
+                          scan_count=100, scan_resolution=400, sample_point_count=10000000, return_gradients=False,
+                          return_surface_pc_normals=False, normalized=False):
+    sample_start = time.time()
+    if surface_point_method == 'sample' and sign_method == 'depth':
+        print("Incompatible methods for sampling points and determining sign, using sign_method='normal' instead.")
+        sign_method = 'normal'
+
+    surface_start = time.time()
+    bound_radius = 1 if normalized else None
+    surface_point_cloud = get_surface_point_cloud(mesh, surface_point_method, bound_radius, scan_count, scan_resolution,
+                                                  sample_point_count,
+                                                  calculate_normals=sign_method == 'normal' or return_gradients)
+
+    surface_end = time.time()
+    print('surface point cloud time cost :', surface_end - surface_start)
+
+    normal_start = time.time()
+    if return_surface_pc_normals:
+        rng = np.random.default_rng()
+        assert surface_point_cloud.points.shape[0] == surface_point_cloud.normals.shape[0]
+        indices = rng.choice(surface_point_cloud.points.shape[0], number_of_points, replace=True)
+        points = surface_point_cloud.points[indices]
+        normals = surface_point_cloud.normals[indices]
+        surface_points = np.concatenate([points, normals], axis=-1)
+    else:
+        surface_points = surface_point_cloud.get_random_surface_points(number_of_points, use_scans=True)
+    normal_end = time.time()
+    print('normal time cost :', normal_end - normal_start)
+    sample_end = time.time()
+    print('sample surface point time cost :', sample_end - sample_start)
+    return surface_points
+
+
+def normalize_vertices(vertices, scale=0.9):
+    bbmin, bbmax = vertices.min(0), vertices.max(0)
+    center = (bbmin + bbmax) * 0.5
+    scale = 2.0 * scale / (bbmax - bbmin).max()
+    vertices = (vertices - center) * scale
+    return vertices, center, scale
+
+
+def export_to_watertight(normalized_mesh, octree_depth: int = 7):
+    """
+        Convert the non-watertight mesh to watertight.
+
+        Args:
+            input_path (str): normalized path
+            octree_depth (int):
+
+        Returns:
+            mesh(trimesh.Trimesh): watertight mesh
+
+        """
+    size = 2 ** octree_depth
+    level = 2 / size
+
+    scaled_vertices, to_orig_center, to_orig_scale = normalize_vertices(normalized_mesh.vertices)
+    sdf = mesh2sdf.core.compute(scaled_vertices, normalized_mesh.faces, size=size)
+    vertices, faces, normals, _ = skimage.measure.marching_cubes(np.abs(sdf), level)
+
+    # watertight mesh
+    vertices = vertices / size * 2 - 1 # -1 to 1
+    vertices = vertices / to_orig_scale + to_orig_center
+    mesh = trimesh.Trimesh(vertices, faces, normals=normals)
+
+    return mesh
+
+
+def process_mesh_to_surface_pc(mesh_list, marching_cubes=False, dilated_offset=0.0, sample_num=10000):
+    # mesh_list : list of trimesh
+    pc_normal_list = []
+    return_mesh_list = []
+    for mesh in mesh_list:
+        if marching_cubes:
+            mesh = export_to_watertight(mesh)
+            print("MC over!")
+        if dilated_offset > 0:
+            new_vertices = mesh.vertices + mesh.vertex_normals * dilated_offset
+            mesh.vertices = new_vertices
+            print("dilate over!")
+
+        mesh.merge_vertices()
+        mesh.update_faces(mesh.unique_faces())
+        mesh.fix_normals()
+
+        return_mesh_list.append(mesh)
+
+        pc_normal = np.asarray(sample_surface_points(mesh, sample_num, return_surface_pc_normals=True))
+        pc_normal_list.append(pc_normal)
+        print("process mesh success")
+    return pc_normal_list, return_mesh_list
+
+
+####    utils
+def euler_to_quat(euler):
+    return Rotation.from_euler('XYZ', euler, degrees=True).as_quat()
+
+def SRT_quat_to_matrix(scale, quat, translation):
+    rotation_matrix = Rotation.from_quat(quat).as_matrix()
+    transform_matrix = np.eye(4)
+    transform_matrix[:3, :3] = rotation_matrix * scale
+    transform_matrix[:3, 3] = translation
+    return transform_matrix
+
+
+def write_output(primitives, name):
+    out_json = {}
+    out_json['operation'] = 0
+    out_json['type'] = 1
+    out_json['scene_id'] = None
+
+    new_group = []
+    model_scene = trimesh.Scene()
+    color_map = sns.color_palette("hls", primitives['type_code'].squeeze().shape[0])
+    color_map = (np.array(color_map) * 255).astype("uint8")
+    for idx, (scale, rotation, translation, type_code) in enumerate(zip(
+        primitives['scale'].squeeze().cpu().numpy(),
+        primitives['rotation'].squeeze().cpu().numpy(),
+        primitives['translation'].squeeze().cpu().numpy(),
+        primitives['type_code'].squeeze().cpu().numpy()
+    )):
+        if type_code == -1:
+            break
+        bs_name = CODE_SHAPE[type_code]
+        new_block = {}
+        new_block['type_id'] = shapename_map[bs_name]
+        new_block['data'] = {}
+        new_block['data']['location'] = translation.tolist()
+        new_block['data']['rotation'] = euler_to_quat(rotation).tolist()
+        new_block['data']['scale'] = scale.tolist()
+        new_block['data']['color'] = ['808080']
+        new_group.append(new_block)
+
+        trans = SRT_quat_to_matrix(scale, euler_to_quat(rotation), translation)
+        bs = mesh_bs[bs_name].copy().apply_transform(trans)
+        new_vertex_colors = np.repeat(color_map[idx:idx+1], bs.visual.vertex_colors.shape[0], axis=0)
+        bs.visual.vertex_colors[:, :3] = new_vertex_colors
+        vertices = bs.vertices.copy()
+        vertices[:, 1] = bs.vertices[:, 2]
+        vertices[:, 2] = -bs.vertices[:, 1]
+        bs.vertices = vertices
+        model_scene.add_geometry(bs)
+    out_json['group'] = new_group
+
+    json_path = os.path.join(LOG_PATH, f'output_{name}.json')
+    with open(json_path, 'w') as json_file:
+        json.dump(out_json, json_file, indent=4)
+
+    glb_path = os.path.join(LOG_PATH, f'output_{name}.glb')
+    model_scene.export(glb_path)
+
+    return glb_path, out_json
+
+
+@torch.no_grad()
+def do_inference(input_3d, dilated_offset=0.0, sample_seed=0, do_sampling=False, do_marching_cubes=False, postprocess='none'):
+    t1 = time.time()
+    set_seed(sample_seed)
+    input_mesh = trimesh.load(input_3d, force='mesh')
+
+    # scale mesh
+    vertices = input_mesh.vertices
+    bounds = np.array([vertices.min(axis=0), vertices.max(axis=0)])
+    vertices = vertices - (bounds[0] + bounds[1])[None, :] / 2
+    vertices = vertices / (bounds[1] - bounds[0]).max() * 1.6
+    input_mesh.vertices = vertices
+
+    pc_list, mesh_list = process_mesh_to_surface_pc(
+        [input_mesh],
+        marching_cubes=do_marching_cubes,
+        dilated_offset=dilated_offset
+    )
+    pc_normal = pc_list[0] # 10000, 6
+    mesh = mesh_list[0]
+
+    pc_coor = pc_normal[:, :3]
+    normals = pc_normal[:, 3:]
+
+    if dilated_offset > 0:
+        # scale mesh and pc
+        vertices = mesh.vertices
+        bounds = np.array([vertices.min(axis=0), vertices.max(axis=0)])
+        vertices = vertices - (bounds[0] + bounds[1])[None, :] / 2
+        vertices = vertices / (bounds[1] - bounds[0]).max() * 1.6
+        mesh.vertices = vertices
+        pc_coor = pc_coor - (bounds[0] + bounds[1])[None, :] / 2
+        pc_coor = pc_coor / (bounds[1] - bounds[0]).max() * 1.6
+
+    input_save_name = os.path.join(LOG_PATH, f'processed_{os.path.basename(input_3d)}')
+    mesh.export(input_save_name)
+
+    assert (np.linalg.norm(normals, axis=-1) > 0.99).all(), 'normals should be unit vectors, something wrong'
+    normalized_pc_normal = np.concatenate([pc_coor, normals], axis=-1, dtype=np.float16)
+
+    input_pc = torch.tensor(normalized_pc_normal, dtype=torch.float16, device=device)[None]
+
+    with accelerator.autocast():
+        if postprocess == 'postprocess1':
+            recon_primitives, mask = transformer.generate_w_recon_loss(pc=input_pc, temperature=temperature, single_directional=True)
+        else:
+            recon_primitives, mask = transformer.generate(pc=input_pc, temperature=temperature)
+
+    output_glb, output_json = write_output(recon_primitives, os.path.basename(input_3d)[:-4])
+
+    return input_save_name, output_glb, output_json
+
+
+import gradio as gr
+
+@spaces.GPU
+def process_3d_model(input_3d, dilated_offset, do_marching_cubes, postprocess_method="postprocess1"):
+    print(f"processing: {input_3d}")
+    # try:
+    preprocess_model_obj, output_model_obj, output_model_json = do_inference(
+        input_3d,
+        dilated_offset=dilated_offset,
+        do_marching_cubes=do_marching_cubes,
+        postprocess=postprocess_method
+    )
+    return output_model_obj
+    # except Exception as e:
+    #     return f"Error processing file: {str(e)}"
+
+# Title and reminder placeholders
+title = "3D Model Processing Demo"
+reminder = "Please upload your 3D model file and adjust parameters as needed."
+
+with gr.Blocks(title=title) as demo:
+    # Title section
+    gr.Markdown(f"# {title}")
+    gr.Markdown(reminder)
+    
+    with gr.Row():
+        with gr.Column():
+            # Input components
+            input_3d = gr.Model3D(label="Upload 3D Model File")
+            dilated_offset = gr.Number(label="Dilated Offset", value=0.015)
+            do_marching_cubes = gr.Checkbox(label="Perform Marching Cubes", value=True)
+            submit_btn = gr.Button("Process Model")
+            
+        with gr.Column():
+            # Output components
+            output = gr.Model3D(label="Primitive Assembly Predition")
+            
+    submit_btn.click(
+        fn=process_3d_model,
+        inputs=[input_3d, dilated_offset, do_marching_cubes],
+        outputs=output
+    )
+    
+
+    # Prepare examples properly
+    example_files = [ [f] for f in glob.glob('./data/demo_glb/*.glb') ]  # Note: wrapped in list and filtered for GLB
+
+    example = gr.Examples(
+        examples=example_files,
+        inputs=[input_3d],  # Only include the Model3D input
+        examples_per_page=14,
+    )
+
+if __name__ == "__main__":
+    demo.launch()

configs/infer.yml

@@ -0,0 +1,52 @@
+dataset:
+  name: base
+  pc_dir: ./data/test_pc
+  bs_dir: data/basic_shapes_norm
+  max_length: 144
+  range_scale: [0, 1]
+  range_rotation: [-180, 180]
+  range_translation: [-1, 1]
+  rotation_type: euler
+  pc_format: pn
+model:
+  attn_depth: 6
+  attn_heads: 6
+  bin_smooth_blur_sigma: -1
+  bs_pc_dir: data/basic_shapes_norm_pc10000
+  coarse_pre_gateloop_depth: 3
+  continuous_range_rotation:
+  - -181
+  - 181
+  continuous_range_scale:
+  - 0
+  - 1
+  continuous_range_translation:
+  - -1
+  - 1
+  dim: 768
+  dim_rotation_embed: 16
+  dim_scale_embed: 16
+  dim_translation_embed: 16
+  dim_type_embed: 48
+  dropout: 0.0
+  embed_order: ctrs
+  gateloop_use_heinsen: false
+  loss_weight:
+    eos: 1.0
+    reconstruction: 1.0
+    rotation: 1.0
+    scale: 1.0
+    translation: 1.0
+    type: 1.0
+  max_primitive_len: 144
+  name: discrete
+  num_discrete_rotation: 181
+  num_discrete_scale: 128
+  num_discrete_translation: 128
+  num_type: 3
+  shape_cond_with_cat: true
+  shape_cond_with_cross_attn: false
+  shape_cond_with_film: false
+  shape_condition_dim: 768
+  shape_condition_len: 77
+  shape_condition_model_type: michelangelo

data/basic_shapes_norm/basic_shapes.json

@@ -0,0 +1,89 @@
+{
+    "SM_GR_BS_CubeBevel_001.ply": {
+        "name": "SM_GR_BS_CubeBevel_001.ply",
+        "tform_bs_to_normalized": [
+            [
+                0.02,
+                0.0,
+                0.0,
+                0.0
+            ],
+            [
+                0.0,
+                0.02,
+                0.0,
+                9.701276818911235e-18
+            ],
+            [
+                0.0,
+                0.0,
+                0.019999999999999997,
+                -0.9999999999999999
+            ],
+            [
+                0.0,
+                0.0,
+                0.0,
+                1.0
+            ]
+        ]
+    },
+    "SM_GR_BS_CylinderSharp_001.ply": {
+        "name": "SM_GR_BS_CylinderSharp_001.ply",
+        "tform_bs_to_normalized": [
+            [
+                0.006666668023003748,
+                0.0,
+                0.0,
+                -2.0345056221459462e-07
+            ],
+            [
+                0.0,
+                0.006666667683919426,
+                0.0,
+                -5.086263794939386e-08
+            ],
+            [
+                0.0,
+                0.0,
+                0.006666665445429783,
+                -0.9999998370794186
+            ],
+            [
+                0.0,
+                0.0,
+                0.0,
+                1.0
+            ]
+        ]
+    },
+    "SM_GR_BS_SphereSharp_001.ply": {
+        "name": "SM_GR_BS_SphereSharp_001.ply",
+        "tform_bs_to_normalized": [
+            [
+                0.006666666666666667,
+                0.0,
+                0.0,
+                0.0
+            ],
+            [
+                0.0,
+                0.006666666666666667,
+                0.0,
+                0.0
+            ],
+            [
+                0.0,
+                0.0,
+                0.006666666666666667,
+                -1.0
+            ],
+            [
+                0.0,
+                0.0,
+                0.0,
+                1.0
+            ]
+        ]
+    }
+}

data/basic_shapes_norm_pc10000/SM_GR_BS_CubeBevel_001.ply

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:ba980c1fb389e30783f09b07d35e788e08a97776d933b8bfd346147c9a7e86a0
+size 510265

data/basic_shapes_norm_pc10000/SM_GR_BS_CylinderSharp_001.ply

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:ab8fb7aa7ec39237474d0a6e77da1d7070742f61af21e6b44dc9998fac1913cc
+size 510265

data/basic_shapes_norm_pc10000/SM_GR_BS_SphereSharp_001.ply

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:8765da7294292422d077267c1b71b9ea055f831aab3840d869656632ee6e8569
+size 510265

data/demo_glb/barbell.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:8a9b9c124c321d6d18342b12407fc7327bdd56c8720d317e7b8694c10c851936
+size 769528

data/demo_glb/book.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:be526e9bca2ce3a74387f2dde7f6a25c9502a7c6d4f9fc671b244d09c18a9d94
+size 5369916

data/demo_glb/bunny.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:f62b2169b7cda3662de660d0b2e8ce2a1ccfe3bd186f243d890440e8cf7a0766
+size 27518016

data/demo_glb/desk.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:4c8fd8041f1e870ba285572f3fb3e129107678ab9a311524a8376cc404cc332e
+size 33679548

data/demo_glb/man.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:063a56d0a56d3866cf36170bbafad93924fd350882ac0f69e727cb43dc203351
+size 31784

data/demo_glb/micky.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:fc6410c2f2a588c5b064f9255a1c1657a9dff061ae6f7342df693c80eef0c69d
+size 294576

data/demo_glb/pac.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:dc39816cc71440fbc31d99c24f24ed42c25daf7319ba07b8dd3e34c1ea083578
+size 274004

data/demo_glb/robot.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:cf201fbe21d73428f88e4a7e428849148ebd500cc4ce6ac3929638a53c5376ae
+size 28116940

data/demo_glb/rocket.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:29601d218e0d51a4dced2f1ed2a898e80f0d223b1e04212d45b0dda4ad670d1c
+size 1426588

data/demo_glb/sheep.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:232b1303e56ec1682536c72bc9409585930985492dcbdfa101cdfb96d0b4fbf2
+size 28732

data/demo_glb/shelf.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d63916c1ef1d5b2fc4d56e20c76316bd977f93819323f73e7f5e1c59df21e284
+size 3091336

data/demo_glb/table.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:743c96d7aa1bef88576c5f28d5e06144f93e27a2e6ea5ef8bd85669d1213af9f
+size 20093692

data/demo_glb/vent.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:98d8d0d2c8d164fc75361d7d1408e9102f9148482c0343e0cc87d21950e20ab1
+size 1785468

data/demo_glb/walkman.glb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:992355cf45609881223561d1081a05483c2bad488ed7148f87243259aa36be1b
+size 158156

pre-requirements.txt

@@ -0,0 +1,36 @@
+--extra-index-url https://download.pytorch.org/whl/cu121
+--extra-index-url https://data.dgl.ai/wheels/torch-2.1/cu121/repo.html
+torch==2.2.0
+torchvision==0.17.0
+dgl
+accelerate
+beartype
+einops
+gateloop_transformer
+matplotlib
+scikit-learn
+pandas
+pytorch_custom_utils
+gradio
+pydantic==2.10.6
+x_transformers
+torch_redstone
+torchdata==0.9.0
+toolz
+environs
+jaxtyping
+omegaconf
+ema_pytorch
+local_attention==1.9.15
+taylor_series_linear_attention
+transformers
+vector_quantize_pytorch
+open3d
+trimesh
+pytorch_lightning
+scikit-image
+opencv-python
+mesh2sdf
+seaborn
+mesh_to_sdf
+point_cloud_utils

primitive_anything/michelangelo/__init__.py

@@ -0,0 +1,51 @@
+import os
+
+from omegaconf import OmegaConf
+import torch
+from torch import nn
+
+from .utils.misc import instantiate_from_config
+from ..utils import default, exists
+
+
+def load_model():
+    model_config = OmegaConf.load(os.path.join(os.path.dirname(__file__), "shapevae-256.yaml"))
+    # print(model_config)
+    if hasattr(model_config, "model"):
+        model_config = model_config.model
+    ckpt_path = "./ckpt/shapevae-256.ckpt"
+
+    model = instantiate_from_config(model_config, ckpt_path=ckpt_path)
+    # model = model.cuda()
+    model = model.eval()
+
+    return model
+
+
+class ShapeConditioner(nn.Module):
+    def __init__(
+        self,
+        *,
+        dim_latent = None
+    ):
+        super().__init__()
+        self.model = load_model()
+
+        self.dim_model_out = 768
+        dim_latent = default(dim_latent, self.dim_model_out)
+        self.dim_latent = dim_latent
+
+    def forward(
+        self,
+        shape = None,
+        shape_embed = None,
+    ):
+        assert exists(shape) ^ exists(shape_embed)
+
+        if not exists(shape_embed):
+            point_feature = self.model.encode_latents(shape)
+            shape_latents = self.model.to_shape_latents(point_feature[:, 1:])
+            shape_head = point_feature[:, 0:1]
+            shape_embed = torch.cat([point_feature[:, 1:], shape_latents], dim=-1)
+            # shape_embed = torch.cat([point_feature[:, 1:], shape_latents], dim=-2) # cat tmp
+        return shape_head, shape_embed

primitive_anything/michelangelo/data/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

primitive_anything/michelangelo/data/templates.json

@@ -0,0 +1,69 @@
+{
+    "shape": [
+        "a point cloud model of {}.",
+        "There is a {} in the scene.",
+        "There is the {} in the scene.",
+        "a photo of a {} in the scene.",
+        "a photo of the {} in the scene.",
+        "a photo of one {} in the scene.",
+        "itap of a {}.",
+        "itap of my {}.",
+        "itap of the {}.",
+        "a photo of a {}.",
+        "a photo of my {}.",
+        "a photo of the {}.",
+        "a photo of one {}.",
+        "a photo of many {}.",
+        "a good photo of a {}.",
+        "a good photo of the {}.",
+        "a bad photo of a {}.",
+        "a bad photo of the {}.",
+        "a photo of a nice {}.",
+        "a photo of the nice {}.",
+        "a photo of a cool {}.",
+        "a photo of the cool {}.",
+        "a photo of a weird {}.",
+        "a photo of the weird {}.",
+        "a photo of a small {}.",
+        "a photo of the small {}.",
+        "a photo of a large {}.",
+        "a photo of the large {}.",
+        "a photo of a clean {}.",
+        "a photo of the clean {}.",
+        "a photo of a dirty {}.",
+        "a photo of the dirty {}.",
+        "a bright photo of a {}.",
+        "a bright photo of the {}.",
+        "a dark photo of a {}.",
+        "a dark photo of the {}.",
+        "a photo of a hard to see {}.",
+        "a photo of the hard to see {}.",
+        "a low resolution photo of a {}.",
+        "a low resolution photo of the {}.",
+        "a cropped photo of a {}.",
+        "a cropped photo of the {}.",
+        "a close-up photo of a {}.",
+        "a close-up photo of the {}.",
+        "a jpeg corrupted photo of a {}.",
+        "a jpeg corrupted photo of the {}.",
+        "a blurry photo of a {}.",
+        "a blurry photo of the {}.",
+        "a pixelated photo of a {}.",
+        "a pixelated photo of the {}.",
+        "a black and white photo of the {}.",
+        "a black and white photo of a {}",
+        "a plastic {}.",
+        "the plastic {}.",
+        "a toy {}.",
+        "the toy {}.",
+        "a plushie {}.",
+        "the plushie {}.",
+        "a cartoon {}.",
+        "the cartoon {}.",
+        "an embroidered {}.",
+        "the embroidered {}.",
+        "a painting of the {}.",
+        "a painting of a {}."
+    ]
+
+}

primitive_anything/michelangelo/data/transforms.py

@@ -0,0 +1,407 @@
+# -*- coding: utf-8 -*-
+import os
+import time
+import numpy as np
+import warnings
+import random
+from omegaconf.listconfig import ListConfig
+from webdataset import pipelinefilter
+import torch
+import torchvision.transforms.functional as TVF
+from torchvision.transforms import InterpolationMode
+from torchvision.transforms.transforms import _interpolation_modes_from_int
+from typing import Sequence
+
+from ..utils import instantiate_from_config
+
+
+def _uid_buffer_pick(buf_dict, rng):
+    uid_keys = list(buf_dict.keys())
+    selected_uid = rng.choice(uid_keys)
+    buf = buf_dict[selected_uid]
+
+    k = rng.randint(0, len(buf) - 1)
+    sample = buf[k]
+    buf[k] = buf[-1]
+    buf.pop()
+
+    if len(buf) == 0:
+        del buf_dict[selected_uid]
+
+    return sample
+
+
+def _add_to_buf_dict(buf_dict, sample):
+    key = sample["__key__"]
+    uid, uid_sample_id = key.split("_")
+    if uid not in buf_dict:
+        buf_dict[uid] = []
+    buf_dict[uid].append(sample)
+
+    return buf_dict
+
+
+def _uid_shuffle(data, bufsize=1000, initial=100, rng=None, handler=None):
+    """Shuffle the data in the stream.
+
+    This uses a buffer of size `bufsize`. Shuffling at
+    startup is less random; this is traded off against
+    yielding samples quickly.
+
+    data: iterator
+    bufsize: buffer size for shuffling
+    returns: iterator
+    rng: either random module or random.Random instance
+
+    """
+    if rng is None:
+        rng = random.Random(int((os.getpid() + time.time()) * 1e9))
+    initial = min(initial, bufsize)
+    buf_dict = dict()
+    current_samples = 0
+    for sample in data:
+        _add_to_buf_dict(buf_dict, sample)
+        current_samples += 1
+
+        if current_samples < bufsize:
+            try:
+                _add_to_buf_dict(buf_dict, next(data))  # skipcq: PYL-R1708
+                current_samples += 1
+            except StopIteration:
+                pass
+
+        if current_samples >= initial:
+            current_samples -= 1
+            yield _uid_buffer_pick(buf_dict, rng)
+
+    while current_samples > 0:
+        current_samples -= 1
+        yield _uid_buffer_pick(buf_dict, rng)
+
+
+uid_shuffle = pipelinefilter(_uid_shuffle)
+
+
+class RandomSample(object):
+    def __init__(self,
+                 num_volume_samples: int = 1024,
+                 num_near_samples: int = 1024):
+
+        super().__init__()
+
+        self.num_volume_samples = num_volume_samples
+        self.num_near_samples = num_near_samples
+
+    def __call__(self, sample):
+        rng = np.random.default_rng()
+
+        # 1. sample surface input
+        total_surface = sample["surface"]
+        ind = rng.choice(total_surface.shape[0], replace=False)
+        surface = total_surface[ind]
+
+        # 2. sample volume/near geometric points
+        vol_points = sample["vol_points"]
+        vol_label = sample["vol_label"]
+        near_points = sample["near_points"]
+        near_label = sample["near_label"]
+
+        ind = rng.choice(vol_points.shape[0], self.num_volume_samples, replace=False)
+        vol_points = vol_points[ind]
+        vol_label = vol_label[ind]
+        vol_points_labels = np.concatenate([vol_points, vol_label[:, np.newaxis]], axis=1)
+
+        ind = rng.choice(near_points.shape[0], self.num_near_samples, replace=False)
+        near_points = near_points[ind]
+        near_label = near_label[ind]
+        near_points_labels = np.concatenate([near_points, near_label[:, np.newaxis]], axis=1)
+
+        # concat sampled volume and near points
+        geo_points = np.concatenate([vol_points_labels, near_points_labels], axis=0)
+
+        sample = {
+            "surface": surface,
+            "geo_points": geo_points
+        }
+
+        return sample
+
+
+class SplitRandomSample(object):
+    def __init__(self,
+                 use_surface_sample: bool = False,
+                 num_surface_samples: int = 4096,
+                 num_volume_samples: int = 1024,
+                 num_near_samples: int = 1024):
+
+        super().__init__()
+
+        self.use_surface_sample = use_surface_sample
+        self.num_surface_samples = num_surface_samples
+        self.num_volume_samples = num_volume_samples
+        self.num_near_samples = num_near_samples
+
+    def __call__(self, sample):
+
+        rng = np.random.default_rng()
+
+        # 1. sample surface input
+        surface = sample["surface"]
+
+        if self.use_surface_sample:
+            replace = surface.shape[0] < self.num_surface_samples
+            ind = rng.choice(surface.shape[0], self.num_surface_samples, replace=replace)
+            surface = surface[ind]
+
+        # 2. sample volume/near geometric points
+        vol_points = sample["vol_points"]
+        vol_label = sample["vol_label"]
+        near_points = sample["near_points"]
+        near_label = sample["near_label"]
+
+        ind = rng.choice(vol_points.shape[0], self.num_volume_samples, replace=False)
+        vol_points = vol_points[ind]
+        vol_label = vol_label[ind]
+        vol_points_labels = np.concatenate([vol_points, vol_label[:, np.newaxis]], axis=1)
+
+        ind = rng.choice(near_points.shape[0], self.num_near_samples, replace=False)
+        near_points = near_points[ind]
+        near_label = near_label[ind]
+        near_points_labels = np.concatenate([near_points, near_label[:, np.newaxis]], axis=1)
+
+        # concat sampled volume and near points
+        geo_points = np.concatenate([vol_points_labels, near_points_labels], axis=0)
+
+        sample = {
+            "surface": surface,
+            "geo_points": geo_points
+        }
+
+        return sample
+
+
+class FeatureSelection(object):
+
+    VALID_SURFACE_FEATURE_DIMS = {
+        "none": [0, 1, 2],                              # xyz
+        "watertight_normal": [0, 1, 2, 3, 4, 5],        # xyz, normal
+        "normal": [0, 1, 2, 6, 7, 8]
+    }
+
+    def __init__(self, surface_feature_type: str):
+
+        self.surface_feature_type = surface_feature_type
+        self.surface_dims = self.VALID_SURFACE_FEATURE_DIMS[surface_feature_type]
+
+    def __call__(self, sample):
+        sample["surface"] = sample["surface"][:, self.surface_dims]
+        return sample
+
+
+class AxisScaleTransform(object):
+    def __init__(self, interval=(0.75, 1.25), jitter=True, jitter_scale=0.005):
+        assert isinstance(interval, (tuple, list, ListConfig))
+        self.interval = interval
+        self.min_val = interval[0]
+        self.max_val = interval[1]
+        self.inter_size = interval[1] - interval[0]
+        self.jitter = jitter
+        self.jitter_scale = jitter_scale
+
+    def __call__(self, sample):
+
+        surface = sample["surface"][..., 0:3]
+        geo_points = sample["geo_points"][..., 0:3]
+
+        scaling = torch.rand(1, 3) * self.inter_size + self.min_val
+        # print(scaling)
+        surface = surface * scaling
+        geo_points = geo_points * scaling
+
+        scale = (1 / torch.abs(surface).max().item()) * 0.999999
+        surface *= scale
+        geo_points *= scale
+
+        if self.jitter:
+            surface += self.jitter_scale * torch.randn_like(surface)
+            surface.clamp_(min=-1.015, max=1.015)
+
+        sample["surface"][..., 0:3] = surface
+        sample["geo_points"][..., 0:3] = geo_points
+
+        return sample
+
+
+class ToTensor(object):
+
+    def __init__(self, tensor_keys=("surface", "geo_points", "tex_points")):
+        self.tensor_keys = tensor_keys
+
+    def __call__(self, sample):
+        for key in self.tensor_keys:
+            if key not in sample:
+                continue
+
+            sample[key] = torch.tensor(sample[key], dtype=torch.float32)
+
+        return sample
+
+
+class AxisScale(object):
+    def __init__(self, interval=(0.75, 1.25), jitter=True, jitter_scale=0.005):
+        assert isinstance(interval, (tuple, list, ListConfig))
+        self.interval = interval
+        self.jitter = jitter
+        self.jitter_scale = jitter_scale
+
+    def __call__(self, surface, *args):
+        scaling = torch.rand(1, 3) * 0.5 + 0.75
+        # print(scaling)
+        surface = surface * scaling
+        scale = (1 / torch.abs(surface).max().item()) * 0.999999
+        surface *= scale
+
+        args_outputs = []
+        for _arg in args:
+            _arg = _arg * scaling * scale
+            args_outputs.append(_arg)
+
+        if self.jitter:
+            surface += self.jitter_scale * torch.randn_like(surface)
+            surface.clamp_(min=-1, max=1)
+
+        if len(args) == 0:
+            return surface
+        else:
+            return surface, *args_outputs
+
+
+class RandomResize(torch.nn.Module):
+    """Apply randomly Resize with a given probability."""
+
+    def __init__(
+        self,
+        size,
+        resize_radio=(0.5, 1),
+        allow_resize_interpolations=(InterpolationMode.BICUBIC, InterpolationMode.BILINEAR, InterpolationMode.BILINEAR),
+        interpolation=InterpolationMode.BICUBIC,
+        max_size=None,
+        antialias=None,
+    ):
+        super().__init__()
+        if not isinstance(size, (int, Sequence)):
+            raise TypeError(f"Size should be int or sequence. Got {type(size)}")
+        if isinstance(size, Sequence) and len(size) not in (1, 2):
+            raise ValueError("If size is a sequence, it should have 1 or 2 values")
+
+        self.size = size
+        self.max_size = max_size
+        # Backward compatibility with integer value
+        if isinstance(interpolation, int):
+            warnings.warn(
+                "Argument 'interpolation' of type int is deprecated since 0.13 and will be removed in 0.15. "
+                "Please use InterpolationMode enum."
+            )
+            interpolation = _interpolation_modes_from_int(interpolation)
+
+        self.interpolation = interpolation
+        self.antialias = antialias
+
+        self.resize_radio = resize_radio
+        self.allow_resize_interpolations = allow_resize_interpolations
+
+    def random_resize_params(self):
+        radio = torch.rand(1) * (self.resize_radio[1] - self.resize_radio[0]) + self.resize_radio[0]
+
+        if isinstance(self.size, int):
+            size = int(self.size * radio)
+        elif isinstance(self.size, Sequence):
+            size = list(self.size)
+            size = (int(size[0] * radio), int(size[1] * radio))
+        else:
+            raise RuntimeError()
+
+        interpolation = self.allow_resize_interpolations[
+            torch.randint(low=0, high=len(self.allow_resize_interpolations), size=(1,))
+        ]
+        return size, interpolation
+
+    def forward(self, img):
+        size, interpolation = self.random_resize_params()
+        img = TVF.resize(img, size, interpolation, self.max_size, self.antialias)
+        img = TVF.resize(img, self.size, self.interpolation, self.max_size, self.antialias)
+        return img
+
+    def __repr__(self) -> str:
+        detail = f"(size={self.size}, interpolation={self.interpolation.value},"
+        detail += f"max_size={self.max_size}, antialias={self.antialias}), resize_radio={self.resize_radio}"
+        return f"{self.__class__.__name__}{detail}"
+
+
+class Compose(object):
+    """Composes several transforms together. This transform does not support torchscript.
+    Please, see the note below.
+
+    Args:
+        transforms (list of ``Transform`` objects): list of transforms to compose.
+
+    Example:
+        >>> transforms.Compose([
+        >>>     transforms.CenterCrop(10),
+        >>>     transforms.ToTensor(),
+        >>> ])
+
+    .. note::
+        In order to script the transformations, please use ``torch.nn.Sequential`` as below.
+
+        >>> transforms = torch.nn.Sequential(
+        >>>     transforms.CenterCrop(10),
+        >>>     transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225)),
+        >>> )
+        >>> scripted_transforms = torch.jit.script(transforms)
+
+        Make sure to use only scriptable transformations, i.e. that work with ``torch.Tensor``, does not require
+        `lambda` functions or ``PIL.Image``.
+
+    """
+
+    def __init__(self, transforms):
+        self.transforms = transforms
+
+    def __call__(self, *args):
+        for t in self.transforms:
+            args = t(*args)
+        return args
+
+    def __repr__(self):
+        format_string = self.__class__.__name__ + '('
+        for t in self.transforms:
+            format_string += '\n'
+            format_string += '    {0}'.format(t)
+        format_string += '\n)'
+        return format_string
+
+
+def identity(*args, **kwargs):
+    if len(args) == 1:
+        return args[0]
+    else:
+        return args
+
+
+def build_transforms(cfg):
+
+    if cfg is None:
+        return identity
+
+    transforms = []
+
+    for transform_name, cfg_instance in cfg.items():
+        transform_instance = instantiate_from_config(cfg_instance)
+        transforms.append(transform_instance)
+        print(f"Build transform: {transform_instance}")
+
+    transforms = Compose(transforms)
+
+    return transforms
+

primitive_anything/michelangelo/data/utils.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+
+import torch
+import numpy as np
+
+
+def worker_init_fn(_):
+    worker_info = torch.utils.data.get_worker_info()
+    worker_id = worker_info.id
+
+    # dataset = worker_info.dataset
+    # split_size = dataset.num_records // worker_info.num_workers
+    # # reset num_records to the true number to retain reliable length information
+    # dataset.sample_ids = dataset.valid_ids[worker_id * split_size:(worker_id + 1) * split_size]
+    # current_id = np.random.choice(len(np.random.get_state()[1]), 1)
+    # return np.random.seed(np.random.get_state()[1][current_id] + worker_id)
+
+    return np.random.seed(np.random.get_state()[1][0] + worker_id)
+
+
+def collation_fn(samples, combine_tensors=True, combine_scalars=True):
+    """
+
+    Args:
+        samples (list[dict]):
+        combine_tensors:
+        combine_scalars:
+
+    Returns:
+
+    """
+
+    result = {}
+
+    keys = samples[0].keys()
+
+    for key in keys:
+        result[key] = []
+
+    for sample in samples:
+        for key in keys:
+            val = sample[key]
+            result[key].append(val)
+
+    for key in keys:
+        val_list = result[key]
+        if isinstance(val_list[0], (int, float)):
+            if combine_scalars:
+                result[key] = np.array(result[key])
+
+        elif isinstance(val_list[0], torch.Tensor):
+            if combine_tensors:
+                result[key] = torch.stack(val_list)
+
+        elif isinstance(val_list[0], np.ndarray):
+            if combine_tensors:
+                result[key] = np.stack(val_list)
+
+    return result

primitive_anything/michelangelo/graphics/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

primitive_anything/michelangelo/graphics/primitives/__init__.py

@@ -0,0 +1,9 @@
+# -*- coding: utf-8 -*-
+
+from .volume import generate_dense_grid_points
+
+from .mesh import (
+    MeshOutput,
+    save_obj,
+    savemeshtes2
+)

primitive_anything/michelangelo/graphics/primitives/mesh.py

@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+
+import os
+import cv2
+import numpy as np
+import PIL.Image
+from typing import Optional
+
+import trimesh
+
+
+def save_obj(pointnp_px3, facenp_fx3, fname):
+    fid = open(fname, "w")
+    write_str = ""
+    for pidx, p in enumerate(pointnp_px3):
+        pp = p
+        write_str += "v %f %f %f\n" % (pp[0], pp[1], pp[2])
+
+    for i, f in enumerate(facenp_fx3):
+        f1 = f + 1
+        write_str += "f %d %d %d\n" % (f1[0], f1[1], f1[2])
+    fid.write(write_str)
+    fid.close()
+    return
+
+
+def savemeshtes2(pointnp_px3, tcoords_px2, facenp_fx3, facetex_fx3, tex_map, fname):
+    fol, na = os.path.split(fname)
+    na, _ = os.path.splitext(na)
+
+    matname = "%s/%s.mtl" % (fol, na)
+    fid = open(matname, "w")
+    fid.write("newmtl material_0\n")
+    fid.write("Kd 1 1 1\n")
+    fid.write("Ka 0 0 0\n")
+    fid.write("Ks 0.4 0.4 0.4\n")
+    fid.write("Ns 10\n")
+    fid.write("illum 2\n")
+    fid.write("map_Kd %s.png\n" % na)
+    fid.close()
+    ####
+
+    fid = open(fname, "w")
+    fid.write("mtllib %s.mtl\n" % na)
+
+    for pidx, p in enumerate(pointnp_px3):
+        pp = p
+        fid.write("v %f %f %f\n" % (pp[0], pp[1], pp[2]))
+
+    for pidx, p in enumerate(tcoords_px2):
+        pp = p
+        fid.write("vt %f %f\n" % (pp[0], pp[1]))
+
+    fid.write("usemtl material_0\n")
+    for i, f in enumerate(facenp_fx3):
+        f1 = f + 1
+        f2 = facetex_fx3[i] + 1
+        fid.write("f %d/%d %d/%d %d/%d\n" % (f1[0], f2[0], f1[1], f2[1], f1[2], f2[2]))
+    fid.close()
+
+    PIL.Image.fromarray(np.ascontiguousarray(tex_map), "RGB").save(
+        os.path.join(fol, "%s.png" % na))
+
+    return
+
+
+class MeshOutput(object):
+
+    def __init__(self,
+                 mesh_v: np.ndarray,
+                 mesh_f: np.ndarray,
+                 vertex_colors: Optional[np.ndarray] = None,
+                 uvs: Optional[np.ndarray] = None,
+                 mesh_tex_idx: Optional[np.ndarray] = None,
+                 tex_map: Optional[np.ndarray] = None):
+
+        self.mesh_v = mesh_v
+        self.mesh_f = mesh_f
+        self.vertex_colors = vertex_colors
+        self.uvs = uvs
+        self.mesh_tex_idx = mesh_tex_idx
+        self.tex_map = tex_map
+
+    def contain_uv_texture(self):
+        return (self.uvs is not None) and (self.mesh_tex_idx is not None) and (self.tex_map is not None)
+
+    def contain_vertex_colors(self):
+        return self.vertex_colors is not None
+
+    def export(self, fname):
+
+        if self.contain_uv_texture():
+            savemeshtes2(
+                self.mesh_v,
+                self.uvs,
+                self.mesh_f,
+                self.mesh_tex_idx,
+                self.tex_map,
+                fname
+            )
+
+        elif self.contain_vertex_colors():
+            mesh_obj = trimesh.Trimesh(vertices=self.mesh_v, faces=self.mesh_f, vertex_colors=self.vertex_colors)
+            mesh_obj.export(fname)
+
+        else:
+            save_obj(
+                self.mesh_v,
+                self.mesh_f,
+                fname
+            )
+
+
+

primitive_anything/michelangelo/graphics/primitives/volume.py

@@ -0,0 +1,21 @@
+# -*- coding: utf-8 -*-
+
+import numpy as np
+
+
+def generate_dense_grid_points(bbox_min: np.ndarray,
+                               bbox_max: np.ndarray,
+                               octree_depth: int,
+                               indexing: str = "ij"):
+    length = bbox_max - bbox_min
+    num_cells = np.exp2(octree_depth)
+    x = np.linspace(bbox_min[0], bbox_max[0], int(num_cells) + 1, dtype=np.float32)
+    y = np.linspace(bbox_min[1], bbox_max[1], int(num_cells) + 1, dtype=np.float32)
+    z = np.linspace(bbox_min[2], bbox_max[2], int(num_cells) + 1, dtype=np.float32)
+    [xs, ys, zs] = np.meshgrid(x, y, z, indexing=indexing)
+    xyz = np.stack((xs, ys, zs), axis=-1)
+    xyz = xyz.reshape(-1, 3)
+    grid_size = [int(num_cells) + 1, int(num_cells) + 1, int(num_cells) + 1]
+
+    return xyz, grid_size, length
+

primitive_anything/michelangelo/models/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

primitive_anything/michelangelo/models/asl_diffusion/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

primitive_anything/michelangelo/models/asl_diffusion/asl_diffuser_pl_module.py

@@ -0,0 +1,483 @@
+# -*- coding: utf-8 -*-
+
+from omegaconf import DictConfig
+from typing import List, Tuple, Dict, Optional, Union
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from torch.optim import lr_scheduler
+import pytorch_lightning as pl
+from pytorch_lightning.utilities import rank_zero_only
+
+from einops import rearrange
+
+from diffusers.schedulers import (
+    DDPMScheduler,
+    DDIMScheduler,
+    KarrasVeScheduler,
+    DPMSolverMultistepScheduler
+)
+
+from ...utils import instantiate_from_config
+# from ..tsal.tsal_base import ShapeAsLatentPLModule
+from ..tsal.tsal_base import AlignedShapeAsLatentPLModule
+from .inference_utils import ddim_sample
+
+SchedulerType = Union[DDIMScheduler, KarrasVeScheduler, DPMSolverMultistepScheduler]
+
+
+def disabled_train(self, mode=True):
+    """Overwrite model.train with this function to make sure train/eval mode
+    does not change anymore."""
+    return self
+
+
+class ASLDiffuser(pl.LightningModule):
+    first_stage_model: Optional[AlignedShapeAsLatentPLModule]
+    # cond_stage_model: Optional[Union[nn.Module, pl.LightningModule]]
+    model: nn.Module
+
+    def __init__(self, *,
+                 first_stage_config,
+                 denoiser_cfg,
+                 scheduler_cfg,
+                 optimizer_cfg,
+                 loss_cfg,
+                 first_stage_key: str = "surface",
+                 cond_stage_key: str = "image",
+                 cond_stage_trainable: bool = True,
+                 scale_by_std: bool = False,
+                 z_scale_factor: float = 1.0,
+                 ckpt_path: Optional[str] = None,
+                 ignore_keys: Union[Tuple[str], List[str]] = ()):
+
+        super().__init__()
+
+        self.first_stage_key = first_stage_key
+        self.cond_stage_key = cond_stage_key
+        self.cond_stage_trainable = cond_stage_trainable
+
+        # 1. initialize first stage. 
+        # Note: the condition model contained in the first stage model.
+        self.first_stage_config = first_stage_config
+        self.first_stage_model = None
+        # self.instantiate_first_stage(first_stage_config)
+
+        # 2. initialize conditional stage
+        # self.instantiate_cond_stage(cond_stage_config)
+        self.cond_stage_model = {
+            "image": self.encode_image,
+            "image_unconditional_embedding": self.empty_img_cond,
+            "text": self.encode_text,
+            "text_unconditional_embedding": self.empty_text_cond,
+            "surface": self.encode_surface,
+            "surface_unconditional_embedding": self.empty_surface_cond,
+        }
+
+        # 3. diffusion model
+        self.model = instantiate_from_config(
+            denoiser_cfg, device=None, dtype=None
+        )
+
+        self.optimizer_cfg = optimizer_cfg
+
+        # 4. scheduling strategy
+        self.scheduler_cfg = scheduler_cfg
+
+        self.noise_scheduler: DDPMScheduler = instantiate_from_config(scheduler_cfg.noise)
+        self.denoise_scheduler: SchedulerType = instantiate_from_config(scheduler_cfg.denoise)
+
+        # 5. loss configures
+        self.loss_cfg = loss_cfg
+
+        self.scale_by_std = scale_by_std
+        if scale_by_std:
+            self.register_buffer("z_scale_factor", torch.tensor(z_scale_factor))
+        else:
+            self.z_scale_factor = z_scale_factor
+
+        self.ckpt_path = ckpt_path
+        if ckpt_path is not None:
+            self.init_from_ckpt(ckpt_path, ignore_keys=ignore_keys)
+
+    def instantiate_first_stage(self, config):
+        model = instantiate_from_config(config)
+        self.first_stage_model = model.eval()
+        self.first_stage_model.train = disabled_train
+        for param in self.first_stage_model.parameters():
+            param.requires_grad = False
+
+        self.first_stage_model = self.first_stage_model.to(self.device)
+
+    # def instantiate_cond_stage(self, config):
+    #     if not self.cond_stage_trainable:
+    #         if config == "__is_first_stage__":
+    #             print("Using first stage also as cond stage.")
+    #             self.cond_stage_model = self.first_stage_model
+    #         elif config == "__is_unconditional__":
+    #             print(f"Training {self.__class__.__name__} as an unconditional model.")
+    #             self.cond_stage_model = None
+    #             # self.be_unconditional = True
+    #         else:
+    #             model = instantiate_from_config(config)
+    #             self.cond_stage_model = model.eval()
+    #             self.cond_stage_model.train = disabled_train
+    #             for param in self.cond_stage_model.parameters():
+    #                 param.requires_grad = False
+    #     else:
+    #         assert config != "__is_first_stage__"
+    #         assert config != "__is_unconditional__"
+    #         model = instantiate_from_config(config)
+    #         self.cond_stage_model = model
+
+    def init_from_ckpt(self, path, ignore_keys=()):
+        state_dict = torch.load(path, map_location="cpu")["state_dict"]
+
+        keys = list(state_dict.keys())
+        for k in keys:
+            for ik in ignore_keys:
+                if k.startswith(ik):
+                    print("Deleting key {} from state_dict.".format(k))
+                    del state_dict[k]
+
+        missing, unexpected = self.load_state_dict(state_dict, strict=False)
+        print(f"Restored from {path} with {len(missing)} missing and {len(unexpected)} unexpected keys")
+        if len(missing) > 0:
+            print(f"Missing Keys: {missing}")
+            print(f"Unexpected Keys: {unexpected}")
+
+    @property
+    def zero_rank(self):
+        if self._trainer:
+            zero_rank = self.trainer.local_rank == 0
+        else:
+            zero_rank = True
+
+        return zero_rank
+
+    def configure_optimizers(self) -> Tuple[List, List]:
+
+        lr = self.learning_rate
+
+        trainable_parameters = list(self.model.parameters())
+        # if the conditional encoder is trainable
+
+        # if self.cond_stage_trainable:
+        #     conditioner_params = [p for p in self.cond_stage_model.parameters() if p.requires_grad]
+        #     trainable_parameters += conditioner_params
+        #     print(f"number of trainable conditional parameters: {len(conditioner_params)}.")
+
+        if self.optimizer_cfg is None:
+            optimizers = [torch.optim.AdamW(trainable_parameters, lr=lr, betas=(0.9, 0.99), weight_decay=1e-3)]
+            schedulers = []
+        else:
+            optimizer = instantiate_from_config(self.optimizer_cfg.optimizer, params=trainable_parameters)
+            scheduler_func = instantiate_from_config(
+                self.optimizer_cfg.scheduler,
+                max_decay_steps=self.trainer.max_steps,
+                lr_max=lr
+            )
+            scheduler = {
+                "scheduler": lr_scheduler.LambdaLR(optimizer, lr_lambda=scheduler_func.schedule),
+                "interval": "step",
+                "frequency": 1
+            }
+            optimizers = [optimizer]
+            schedulers = [scheduler]
+
+        return optimizers, schedulers
+
+    @torch.no_grad()
+    def encode_text(self, text):
+
+        b = text.shape[0]
+        text_tokens = rearrange(text, "b t l -> (b t) l")
+        text_embed = self.first_stage_model.model.encode_text_embed(text_tokens)
+        text_embed = rearrange(text_embed, "(b t) d -> b t d", b=b)
+        text_embed = text_embed.mean(dim=1)
+        text_embed = text_embed / text_embed.norm(dim=-1, keepdim=True)
+
+        return text_embed
+
+    @torch.no_grad()
+    def encode_image(self, img):
+
+        return self.first_stage_model.model.encode_image_embed(img)
+
+    @torch.no_grad()
+    def encode_surface(self, surface):
+
+        return self.first_stage_model.model.encode_shape_embed(surface, return_latents=False)
+
+    @torch.no_grad()
+    def empty_text_cond(self, cond):
+
+        return torch.zeros_like(cond, device=cond.device)
+
+    @torch.no_grad()
+    def empty_img_cond(self, cond):
+
+        return torch.zeros_like(cond, device=cond.device)
+
+    @torch.no_grad()
+    def empty_surface_cond(self, cond):
+
+        return torch.zeros_like(cond, device=cond.device)
+
+    @torch.no_grad()
+    def encode_first_stage(self, surface: torch.FloatTensor, sample_posterior=True):
+
+        z_q = self.first_stage_model.encode(surface, sample_posterior)
+        z_q = self.z_scale_factor * z_q
+
+        return z_q
+
+    @torch.no_grad()
+    def decode_first_stage(self, z_q: torch.FloatTensor, **kwargs):
+
+        z_q = 1. / self.z_scale_factor * z_q
+        latents = self.first_stage_model.decode(z_q, **kwargs)
+        return latents
+
+    @rank_zero_only
+    @torch.no_grad()
+    def on_train_batch_start(self, batch, batch_idx):
+        # only for very first batch
+        if self.scale_by_std and self.current_epoch == 0 and self.global_step == 0 \
+                and batch_idx == 0 and self.ckpt_path is None:
+            # set rescale weight to 1./std of encodings
+            print("### USING STD-RESCALING ###")
+
+            z_q = self.encode_first_stage(batch[self.first_stage_key])
+            z = z_q.detach()
+
+            del self.z_scale_factor
+            self.register_buffer("z_scale_factor", 1. / z.flatten().std())
+            print(f"setting self.z_scale_factor to {self.z_scale_factor}")
+
+            print("### USING STD-RESCALING ###")
+
+    def compute_loss(self, model_outputs, split):
+        """
+
+        Args:
+            model_outputs (dict):
+                - x_0:
+                - noise:
+                - noise_prior:
+                - noise_pred:
+                - noise_pred_prior:
+
+            split (str):
+
+        Returns:
+
+        """
+
+        pred = model_outputs["pred"]
+
+        if self.noise_scheduler.prediction_type == "epsilon":
+            target = model_outputs["noise"]
+        elif self.noise_scheduler.prediction_type == "sample":
+            target = model_outputs["x_0"]
+        else:
+            raise NotImplementedError(f"Prediction Type: {self.noise_scheduler.prediction_type} not yet supported.")
+
+        if self.loss_cfg.loss_type == "l1":
+            simple = F.l1_loss(pred, target, reduction="mean")
+        elif self.loss_cfg.loss_type in ["mse", "l2"]:
+            simple = F.mse_loss(pred, target, reduction="mean")
+        else:
+            raise NotImplementedError(f"Loss Type: {self.loss_cfg.loss_type} not yet supported.")
+
+        total_loss = simple
+
+        loss_dict = {
+            f"{split}/total_loss": total_loss.clone().detach(),
+            f"{split}/simple": simple.detach(),
+        }
+
+        return total_loss, loss_dict
+
+    def forward(self, batch):
+        """
+
+        Args:
+            batch:
+
+        Returns:
+
+        """
+
+        if self.first_stage_model is None:
+            self.instantiate_first_stage(self.first_stage_config)
+
+        latents = self.encode_first_stage(batch[self.first_stage_key])
+
+        # conditions = self.cond_stage_model.encode(batch[self.cond_stage_key])
+
+        conditions = self.cond_stage_model[self.cond_stage_key](batch[self.cond_stage_key]).unsqueeze(1)
+
+        mask = torch.rand((len(conditions), 1, 1), device=conditions.device, dtype=conditions.dtype) >= 0.1
+        conditions = conditions * mask.to(conditions)
+
+        # Sample noise that we"ll add to the latents
+        # [batch_size, n_token, latent_dim]
+        noise = torch.randn_like(latents)
+        bs = latents.shape[0]
+        # Sample a random timestep for each motion
+        timesteps = torch.randint(
+            0,
+            self.noise_scheduler.config.num_train_timesteps,
+            (bs,),
+            device=latents.device,
+        )
+        timesteps = timesteps.long()
+        # Add noise to the latents according to the noise magnitude at each timestep
+        noisy_z = self.noise_scheduler.add_noise(latents, noise, timesteps)
+
+        # diffusion model forward
+        noise_pred = self.model(noisy_z, timesteps, conditions)
+
+        diffusion_outputs = {
+            "x_0": noisy_z,
+            "noise": noise,
+            "pred": noise_pred
+        }
+
+        return diffusion_outputs
+
+    def training_step(self, batch: Dict[str, Union[torch.FloatTensor, List[str]]],
+                      batch_idx: int, optimizer_idx: int = 0) -> torch.FloatTensor:
+        """
+
+        Args:
+            batch (dict): the batch sample, and it contains:
+                - surface (torch.FloatTensor):
+                - image (torch.FloatTensor): if provide, [bs, 3, h, w], item range [0, 1]
+                - depth (torch.FloatTensor): if provide, [bs, 1, h, w], item range [-1, 1]
+                - normal (torch.FloatTensor): if provide, [bs, 3, h, w], item range [-1, 1]
+                - text (list of str):
+
+            batch_idx (int):
+
+            optimizer_idx (int):
+
+        Returns:
+            loss (torch.FloatTensor):
+
+        """
+
+        diffusion_outputs = self(batch)
+
+        loss, loss_dict = self.compute_loss(diffusion_outputs, "train")
+        self.log_dict(loss_dict, prog_bar=True, logger=True, sync_dist=False, rank_zero_only=True)
+
+        return loss
+
+    def validation_step(self, batch: Dict[str, torch.FloatTensor],
+                        batch_idx: int, optimizer_idx: int = 0) -> torch.FloatTensor:
+        """
+
+        Args:
+            batch (dict): the batch sample, and it contains:
+                - surface_pc (torch.FloatTensor): [n_pts, 4]
+                - surface_feats (torch.FloatTensor): [n_pts, c]
+                - text (list of str):
+
+            batch_idx (int):
+
+            optimizer_idx (int):
+
+        Returns:
+            loss (torch.FloatTensor):
+
+        """
+
+        diffusion_outputs = self(batch)
+
+        loss, loss_dict = self.compute_loss(diffusion_outputs, "val")
+        self.log_dict(loss_dict, prog_bar=True, logger=True, sync_dist=False, rank_zero_only=True)
+
+        return loss
+
+    @torch.no_grad()
+    def sample(self,
+               batch: Dict[str, Union[torch.FloatTensor, List[str]]],
+               sample_times: int = 1,
+               steps: Optional[int] = None,
+               guidance_scale: Optional[float] = None,
+               eta: float = 0.0,
+               return_intermediates: bool = False, **kwargs):
+
+        if self.first_stage_model is None:
+            self.instantiate_first_stage(self.first_stage_config)
+
+        if steps is None:
+            steps = self.scheduler_cfg.num_inference_steps
+
+        if guidance_scale is None:
+            guidance_scale = self.scheduler_cfg.guidance_scale
+        do_classifier_free_guidance = guidance_scale > 0
+
+        # conditional encode
+        xc = batch[self.cond_stage_key]
+        # cond = self.cond_stage_model[self.cond_stage_key](xc)
+        cond = self.cond_stage_model[self.cond_stage_key](xc).unsqueeze(1)
+
+        if do_classifier_free_guidance:
+            """
+            Note: There are two kinds of uncond for text. 
+            1: using "" as uncond text; (in SAL diffusion)
+            2: zeros_like(cond) as uncond text; (in MDM)
+            """
+            # un_cond = self.cond_stage_model.unconditional_embedding(batch_size=len(xc))
+            un_cond = self.cond_stage_model[f"{self.cond_stage_key}_unconditional_embedding"](cond)
+            # un_cond = torch.zeros_like(cond, device=cond.device)
+            cond = torch.cat([un_cond, cond], dim=0)
+
+        outputs = []
+        latents = None
+
+        if not return_intermediates:
+            for _ in range(sample_times):
+                sample_loop = ddim_sample(
+                    self.denoise_scheduler,
+                    self.model,
+                    shape=self.first_stage_model.latent_shape,
+                    cond=cond,
+                    steps=steps,
+                    guidance_scale=guidance_scale,
+                    do_classifier_free_guidance=do_classifier_free_guidance,
+                    device=self.device,
+                    eta=eta,
+                    disable_prog=not self.zero_rank
+                )
+                for sample, t in sample_loop:
+                    latents = sample
+                outputs.append(self.decode_first_stage(latents, **kwargs))
+        else:
+
+            sample_loop = ddim_sample(
+                self.denoise_scheduler,
+                self.model,
+                shape=self.first_stage_model.latent_shape,
+                cond=cond,
+                steps=steps,
+                guidance_scale=guidance_scale,
+                do_classifier_free_guidance=do_classifier_free_guidance,
+                device=self.device,
+                eta=eta,
+                disable_prog=not self.zero_rank
+            )
+
+            iter_size = steps // sample_times
+            i = 0
+            for sample, t in sample_loop:
+                latents = sample
+                if i % iter_size == 0 or i == steps - 1:
+                    outputs.append(self.decode_first_stage(latents, **kwargs))
+                i += 1
+
+        return outputs

primitive_anything/michelangelo/models/asl_diffusion/asl_udt.py

@@ -0,0 +1,104 @@
+# -*- coding: utf-8 -*-
+
+import torch
+import torch.nn as nn
+from typing import Optional
+from diffusers.models.embeddings import Timesteps
+import math
+
+from ..modules.transformer_blocks import MLP
+from ..modules.diffusion_transformer import UNetDiffusionTransformer
+
+
+class ConditionalASLUDTDenoiser(nn.Module):
+
+    def __init__(self, *,
+                 device: Optional[torch.device],
+                 dtype: Optional[torch.dtype],
+                 input_channels: int,
+                 output_channels: int,
+                 n_ctx: int,
+                 width: int,
+                 layers: int,
+                 heads: int,
+                 context_dim: int,
+                 context_ln: bool = True,
+                 skip_ln: bool = False,
+                 init_scale: float = 0.25,
+                 flip_sin_to_cos: bool = False,
+                 use_checkpoint: bool = False):
+        super().__init__()
+
+        self.use_checkpoint = use_checkpoint
+
+        init_scale = init_scale * math.sqrt(1.0 / width)
+
+        self.backbone = UNetDiffusionTransformer(
+            device=device,
+            dtype=dtype,
+            n_ctx=n_ctx,
+            width=width,
+            layers=layers,
+            heads=heads,
+            skip_ln=skip_ln,
+            init_scale=init_scale,
+            use_checkpoint=use_checkpoint
+        )
+        self.ln_post = nn.LayerNorm(width, device=device, dtype=dtype)
+        self.input_proj = nn.Linear(input_channels, width, device=device, dtype=dtype)
+        self.output_proj = nn.Linear(width, output_channels, device=device, dtype=dtype)
+
+        # timestep embedding
+        self.time_embed = Timesteps(width, flip_sin_to_cos=flip_sin_to_cos, downscale_freq_shift=0)
+        self.time_proj = MLP(
+            device=device, dtype=dtype, width=width, init_scale=init_scale
+        )
+
+        self.context_embed = nn.Sequential(
+            nn.LayerNorm(context_dim, device=device, dtype=dtype),
+            nn.Linear(context_dim, width, device=device, dtype=dtype),
+        )
+
+        if context_ln:
+            self.context_embed = nn.Sequential(
+                nn.LayerNorm(context_dim, device=device, dtype=dtype),
+                nn.Linear(context_dim, width, device=device, dtype=dtype),
+            )
+        else:
+            self.context_embed = nn.Linear(context_dim, width, device=device, dtype=dtype)
+
+    def forward(self,
+                model_input: torch.FloatTensor,
+                timestep: torch.LongTensor,
+                context: torch.FloatTensor):
+
+        r"""
+        Args:
+            model_input (torch.FloatTensor): [bs, n_data, c]
+            timestep (torch.LongTensor): [bs,]
+            context (torch.FloatTensor): [bs, context_tokens, c]
+
+        Returns:
+            sample (torch.FloatTensor): [bs, n_data, c]
+
+        """
+
+        _, n_data, _ = model_input.shape
+
+        # 1. time
+        t_emb = self.time_proj(self.time_embed(timestep)).unsqueeze(dim=1)
+
+        # 2. conditions projector
+        context = self.context_embed(context)
+
+        # 3. denoiser
+        x = self.input_proj(model_input)
+        x = torch.cat([t_emb, context, x], dim=1)
+        x = self.backbone(x)
+        x = self.ln_post(x)
+        x = x[:, -n_data:]
+        sample = self.output_proj(x)
+
+        return sample
+
+

primitive_anything/michelangelo/models/asl_diffusion/base.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+
+import torch
+import torch.nn as nn
+
+
+class BaseDenoiser(nn.Module):
+
+    def __init__(self):
+        super().__init__()
+
+    def forward(self, x, t, context):
+        raise NotImplementedError

primitive_anything/michelangelo/models/asl_diffusion/clip_asl_diffuser_pl_module.py

@@ -0,0 +1,393 @@
+# -*- coding: utf-8 -*-
+
+from omegaconf import DictConfig
+from typing import List, Tuple, Dict, Optional, Union
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from torch.optim import lr_scheduler
+import pytorch_lightning as pl
+from pytorch_lightning.utilities import rank_zero_only
+
+from diffusers.schedulers import (
+    DDPMScheduler,
+    DDIMScheduler,
+    KarrasVeScheduler,
+    DPMSolverMultistepScheduler
+)
+
+from ...utils import instantiate_from_config
+from ..tsal.tsal_base import AlignedShapeAsLatentPLModule
+from .inference_utils import ddim_sample
+
+SchedulerType = Union[DDIMScheduler, KarrasVeScheduler, DPMSolverMultistepScheduler]
+
+
+def disabled_train(self, mode=True):
+    """Overwrite model.train with this function to make sure train/eval mode
+    does not change anymore."""
+    return self
+
+
+class ClipASLDiffuser(pl.LightningModule):
+    first_stage_model: Optional[AlignedShapeAsLatentPLModule]
+    cond_stage_model: Optional[Union[nn.Module, pl.LightningModule]]
+    model: nn.Module
+
+    def __init__(self, *,
+                 first_stage_config,
+                 cond_stage_config,
+                 denoiser_cfg,
+                 scheduler_cfg,
+                 optimizer_cfg,
+                 loss_cfg,
+                 first_stage_key: str = "surface",
+                 cond_stage_key: str = "image",
+                 scale_by_std: bool = False,
+                 z_scale_factor: float = 1.0,
+                 ckpt_path: Optional[str] = None,
+                 ignore_keys: Union[Tuple[str], List[str]] = ()):
+
+        super().__init__()
+
+        self.first_stage_key = first_stage_key
+        self.cond_stage_key = cond_stage_key
+
+        # 1. lazy initialize first stage
+        self.instantiate_first_stage(first_stage_config)
+
+        # 2. initialize conditional stage
+        self.instantiate_cond_stage(cond_stage_config)
+
+        # 3. diffusion model
+        self.model = instantiate_from_config(
+            denoiser_cfg, device=None, dtype=None
+        )
+
+        self.optimizer_cfg = optimizer_cfg
+
+        # 4. scheduling strategy
+        self.scheduler_cfg = scheduler_cfg
+
+        self.noise_scheduler: DDPMScheduler = instantiate_from_config(scheduler_cfg.noise)
+        self.denoise_scheduler: SchedulerType = instantiate_from_config(scheduler_cfg.denoise)
+
+        # 5. loss configures
+        self.loss_cfg = loss_cfg
+
+        self.scale_by_std = scale_by_std
+        if scale_by_std:
+            self.register_buffer("z_scale_factor", torch.tensor(z_scale_factor))
+        else:
+            self.z_scale_factor = z_scale_factor
+
+        self.ckpt_path = ckpt_path
+        if ckpt_path is not None:
+            self.init_from_ckpt(ckpt_path, ignore_keys=ignore_keys)
+
+    def instantiate_non_trainable_model(self, config):
+        model = instantiate_from_config(config)
+        model = model.eval()
+        model.train = disabled_train
+        for param in model.parameters():
+            param.requires_grad = False
+
+        return model
+
+    def instantiate_first_stage(self, first_stage_config):
+        self.first_stage_model = self.instantiate_non_trainable_model(first_stage_config)
+        self.first_stage_model.set_shape_model_only()
+
+    def instantiate_cond_stage(self, cond_stage_config):
+        self.cond_stage_model = self.instantiate_non_trainable_model(cond_stage_config)
+
+    def init_from_ckpt(self, path, ignore_keys=()):
+        state_dict = torch.load(path, map_location="cpu")["state_dict"]
+
+        keys = list(state_dict.keys())
+        for k in keys:
+            for ik in ignore_keys:
+                if k.startswith(ik):
+                    print("Deleting key {} from state_dict.".format(k))
+                    del state_dict[k]
+
+        missing, unexpected = self.load_state_dict(state_dict, strict=False)
+        print(f"Restored from {path} with {len(missing)} missing and {len(unexpected)} unexpected keys")
+        if len(missing) > 0:
+            print(f"Missing Keys: {missing}")
+            print(f"Unexpected Keys: {unexpected}")
+
+    @property
+    def zero_rank(self):
+        if self._trainer:
+            zero_rank = self.trainer.local_rank == 0
+        else:
+            zero_rank = True
+
+        return zero_rank
+
+    def configure_optimizers(self) -> Tuple[List, List]:
+
+        lr = self.learning_rate
+
+        trainable_parameters = list(self.model.parameters())
+        if self.optimizer_cfg is None:
+            optimizers = [torch.optim.AdamW(trainable_parameters, lr=lr, betas=(0.9, 0.99), weight_decay=1e-3)]
+            schedulers = []
+        else:
+            optimizer = instantiate_from_config(self.optimizer_cfg.optimizer, params=trainable_parameters)
+            scheduler_func = instantiate_from_config(
+                self.optimizer_cfg.scheduler,
+                max_decay_steps=self.trainer.max_steps,
+                lr_max=lr
+            )
+            scheduler = {
+                "scheduler": lr_scheduler.LambdaLR(optimizer, lr_lambda=scheduler_func.schedule),
+                "interval": "step",
+                "frequency": 1
+            }
+            optimizers = [optimizer]
+            schedulers = [scheduler]
+
+        return optimizers, schedulers
+
+    @torch.no_grad()
+    def encode_first_stage(self, surface: torch.FloatTensor, sample_posterior=True):
+
+        z_q = self.first_stage_model.encode(surface, sample_posterior)
+        z_q = self.z_scale_factor * z_q
+
+        return z_q
+
+    @torch.no_grad()
+    def decode_first_stage(self, z_q: torch.FloatTensor, **kwargs):
+
+        z_q = 1. / self.z_scale_factor * z_q
+        latents = self.first_stage_model.decode(z_q, **kwargs)
+        return latents
+
+    @rank_zero_only
+    @torch.no_grad()
+    def on_train_batch_start(self, batch, batch_idx):
+        # only for very first batch
+        if self.scale_by_std and self.current_epoch == 0 and self.global_step == 0 \
+                and batch_idx == 0 and self.ckpt_path is None:
+            # set rescale weight to 1./std of encodings
+            print("### USING STD-RESCALING ###")
+
+            z_q = self.encode_first_stage(batch[self.first_stage_key])
+            z = z_q.detach()
+
+            del self.z_scale_factor
+            self.register_buffer("z_scale_factor", 1. / z.flatten().std())
+            print(f"setting self.z_scale_factor to {self.z_scale_factor}")
+
+            print("### USING STD-RESCALING ###")
+
+    def compute_loss(self, model_outputs, split):
+        """
+
+        Args:
+            model_outputs (dict):
+                - x_0:
+                - noise:
+                - noise_prior:
+                - noise_pred:
+                - noise_pred_prior:
+
+            split (str):
+
+        Returns:
+
+        """
+
+        pred = model_outputs["pred"]
+
+        if self.noise_scheduler.prediction_type == "epsilon":
+            target = model_outputs["noise"]
+        elif self.noise_scheduler.prediction_type == "sample":
+            target = model_outputs["x_0"]
+        else:
+            raise NotImplementedError(f"Prediction Type: {self.noise_scheduler.prediction_type} not yet supported.")
+
+        if self.loss_cfg.loss_type == "l1":
+            simple = F.l1_loss(pred, target, reduction="mean")
+        elif self.loss_cfg.loss_type in ["mse", "l2"]:
+            simple = F.mse_loss(pred, target, reduction="mean")
+        else:
+            raise NotImplementedError(f"Loss Type: {self.loss_cfg.loss_type} not yet supported.")
+
+        total_loss = simple
+
+        loss_dict = {
+            f"{split}/total_loss": total_loss.clone().detach(),
+            f"{split}/simple": simple.detach(),
+        }
+
+        return total_loss, loss_dict
+
+    def forward(self, batch):
+        """
+
+        Args:
+            batch:
+
+        Returns:
+
+        """
+
+        latents = self.encode_first_stage(batch[self.first_stage_key])
+        conditions = self.cond_stage_model.encode(batch[self.cond_stage_key])
+
+        # Sample noise that we"ll add to the latents
+        # [batch_size, n_token, latent_dim]
+        noise = torch.randn_like(latents)
+        bs = latents.shape[0]
+        # Sample a random timestep for each motion
+        timesteps = torch.randint(
+            0,
+            self.noise_scheduler.config.num_train_timesteps,
+            (bs,),
+            device=latents.device,
+        )
+        timesteps = timesteps.long()
+        # Add noise to the latents according to the noise magnitude at each timestep
+        noisy_z = self.noise_scheduler.add_noise(latents, noise, timesteps)
+
+        # diffusion model forward
+        noise_pred = self.model(noisy_z, timesteps, conditions)
+
+        diffusion_outputs = {
+            "x_0": noisy_z,
+            "noise": noise,
+            "pred": noise_pred
+        }
+
+        return diffusion_outputs
+
+    def training_step(self, batch: Dict[str, Union[torch.FloatTensor, List[str]]],
+                      batch_idx: int, optimizer_idx: int = 0) -> torch.FloatTensor:
+        """
+
+        Args:
+            batch (dict): the batch sample, and it contains:
+                - surface (torch.FloatTensor):
+                - image (torch.FloatTensor): if provide, [bs, 3, h, w], item range [0, 1]
+                - depth (torch.FloatTensor): if provide, [bs, 1, h, w], item range [-1, 1]
+                - normal (torch.FloatTensor): if provide, [bs, 3, h, w], item range [-1, 1]
+                - text (list of str):
+
+            batch_idx (int):
+
+            optimizer_idx (int):
+
+        Returns:
+            loss (torch.FloatTensor):
+
+        """
+
+        diffusion_outputs = self(batch)
+
+        loss, loss_dict = self.compute_loss(diffusion_outputs, "train")
+        self.log_dict(loss_dict, prog_bar=True, logger=True, sync_dist=False, rank_zero_only=True)
+
+        return loss
+
+    def validation_step(self, batch: Dict[str, torch.FloatTensor],
+                        batch_idx: int, optimizer_idx: int = 0) -> torch.FloatTensor:
+        """
+
+        Args:
+            batch (dict): the batch sample, and it contains:
+                - surface_pc (torch.FloatTensor): [n_pts, 4]
+                - surface_feats (torch.FloatTensor): [n_pts, c]
+                - text (list of str):
+
+            batch_idx (int):
+
+            optimizer_idx (int):
+
+        Returns:
+            loss (torch.FloatTensor):
+
+        """
+
+        diffusion_outputs = self(batch)
+
+        loss, loss_dict = self.compute_loss(diffusion_outputs, "val")
+        self.log_dict(loss_dict, prog_bar=True, logger=True, sync_dist=False, rank_zero_only=True)
+
+        return loss
+
+    @torch.no_grad()
+    def sample(self,
+               batch: Dict[str, Union[torch.FloatTensor, List[str]]],
+               sample_times: int = 1,
+               steps: Optional[int] = None,
+               guidance_scale: Optional[float] = None,
+               eta: float = 0.0,
+               return_intermediates: bool = False, **kwargs):
+
+        if steps is None:
+            steps = self.scheduler_cfg.num_inference_steps
+
+        if guidance_scale is None:
+            guidance_scale = self.scheduler_cfg.guidance_scale
+        do_classifier_free_guidance = guidance_scale > 0
+
+        # conditional encode
+        xc = batch[self.cond_stage_key]
+
+        # print(self.first_stage_model.device, self.cond_stage_model.device, self.device)
+
+        cond = self.cond_stage_model(xc)
+
+        if do_classifier_free_guidance:
+            un_cond = self.cond_stage_model.unconditional_embedding(batch_size=len(xc))
+            cond = torch.cat([un_cond, cond], dim=0)
+
+        outputs = []
+        latents = None
+
+        if not return_intermediates:
+            for _ in range(sample_times):
+                sample_loop = ddim_sample(
+                    self.denoise_scheduler,
+                    self.model,
+                    shape=self.first_stage_model.latent_shape,
+                    cond=cond,
+                    steps=steps,
+                    guidance_scale=guidance_scale,
+                    do_classifier_free_guidance=do_classifier_free_guidance,
+                    device=self.device,
+                    eta=eta,
+                    disable_prog=not self.zero_rank
+                )
+                for sample, t in sample_loop:
+                    latents = sample
+                outputs.append(self.decode_first_stage(latents, **kwargs))
+        else:
+
+            sample_loop = ddim_sample(
+                self.denoise_scheduler,
+                self.model,
+                shape=self.first_stage_model.latent_shape,
+                cond=cond,
+                steps=steps,
+                guidance_scale=guidance_scale,
+                do_classifier_free_guidance=do_classifier_free_guidance,
+                device=self.device,
+                eta=eta,
+                disable_prog=not self.zero_rank
+            )
+
+            iter_size = steps // sample_times
+            i = 0
+            for sample, t in sample_loop:
+                latents = sample
+                if i % iter_size == 0 or i == steps - 1:
+                    outputs.append(self.decode_first_stage(latents, **kwargs))
+                i += 1
+
+        return outputs

primitive_anything/michelangelo/models/asl_diffusion/inference_utils.py

@@ -0,0 +1,80 @@
+# -*- coding: utf-8 -*-
+
+import torch
+from tqdm import tqdm
+from typing import Tuple, List, Union, Optional
+from diffusers.schedulers import DDIMScheduler
+
+
+__all__ = ["ddim_sample"]
+
+
+def ddim_sample(ddim_scheduler: DDIMScheduler,
+                diffusion_model: torch.nn.Module,
+                shape: Union[List[int], Tuple[int]],
+                cond: torch.FloatTensor,
+                steps: int,
+                eta: float = 0.0,
+                guidance_scale: float = 3.0,
+                do_classifier_free_guidance: bool = True,
+                generator: Optional[torch.Generator] = None,
+                device: torch.device = "cuda:0",
+                disable_prog: bool = True):
+
+    assert steps > 0, f"{steps} must > 0."
+
+    # init latents
+    bsz = cond.shape[0]
+    if do_classifier_free_guidance:
+        bsz = bsz // 2
+
+    latents = torch.randn(
+        (bsz, *shape),
+        generator=generator,
+        device=cond.device,
+        dtype=cond.dtype,
+    )
+    # scale the initial noise by the standard deviation required by the scheduler
+    latents = latents * ddim_scheduler.init_noise_sigma
+    # set timesteps
+    ddim_scheduler.set_timesteps(steps)
+    timesteps = ddim_scheduler.timesteps.to(device)
+    # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
+    # eta (η) is only used with the DDIMScheduler, and between [0, 1]
+    extra_step_kwargs = {
+        "eta": eta,
+        "generator": generator
+    }
+
+    # reverse
+    for i, t in enumerate(tqdm(timesteps, disable=disable_prog, desc="DDIM Sampling:", leave=False)):
+        # expand the latents if we are doing classifier free guidance
+        latent_model_input = (
+            torch.cat([latents] * 2)
+            if do_classifier_free_guidance
+            else latents
+        )
+        # latent_model_input = scheduler.scale_model_input(latent_model_input, t)
+        # predict the noise residual
+        timestep_tensor = torch.tensor([t], dtype=torch.long, device=device)
+        timestep_tensor = timestep_tensor.expand(latent_model_input.shape[0])
+        noise_pred = diffusion_model.forward(latent_model_input, timestep_tensor, cond)
+
+        # perform guidance
+        if do_classifier_free_guidance:
+            noise_pred_uncond, noise_pred_text = noise_pred.chunk(2)
+            noise_pred = noise_pred_uncond + guidance_scale * (
+                    noise_pred_text - noise_pred_uncond
+            )
+            # text_embeddings_for_guidance = encoder_hidden_states.chunk(
+            #     2)[1] if do_classifier_free_guidance else encoder_hidden_states
+        # compute the previous noisy sample x_t -> x_t-1
+        latents = ddim_scheduler.step(
+            noise_pred, t, latents, **extra_step_kwargs
+        ).prev_sample
+
+        yield latents, t
+
+
+def karra_sample():
+    pass

primitive_anything/michelangelo/models/conditional_encoders/__init__.py

@@ -0,0 +1,3 @@
+# -*- coding: utf-8 -*-
+
+from .clip import CLIPEncoder

primitive_anything/michelangelo/models/conditional_encoders/clip.py

@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+
+import torch
+import numpy as np
+from PIL import Image
+from dataclasses import dataclass
+from torchvision.transforms import Normalize
+from transformers import CLIPModel, CLIPTokenizer
+from transformers.utils import ModelOutput
+from typing import Iterable, Optional, Union, List
+
+
+ImageType = Union[np.ndarray, torch.Tensor, Image.Image]
+
+
+@dataclass
+class CLIPEmbedOutput(ModelOutput):
+    last_hidden_state: torch.FloatTensor = None
+    pooler_output: torch.FloatTensor = None
+    embeds: torch.FloatTensor = None
+
+
+class CLIPEncoder(torch.nn.Module):
+
+    def __init__(self, model_path="openai/clip-vit-base-patch32"):
+
+        super().__init__()
+
+        # Load the CLIP model and processor
+        self.model: CLIPModel = CLIPModel.from_pretrained(model_path)
+        self.tokenizer = CLIPTokenizer.from_pretrained(model_path)
+        self.image_preprocess = Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
+
+        self.model.training = False
+        for p in self.model.parameters():
+            p.requires_grad = False
+
+    @torch.no_grad()
+    def encode_image(self, images: Iterable[Optional[ImageType]]):
+        pixel_values = self.image_preprocess(images)
+
+        vision_outputs = self.model.vision_model(pixel_values=pixel_values)
+
+        pooler_output = vision_outputs[1]  # pooled_output
+        image_features = self.model.visual_projection(pooler_output)
+
+        visual_embeds = CLIPEmbedOutput(
+            last_hidden_state=vision_outputs.last_hidden_state,
+            pooler_output=pooler_output,
+            embeds=image_features
+        )
+
+        return visual_embeds
+
+    @torch.no_grad()
+    def encode_text(self, texts: List[str]):
+        text_inputs = self.tokenizer(texts, padding=True, return_tensors="pt")
+
+        text_outputs = self.model.text_model(input_ids=text_inputs)
+
+        pooler_output = text_outputs[1]  # pooled_output
+        text_features = self.model.text_projection(pooler_output)
+
+        text_embeds = CLIPEmbedOutput(
+            last_hidden_state=text_outputs.last_hidden_state,
+            pooler_output=pooler_output,
+            embeds=text_features
+        )
+
+        return text_embeds
+
+    def forward(self,
+                images: Iterable[Optional[ImageType]],
+                texts: List[str]):
+
+        visual_embeds = self.encode_image(images)
+        text_embeds = self.encode_text(texts)
+
+        return visual_embeds, text_embeds
+
+
+
+
+
+
+
+
+
+

primitive_anything/michelangelo/models/conditional_encoders/encoder_factory.py

@@ -0,0 +1,562 @@
+# -*- coding: utf-8 -*-
+import os
+
+import torch
+import torch.nn as nn
+from torchvision import transforms
+from transformers import CLIPModel, CLIPTokenizer
+from collections import OrderedDict
+
+from ...data.transforms import RandomResize
+
+
+class AbstractEncoder(nn.Module):
+    embedding_dim: int
+
+    def __init__(self):
+        super().__init__()
+
+    def encode(self, *args, **kwargs):
+        raise NotImplementedError
+
+
+class ClassEmbedder(nn.Module):
+    def __init__(self, embed_dim, n_classes=1000, key="class"):
+        super().__init__()
+        self.key = key
+        self.embedding = nn.Embedding(n_classes, embed_dim)
+
+    def forward(self, batch, key=None):
+        if key is None:
+            key = self.key
+        # this is for use in crossattn
+        c = batch[key][:, None]
+        c = self.embedding(c)
+        return c
+
+
+class FrozenCLIPTextEmbedder(AbstractEncoder):
+    """Uses the CLIP transformer encoder for text (from Hugging Face)"""
+
+    def __init__(
+        self,
+        version="openai/clip-vit-large-patch14",
+        tokenizer_version=None,
+        device="cuda",
+        max_length=77,
+        zero_embedding_radio: float = 0.1,
+    ):
+        super().__init__()
+        self.tokenizer = CLIPTokenizer.from_pretrained(tokenizer_version or version)
+
+        self.device = device
+        self.max_length = max_length
+        self.zero_embedding_radio = zero_embedding_radio
+
+        self.clip_dict = OrderedDict()
+        self.clip_name = os.path.split(version)[-1]
+
+        transformer = CLIPModel.from_pretrained(version).text_model
+
+        for param in transformer.parameters():
+            param.requires_grad = False
+        self.clip_dict[self.clip_name] = transformer
+
+        self._move_flag = False
+
+    @property
+    def clip(self):
+        return self.clip_dict[self.clip_name]
+
+    def move(self):
+        if self._move_flag:
+            return
+
+        self.clip_dict[self.clip_name] = self.clip_dict[self.clip_name].to(self.device)
+        self._move_flag = True
+
+    def unconditional_embedding(self, batch_size):
+        empty_text = [""] * batch_size
+        empty_z = self.forward(empty_text)
+        return empty_z
+
+    def forward(self, text):
+        self.move()
+
+        batch_encoding = self.tokenizer(
+            text,
+            truncation=True,
+            max_length=self.max_length,
+            return_length=True,
+            return_overflowing_tokens=False,
+            padding="max_length",
+            return_tensors="pt",
+        )
+
+        tokens = batch_encoding["input_ids"].to(self.device)
+        outputs = self.clip(input_ids=tokens)
+
+        z = outputs.last_hidden_state
+        return z
+
+    def encode(self, text):
+        batch_size = len(text)
+        batch_mask = torch.rand((batch_size,))
+        for i in range(batch_size):
+            if batch_mask[i] < self.zero_embedding_radio:
+                text[i] = ""
+
+        return self(text)
+
+class FrozenAlignedCLIPTextEmbedder(AbstractEncoder):
+    """Uses the CLIP transformer encoder for text (from Hugging Face)"""
+
+    def __init__(
+        self,
+        version="openai/clip-vit-large-patch14",
+        tokenizer_version=None,
+        device="cuda",
+        max_length=77,
+        zero_embedding_radio: float = 0.1,
+    ):
+        super().__init__()
+        self.tokenizer = CLIPTokenizer.from_pretrained(tokenizer_version or version)
+
+        self.device = device
+        self.max_length = max_length
+        self.zero_embedding_radio = zero_embedding_radio
+
+        self.clip_dict = OrderedDict()
+        self.clip_name = os.path.split(version)[-1]
+
+        transformer = CLIPModel.from_pretrained(version).text_model
+
+        for param in transformer.parameters():
+            param.requires_grad = False
+        self.clip_dict[self.clip_name] = transformer
+
+        self._move_flag = False
+
+    @property
+    def clip(self):
+        return self.clip_dict[self.clip_name]
+
+    def move(self):
+        if self._move_flag:
+            return
+
+        self.clip_dict[self.clip_name] = self.clip_dict[self.clip_name].to(self.device)
+        self._move_flag = True
+
+    def unconditional_embedding(self, batch_size):
+        empty_text = [""] * batch_size
+        empty_z = self.forward(empty_text)
+        return empty_z
+
+    def forward(self, text):
+        self.move()
+
+        batch_encoding = self.tokenizer(
+            text,
+            truncation=True,
+            max_length=self.max_length,
+            return_length=True,
+            return_overflowing_tokens=False,
+            padding="max_length",
+            return_tensors="pt",
+        )
+
+        tokens = batch_encoding["input_ids"].to(self.device)
+        outputs = self.clip(input_ids=tokens)
+
+        z = outputs.last_hidden_state
+        return z
+
+    def encode(self, text):
+        batch_size = len(text)
+        batch_mask = torch.rand((batch_size,))
+        for i in range(batch_size):
+            if batch_mask[i] < self.zero_embedding_radio:
+                text[i] = ""
+
+        return self(text)
+
+
+class FrozenCLIPImageEmbedder(AbstractEncoder):
+    """Uses the CLIP transformer encoder for text (from Hugging Face)"""
+
+    def __init__(
+            self,
+            version="openai/clip-vit-large-patch14",
+            device="cuda",
+            zero_embedding_radio=0.1,
+            normalize_embedding=True,
+            num_projection_vector=0,
+            linear_mapping_bias=True,
+            reverse_visual_projection=False,
+    ):
+        super().__init__()
+
+        self.device = device
+
+        self.clip_dict = OrderedDict()
+        self.clip_name = os.path.split(version)[-1]
+
+        clip_model = CLIPModel.from_pretrained(version)
+        clip_model.text_model = None
+        clip_model.text_projection = None
+        clip_model = clip_model.eval()
+        for param in self.parameters():
+            param.requires_grad = False
+        self.clip_dict[self.clip_name] = clip_model
+
+        self.transform = transforms.Compose(
+            [
+                transforms.Resize(224, transforms.InterpolationMode.BICUBIC, antialias=True),
+                transforms.CenterCrop(224),  # crop a (224, 224) square
+                transforms.Normalize(
+                    mean=[0.48145466, 0.4578275, 0.40821073],
+                    std=[0.26862954, 0.26130258, 0.27577711],
+                ),
+            ]
+        )
+        self.zero_embedding_radio = zero_embedding_radio
+
+        self.num_projection_vector = num_projection_vector
+        self.reverse_visual_projection = reverse_visual_projection
+        self.normalize_embedding = normalize_embedding
+
+        embedding_dim = (
+            clip_model.visual_projection.in_features
+            if reverse_visual_projection
+            else clip_model.visual_projection.out_features
+        )
+        self.embedding_dim = embedding_dim
+        if self.num_projection_vector > 0:
+            self.projection = nn.Linear(
+                embedding_dim,
+                clip_model.visual_projection.out_features * num_projection_vector,
+                bias=linear_mapping_bias,
+            )
+            nn.init.normal_(self.projection.weight, std=embedding_dim ** -0.5)
+
+        self._move_flag = False
+
+    @property
+    def clip(self):
+        return self.clip_dict[self.clip_name]
+
+    def unconditional_embedding(self, batch_size):
+        zero = torch.zeros(
+            batch_size,
+            1,
+            self.embedding_dim,
+            device=self.device,
+            dtype=self.clip.visual_projection.weight.dtype,
+        )
+        if self.num_projection_vector > 0:
+            zero = self.projection(zero).view(batch_size, self.num_projection_vector, -1)
+        return zero
+
+    def forward(self, image, value_range=(-1, 1), zero_embedding_radio=0):
+        if value_range is not None:
+            low, high = value_range
+            image = (image - low) / (high - low)
+
+        image = image.to(self.device, dtype=self.clip.visual_projection.weight.dtype)
+
+        if self.reverse_visual_projection:
+            z = self.clip.vision_model(self.transform(image))[1]
+        else:
+            z = self.clip.get_image_features(self.transform(image))
+
+        if self.normalize_embedding:
+            z = z / z.norm(dim=-1, keepdim=True)
+        if z.ndim == 2:
+            z = z.unsqueeze(dim=-2)
+
+        if zero_embedding_radio > 0:
+            mask = torch.rand((len(image), 1, 1), device=z.device, dtype=z.dtype) < zero_embedding_radio
+            z = z * mask.to(z)
+
+        if self.num_projection_vector > 0:
+            z = self.projection(z).view(len(image), self.num_projection_vector, -1)
+
+        return z
+
+    def move(self):
+        if self._move_flag:
+            return
+
+        self.clip_dict[self.clip_name] = self.clip_dict[self.clip_name].to(self.device)
+        self._move_flag = True
+
+    def encode(self, image):
+        self.move()
+        return self(image, zero_embedding_radio=self.zero_embedding_radio)
+
+
+class FrozenCLIPImageGridEmbedder(AbstractEncoder):
+
+    def __init__(
+            self,
+            version="openai/clip-vit-large-patch14",
+            device="cuda",
+            zero_embedding_radio=0.1,
+    ):
+        super().__init__()
+
+        self.device = device
+
+        self.clip_dict = OrderedDict()
+        self.clip_name = os.path.split(version)[-1]
+
+        clip_model: CLIPModel = CLIPModel.from_pretrained(version)
+        clip_model.text_model = None
+        clip_model.text_projection = None
+        clip_model = clip_model.eval()
+        for param in self.parameters():
+            param.requires_grad = False
+        self.clip_dict[self.clip_name] = clip_model
+
+        self.transform = transforms.Compose(
+            [
+                transforms.Resize(224, transforms.InterpolationMode.BILINEAR, antialias=True),
+                transforms.CenterCrop(224),  # crop a (224, 224) square
+                transforms.Normalize(
+                    mean=[0.48145466, 0.4578275, 0.40821073],
+                    std=[0.26862954, 0.26130258, 0.27577711],
+                ),
+            ]
+        )
+        self.zero_embedding_radio = zero_embedding_radio
+        self.embedding_dim = clip_model.vision_embed_dim
+
+        self._move_flag = False
+
+    @property
+    def clip(self):
+        return self.clip_dict[self.clip_name]
+
+    def move(self):
+        if self._move_flag:
+            return
+
+        self.clip_dict[self.clip_name] = self.clip_dict[self.clip_name].to(self.device)
+        self._move_flag = True
+
+    def unconditional_embedding(self, batch_size):
+        zero = torch.zeros(
+            batch_size,
+            self.clip.vision_model.embeddings.num_positions,
+            self.embedding_dim,
+            device=self.device,
+            dtype=self.clip.visual_projection.weight.dtype,
+        )
+        return zero
+
+    def forward(self, image, value_range=(-1, 1), zero_embedding_radio=0):
+        self.move()
+
+        if value_range is not None:
+            low, high = value_range
+            image = (image - low) / (high - low)
+
+        image = image.to(self.device, dtype=self.clip.visual_projection.weight.dtype)
+
+        z = self.clip.vision_model(self.transform(image)).last_hidden_state
+
+        if zero_embedding_radio > 0:
+            mask = torch.rand((len(image), 1, 1), device=z.device, dtype=z.dtype) >= zero_embedding_radio
+            z = z * mask.to(z)
+
+        return z
+
+    def encode(self, image):
+        return self(image, zero_embedding_radio=self.zero_embedding_radio)
+
+
+class MoECLIPImageEncoder(nn.Module):
+    def __init__(
+            self,
+            versions,
+            hidden_state_dim,
+            num_projection_vector=8,
+            zero_embedding_radio=0.1,
+            device="cuda",
+            precision="fp16",
+            normalize=False,
+            clip_max=0,
+            transform_type="base",
+            argument_p=0.2,
+    ):
+        super().__init__()
+
+        self.device = torch.device(device)
+        self.hidden_state_dim = hidden_state_dim
+        self.zero_embedding_radio = zero_embedding_radio
+        self.num_projection_vector = num_projection_vector
+        self.dtype = dict(fp16=torch.float16, fp32=torch.float32, bf16=torch.bfloat16)[precision]
+        self.normalize = normalize
+        self.clip_max = clip_max
+
+        if transform_type == "base":
+            self.transform = transforms.Compose(
+                [
+                    transforms.Resize(224, transforms.InterpolationMode.BICUBIC, antialias=True),
+                    transforms.CenterCrop(224),  # crop a (224, 224) square
+                    transforms.Normalize(
+                        mean=[0.48145466, 0.4578275, 0.40821073],
+                        std=[0.26862954, 0.26130258, 0.27577711],
+                    ),
+                ]
+            )
+        elif transform_type == "crop_blur_resize":
+            self.transform = transforms.Compose(
+                [
+                    transforms.Resize(224, transforms.InterpolationMode.BICUBIC, antialias=True),
+                    transforms.CenterCrop(224),  # crop a (224, 224) square
+                    transforms.RandomApply(
+                        transforms=[
+                            transforms.RandomResizedCrop(
+                                size=224,
+                                scale=(0.8, 1.0),
+                                ratio=(0.99, 1.01),
+                                interpolation=transforms.InterpolationMode.BICUBIC,
+                            ),
+                        ],
+                        p=argument_p,
+                    ),
+                    transforms.RandomApply(
+                        transforms=[
+                            transforms.GaussianBlur(kernel_size=9, sigma=(0.1, 5)),
+                        ],
+                        p=argument_p,
+                    ),
+                    transforms.RandomApply(
+                        transforms=[
+                            RandomResize(size=224, resize_radio=(0.2, 1)),
+                        ],
+                        p=argument_p,
+                    ),
+                    transforms.Normalize(
+                        mean=[0.48145466, 0.4578275, 0.40821073],
+                        std=[0.26862954, 0.26130258, 0.27577711],
+                    ),
+                ]
+            )
+        else:
+            raise ValueError(f"invalid {transform_type=}")
+
+        if isinstance(versions, str):
+            versions = (versions,)
+
+        # 如果直接把clips定位为当前类的子module，1. 会在保存ckp时存无用的多个权重。 2. pl会调用to，导致layer_norm的权重也被转换成fp16
+        clips = OrderedDict()
+
+        for v in versions:
+            # 因为clips不是子module，直接指定device="cuda"会错误地导致clip模型权重都被放到cuda:0上。
+            clips[v], _ = clip.load(name=v, device="cpu", jit=False, download_root=None)
+            delattr(clips[v], "transformer")
+            clips[v].eval()
+            clips[v].requires_grad_(False)
+
+        self.clips_hidden_dim = sum(clips[v].ln_final.weight.size(0) for v in clips)
+
+        if self.num_projection_vector == 0:
+            self.projection = nn.Identity()
+        else:
+            self.projection = nn.Linear(self.clips_hidden_dim, hidden_state_dim * self.num_projection_vector, bias=True)
+            self.projection.to(dtype=self.dtype)
+            nn.init.normal_(self.projection.weight, std=self.clips_hidden_dim ** -0.5)
+
+        self.clips = clips
+
+        self._move_flag = False
+
+    def move(self):
+        if self._move_flag:
+            return
+
+        def convert_weights(model: nn.Module):
+            """Convert applicable model parameters to fp16"""
+
+            def _convert_weights_to_fp16(l):
+                if isinstance(l, (nn.Conv1d, nn.Conv2d, nn.Linear)):
+                    l.weight.data = l.weight.data.type(self.dtype)
+                    if l.bias is not None:
+                        l.bias.data = l.bias.data.type(self.dtype)
+
+                if isinstance(l, nn.MultiheadAttention):
+                    for attr in [
+                        *[f"{s}_proj_weight" for s in ["in", "q", "k", "v"]],
+                        "in_proj_bias",
+                        "bias_k",
+                        "bias_v",
+                    ]:
+                        tensor = getattr(l, attr)
+                        if tensor is not None:
+                            tensor.data = tensor.data.type(self.dtype)
+
+                for name in ["text_projection", "proj"]:
+                    if hasattr(l, name):
+                        attr = getattr(l, name)
+                        if attr is not None:
+                            attr.data = attr.data.type(self.dtype)
+
+            model.apply(_convert_weights_to_fp16)
+
+        for k in self.clips:
+            self.clips[k].to(self.device)
+            convert_weights(self.clips[k])  # fp32 -> self.dtype
+        self._move_flag = True
+
+    def unconditional_embedding(self, batch_size=None):
+        zero = torch.zeros(
+            batch_size,
+            self.clips_hidden_dim,
+            device=self.device,
+            dtype=self.dtype,
+        )
+        if self.num_projection_vector > 0:
+            zero = self.projection(zero).view(batch_size, self.num_projection_vector, -1)
+        return zero
+
+    def convert_embedding(self, z):
+        if self.num_projection_vector > 0:
+            z = self.projection(z.type(self.projection.weight.dtype)).view(len(z), self.num_projection_vector, -1)
+        return z
+
+    def forward(self, image, value_range=(-1, 1), zero_embedding_radio=0):
+        if value_range is not None:
+            low, high = value_range
+            image = (image - low) / (high - low)
+
+        image = self.transform(image)
+
+        with torch.no_grad():
+            embs = []
+            for v in self.clips:
+                x = self.clips[v].encode_image(image)
+                if self.normalize:
+                    x = x / x.norm(p=2, dim=-1, keepdim=True) * (x.size(-1) ** 0.5)
+                    # clip_max only works with normalization
+                    if self.clip_max > 0:
+                        x = x.clamp(-self.clip_max, self.clip_max)
+                embs.append(x)
+
+            z = torch.cat(embs, dim=-1)
+            if self.normalize:
+                z /= z.size(-1) ** 0.5
+
+        if zero_embedding_radio > 0:
+            mask = torch.rand((len(image), 1, 1), device=z.device, dtype=z.dtype) >= zero_embedding_radio
+            z = z + mask.to(z)
+
+        if self.num_projection_vector > 0:
+            z = self.projection(z).view(len(image), self.num_projection_vector, -1)
+        return z
+
+    def encode(self, image):
+        self.move()
+        return self(image, zero_embedding_radio=self.zero_embedding_radio)

primitive_anything/michelangelo/models/modules/__init__.py

@@ -0,0 +1,3 @@
+# -*- coding: utf-8 -*-
+
+from .checkpoint import checkpoint

primitive_anything/michelangelo/models/modules/checkpoint.py

@@ -0,0 +1,69 @@
+# -*- coding: utf-8 -*-
+"""
+Adapted from: https://github.com/openai/guided-diffusion/blob/22e0df8183507e13a7813f8d38d51b072ca1e67c/guided_diffusion/nn.py#L124
+"""
+
+import torch
+from typing import Callable, Iterable, Sequence, Union
+
+
+def checkpoint(
+    func: Callable[..., Union[torch.Tensor, Sequence[torch.Tensor]]],
+    inputs: Sequence[torch.Tensor],
+    params: Iterable[torch.Tensor],
+    flag: bool,
+    use_deepspeed: bool = False
+):
+    """
+    Evaluate a function without caching intermediate activations, allowing for
+    reduced memory at the expense of extra compute in the backward pass.
+    :param func: the function to evaluate.
+    :param inputs: the argument sequence to pass to `func`.
+    :param params: a sequence of parameters `func` depends on but does not
+                   explicitly take as arguments.
+    :param flag: if False, disable gradient checkpointing.
+    :param use_deepspeed: if True, use deepspeed
+    """
+    if flag:
+        if use_deepspeed:
+            import deepspeed
+            return deepspeed.checkpointing.checkpoint(func, *inputs)
+
+        args = tuple(inputs) + tuple(params)
+        return CheckpointFunction.apply(func, len(inputs), *args)
+    else:
+        return func(*inputs)
+
+
+class CheckpointFunction(torch.autograd.Function):
+    @staticmethod
+    @torch.cuda.amp.custom_fwd
+    def forward(ctx, run_function, length, *args):
+        ctx.run_function = run_function
+        ctx.input_tensors = list(args[:length])
+        ctx.input_params = list(args[length:])
+
+        with torch.no_grad():
+            output_tensors = ctx.run_function(*ctx.input_tensors)
+        return output_tensors
+
+    @staticmethod
+    @torch.cuda.amp.custom_bwd
+    def backward(ctx, *output_grads):
+        ctx.input_tensors = [x.detach().requires_grad_(True) for x in ctx.input_tensors]
+        with torch.enable_grad():
+            # Fixes a bug where the first op in run_function modifies the
+            # Tensor storage in place, which is not allowed for detach()'d
+            # Tensors.
+            shallow_copies = [x.view_as(x) for x in ctx.input_tensors]
+            output_tensors = ctx.run_function(*shallow_copies)
+        input_grads = torch.autograd.grad(
+            output_tensors,
+            ctx.input_tensors + ctx.input_params,
+            output_grads,
+            allow_unused=True,
+        )
+        del ctx.input_tensors
+        del ctx.input_params
+        del output_tensors
+        return (None, None) + input_grads

primitive_anything/michelangelo/models/modules/diffusion_transformer.py

@@ -0,0 +1,218 @@
+# -*- coding: utf-8 -*-
+
+import math
+import torch
+import torch.nn as nn
+from typing import Optional
+
+from .checkpoint import checkpoint
+from .transformer_blocks import (
+    init_linear,
+    MLP,
+    MultiheadCrossAttention,
+    MultiheadAttention,
+    ResidualAttentionBlock
+)
+
+
+class AdaLayerNorm(nn.Module):
+    def __init__(self,
+                 device: torch.device,
+                 dtype: torch.dtype,
+                 width: int):
+
+        super().__init__()
+
+        self.silu = nn.SiLU(inplace=True)
+        self.linear = nn.Linear(width, width * 2, device=device, dtype=dtype)
+        self.layernorm = nn.LayerNorm(width, elementwise_affine=False, device=device, dtype=dtype)
+
+    def forward(self, x, timestep):
+        emb = self.linear(timestep)
+        scale, shift = torch.chunk(emb, 2, dim=2)
+        x = self.layernorm(x) * (1 + scale) + shift
+        return x
+
+
+class DitBlock(nn.Module):
+    def __init__(
+            self,
+            *,
+            device: torch.device,
+            dtype: torch.dtype,
+            n_ctx: int,
+            width: int,
+            heads: int,
+            context_dim: int,
+            qkv_bias: bool = False,
+            init_scale: float = 1.0,
+            use_checkpoint: bool = False
+    ):
+        super().__init__()
+
+        self.use_checkpoint = use_checkpoint
+
+        self.attn = MultiheadAttention(
+            device=device,
+            dtype=dtype,
+            n_ctx=n_ctx,
+            width=width,
+            heads=heads,
+            init_scale=init_scale,
+            qkv_bias=qkv_bias
+        )
+        self.ln_1 = AdaLayerNorm(device, dtype, width)
+
+        if context_dim is not None:
+            self.ln_2 = AdaLayerNorm(device, dtype, width)
+            self.cross_attn = MultiheadCrossAttention(
+                device=device,
+                dtype=dtype,
+                width=width,
+                heads=heads,
+                data_width=context_dim,
+                init_scale=init_scale,
+                qkv_bias=qkv_bias
+            )
+
+        self.mlp = MLP(device=device, dtype=dtype, width=width, init_scale=init_scale)
+        self.ln_3 = AdaLayerNorm(device, dtype, width)
+
+    def forward(self, x: torch.Tensor, t: torch.Tensor, context: Optional[torch.Tensor] = None):
+        return checkpoint(self._forward, (x, t, context), self.parameters(), self.use_checkpoint)
+
+    def _forward(self, x: torch.Tensor, t: torch.Tensor, context: Optional[torch.Tensor] = None):
+        x = x + self.attn(self.ln_1(x, t))
+        if context is not None:
+            x = x + self.cross_attn(self.ln_2(x, t), context)
+        x = x + self.mlp(self.ln_3(x, t))
+        return x
+
+
+class DiT(nn.Module):
+    def __init__(
+            self,
+            *,
+            device: Optional[torch.device],
+            dtype: Optional[torch.dtype],
+            n_ctx: int,
+            width: int,
+            layers: int,
+            heads: int,
+            context_dim: int,
+            init_scale: float = 0.25,
+            qkv_bias: bool = False,
+            use_checkpoint: bool = False
+    ):
+        super().__init__()
+        self.n_ctx = n_ctx
+        self.width = width
+        self.layers = layers
+
+        self.resblocks = nn.ModuleList(
+            [
+                DitBlock(
+                    device=device,
+                    dtype=dtype,
+                    n_ctx=n_ctx,
+                    width=width,
+                    heads=heads,
+                    context_dim=context_dim,
+                    qkv_bias=qkv_bias,
+                    init_scale=init_scale,
+                    use_checkpoint=use_checkpoint
+                )
+                for _ in range(layers)
+            ]
+        )
+
+    def forward(self, x: torch.Tensor, t: torch.Tensor, context: Optional[torch.Tensor] = None):
+        for block in self.resblocks:
+            x = block(x, t, context)
+        return x
+
+
+class UNetDiffusionTransformer(nn.Module):
+    def __init__(
+            self,
+            *,
+            device: Optional[torch.device],
+            dtype: Optional[torch.dtype],
+            n_ctx: int,
+            width: int,
+            layers: int,
+            heads: int,
+            init_scale: float = 0.25,
+            qkv_bias: bool = False,
+            skip_ln: bool = False,
+            use_checkpoint: bool = False
+    ):
+        super().__init__()
+
+        self.n_ctx = n_ctx
+        self.width = width
+        self.layers = layers
+
+        self.encoder = nn.ModuleList()
+        for _ in range(layers):
+            resblock = ResidualAttentionBlock(
+                device=device,
+                dtype=dtype,
+                n_ctx=n_ctx,
+                width=width,
+                heads=heads,
+                init_scale=init_scale,
+                qkv_bias=qkv_bias,
+                use_checkpoint=use_checkpoint
+            )
+            self.encoder.append(resblock)
+
+        self.middle_block = ResidualAttentionBlock(
+            device=device,
+            dtype=dtype,
+            n_ctx=n_ctx,
+            width=width,
+            heads=heads,
+            init_scale=init_scale,
+            qkv_bias=qkv_bias,
+            use_checkpoint=use_checkpoint
+        )
+
+        self.decoder = nn.ModuleList()
+        for _ in range(layers):
+            resblock = ResidualAttentionBlock(
+                device=device,
+                dtype=dtype,
+                n_ctx=n_ctx,
+                width=width,
+                heads=heads,
+                init_scale=init_scale,
+                qkv_bias=qkv_bias,
+                use_checkpoint=use_checkpoint
+            )
+            linear = nn.Linear(width * 2, width, device=device, dtype=dtype)
+            init_linear(linear, init_scale)
+
+            layer_norm = nn.LayerNorm(width, device=device, dtype=dtype) if skip_ln else None
+
+            self.decoder.append(nn.ModuleList([resblock, linear, layer_norm]))
+
+    def forward(self, x: torch.Tensor):
+
+        enc_outputs = []
+        for block in self.encoder:
+            x = block(x)
+            enc_outputs.append(x)
+
+        x = self.middle_block(x)
+
+        for i, (resblock, linear, layer_norm) in enumerate(self.decoder):
+            x = torch.cat([enc_outputs.pop(), x], dim=-1)
+            x = linear(x)
+
+            if layer_norm is not None:
+                x = layer_norm(x)
+
+            x = resblock(x)
+
+        return x