Spaces:
Running
Running
import os | |
import torch | |
import time | |
import threading | |
import json | |
import gc | |
from flask import Flask, request, jsonify, send_file, Response, stream_with_context | |
from werkzeug.utils import secure_filename | |
from PIL import Image | |
import io | |
import zipfile | |
import uuid | |
import traceback | |
from huggingface_hub import snapshot_download, login, HfFileSystem | |
from flask_cors import CORS | |
import numpy as np | |
import trimesh | |
from transformers import pipeline | |
from scipy.ndimage import gaussian_filter | |
from scipy import interpolate | |
import cv2 | |
app = Flask(__name__) | |
CORS(app) | |
# Configure directories | |
UPLOAD_FOLDER = '/tmp/uploads' | |
RESULTS_FOLDER = '/tmp/results' | |
CACHE_DIR = '/tmp/huggingface' | |
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg'} | |
os.makedirs(UPLOAD_FOLDER, exist_ok=True) | |
os.makedirs(RESULTS_FOLDER, exist_ok=True) | |
os.makedirs(CACHE_DIR, exist_ok=True) | |
os.environ['HF_HOME'] = CACHE_DIR | |
os.environ['TRANSFORMERS_CACHE'] = os.path.join(CACHE_DIR, 'transformers') | |
os.environ['HF_DATASETS_CACHE'] = os.path.join(CACHE_DIR, 'datasets') | |
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER | |
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 | |
# Job tracking | |
processing_jobs = {} | |
# Model variables | |
dpt_estimator = None | |
model_loaded = False | |
model_loading = False | |
TIMEOUT_SECONDS = 240 | |
MAX_DIMENSION = 518 | |
class TimeoutError(Exception): | |
pass | |
def process_with_timeout(function, args, timeout): | |
result = [None] | |
error = [None] | |
completed = [False] | |
def target(): | |
try: | |
result[0] = function(*args) | |
completed[0] = True | |
except Exception as e: | |
error[0] = e | |
thread = threading.Thread(target=target) | |
thread.daemon = True | |
thread.start() | |
thread.join(timeout) | |
if not completed[0]: | |
if thread.is_alive(): | |
return None, TimeoutError(f"Processing timed out after {timeout} seconds") | |
elif error[0]: | |
return None, error[0] | |
if error[0]: | |
return None, error[0] | |
return result[0], None | |
def allowed_file(filename): | |
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS | |
def remove_background(image_path): | |
try: | |
# Load image | |
img = cv2.imread(image_path) | |
img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) | |
# Initialize mask and models for GrabCut | |
mask = np.zeros(img.shape[:2], np.uint8) | |
bgd_model = np.zeros((1, 65), np.float64) | |
fgd_model = np.zeros((1, 65), np.float64) | |
# Define initial rectangle (10% border margin) | |
h, w = img.shape[:2] | |
margin = int(min(w, h) * 0.1) | |
rect = (margin, margin, w - 2 * margin, h - 2 * margin) | |
# Run GrabCut | |
cv2.grabCut(img, mask, rect, bgd_model, fgd_model, 5, cv2.GC_INIT_WITH_RECT) | |
# Create final mask (0 for background, 1 for foreground) | |
mask2 = np.where((mask == 2) | (mask == 0), 0, 1).astype('uint8') | |
# Check if foreground exists | |
if np.sum(mask2) == 0: | |
print(f"Warning: No foreground detected in {image_path}") | |
return None | |
# Apply mask and set background to black | |
img = img * mask2[:, :, np.newaxis] | |
img_pil = Image.fromarray(img).convert("RGB") | |
return img_pil | |
except Exception as e: | |
print(f"Error in remove_background for {image_path}: {str(e)}") | |
raise | |
def preprocess_image(image_path): | |
img = remove_background(image_path) | |
if img is None: | |
raise ValueError("No foreground detected in image") | |
if img.width > MAX_DIMENSION or img.height > MAX_DIMENSION: | |
if img.width > img.height: | |
new_width = MAX_DIMENSION | |
new_height = int(img.height * (MAX_DIMENSION / img.width)) | |
else: | |
new_height = MAX_DIMENSION | |
new_width = int(img.width * (MAX_DIMENSION / img.height)) | |
img = img.resize((new_width, new_height), Image.LANCZOS) | |
img_array = np.array(img) | |
if len(img_array.shape) == 3 and img_array.shape[2] == 3: | |
lab = cv2.cvtColor(img_array, cv2.COLOR_RGB2LAB) | |
l, a, b = cv2.split(lab) | |
clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)) | |
cl = clahe.apply(l) | |
enhanced_lab = cv2.merge((cl, a, b)) | |
img_array = cv2.cvtColor(enhanced_lab, cv2.COLOR_LAB2RGB) | |
img = Image.fromarray(img_array) | |
return img | |
def load_models(): | |
global dpt_estimator, model_loaded, model_loading | |
if model_loaded: | |
return dpt_estimator | |
if model_loading: | |
while model_loading and not model_loaded: | |
time.sleep(0.5) | |
return dpt_estimator | |
try: | |
model_loading = True | |
print("Loading models...") | |
hf_token = os.environ.get('HF_TOKEN') | |
if hf_token: | |
print("HF_TOKEN found, attempting login...") | |
login(token=hf_token) | |
print("Authenticated with Hugging Face token") | |
else: | |
print("Error: HF_TOKEN not found in environment. Intel/dpt-large requires authentication.") | |
raise ValueError("HF_TOKEN is required for Intel/dpt-large") | |
dpt_model_name = "Intel/dpt-large" | |
fs = HfFileSystem(token=hf_token) | |
model_cached = os.path.exists(os.path.join(CACHE_DIR, "hub", "models--Intel--dpt-large")) | |
if not model_cached: | |
max_retries = 3 | |
retry_delay = 5 | |
for attempt in range(max_retries): | |
try: | |
print(f"Attempting to download {dpt_model_name}, attempt {attempt+1}") | |
snapshot_download( | |
repo_id=dpt_model_name, | |
cache_dir=CACHE_DIR, | |
resume_download=True, | |
token=hf_token | |
) | |
print(f"Successfully downloaded {dpt_model_name}") | |
break | |
except Exception as e: | |
if attempt < max_retries - 1: | |
print(f"DPT download attempt {attempt+1} failed: {str(e)}. Retrying after {retry_delay}s...") | |
time.sleep(retry_delay) | |
retry_delay *= 2 | |
else: | |
raise | |
else: | |
print(f"{dpt_model_name} already cached in {CACHE_DIR}") | |
dpt_estimator = pipeline( | |
"depth-estimation", | |
model=dpt_model_name, | |
device=-1, | |
cache_dir=CACHE_DIR, | |
use_fast=True | |
) | |
print("DPT-Large loaded") | |
gc.collect() | |
model_loaded = True | |
return dpt_estimator | |
except Exception as e: | |
print(f"Error loading models: {str(e)}") | |
print(traceback.format_exc()) | |
raise | |
finally: | |
model_loading = False | |
def enhance_depth_map(depth_map, detail_level='medium'): | |
enhanced_depth = depth_map.copy().astype(np.float32) | |
p_low, p_high = np.percentile(enhanced_depth, [1, 99]) | |
enhanced_depth = np.clip(enhanced_depth, p_low, p_high) | |
enhanced_depth = (enhanced_depth - p_low) / (p_high - p_low) if p_high > p_low else enhanced_depth | |
if detail_level == 'high': | |
blurred = gaussian_filter(enhanced_depth, sigma=1.5) | |
mask = enhanced_depth - blurred | |
enhanced_depth = enhanced_depth + 1.5 * mask | |
smooth1 = gaussian_filter(enhanced_depth, sigma=0.5) | |
smooth2 = gaussian_filter(enhanced_depth, sigma=2.0) | |
edge_mask = enhanced_depth - smooth2 | |
enhanced_depth = smooth1 + 1.2 * edge_mask | |
elif detail_level == 'medium': | |
blurred = gaussian_filter(enhanced_depth, sigma=1.0) | |
mask = enhanced_depth - blurred | |
enhanced_depth = enhanced_depth + 0.8 * mask | |
enhanced_depth = gaussian_filter(enhanced_depth, sigma=0.5) | |
else: | |
enhanced_depth = gaussian_filter(enhanced_depth, sigma=0.7) | |
enhanced_depth = np.clip(enhanced_depth, 0, 1) | |
return enhanced_depth | |
def depth_to_mesh(depth_map, image, resolution=80, detail_level='medium', view_angle=0): | |
enhanced_depth = enhance_depth_map(depth_map, detail_level) | |
h, w = enhanced_depth.shape | |
x = np.linspace(0, w-1, resolution) | |
y = np.linspace(0, h-1, resolution) | |
x_grid, y_grid = np.meshgrid(x, y) | |
interp_func = interpolate.RectBivariateSpline( | |
np.arange(h), np.arange(w), enhanced_depth, kx=3, ky=3 | |
) | |
z_values = interp_func(y, x, grid=True) | |
if detail_level == 'high': | |
dx = np.gradient(z_values, axis=1) | |
dy = np.gradient(z_values, axis=0) | |
gradient_magnitude = np.sqrt(dx**2 + dy**2) | |
edge_mask = np.clip(gradient_magnitude * 5, 0, 0.2) | |
z_values = z_values + edge_mask * (z_values - gaussian_filter(z_values, sigma=1.0)) | |
z_min, z_max = np.percentile(z_values, [2, 98]) | |
z_values = (z_values - z_min) / (z_max - z_min) if z_max > z_min else z_values | |
z_scaling = 2.5 if detail_level == 'high' else 2.0 if detail_level == 'medium' else 1.5 | |
z_values = z_values * z_scaling | |
x_grid = (x_grid / w - 0.5) * 2.0 | |
y_grid = (y_grid / h - 0.5) * 2.0 | |
vertices = np.vstack([x_grid.flatten(), -y_grid.flatten(), -z_values.flatten()]).T | |
if view_angle != 0: | |
rotation_matrix = trimesh.transformations.rotation_matrix(view_angle, [0, 1, 0]) | |
vertices = trimesh.transform_points(vertices, rotation_matrix) | |
faces = [] | |
for i in range(resolution-1): | |
for j in range(resolution-1): | |
p1 = i * resolution + j | |
p2 = i * resolution + (j + 1) | |
p3 = (i + 1) * resolution + j | |
p4 = (i + 1) * resolution + (j + 1) | |
v1 = vertices[p1] | |
v2 = vertices[p2] | |
v3 = vertices[p3] | |
v4 = vertices[p4] | |
norm1 = np.cross(v2-v1, v4-v1) | |
norm2 = np.cross(v4-v3, v1-v3) | |
if np.dot(norm1, norm2) >= 0: | |
faces.append([p1, p2, p4]) | |
faces.append([p1, p4, p3]) | |
else: | |
faces.append([p1, p2, p3]) | |
faces.append([p2, p4, p3]) | |
faces = np.array(faces) | |
mesh = trimesh.Trimesh(vertices=vertices, faces=faces) | |
if image: | |
img_array = np.array(image) | |
vertex_colors = np.zeros((vertices.shape[0], 4), dtype=np.uint8) | |
for i in range(resolution): | |
for j in range(resolution): | |
img_x = j * (img_array.shape[1] - 1) / (resolution - 1) | |
img_y = i * (img_array.shape[0] - 1) / (resolution - 1) | |
x0, y0 = int(img_x), int(img_y) | |
x1, y1 = min(x0 + 1, img_array.shape[1] - 1), min(y0 + 1, img_array.shape[0] - 1) | |
wx = img_x - x0 | |
wy = img_y - y0 | |
vertex_idx = i * resolution + j | |
if len(img_array.shape) == 3 and img_array.shape[2] == 3: | |
r = int((1-wx)*(1-wy)*img_array[y0, x0, 0] + wx*(1-wy)*img_array[y0, x1, 0] + | |
(1-wx)*wy*img_array[y1, x0, 0] + wx*wy*img_array[y1, x1, 0]) | |
g = int((1-wx)*(1-wy)*img_array[y0, x0, 1] + wx*(1-wy)*img_array[y0, x1, 1] + | |
(1-wx)*wy*img_array[y1, x0, 1] + wx*wy*img_array[y1, x1, 1]) | |
b = int((1-wx)*(1-wy)*img_array[y0, x0, 2] + wx*(1-wy)*img_array[y0, x1, 2] + | |
(1-wx)*wy*img_array[y1, x0, 2] + wx*wy*img_array[y1, x1, 2]) | |
vertex_colors[vertex_idx, :3] = [r, g, b] | |
vertex_colors[vertex_idx, 3] = 255 | |
else: | |
gray = int((1-wx)*(1-wy)*img_array[y0, x0] + wx*(1-wy)*img_array[y0, x1] + | |
(1-wx)*wy*img_array[y1, x0] + wx*wy*img_array[y1, x1]) | |
vertex_colors[vertex_idx, :3] = [gray, gray, gray] | |
vertex_colors[vertex_idx, 3] = 255 | |
mesh.visual.vertex_colors = vertex_colors | |
if detail_level != 'high': | |
mesh = mesh.smoothed(method='laplacian', iterations=1) | |
mesh.fix_normals() | |
return mesh | |
def combine_meshes(meshes): | |
if len(meshes) == 1: | |
return meshes[0] | |
combined_vertices = [] | |
combined_faces = [] | |
vertex_offset = 0 | |
for mesh in meshes: | |
combined_vertices.append(mesh.vertices) | |
combined_faces.append(mesh.faces + vertex_offset) | |
vertex_offset += len(mesh.vertices) | |
combined_vertices = np.vstack(combined_vertices) | |
combined_faces = np.vstack(combined_faces) | |
combined_mesh = trimesh.Trimesh(vertices=combined_vertices, faces=combined_faces) | |
combined_mesh = combined_mesh.subdivide_to_size(max_edge=0.05) | |
combined_mesh = combined_mesh.smoothed(method='laplacian', iterations=2) | |
combined_mesh.fill_holes() | |
combined_mesh.fix_normals() | |
return combined_mesh | |
def health_check(): | |
return jsonify({ | |
"status": "healthy", | |
"model": "DPT-Large (Multi-View)", | |
"device": "cpu" | |
}), 200 | |
def progress(job_id): | |
def generate(): | |
if job_id not in processing_jobs: | |
yield f"data: {json.dumps({'error': 'Job not found'})}\n\n" | |
return | |
job = processing_jobs[job_id] | |
yield f"data: {json.dumps({'status': 'processing', 'progress': job['progress']})}\n\n" | |
last_progress = job['progress'] | |
check_count = 0 | |
while job['status'] == 'processing': | |
if job['progress'] != last_progress: | |
yield f"data: {json.dumps({'status': 'processing', 'progress': job['progress']})}\n\n" | |
last_progress = job['progress'] | |
time.sleep(0.5) | |
check_count += 1 | |
if check_count > 60: | |
if 'thread_alive' in job and not job['thread_alive'](): | |
job['status'] = 'error' | |
job['error'] = 'Processing thread died unexpectedly' | |
break | |
check_count = 0 | |
if job['status'] == 'completed': | |
yield f"data: {json.dumps({'status': 'completed', 'progress': 100, 'result_url': job['result_url'], 'preview_url': job['preview_url']})}\n\n" | |
else: | |
yield f"data: {json.dumps({'status': 'error', 'error': job['error']})}\n\n" | |
return Response(stream_with_context(generate()), mimetype='text/event-stream') | |
def convert_image_to_3d(): | |
required_views = ['front', 'back'] | |
optional_views = ['left', 'right'] | |
view_files = {} | |
for view in required_views + optional_views: | |
if view in request.files and request.files[view].filename != '': | |
view_files[view] = request.files[view] | |
if not all(view in view_files for view in required_views): | |
return jsonify({"error": "Front and back images are required"}), 400 | |
for view, file in view_files.items(): | |
if not allowed_file(file.filename): | |
return jsonify({"error": f"File type not allowed for {view}. Supported types: {', '.join(ALLOWED_EXTENSIONS)}"}), 400 | |
try: | |
mesh_resolution = min(int(request.form.get('mesh_resolution', 80)), 120) | |
output_format = request.form.get('output_format', 'glb').lower() | |
detail_level = request.form.get('detail_level', 'medium').lower() | |
texture_quality = request.form.get('texture_quality', 'medium').lower() | |
except ValueError: | |
return jsonify({"error": "Invalid parameter values"}), 400 | |
if output_format not in ['obj', 'glb']: | |
return jsonify({"error": "Unsupported output format. Use 'obj' or 'glb'"}), 400 | |
if detail_level == 'high': | |
mesh_resolution = min(int(mesh_resolution * 1.5), 120) | |
elif detail_level == 'low': | |
mesh_resolution = max(int(mesh_resolution * 0.7), 50) | |
job_id = str(uuid.uuid4()) | |
output_dir = os.path.join(RESULTS_FOLDER, job_id) | |
os.makedirs(output_dir, exist_ok=True) | |
filepaths = {} | |
for view, file in view_files.items(): | |
filename = secure_filename(file.filename) | |
filepath = os.path.join(app.config['UPLOAD_FOLDER'], f"{job_id}_{view}_{filename}") | |
file.save(filepath) | |
filepaths[view] = filepath | |
processing_jobs[job_id] = { | |
'status': 'processing', | |
'progress': 0, | |
'result_url': None, | |
'preview_url': None, | |
'error': None, | |
'output_format': output_format, | |
'created_at': time.time() | |
} | |
def process_images(): | |
thread = threading.current_thread() | |
processing_jobs[job_id]['thread_alive'] = lambda: thread.is_alive() | |
try: | |
processing_jobs[job_id]['progress'] = 5 | |
images = {} | |
for view, filepath in filepaths.items(): | |
try: | |
images[view] = preprocess_image(filepath) | |
except ValueError as e: | |
processing_jobs[job_id]['status'] = 'error' | |
processing_jobs[job_id]['error'] = f"Error preprocessing {view} image: {str(e)}" | |
return | |
processing_jobs[job_id]['progress'] = 10 | |
try: | |
dpt_model = load_models() | |
processing_jobs[job_id]['progress'] = 20 | |
except Exception as e: | |
processing_jobs[job_id]['status'] = 'error' | |
processing_jobs[job_id]['error'] = f"Error loading models: {str(e)}" | |
return | |
try: | |
def estimate_depths(): | |
meshes = [] | |
view_angles = {'front': 0, 'back': np.pi, 'left': np.pi/2, 'right': -np.pi/2} | |
with torch.no_grad(): | |
for view, image in images.items(): | |
dpt_result = dpt_model(image) | |
dpt_depth = dpt_result["depth"] | |
depth_map = np.array(dpt_depth) if isinstance(dpt_depth, Image.Image) else dpt_depth | |
if len(depth_map.shape) > 2: | |
depth_map = np.mean(depth_map, axis=2) | |
p_low, p_high = np.percentile(depth_map, [1, 99]) | |
depth_map = np.clip((depth_map - p_low) / (p_high - p_low), 0, 1) if p_high > p_low else depth_map | |
mesh = depth_to_mesh(depth_map, image, resolution=mesh_resolution, detail_level=detail_level, view_angle=view_angles[view]) | |
meshes.append(mesh) | |
gc.collect() | |
combined_mesh = combine_meshes(meshes) | |
return combined_mesh | |
combined_mesh, error = process_with_timeout(estimate_depths, [], TIMEOUT_SECONDS) | |
if error: | |
if isinstance(error, TimeoutError): | |
processing_jobs[job_id]['status'] = 'error' | |
processing_jobs[job_id]['error'] = f"Processing timed out after {TIMEOUT_SECONDS} seconds" | |
return | |
else: | |
raise error | |
processing_jobs[job_id]['progress'] = 80 | |
if output_format == 'obj': | |
obj_path = os.path.join(output_dir, "model.obj") | |
combined_mesh.export( | |
obj_path, | |
file_type='obj', | |
include_normals=True, | |
include_texture=True | |
) | |
zip_path = os.path.join(output_dir, "model.zip") | |
with zipfile.ZipFile(zip_path, 'w') as zipf: | |
zipf.write(obj_path, arcname="model.obj") | |
mtl_path = os.path.join(output_dir, "model.mtl") | |
if os.path.exists(mtl_path): | |
zipf.write(mtl_path, arcname="model.mtl") | |
texture_path = os.path.join(output_dir, "model.png") | |
if os.path.exists(texture_path): | |
zipf.write(texture_path, arcname="model.png") | |
processing_jobs[job_id]['result_url'] = f"/download/{job_id}" | |
processing_jobs[job_id]['preview_url'] = f"/preview/{job_id}" | |
elif output_format == 'glb': | |
glb_path = os.path.join(output_dir, "model.glb") | |
combined_mesh.export( | |
glb_path, | |
file_type='glb' | |
) | |
processing_jobs[job_id]['result_url'] = f"/download/{job_id}" | |
processing_jobs[job_id]['preview_url'] = f"/preview/{job_id}" | |
processing_jobs[job_id]['status'] = 'completed' | |
processing_jobs[job_id]['progress'] = 100 | |
print(f"Job {job_id} completed") | |
except Exception as e: | |
error_details = traceback.format_exc() | |
processing_jobs[job_id]['status'] = 'error' | |
processing_jobs[job_id]['error'] = f"Error during processing: {str(e)}" | |
print(f"Error processing job {job_id}: {str(e)}") | |
print(error_details) | |
return | |
for filepath in filepaths.values(): | |
if os.path.exists(filepath): | |
os.remove(filepath) | |
gc.collect() | |
except Exception as e: | |
error_details = traceback.format_exc() | |
processing_jobs[job_id]['status'] = 'error' | |
processing_jobs[job_id]['error'] = f"{str(e)}\n{error_details}" | |
print(f"Error processing job {job_id}: {str(e)}") | |
print(error_details) | |
for filepath in filepaths.values(): | |
if os.path.exists(filepath): | |
os.remove(filepath) | |
processing_thread = threading.Thread(target=process_images) | |
processing_thread.daemon = True | |
processing_thread.start() | |
return jsonify({"job_id": job_id}), 202 | |
def download_model(job_id): | |
if job_id not in processing_jobs or processing_jobs[job_id]['status'] != 'completed': | |
return jsonify({"error": "Model not found or processing not complete"}), 404 | |
output_dir = os.path.join(RESULTS_FOLDER, job_id) | |
output_format = processing_jobs[job_id].get('output_format', 'glb') | |
if output_format == 'obj': | |
zip_path = os.path.join(output_dir, "model.zip") | |
if os.path.exists(zip_path): | |
return send_file(zip_path, as_attachment=True, download_name="model.zip") | |
else: | |
glb_path = os.path.join(output_dir, "model.glb") | |
if os.path.exists(glb_path): | |
return send_file(glb_path, as_attachment=True, download_name="model.glb") | |
return jsonify({"error": "File not found"}), 404 | |
def preview_model(job_id): | |
if job_id not in processing_jobs or processing_jobs[job_id]['status'] != 'completed': | |
return jsonify({"error": "Model not found or processing not complete"}), 404 | |
output_dir = os.path.join(RESULTS_FOLDER, job_id) | |
output_format = processing_jobs[job_id].get('output_format', 'glb') | |
if output_format == 'obj': | |
obj_path = os.path.join(output_dir, "model.obj") | |
if os.path.exists(obj_path): | |
return send_file(obj_path, mimetype='model/obj') | |
else: | |
glb_path = os.path.join(output_dir, "model.glb") | |
if os.path.exists(glb_path): | |
return send_file(glb_path, mimetype='model/gltf-binary') | |
return jsonify({"error": "File not found"}), 404 | |
def cleanup_old_jobs(): | |
current_time = time.time() | |
job_ids_to_remove = [] | |
for job_id, job_data in processing_jobs.items(): | |
if job_data['status'] == 'completed' and (current_time - job_data.get('created_at', 0)) > 3600: | |
job_ids_to_remove.append(job_id) | |
elif job_data['status'] == 'error' and (current_time - job_data.get('created_at', 0)) > 1800: | |
job_ids_to_remove.append(job_id) | |
for job_id in job_ids_to_remove: | |
output_dir = os.path.join(RESULTS_FOLDER, job_id) | |
try: | |
import shutil | |
if os.path.exists(output_dir): | |
shutil.rmtree(output_dir) | |
except Exception as e: | |
print(f"Error cleaning up job {job_id}: {str(e)}") | |
if job_id in processing_jobs: | |
del processing_jobs[job_id] | |
threading.Timer(300, cleanup_old_jobs).start() | |
def model_info(job_id): | |
if job_id not in processing_jobs: | |
return jsonify({"error": "Model not found"}), 404 | |
job = processing_jobs[job_id] | |
if job['status'] != 'completed': | |
return jsonify({ | |
"status": job['status'], | |
"progress": job['progress'], | |
"error": job.get('error') | |
}), 200 | |
output_dir = os.path.join(RESULTS_FOLDER, job_id) | |
model_stats = {} | |
if job['output_format'] == 'obj': | |
obj_path = os.path.join(output_dir, "model.obj") | |
zip_path = os.path.join(output_dir, "model.zip") | |
if os.path.exists(obj_path): | |
model_stats['obj_size'] = os.path.getsize(obj_path) | |
if os.path.exists(zip_path): | |
model_stats['package_size'] = os.path.getsize(zip_path) | |
else: | |
glb_path = os.path.join(output_dir, "model.glb") | |
if os.path.exists(glb_path): | |
model_stats['model_size'] = os.path.getsize(glb_path) | |
return jsonify({ | |
"status": job['status'], | |
"model_format": job['output_format'], | |
"download_url": job['result_url'], | |
"preview_url": job['preview_url'], | |
"model_stats": model_stats, | |
"created_at": job.get('created_at'), | |
"completed_at": job.get('completed_at') | |
}), 200 | |
def index(): | |
return jsonify({ | |
"message": "Multi-View Image to 3D API (DPT-Large)", | |
"endpoints": [ | |
"/convert", | |
"/progress/<job_id>", | |
"/download/<job_id>", | |
"/preview/<job_id>", | |
"/model-info/<job_id>" | |
], | |
"parameters": { | |
"front": "Image file (required)", | |
"back": "Image file (required)", | |
"left": "Image file (optional)", | |
"right": "Image file (optional)", | |
"mesh_resolution": "Integer (50-120)", | |
"output_format": "obj or glb", | |
"detail_level": "low, medium, or high", | |
"texture_quality": "low, medium, or high" | |
}, | |
"description": "Creates 3D models from multiple 2D images using Intel DPT-Large with custom background removal." | |
}), 200 | |
if __name__ == '__main__': | |
cleanup_old_jobs() | |
port = int(os.environ.get('PORT', 7860)) | |
app.run(host='0.0.0.0', port=port) |