rightnight / app.py
mac9087's picture
Update app.py
27f2066 verified
raw
history blame
27.4 kB
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
@app.route('/health', methods=['GET'])
def health_check():
return jsonify({
"status": "healthy",
"model": "DPT-Large (Multi-View)",
"device": "cpu"
}), 200
@app.route('/progress/<job_id>', methods=['GET'])
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')
@app.route('/convert', methods=['POST'])
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
@app.route('/download/<job_id>', methods=['GET'])
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
@app.route('/preview/<job_id>', methods=['GET'])
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()
@app.route('/model-info/<job_id>', methods=['GET'])
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
@app.route('/', methods=['GET'])
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)