Spaces:
Sleeping
Sleeping
| """ | |
| Simple 3D scan pipeline: run photogrammetry on sample images via COLMAP + Open3D | |
| """ | |
| import threading | |
| import os | |
| import shutil | |
| from plugins.three_d_scan.colmap_wrapper import COLMAPWrapper | |
| from plugins.three_d_scan.open3d_pipeline import clean_mesh | |
| # Global mission status | |
| _mission_status = { | |
| 'status': 'idle', | |
| 'log': [], | |
| 'mesh_path': None | |
| } | |
| def _run_pipeline(): | |
| # Reset status | |
| _mission_status['log'].clear() | |
| _mission_status['mesh_path'] = None | |
| try: | |
| _mission_status['status'] = 'preparing images' | |
| src = os.path.join(os.getcwd(), 'data', 'external_faces') | |
| work = os.path.join(os.getcwd(), 'plugins', 'three_d_scan', 'workspace') | |
| if os.path.exists(work): | |
| shutil.rmtree(work) | |
| os.makedirs(work, exist_ok=True) | |
| # Copy images | |
| for fn in os.listdir(src): | |
| if fn.lower().endswith(('.jpg', '.jpeg', '.png')): | |
| shutil.copy(os.path.join(src, fn), work) | |
| _mission_status['log'].append('Images prepared') | |
| # Run COLMAP pipeline | |
| _mission_status['status'] = 'running COLMAP' | |
| colmap = COLMAPWrapper() | |
| fused = colmap.run_full_pipeline(image_dir=work, workspace_dir=work) | |
| _mission_status['log'].append('COLMAP complete') | |
| # Clean mesh | |
| _mission_status['status'] = 'cleaning mesh' | |
| cleaned = clean_mesh(fused, os.path.join(work, 'cleaned.ply')) | |
| _mission_status['mesh_path'] = os.path.basename(cleaned) | |
| _mission_status['log'].append('Mesh cleaned') | |
| _mission_status['status'] = 'complete' | |
| except Exception as e: | |
| msg = str(e) | |
| _mission_status['status'] = f'error: {msg}' | |
| _mission_status['log'].append(msg) | |
| def start_mission(): | |
| """Launch pipeline in background thread""" | |
| if _mission_status.get('status') == 'running': | |
| return False | |
| thread = threading.Thread(target=_run_pipeline, daemon=True) | |
| thread.start() | |
| _mission_status['status'] = 'running' | |
| return True | |
| def get_mission_status(): | |
| """Return mission status snapshot""" | |
| return dict(_mission_status) |