"""Public CANIS launcher; proprietary runtime code is loaded from a read-only volume.""" from __future__ import annotations import spaces # Must precede torch/CUDA imports for ZeroGPU emulation. 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 # Startup preload stays disabled so deserialization begins only after # ZeroGPU enters this worker; later calls reuse the cached CPU runtime. # This avoids restoring a packed tensor snapshot before callback START. 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) # Keep the gallery assets unchanged while placing the three requested # samples together in the third row (positions 7-9). 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()