Updated Handler with support for custom 'commands' and seperated code into methods for more efficient calling.
Browse files- handler.py +63 -26
handler.py
CHANGED
|
@@ -62,15 +62,36 @@ class EndpointHandler():
|
|
| 62 |
print('loading SDF model...')
|
| 63 |
self.sdf_model.load_state_dict(load_checkpoint(self.sdf_name, device))
|
| 64 |
|
| 65 |
-
def __call__(self,
|
| 66 |
-
|
| 67 |
-
|
| 68 |
-
|
| 69 |
-
|
| 70 |
-
|
| 71 |
-
|
| 72 |
-
""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 73 |
|
|
|
|
|
|
|
| 74 |
use_image = False
|
| 75 |
|
| 76 |
#Checks if an image key has been provided, and if so, uses the image data instead of text input
|
|
@@ -123,11 +144,40 @@ class EndpointHandler():
|
|
| 123 |
#image = self.pipe(inputs, guidance_scale=7.5)["sample"][0]
|
| 124 |
|
| 125 |
pc = sampler.output_to_point_clouds(samples)[0]
|
| 126 |
-
print('type of pc: ', type(pc))
|
| 127 |
|
| 128 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 129 |
mesh = marching_cubes_mesh(
|
| 130 |
-
pc=
|
| 131 |
model=self.sdf_model,
|
| 132 |
batch_size=4096,
|
| 133 |
grid_size=32, # increase to 128 for resolution used in evals
|
|
@@ -138,19 +188,6 @@ class EndpointHandler():
|
|
| 138 |
with open('mesh.ply', 'wb') as f:
|
| 139 |
mesh.write_ply(f)
|
| 140 |
print(mesh)
|
| 141 |
-
|
| 142 |
-
pc_dict = {}
|
| 143 |
-
|
| 144 |
-
data_list = pc.coords.tolist()
|
| 145 |
-
json_string = json.dumps(data_list)
|
| 146 |
-
pc_dict['data'] = json_string
|
| 147 |
-
|
| 148 |
-
# Convert NumPy arrays to Python lists for serializing
|
| 149 |
-
serializable_channels = {key: value.tolist() for key, value in pc.channels.items()}
|
| 150 |
-
|
| 151 |
-
# Serialize the dictionary to a JSON-formatted string
|
| 152 |
-
channel_data = json.dumps(serializable_channels)
|
| 153 |
-
pc_dict['channels'] = channel_data
|
| 154 |
-
|
| 155 |
-
#return pc_dict
|
| 156 |
return mesh
|
|
|
|
|
|
| 62 |
print('loading SDF model...')
|
| 63 |
self.sdf_model.load_state_dict(load_checkpoint(self.sdf_name, device))
|
| 64 |
|
| 65 |
+
def __call__(self, input_data: Any) -> Any:
|
| 66 |
+
|
| 67 |
+
command = "null"
|
| 68 |
+
|
| 69 |
+
if "command" in input_data:
|
| 70 |
+
command = input_data["command"]
|
| 71 |
+
|
| 72 |
+
print(f"the command is: {command}")
|
| 73 |
+
|
| 74 |
+
#Assume the user app is still running the old version, and send the data back as it is being expected
|
| 75 |
+
#Currently, the App expects a .ply Mesh to be sent back, and will not have a command input sent with it
|
| 76 |
+
if command == "null":
|
| 77 |
+
temp_pc = self.generate_point_cloud(input_data)
|
| 78 |
+
return self.generate_mesh_from_pc(temp_pc)
|
| 79 |
+
elif command == "generate_pc":
|
| 80 |
+
return self.generate_point_cloud(input_data)
|
| 81 |
+
elif command == "generate_mesh":
|
| 82 |
+
return self.generate_mesh_from_pc(input_data["raw_pc"])
|
| 83 |
+
elif command == "status":
|
| 84 |
+
return self.check_status()
|
| 85 |
+
|
| 86 |
+
|
| 87 |
+
|
| 88 |
+
|
| 89 |
+
def check_status(self) -> bool:
|
| 90 |
+
return self.active
|
| 91 |
+
|
| 92 |
|
| 93 |
+
def generate_point_cloud(self, data: Any) -> Dict[str, Dict[str, float]]:
|
| 94 |
+
print("generate pc called...")
|
| 95 |
use_image = False
|
| 96 |
|
| 97 |
#Checks if an image key has been provided, and if so, uses the image data instead of text input
|
|
|
|
| 144 |
#image = self.pipe(inputs, guidance_scale=7.5)["sample"][0]
|
| 145 |
|
| 146 |
pc = sampler.output_to_point_clouds(samples)[0]
|
|
|
|
| 147 |
|
| 148 |
+
pc_dict = {}
|
| 149 |
+
|
| 150 |
+
data_list = pc.coords.tolist()
|
| 151 |
+
json_string = json.dumps(data_list)
|
| 152 |
+
pc_dict['data'] = json_string
|
| 153 |
+
|
| 154 |
+
# Convert NumPy arrays to Python lists for serializing
|
| 155 |
+
serializable_channels = {key: value.tolist() for key, value in pc.channels.items()}
|
| 156 |
+
|
| 157 |
+
# Serialize the dictionary to a JSON-formatted string
|
| 158 |
+
channel_data = json.dumps(serializable_channels)
|
| 159 |
+
pc_dict['channels'] = channel_data
|
| 160 |
+
|
| 161 |
+
return pc_dict
|
| 162 |
+
|
| 163 |
+
|
| 164 |
+
def generate_mesh_from_pc(self, pc_data: Any) -> Any:
|
| 165 |
+
# Produce a mesh (with vertex colors)
|
| 166 |
+
print("generate mesh called...")
|
| 167 |
+
|
| 168 |
+
#De-serialize both the coords and channel data
|
| 169 |
+
coords_list = json.loads(pc_data['data'])
|
| 170 |
+
channels_dict = json.loads(pc_data['channels'])
|
| 171 |
+
|
| 172 |
+
# Reconstruct the PointCloud object
|
| 173 |
+
# Make sure to use .items() for the dictionary to output the key-value pairs together
|
| 174 |
+
point_cloud = PointCloud(
|
| 175 |
+
coords=np.array(coords_list, dtype=np.float32),
|
| 176 |
+
channels={name: np.array(array, dtype=np.float32) for name, array in channels_dict.items()}
|
| 177 |
+
)
|
| 178 |
+
|
| 179 |
mesh = marching_cubes_mesh(
|
| 180 |
+
pc=point_cloud,
|
| 181 |
model=self.sdf_model,
|
| 182 |
batch_size=4096,
|
| 183 |
grid_size=32, # increase to 128 for resolution used in evals
|
|
|
|
| 188 |
with open('mesh.ply', 'wb') as f:
|
| 189 |
mesh.write_ply(f)
|
| 190 |
print(mesh)
|
| 191 |
+
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 192 |
return mesh
|
| 193 |
+
|