Spaces:
Running
Running
| import json | |
| import logging | |
| import shutil | |
| import time | |
| import uuid | |
| from pathlib import Path | |
| from runtime import ROOT, MODEL_ROOT, configure_storage, startup_checks | |
| configure_storage() | |
| logging.basicConfig(filename=ROOT/"logs/workflow.log",level=logging.INFO, | |
| format="%(asctime)s %(levelname)s %(message)s") | |
| runtime_report = startup_checks() | |
| import gradio as gr | |
| import torch | |
| import trimesh | |
| from preflight import inspect_glb | |
| import upstream.app as mia | |
| mia.init_models() | |
| # Initialize existing upstream component references used by its workflow. | |
| # The original UI is not launched. | |
| upstream_blocks = mia.init_blocks() | |
| print("MODELS_READY dedicated CUDA inference",flush=True) | |
| def preserve_and_inspect(path): | |
| if not path: | |
| raise gr.Error("Upload a GLB first.") | |
| folder = ROOT/"inputs"/uuid.uuid4().hex | |
| folder.mkdir() | |
| original = folder/"original.glb" | |
| shutil.copyfile(path,original) | |
| try: | |
| report = inspect_glb(original) | |
| except (ValueError,KeyError,IndexError,TypeError) as exc: | |
| report = {"inference_allowed":False,"warning":"Invalid or unsupported GLB: "+str(exc)} | |
| report.update(original=str(original),cuda_available=torch.cuda.is_available(),gpu=runtime_report["gpu"]) | |
| (folder/"preflight.json").write_text(json.dumps(report,indent=2)) | |
| return str(original),report,str(original) | |
| def execute(original,report): | |
| if not original or not report or not report.get("inference_allowed"): | |
| raise gr.Error((report or {}).get("warning") or "Inspect an eligible GLB before inference.") | |
| if not torch.cuda.is_available(): | |
| raise gr.Error("CUDA is unavailable; CPU inference is disabled.") | |
| # Recheck the preserved file before expensive processing. | |
| checked=inspect_glb(original) | |
| if not checked["inference_allowed"]: | |
| raise gr.Error(checked["warning"]) | |
| job=ROOT/"outputs"/uuid.uuid4().hex | |
| job.mkdir() | |
| working=job/"character.glb" | |
| shutil.copyfile(original,working) | |
| db=mia.DB() | |
| started=time.time() | |
| stages=("Loading mesh","Preparing coordinates","Predicting skeleton and weights", | |
| "Preparing rest pose","Exporting rigged asset","Finishing") | |
| try: | |
| for stage,_ in zip(stages,mia._pipeline(input_path=str(working),db=db, | |
| reset_to_rest=True,rest_pose_type=None)): | |
| yield stage,None,None,None | |
| for p in (db.anim_path,db.anim_vis_path): | |
| if not p or not Path(p).is_file() or Path(p).stat().st_size==0: | |
| raise RuntimeError("Upstream export did not produce a nonempty asset") | |
| # Validate export parsing without changing it. | |
| result=trimesh.load(db.anim_vis_path,force="scene",process=False) | |
| if not result.geometry: | |
| raise RuntimeError("Exported GLB contains no geometry") | |
| with open(db.anim_vis_path,"rb") as stream: | |
| import struct | |
| stream.seek(12) | |
| length,kind=struct.unpack("<II",stream.read(8)) | |
| doc=json.loads(stream.read(length)) | |
| if not doc.get("skins") or not any("skin" in n for n in doc.get("nodes",[])): | |
| raise RuntimeError("Exported GLB has no skin binding") | |
| evidence=dict(status="PASS",original=original,glb=db.anim_vis_path,fbx=db.anim_path, | |
| elapsed_seconds=time.time()-started,export_meshes=len(result.geometry), | |
| skin_count=len(doc["skins"]),runtime=runtime_report["startup_id"]) | |
| (job/"result.json").write_text(json.dumps(evidence,indent=2)) | |
| logging.info("WORKFLOW_PASS %s",json.dumps(evidence)) | |
| print("WORKFLOW_PASS "+json.dumps(evidence),flush=True) | |
| yield "Complete: rigged GLB and FBX exported.",db.anim_vis_path,db.anim_vis_path,db.anim_path | |
| except Exception as exc: | |
| logging.exception("Workflow failed in %s",job) | |
| (job/"error.txt").write_text(str(exc)) | |
| raise gr.Error("Workflow failed: "+str(exc)+". Original preserved; details in /data/logs/workflow.log.") | |
| finally: | |
| mia.clear(db) | |
| def example(): | |
| return preserve_and_inspect(MODEL_ROOT/"data/examples/ninja.glb") | |
| def dense_character_test(): | |
| # Densify an existing character only for preflight; never modify its source. | |
| source=MODEL_ROOT/"data/examples/ninja.glb" | |
| mesh=trimesh.load(source,force="mesh",process=False) | |
| while len(mesh.faces)<=500_000: | |
| mesh=mesh.subdivide() | |
| folder=ROOT/"inputs"/("dense-test-"+uuid.uuid4().hex) | |
| folder.mkdir() | |
| path=folder/"dense-character.glb" | |
| mesh.export(path) | |
| report=inspect_glb(path) | |
| assert not report["inference_allowed"],"Dense character must be blocked before inference" | |
| report["test"]="PASS: dense character preprocessing warning" | |
| report["source"]=str(source) | |
| report["original"]=str(path) | |
| (folder/"preflight.json").write_text(json.dumps(report,indent=2)) | |
| print("DENSE_CHARACTER_TEST "+json.dumps(report),flush=True) | |
| return report | |
| with gr.Blocks(title="Make-It-Animatable") as demo: | |
| gr.Markdown("# Motion Rig\nUpload a character GLB, inspect it, then prepare and export its rig.") | |
| original=gr.State() | |
| report_state=gr.State() | |
| upload=gr.File(label="Upload GLB",file_types=[".glb"],type="filepath") | |
| inspect_button=gr.Button("Inspect / preflight") | |
| report_view=gr.JSON(label="Mesh preflight") | |
| original_download=gr.File(label="Preserved original",interactive=False) | |
| run=gr.Button("Prepare and export rig",variant="primary") | |
| status=gr.Textbox(label="Status",interactive=False) | |
| preview=gr.Model3D(label="Export preview",interactive=False) | |
| with gr.Row(): | |
| glb=gr.File(label="Download rigged GLB",interactive=False) | |
| fbx=gr.File(label="Download rigged FBX",interactive=False) | |
| with gr.Accordion("Runtime and acceptance checks",open=False): | |
| gr.JSON(value=runtime_report,label="Startup GPU and dependency checks") | |
| normal_test=gr.Button("Load bundled normal character for testing") | |
| dense_test=gr.Button("Test dense character preflight") | |
| dense_result=gr.JSON(label="Dense test result") | |
| inspect_button.click(preserve_and_inspect,inputs=upload, | |
| outputs=[original,report_view,original_download]).success( | |
| lambda x:x,inputs=report_view,outputs=report_state) | |
| normal_test.click(example,outputs=[original,report_view,original_download]).success( | |
| lambda x:x,inputs=report_view,outputs=report_state) | |
| run.click(execute,inputs=[original,report_state],outputs=[status,preview,glb,fbx], | |
| concurrency_limit=1,concurrency_id="workflow") | |
| dense_test.click(dense_character_test,outputs=dense_result, | |
| concurrency_limit=1,concurrency_id="workflow") | |
| demo.queue(max_size=4,default_concurrency_limit=1).launch( | |
| server_name="0.0.0.0",server_port=7860,show_error=True,ssr_mode=False, | |
| allowed_paths=[str(ROOT/"inputs"),str(ROOT/"outputs")]) | |