| """Public CANIS launcher; proprietary runtime code is loaded from a read-only volume.""" |
|
|
| from __future__ import annotations |
|
|
| import spaces |
|
|
| import builtins |
| import collections |
| import collections.abc |
| import hashlib |
| import ipaddress |
| import importlib.util |
| import inspect |
| import json |
| import os |
| import sys |
| import threading |
| import time |
| import types |
| from contextlib import contextmanager |
| from datetime import datetime, timezone |
| from functools import wraps |
| from pathlib import Path |
|
|
|
|
| PUBLIC_ROOT = Path(__file__).resolve().parent |
| MODEL_ROOT = Path(os.environ.get("CANIS_MODEL_ROOT", "/models/canis")) |
| RUNTIME_ROOT = MODEL_ROOT / "runtime_bundle" |
| ORIGINAL_APP = RUNTIME_ROOT / "scripts" / "demo" / "app_src_chain_gradio.py" |
| DINOV2_SOURCE_ROOT = MODEL_ROOT / "dinov2" / "repo" |
| DINOV2_HF_REPO = "facebook/ShapeR" |
| DINOV2_HF_REVISION = "a7ffd10934ba5057c521208651c1a11446ad6f30" |
| DINOV2_CHECKPOINT = "dinov2_vitl14_reg4_pretrain.pth" |
| DINOV2_CHECKPOINT_SIZE = 1_217_607_321 |
| OUTPUT_ROOT = Path(os.environ.get("CANIS_OUTPUT_ROOT", "/tmp/canis_outputs")) |
| POOL_ROOT = Path(os.environ.get("CANIS_POOL_ROOT", "/tmp/canis_empty_pool")) |
| PUBLIC_TRELLIS_REPO = "Louischong/Trellis-OA" |
| PIPELINE_CONFIG_ROOT = Path("/tmp/canis_pipeline_config") |
| QUOTA_ROOT = Path(os.environ.get("CANIS_QUOTA_ROOT", "/data/canis-usage")) |
| DAILY_CANONICALIZE_LIMIT = int(os.environ.get("CANIS_DAILY_LIMIT", "5")) |
| QUEUE_MAX_SIZE = int(os.environ.get("CANIS_QUEUE_MAX_SIZE", "3")) |
| DIAGNOSTIC_MODE = "production_fcgf" |
| IS_ZERO_GPU = os.environ.get("SPACES_ZERO_GPU") is not None |
|
|
| _quota_lock = threading.RLock() |
| _active_visitors: set[tuple[str, str]] = set() |
| @spaces.GPU(duration=120) |
| def _canonicalize_on_gpu( |
| selected_glb_path, |
| uploaded_glb, |
| rx, |
| ry, |
| rz, |
| ): |
| """Stable module-level ZeroGPU boundary around the private callback.""" |
| started = time.perf_counter() |
| _trace(f"phase=gpu_callback event=START mode={DIAGNOSTIC_MODE}") |
| if DIAGNOSTIC_MODE == "zero_pack_pure_python_probe": |
| elapsed = time.perf_counter() - started |
| _trace(f"phase=gpu_callback event=DONE mode={DIAGNOSTIC_MODE} elapsed={elapsed:.3f}s") |
| return ( |
| f"Pure Python ZeroGPU worker probe passed in {elapsed:.3f}s", |
| None, |
| None, |
| "", |
| "", |
| None, |
| None, |
| None, |
| None, |
| None, |
| ) |
| _trace_gpu_environment() |
| if DIAGNOSTIC_MODE in {"zero_pack_native_probe", "zero_pack_deferred_native_probe"}: |
| if DIAGNOSTIC_MODE == "zero_pack_deferred_native_probe": |
| _install_pytorch3d_compatibility() |
| _install_blackwell_attention_compatibility() |
| return _run_zero_pack_native_probe(started) |
| runtime = None |
| try: |
| import torch |
| from canis_job_entry import run_job |
|
|
| |
| |
| |
| import canis_job_entry |
|
|
| app = canis_job_entry._APP |
| args = canis_job_entry._ARGS |
| if app is None or args is None: |
| raise RuntimeError("Private CANIS job context is not initialized") |
| runtime = app.load_runtime(args, need_fcgf=True) |
| move_started = time.perf_counter() |
| _trace("phase=runtime.trellis.move_cpu_to_cuda event=START") |
| runtime.ctx.pipeline.to(torch.device("cuda:0")) |
| runtime.ctx.device = torch.device("cuda:0") |
| torch.cuda.synchronize() |
| _trace( |
| "phase=runtime.trellis.move_cpu_to_cuda event=DONE " |
| f"elapsed={time.perf_counter() - move_started:.3f}s" |
| ) |
|
|
| result = run_job(selected_glb_path, uploaded_glb, rx, ry, rz) |
| except Exception as exc: |
| _trace( |
| "phase=gpu_callback event=ERROR " |
| f"elapsed={time.perf_counter() - started:.3f}s " |
| f"type={type(exc).__name__} message={str(exc)[:300]!r}" |
| ) |
| raise |
| finally: |
| if runtime is not None and IS_ZERO_GPU: |
| import torch |
|
|
| move_started = time.perf_counter() |
| _trace("phase=runtime.trellis.move_cuda_to_cpu event=START") |
| runtime.ctx.pipeline.to(torch.device("cpu")) |
| runtime.ctx.device = torch.device("cpu") |
| torch.cuda.synchronize() |
| torch.cuda.empty_cache() |
| _trace( |
| "phase=runtime.trellis.move_cuda_to_cpu event=DONE " |
| f"elapsed={time.perf_counter() - move_started:.3f}s" |
| ) |
| elapsed = time.perf_counter() - started |
| _trace(f"phase=gpu_callback event=DONE elapsed={elapsed:.3f}s") |
| return _sanitize_job_result(result, elapsed) |
|
|
|
|
| def _probe_phase(label: str, function) -> str: |
| """Run one worker-only diagnostic phase and preserve its last durable marker.""" |
| started = time.perf_counter() |
| _trace(f"phase=diagnostic.{label} event=START") |
| try: |
| detail = function() |
| except Exception as exc: |
| _trace( |
| f"phase=diagnostic.{label} event=ERROR " |
| f"elapsed={time.perf_counter() - started:.3f}s " |
| f"type={type(exc).__name__} message={str(exc)[:300]!r}" |
| ) |
| raise |
| elapsed = time.perf_counter() - started |
| _trace(f"phase=diagnostic.{label} event=DONE elapsed={elapsed:.3f}s detail={detail!r}") |
| return f"{label}={elapsed:.3f}s" |
|
|
|
|
| def _run_zero_pack_native_probe(started: float): |
| """Exercise the ZeroGPU worker and native CUDA wheels without loading CANIS models.""" |
| import torch |
|
|
| timings: list[str] = [] |
|
|
| def cuda_basic(): |
| left = torch.arange(4096, device="cuda", dtype=torch.float32).reshape(64, 64) |
| value = float((left @ left.T).sum().item()) |
| torch.cuda.synchronize() |
| return {"value_finite": bool(value == value), "memory": torch.cuda.memory_allocated(0)} |
|
|
| def xformers_kernel(): |
| import xformers |
| import xformers.ops as xops |
|
|
| query = torch.randn((1, 32, 2, 64), device="cuda", dtype=torch.float16) |
| output = xops.memory_efficient_attention(query, query, query) |
| torch.cuda.synchronize() |
| return {"version": getattr(xformers, "__version__", "unknown"), "shape": tuple(output.shape)} |
|
|
| def spconv_kernel(): |
| import spconv |
| import spconv.pytorch as spconv_torch |
|
|
| indices = torch.tensor( |
| [[0, 0, 0, 0], [0, 0, 0, 1], [0, 0, 1, 0], [0, 1, 0, 0]], |
| device="cuda", |
| dtype=torch.int32, |
| ) |
| features = torch.randn((4, 4), device="cuda", dtype=torch.float16) |
| sparse = spconv_torch.SparseConvTensor(features, indices, [4, 4, 4], 1) |
| layer = spconv_torch.SubMConv3d(4, 4, 3, padding=1, bias=False).cuda().half() |
| output = layer(sparse) |
| value = float(output.features.float().sum().item()) |
| torch.cuda.synchronize() |
| return {"version": getattr(spconv, "__version__", "unknown"), "value_finite": bool(value == value)} |
|
|
| def pytorch3d_kernel(): |
| import pytorch3d |
| from pytorch3d.ops import sample_points_from_meshes |
| from pytorch3d.structures import Meshes |
|
|
| vertices = torch.tensor( |
| [[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0]], |
| device="cuda", |
| dtype=torch.float32, |
| ) |
| faces = torch.tensor([[0, 1, 2]], device="cuda", dtype=torch.int64) |
| output = sample_points_from_meshes(Meshes(verts=[vertices], faces=[faces]), 16) |
| value = float(output.sum().item()) |
| torch.cuda.synchronize() |
| return { |
| "version": getattr(pytorch3d, "__version__", "unknown"), |
| "shape": tuple(output.shape), |
| "value_finite": bool(value == value), |
| } |
|
|
| def vox2seq_kernel(): |
| import vox2seq |
|
|
| coords = torch.tensor( |
| [[0, 0, 0], [1, 2, 3], [7, 6, 5], [15, 15, 15]], |
| device="cuda", |
| dtype=torch.int32, |
| ) |
| encoded = vox2seq.encode(coords, mode="z_order") |
| decoded = vox2seq.decode(encoded, mode="z_order") |
| torch.cuda.synchronize() |
| return {"encoded_shape": tuple(encoded.shape), "decoded_shape": tuple(decoded.shape)} |
|
|
| for label, function in ( |
| ("cuda_basic", cuda_basic), |
| ("xformers_kernel", xformers_kernel), |
| ("spconv_kernel", spconv_kernel), |
| ("pytorch3d_kernel", pytorch3d_kernel), |
| ("vox2seq_kernel", vox2seq_kernel), |
| ): |
| timings.append(_probe_phase(label, function)) |
|
|
| elapsed = time.perf_counter() - started |
| _trace(f"phase=gpu_callback event=DONE mode={DIAGNOSTIC_MODE} elapsed={elapsed:.3f}s") |
| status = "Zero-pack diagnostic passed | " + " | ".join(timings) |
| return (status, None, None, "", "", None, None, None, None, None) |
|
|
|
|
| def _trace(message: str) -> None: |
| """Emit one durable phase marker without exposing paths or user data.""" |
| print( |
| f"[CANIS_TRACE] t={time.time():.3f} pid={os.getpid()} " |
| f"thread={threading.current_thread().name} {message}", |
| flush=True, |
| ) |
|
|
|
|
| def _traced_phase(label: str, function): |
| if getattr(function, "_canis_phase_traced", False): |
| return function |
|
|
| @wraps(function) |
| def traced(*args, **kwargs): |
| started = time.perf_counter() |
| _trace(f"phase={label} event=START") |
| try: |
| result = function(*args, **kwargs) |
| except Exception as exc: |
| _trace( |
| f"phase={label} event=ERROR elapsed={time.perf_counter() - started:.3f}s " |
| f"type={type(exc).__name__} message={str(exc)[:300]!r}" |
| ) |
| raise |
| _trace(f"phase={label} event=DONE elapsed={time.perf_counter() - started:.3f}s") |
| return result |
|
|
| traced._canis_phase_traced = True |
| return traced |
|
|
|
|
| @contextmanager |
| def _trace_runtime_imports_and_calls(): |
| selected_imports = { |
| "torch", |
| "open3d", |
| "spconv", |
| "xformers", |
| "MinkowskiEngine", |
| "pytorch3d", |
| "vox2seq", |
| "src.env", |
| "src.mise.fcgf", |
| "src_chain.run_object_pipeline", |
| "trellis", |
| "trellis.pipelines", |
| "trellis.pipelines.trellis_image_to_3d", |
| } |
| seen: set[str] = set() |
| original_import = builtins.__import__ |
|
|
| def traced_import(name, globals=None, locals=None, fromlist=(), level=0): |
| should_trace = level == 0 and name in selected_imports and name not in seen |
| if not should_trace: |
| return original_import(name, globals, locals, fromlist, level) |
| seen.add(name) |
| started = time.perf_counter() |
| _trace(f"phase=runtime.import.{name} event=START") |
| try: |
| module = original_import(name, globals, locals, fromlist, level) |
| except Exception as exc: |
| _trace( |
| f"phase=runtime.import.{name} event=ERROR " |
| f"elapsed={time.perf_counter() - started:.3f}s " |
| f"type={type(exc).__name__} message={str(exc)[:300]!r}" |
| ) |
| raise |
| if name == "src_chain.run_object_pipeline": |
| target = sys.modules.get(name) |
| function = getattr(target, "init_trellis_context", None) if target else None |
| if callable(function): |
| setattr( |
| target, |
| "init_trellis_context", |
| _traced_phase("runtime.init_trellis_context", function), |
| ) |
| _install_trellis_load_tracing(target) |
| _trace( |
| f"phase=runtime.import.{name} event=DONE " |
| f"elapsed={time.perf_counter() - started:.3f}s" |
| ) |
| return module |
|
|
| builtins.__import__ = traced_import |
| try: |
| yield |
| finally: |
| builtins.__import__ = original_import |
|
|
|
|
| def _install_trellis_load_tracing(run_pipeline_module) -> None: |
| if run_pipeline_module is None: |
| return |
| pipeline_class = getattr(run_pipeline_module, "TrellisImageTo3DPipeline", None) |
| if pipeline_class is None: |
| return |
| load_function = getattr(pipeline_class, "from_pretrained", None) |
| if callable(load_function) and not getattr(load_function, "_canis_phase_traced", False): |
| setattr( |
| pipeline_class, |
| "from_pretrained", |
| staticmethod(_traced_phase("runtime.trellis.deserialize", load_function)), |
| ) |
| base_class = next( |
| (candidate for candidate in pipeline_class.__mro__[1:] if "to" in candidate.__dict__), |
| None, |
| ) |
| if base_class is not None: |
| move_function = getattr(base_class, "to", None) |
| if callable(move_function) and not getattr(move_function, "_canis_phase_traced", False): |
| setattr(base_class, "to", _traced_phase("runtime.trellis.move_to_cuda", move_function)) |
|
|
|
|
| def _wrap_module_phases(module_name: str, names: dict[str, str]) -> None: |
| module = sys.modules.get(module_name) |
| if module is None: |
| _trace(f"phase_instrumentation module={module_name} event=NOT_IMPORTED") |
| return |
| for function_name, label in names.items(): |
| function = getattr(module, function_name, None) |
| if callable(function): |
| setattr(module, function_name, _traced_phase(label, function)) |
| else: |
| _trace( |
| f"phase_instrumentation module={module_name} " |
| f"function={function_name} event=NOT_FOUND" |
| ) |
|
|
|
|
| def _install_phase_tracing(namespace: dict[str, object]) -> None: |
| main_function = namespace.get("main") |
| if not callable(main_function): |
| raise RuntimeError("Private CANIS main function is unavailable") |
| runtime_globals = main_function.__globals__ |
|
|
| for function_name, label in { |
| "load_fcgf_model": "runtime.load_fcgf", |
| "run_src_chain": "canonicalize.src_chain", |
| "run_registration": "canonicalize.registration", |
| "save_canonicalized_glb": "canonicalize.save_glb", |
| "save_front_view_glb": "canonicalize.save_front_view", |
| "result_tuple": "canonicalize.format_outputs", |
| }.items(): |
| function = runtime_globals.get(function_name) |
| if callable(function): |
| runtime_globals[function_name] = _traced_phase(label, function) |
|
|
| original_load_runtime = runtime_globals.get("load_runtime") |
| if not callable(original_load_runtime): |
| raise RuntimeError("Private CANIS load_runtime function is unavailable") |
|
|
| @wraps(original_load_runtime) |
| def traced_load_runtime(*args, **kwargs): |
| started = time.perf_counter() |
| _trace("phase=runtime.load_all_models event=START") |
| try: |
| with _trace_runtime_imports_and_calls(): |
| result = original_load_runtime(*args, **kwargs) |
| except Exception as exc: |
| _trace( |
| "phase=runtime.load_all_models event=ERROR " |
| f"elapsed={time.perf_counter() - started:.3f}s " |
| f"type={type(exc).__name__} message={str(exc)[:300]!r}" |
| ) |
| raise |
| _install_imported_module_tracing() |
| _trace( |
| "phase=runtime.load_all_models event=DONE " |
| f"elapsed={time.perf_counter() - started:.3f}s" |
| ) |
| return result |
|
|
| runtime_globals["load_runtime"] = traced_load_runtime |
| _trace("phase_instrumentation event=READY") |
|
|
|
|
| def _install_imported_module_tracing() -> None: |
| _wrap_module_phases( |
| "src_chain.input_assets", |
| {"prepare_input_asset": "input.prepare_asset"}, |
| ) |
| _wrap_module_phases( |
| "src_chain.run_object_pipeline", |
| { |
| "init_trellis_context": "runtime.load_trellis", |
| "run_once": "src_chain.run_once", |
| "encode_sparse_structure": "src_chain.encode_sparse_structure", |
| "render_and_select_views": "src_chain.render_views", |
| "run_trellis_multi_image": "src_chain.trellis_inference", |
| "build_patch_block_records_from_masks": "src_chain.patch_blocks", |
| "build_render_3d_records_from_patch_blocks": "src_chain.render_3d_records", |
| "build_projection_info_by_ind": "src_chain.projection_fit", |
| "build_trellis_3d_records_chain": "src_chain.correspondence", |
| "save_chain_visualizations": "src_chain.visualization", |
| "extract_pairs_by_patch_per_ind": "src_chain.extract_pairs", |
| "flatten_pairs_to_matches": "src_chain.flatten_matches", |
| "add_scaled_centered_trellis_xyz_to_matches": "src_chain.scale_matches", |
| }, |
| ) |
| _wrap_module_phases( |
| "src.mac_fcgf_anchor", |
| { |
| "extract_fcgf_features": "registration.fcgf_features", |
| "Registration_and_estimationT_hyper": "registration.mac", |
| "Refine_icp": "registration.icp", |
| }, |
| ) |
| _trace("deep_phase_instrumentation event=READY") |
|
|
|
|
| def _trace_gpu_environment() -> None: |
| import torch |
|
|
| details = { |
| "torch": torch.__version__, |
| "torch_cuda": torch.version.cuda, |
| "cuda_available": torch.cuda.is_available(), |
| "device_count": torch.cuda.device_count(), |
| } |
| if torch.cuda.is_available(): |
| details.update( |
| { |
| "device": torch.cuda.get_device_name(0), |
| "capability": torch.cuda.get_device_capability(0), |
| "allocated": torch.cuda.memory_allocated(0), |
| "reserved": torch.cuda.memory_reserved(0), |
| } |
| ) |
| _trace(f"gpu_environment {details!r}") |
|
|
|
|
| def _sha256(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as handle: |
| for block in iter(lambda: handle.read(1024 * 1024), b""): |
| digest.update(block) |
| return digest.hexdigest() |
|
|
|
|
| def _verify_private_runtime() -> None: |
| manifest_path = RUNTIME_ROOT / "runtime_manifest.json" |
| if not manifest_path.is_file(): |
| raise FileNotFoundError("Private CANIS runtime manifest is unavailable") |
| manifest = json.loads(manifest_path.read_text(encoding="utf-8")) |
| files = manifest.get("files") |
| if not isinstance(files, list) or not files: |
| raise RuntimeError("Private CANIS runtime manifest is invalid") |
| runtime_resolved = RUNTIME_ROOT.resolve() |
| for record in files: |
| relative = Path(record["path"]) |
| target = (RUNTIME_ROOT / relative).resolve() |
| if runtime_resolved not in target.parents or not target.is_file(): |
| raise RuntimeError("Private CANIS runtime integrity check failed") |
| if target.stat().st_size != int(record["size"]): |
| raise RuntimeError("Private CANIS runtime integrity check failed") |
| if _sha256(target) != record["sha256"]: |
| raise RuntimeError("Private CANIS runtime integrity check failed") |
|
|
|
|
| def _install_python_compatibility() -> None: |
| if not hasattr(collections, "Sequence"): |
| collections.Sequence = collections.abc.Sequence |
|
|
|
|
| def _install_trusted_checkpoint_compatibility() -> None: |
| import torch |
|
|
| original_load = torch.load |
|
|
| @wraps(original_load) |
| def compatible_load(file, *args, **kwargs): |
| try: |
| filename = Path(os.fspath(file)).name |
| except TypeError: |
| filename = "" |
| if filename == "ResUNetBN2C-feat32-3dmatch-v0.05.pth": |
| kwargs.setdefault("weights_only", False) |
| return original_load(file, *args, **kwargs) |
|
|
| torch.load = compatible_load |
|
|
|
|
| def _install_pytorch3d_compatibility() -> None: |
| from pytorch3d import _C |
|
|
| if not hasattr(_C, "MAX_UINT"): |
| _C.MAX_UINT = 2**32 - 1 |
| if not hasattr(_C, "EPS"): |
| _C.EPS = 1e-6 |
|
|
|
|
| def _install_blackwell_attention_compatibility() -> None: |
| import xformers.ops as xops |
|
|
| original_attention = xops.memory_efficient_attention |
| cutlass_op = (xops.fmha.cutlass.FwOp, xops.fmha.cutlass.BwOp) |
| for operator in cutlass_op: |
| operator.CUDA_MAXIMUM_COMPUTE_CAPABILITY = (12, 0) |
|
|
| @wraps(original_attention) |
| def cutlass_attention(*args, **kwargs): |
| kwargs.setdefault("op", cutlass_op) |
| return original_attention(*args, **kwargs) |
|
|
| xops.memory_efficient_attention = cutlass_attention |
|
|
|
|
| def _install_fpfh_only_compatibility() -> None: |
| """Keep the dormant FCGF branch from importing MinkowskiEngine in FPFH mode.""" |
| module_name = "src.mise.cal_fcgf" |
| if module_name in sys.modules: |
| raise RuntimeError("FCGF module was imported before FPFH-only setup") |
| placeholder = types.ModuleType(module_name) |
|
|
| def disabled_extract_features(*args, **kwargs): |
| raise RuntimeError("FCGF feature extraction is disabled in the FPFH deployment") |
|
|
| placeholder.extract_features = disabled_extract_features |
| sys.modules[module_name] = placeholder |
| _trace("phase=runtime.registration_mode event=READY mode=fpfh minkowski=disabled") |
|
|
|
|
| def _validate_models() -> Path: |
| pipeline = MODEL_ROOT / "pipeline.json" |
| if not pipeline.is_file(): |
| raise FileNotFoundError("Required private model assets are unavailable") |
| return MODEL_ROOT |
|
|
|
|
| def _prepare_cached_pipeline_config(private_checkpoint: Path) -> Path: |
| """Use public copies of identical TRELLIS weights and keep CANIS-only assets private.""" |
| source_config = private_checkpoint / "pipeline.json" |
| config = json.loads(source_config.read_text(encoding="utf-8")) |
| model_config = config.get("args", {}).get("models") |
| if not isinstance(model_config, dict) or "sparse_structure_encoder" not in model_config: |
| raise RuntimeError("Private CANIS pipeline configuration is invalid") |
|
|
| hybrid_models = {} |
| for name, relative_model in model_config.items(): |
| if name == "sparse_structure_encoder": |
| private_model = private_checkpoint / str(relative_model) |
| if not private_model.with_suffix(".json").is_file(): |
| raise FileNotFoundError("Private sparse encoder configuration is unavailable") |
| if not private_model.with_suffix(".safetensors").is_file(): |
| raise FileNotFoundError("Private sparse encoder weights are unavailable") |
| hybrid_models[name] = str(private_model) |
| else: |
| hybrid_models[name] = f"{PUBLIC_TRELLIS_REPO}/{relative_model}" |
|
|
| config["args"]["models"] = hybrid_models |
| PIPELINE_CONFIG_ROOT.mkdir(parents=True, exist_ok=True) |
| target = PIPELINE_CONFIG_ROOT / "pipeline.json" |
| target.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8") |
| _trace( |
| "phase=runtime.prepare_model_sources event=DONE " |
| f"public_models={len(hybrid_models) - 1} private_models=1" |
| ) |
| return PIPELINE_CONFIG_ROOT |
|
|
|
|
| def _install_hf_dinov2_loader() -> None: |
| """Load the native DINOv2 code locally and its exact weight from HF Hub.""" |
| hubconf = DINOV2_SOURCE_ROOT / "hubconf.py" |
| if not hubconf.is_file(): |
| raise FileNotFoundError("Local DINOv2 source is unavailable") |
|
|
| import torch |
| from huggingface_hub import hf_hub_download |
|
|
| started = time.perf_counter() |
| _trace( |
| "phase=runtime.dinov2.hf_download event=START " |
| f"repo={DINOV2_HF_REPO} revision={DINOV2_HF_REVISION[:12]}" |
| ) |
| cached_checkpoint = Path( |
| hf_hub_download( |
| repo_id=DINOV2_HF_REPO, |
| filename=DINOV2_CHECKPOINT, |
| revision=DINOV2_HF_REVISION, |
| cache_dir="/tmp/canis_hf_cache", |
| ) |
| ) |
| checkpoint_size = cached_checkpoint.stat().st_size |
| if checkpoint_size != DINOV2_CHECKPOINT_SIZE: |
| raise RuntimeError( |
| f"DINOv2 checkpoint size mismatch: {checkpoint_size} != " |
| f"{DINOV2_CHECKPOINT_SIZE}" |
| ) |
|
|
| torch_hub_root = Path("/tmp/canis_torch_hub") |
| checkpoint_root = torch_hub_root / "checkpoints" |
| checkpoint_root.mkdir(parents=True, exist_ok=True) |
| checkpoint_link = checkpoint_root / DINOV2_CHECKPOINT |
| checkpoint_link.unlink(missing_ok=True) |
| checkpoint_link.symlink_to(cached_checkpoint) |
| torch.hub.set_dir(str(torch_hub_root)) |
| _trace( |
| "phase=runtime.dinov2.hf_download event=DONE " |
| f"elapsed={time.perf_counter() - started:.3f}s bytes={checkpoint_size}" |
| ) |
|
|
| original_torch_hub_load = torch.hub.load |
| if getattr(original_torch_hub_load, "_canis_dinov2_local", False): |
| return |
|
|
| @wraps(original_torch_hub_load) |
| def controlled_torch_hub_load(repo_or_dir, model, *args, **kwargs): |
| if str(repo_or_dir) != "facebookresearch/dinov2": |
| return original_torch_hub_load(repo_or_dir, model, *args, **kwargs) |
| kwargs["source"] = "local" |
| kwargs.pop("trust_repo", None) |
| kwargs.pop("skip_validation", None) |
| _trace( |
| "phase=runtime.dinov2.local_source event=READY " |
| f"model={model} source_commit=2302b6bf4695" |
| ) |
| return original_torch_hub_load( |
| str(DINOV2_SOURCE_ROOT), model, *args, **kwargs |
| ) |
|
|
| controlled_torch_hub_load._canis_dinov2_local = True |
| torch.hub.load = controlled_torch_hub_load |
|
|
|
|
| def _sanitize_job_result(result, elapsed: float): |
| if not isinstance(result, (tuple, list)) or not result: |
| return result |
| status = result[0] |
| if isinstance(status, str) and ("Traceback" in status or "File \"" in status): |
| sanitized = list(result) |
| sanitized[0] = "Canonicalization failed. The owner can inspect the private runtime logs." |
| print(f"[CANIS] job failed after {elapsed:.2f}s; internal traceback suppressed", flush=True) |
| return tuple(sanitized) |
| print(f"[CANIS] job completed in {elapsed:.2f}s", flush=True) |
| return result |
|
|
|
|
|
|
| def _quota_message(message: str): |
| """Return a status-only result matching the private callback's ten outputs.""" |
| return ( |
| message, |
| None, |
| None, |
| "", |
| "", |
| None, |
| None, |
| None, |
| None, |
| None, |
| ) |
|
|
|
|
| def _quota_secret() -> bytes: |
| """Return a persistent private salt without storing visitor addresses.""" |
| if not QUOTA_ROOT.is_dir(): |
| raise RuntimeError("persistent quota volume is unavailable") |
| path = QUOTA_ROOT / ".anonymous-quota-secret" |
| try: |
| return path.read_bytes() |
| except FileNotFoundError: |
| secret = os.urandom(32) |
| try: |
| descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) |
| except FileExistsError: |
| return path.read_bytes() |
| with os.fdopen(descriptor, "wb") as handle: |
| handle.write(secret) |
| handle.flush() |
| os.fsync(handle.fileno()) |
| return secret |
| except OSError as exc: |
| raise RuntimeError("anonymous quota secret is unavailable") from exc |
|
|
|
|
| def _normalized_ip(value: str | None) -> str | None: |
| if not value: |
| return None |
| candidate = value.strip() |
| if candidate.startswith("[") and "]" in candidate: |
| candidate = candidate[1:candidate.index("]")] |
| elif candidate.count(":") == 1 and "." in candidate: |
| candidate = candidate.rsplit(":", 1)[0] |
| try: |
| return ipaddress.ip_address(candidate).compressed |
| except ValueError: |
| return None |
|
|
|
|
| def _anonymous_identity(request) -> tuple[str, str]: |
| """Build a non-reversible daily-quota key from the proxy-reported client IP.""" |
| if request is None: |
| raise RuntimeError("request metadata is unavailable") |
| headers = getattr(request, "headers", {}) or {} |
| candidates = [] |
| forwarded = headers.get("x-forwarded-for") |
| if forwarded: |
| candidates.extend(part.strip() for part in forwarded.split(",")) |
| candidates.extend( |
| headers.get(name) |
| for name in ("cf-connecting-ip", "x-real-ip") |
| if headers.get(name) |
| ) |
| client = getattr(request, "client", None) |
| candidates.append(getattr(client, "host", None)) |
| address = next((ip for raw in candidates if (ip := _normalized_ip(raw))), None) |
| if address is None: |
| raise RuntimeError("client address is unavailable") |
| user_key = hashlib.sha256( |
| b"canis-anonymous-ip-quota-v1:" + _quota_secret() + b":" + address.encode("ascii") |
| ).hexdigest() |
| return "Visitor", user_key |
|
|
|
|
| def _quota_day() -> str: |
| return datetime.now(timezone.utc).date().isoformat() |
|
|
|
|
| def _quota_path(day: str, user_key: str) -> Path: |
| if not QUOTA_ROOT.is_dir(): |
| raise RuntimeError("persistent quota volume is unavailable") |
| if len(user_key) != 64 or any(ch not in "0123456789abcdef" for ch in user_key): |
| raise RuntimeError("invalid quota account key") |
| daily_root = QUOTA_ROOT / "daily-anonymous-ip-v1" / day |
| daily_root.mkdir(parents=True, exist_ok=True) |
| return daily_root / f"{user_key}.json" |
|
|
|
|
| def _read_quota_count(day: str, user_key: str) -> int: |
| path = _quota_path(day, user_key) |
| if not path.exists(): |
| return 0 |
| try: |
| payload = json.loads(path.read_text(encoding="utf-8")) |
| except (OSError, json.JSONDecodeError) as exc: |
| raise RuntimeError("daily quota data could not be read") from exc |
| if not isinstance(payload, dict) or payload.get("date") != day: |
| raise RuntimeError("daily quota data is invalid") |
| count = payload.get("count") |
| if not isinstance(count, int) or count < 0: |
| raise RuntimeError("daily quota count is invalid") |
| return count |
|
|
|
|
| def _write_quota_count(day: str, user_key: str, count: int) -> None: |
| path = _quota_path(day, user_key) |
| temp = path.with_name( |
| f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp" |
| ) |
| payload = { |
| "schema": 3, |
| "date": day, |
| "count": count, |
| "limit": DAILY_CANONICALIZE_LIMIT, |
| "updated_at": datetime.now(timezone.utc).isoformat(), |
| } |
| try: |
| with temp.open("x", encoding="utf-8") as handle: |
| json.dump(payload, handle, sort_keys=True, separators=(",", ":")) |
| handle.write("\n") |
| handle.flush() |
| os.fsync(handle.fileno()) |
| os.replace(temp, path) |
| finally: |
| try: |
| temp.unlink(missing_ok=True) |
| except OSError: |
| pass |
|
|
|
|
| def _consume_daily_slot(user_key: str, day: str) -> tuple[str, int]: |
| """Atomically consume one attempt and prevent concurrent runs per visitor.""" |
| quota_key = (day, user_key) |
| with _quota_lock: |
| if quota_key in _active_visitors: |
| return "active", _read_quota_count(day, user_key) |
| used = _read_quota_count(day, user_key) |
| if used >= DAILY_CANONICALIZE_LIMIT: |
| return "limit", used |
| used += 1 |
| _write_quota_count(day, user_key, used) |
| _active_visitors.add(quota_key) |
| _trace( |
| "phase=quota.consume event=ALLOW " |
| f"day={day} user={user_key[:12]} used={used}" |
| ) |
| return "allowed", used |
|
|
|
|
| def _release_visitor(user_key: str, day: str) -> None: |
| with _quota_lock: |
| _active_visitors.discard((day, user_key)) |
|
|
|
|
| def _append_usage(result, display_name: str, used: int): |
| if not isinstance(result, (tuple, list)) or not result or not isinstance(result[0], str): |
| return _quota_message( |
| "Canonicalization returned an invalid response. This accepted attempt was counted." |
| ) |
| updated = list(result) |
| remaining = max(0, DAILY_CANONICALIZE_LIMIT - used) |
| updated[0] = ( |
| f"{updated[0]} | {display_name} | " |
| f"daily attempts {used}/{DAILY_CANONICALIZE_LIMIT}, remaining {remaining} " |
| "(resets 00:00 UTC)" |
| ) |
| return tuple(updated) |
|
|
| def main() -> None: |
| os.environ.setdefault("ATTN_BACKEND", "xformers") |
| os.environ.setdefault("SPARSE_ATTN", "xformers") |
| os.environ.setdefault("SPCONV_ALGO", "native") |
| os.environ.setdefault("NUMBA_THREADING_LAYER", "workqueue") |
| os.environ.setdefault("NUMBA_NUM_THREADS", "1") |
| os.environ.setdefault("OMP_NUM_THREADS", "12") |
| os.environ.setdefault("MPLCONFIGDIR", "/tmp/matplotlib") |
| os.environ.setdefault("NUMBA_CACHE_DIR", "/tmp/numba") |
| os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False") |
| _trace(f"phase=deployment_mode event=READY mode={DIAGNOSTIC_MODE}") |
|
|
| for directory in (OUTPUT_ROOT, POOL_ROOT, Path("/tmp/matplotlib"), Path("/tmp/numba")): |
| directory.mkdir(parents=True, exist_ok=True) |
|
|
| _verify_private_runtime() |
| _install_python_compatibility() |
| if DIAGNOSTIC_MODE != "zero_pack_pure_python_probe": |
| _install_trusted_checkpoint_compatibility() |
| if DIAGNOSTIC_MODE not in {"zero_pack_deferred_native_probe", "zero_pack_pure_python_probe"}: |
| _install_pytorch3d_compatibility() |
| _install_blackwell_attention_compatibility() |
| _trace("phase=runtime.registration_mode event=READY mode=fcgf minkowski=enabled") |
| private_checkpoint = _validate_models() |
| checkpoint = _prepare_cached_pipeline_config(private_checkpoint) |
| _trace("phase=runtime.model_source event=READY source=public_trellis_hybrid") |
|
|
| sys.path.insert(0, str(RUNTIME_ROOT)) |
| sys.path.insert(0, str(RUNTIME_ROOT / "deploy")) |
| import gradio as gr |
|
|
| def canonicalize_with_quota( |
| selected_glb_path, |
| uploaded_glb, |
| rx, |
| ry, |
| rz, |
| request, |
| ): |
| try: |
| display_name, user_key = _anonymous_identity(request) |
| except Exception as exc: |
| _trace( |
| "phase=quota.identity event=ERROR " |
| f"type={type(exc).__name__} message={str(exc)[:160]!r}" |
| ) |
| return _quota_message( |
| "This visitor's daily quota could not be verified, so Canonicalize is blocked " |
| "to protect the service. No GPU job was started." |
| ) |
| day = _quota_day() |
| try: |
| decision, used = _consume_daily_slot(user_key, day) |
| except Exception as exc: |
| _trace( |
| "phase=quota.consume event=ERROR " |
| f"type={type(exc).__name__} message={str(exc)[:160]!r}" |
| ) |
| return _quota_message( |
| "Daily quota storage is temporarily unavailable, so Canonicalize is blocked " |
| "to protect the service. No GPU job was started." |
| ) |
| if decision == "active": |
| _trace( |
| "phase=quota.consume event=DENY reason=active " |
| f"day={day} user={user_key[:12]} used={used}" |
| ) |
| return _quota_message( |
| "This visitor already has a Canonicalize job running. Wait for it to finish " |
| "before submitting another one." |
| ) |
| if decision == "limit": |
| _trace( |
| "phase=quota.consume event=DENY reason=limit " |
| f"day={day} user={user_key[:12]} used={used}" |
| ) |
| return _quota_message( |
| f"Daily limit reached: {used}/{DAILY_CANONICALIZE_LIMIT} " |
| "accepted attempts used. The quota resets at 00:00 UTC." |
| ) |
|
|
| try: |
| result = _canonicalize_on_gpu( |
| selected_glb_path, |
| uploaded_glb, |
| rx, |
| ry, |
| rz, |
| ) |
| except Exception as exc: |
| _trace( |
| "phase=quota.job event=ERROR " |
| f"day={day} user={user_key[:12]} " |
| f"type={type(exc).__name__} message={str(exc)[:160]!r}" |
| ) |
| return _quota_message( |
| "Canonicalization failed. The owner can inspect the private runtime logs. " |
| f"This accepted attempt was counted ({used}/{DAILY_CANONICALIZE_LIMIT})." |
| ) |
| finally: |
| _release_visitor(user_key, day) |
|
|
| return _append_usage(result, display_name, used) |
|
|
| canonicalize_with_quota.__annotations__["request"] = gr.Request |
|
|
| original_click = gr.Button.click |
| original_queue = gr.Blocks.queue |
| original_blocks_init = gr.Blocks.__init__ |
| queue_configured = False |
| launch_ui_kwargs = {} |
|
|
| @wraps(original_blocks_init) |
| def controlled_blocks_init(blocks, *args, **kwargs): |
| kwargs["css"] = ( |
| (kwargs.get("css") or "") |
| + "\nfooter{display:none!important}" |
| + "\n.sample-label{visibility:hidden!important}" |
| ) |
| signature = inspect.signature(original_blocks_init) |
| for name in ("theme", "css", "css_paths", "js", "head", "head_paths"): |
| if name in kwargs and name not in signature.parameters: |
| value = kwargs.pop(name) |
| if value is not None: |
| launch_ui_kwargs[name] = value |
| return original_blocks_init(blocks, *args, **kwargs) |
|
|
| gr.Blocks.__init__ = controlled_blocks_init |
|
|
| def controlled_click(button, fn=None, *args, **kwargs): |
| if callable(fn) and getattr(fn, "__name__", "") == "run_job": |
| button.value = "Canonicalize" |
| fn = canonicalize_with_quota |
| if "api_visibility" in inspect.signature(original_click).parameters: |
| kwargs.pop("api_name", None) |
| kwargs["api_visibility"] = "private" |
| else: |
| kwargs["api_name"] = False |
| kwargs["concurrency_limit"] = 1 |
| kwargs["concurrency_id"] = "canis-gpu" |
| kwargs["trigger_mode"] = "once" |
| elif callable(fn) and getattr(fn, "__name__", "") == "preview_rotated": |
| if "api_visibility" in inspect.signature(original_click).parameters: |
| kwargs.pop("api_name", None) |
| kwargs["api_visibility"] = "private" |
| else: |
| kwargs["api_name"] = False |
| return original_click(button, fn, *args, **kwargs) |
|
|
| gr.Button.click = controlled_click |
|
|
| @wraps(original_queue) |
| def controlled_queue(blocks, *args, **kwargs): |
| nonlocal queue_configured |
| signature = inspect.signature(original_queue) |
| bound = signature.bind_partial(blocks, *args, **kwargs) |
| enforced = { |
| "max_size": QUEUE_MAX_SIZE, |
| "default_concurrency_limit": 1, |
| "api_open": False, |
| } |
| for name, value in enforced.items(): |
| if name in signature.parameters: |
| bound.arguments[name] = value |
| queue_configured = True |
| _trace( |
| "phase=frontend.queue event=READY " |
| f"max_size={QUEUE_MAX_SIZE} concurrency=1 api_open=false" |
| ) |
| return original_queue(*bound.args, **bound.kwargs) |
|
|
| gr.Blocks.queue = controlled_queue |
|
|
| if not IS_ZERO_GPU: |
| original_launch = gr.Blocks.launch |
|
|
| @wraps(original_launch) |
| def controlled_launch(blocks, *args, **kwargs): |
| if not queue_configured: |
| controlled_queue(blocks) |
| for name, value in launch_ui_kwargs.items(): |
| kwargs.setdefault(name, value) |
| kwargs["ssr_mode"] = False |
| _trace("phase=frontend.ssr event=DISABLED hardware=dedicated") |
| return original_launch(blocks, *args, **kwargs) |
|
|
| gr.Blocks.launch = controlled_launch |
| argv = [ |
| sys.executable, |
| str(ORIGINAL_APP), |
| "--glb-root", str(PUBLIC_ROOT / "assets" / "examples"), |
| "--pool-root", str(POOL_ROOT), |
| "--out-root", str(OUTPUT_ROOT), |
| "--ckpt", str(checkpoint), |
| "--device", "cuda:0", |
| "--cuda-visible-devices", "0", |
| "--attn-backend", "xformers", |
| "--spconv-algo", "native", |
| "--server-name", "0.0.0.0", |
| "--server-port", os.environ.get("PORT", "7860"), |
| "--glb-limit", "12", |
| "--registration-desc", "fcgf", |
| ] |
| if not IS_ZERO_GPU and DIAGNOSTIC_MODE == "production_fcgf": |
| argv.append("--preload") |
| os.chdir(RUNTIME_ROOT) |
| sys.argv = argv[1:] |
| module_name = "canis_private_app" |
| spec = importlib.util.spec_from_file_location(module_name, ORIGINAL_APP) |
| if spec is None or spec.loader is None: |
| raise RuntimeError("Private CANIS app module cannot be loaded") |
| private_app = importlib.util.module_from_spec(spec) |
| sys.modules[module_name] = private_app |
| spec.loader.exec_module(private_app) |
|
|
| |
| |
| private_app.PREFERRED_GLB_STEMS = ( |
| "koalasample", |
| "cartoonelephantsample", |
| "03_1483a55c785d4379a25441681153cfce", |
| "06_2a96032b9a3b493d84a039d3d03452a2", |
| "16_9addb6a87d25402d994488ad84e32391", |
| "17_a337046823fc4e41a3cc7187d263d046", |
| "plushwolfsample", |
| "reindeersample", |
| "01_032b4472d1364069bb58bf801b59dbff", |
| "18_a8989077813b44e8837fa1b283d3c050", |
| "22_f81c470b8a6b4b5b9b318adbada57341", |
| "23_fa171cca6a624e70a6275f0167e5f61f", |
| ) |
|
|
| import canis_job_entry |
|
|
| canis_job_entry.configure(private_app, private_app.parse_args()) |
| _install_phase_tracing(vars(private_app)) |
| _install_hf_dinov2_loader() |
| private_app.main() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
