"""Generate a depth map for the accessibility geometry pipeline. This script keeps depth estimation separate from reconstruction so the core pipeline can run in lightweight environments and can optionally use heavier models such as Depth Anything V2 when their dependencies are installed. """ from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any import numpy as np from PIL import Image, ImageOps DEFAULT_DEPTH_MODEL = 'depth-anything/Depth-Anything-V2-Small-hf' DEFAULT_VGGT_MODEL = 'facebook/VGGT-1B' def depth_model_identity( *, engine: str, model: str, checkpoint: str | Path | None, encoder: str, metric_depth: bool, vggt_model: str, ) -> dict[str, Any]: """Return an auditable third-party model identity without guessing revision.""" if engine == 'depth_anything_v2': size = { 'vits': 'Small', 'vitb': 'Base', 'vitl': 'Large', 'vitg': 'Giant', }.get(encoder, encoder) checkpoint_name = Path(checkpoint).name.lower() if checkpoint else '' dataset = ( 'Hypersim' if 'hypersim' in checkpoint_name else 'VKITTI' if 'vkitti' in checkpoint_name else None ) if metric_depth and dataset: model_id = f'depth-anything/Depth-Anything-V2-Metric-{dataset}-{size}' else: model_id = f'depth-anything/Depth-Anything-V2-{size}' return { 'model_id': model_id, 'family': 'Depth Anything V2', 'encoder': encoder, 'metric_training_dataset': dataset, 'license': 'Apache-2.0' if encoder == 'vits' else 'CC-BY-NC-4.0', 'third_party': True, 'revision': 'local_checkpoint_revision_unrecorded', 'revision_review_required_before_public_release': True, } if engine == 'transformers': return { 'model_id': model, 'family': 'Transformers depth-estimation pipeline', 'license': 'see_upstream_model_card', 'third_party': True, 'revision': 'runtime_default_or_local_cache', 'revision_review_required_before_public_release': True, } if engine == 'vggt': return { 'model_id': vggt_model, 'family': 'VGGT', 'license': 'see_upstream_model_card', 'third_party': True, 'revision': 'local_checkpoint_or_runtime_default', 'revision_review_required_before_public_release': True, } return { 'model_id': 'opencv_depth_fallback', 'family': 'nonlearned_fallback', 'third_party': True, } def read_rgb(path: str | Path, max_size: int | None = None) -> Image.Image: image = ImageOps.exif_transpose(Image.open(path)).convert('RGB') if max_size and max(image.size) > max_size: scale = max_size / max(image.size) size = (round(image.size[0] * scale), round(image.size[1] * scale)) image = image.resize(size, Image.Resampling.LANCZOS) return image def normalize_depth(depth: np.ndarray, near: float, far: float, invert: bool) -> np.ndarray: depth = depth.astype(np.float32) valid = np.isfinite(depth) if not np.any(valid): raise ValueError('Depth prediction contains no finite values') values = depth[valid] lo, hi = np.percentile(values, [1, 99]) if hi <= lo: hi = lo + 1.0 norm = np.clip((depth - lo) / (hi - lo), 0.0, 1.0) if invert: norm = 1.0 - norm return near + norm * (far - near) def resize_float_map(values: np.ndarray, size: tuple[int, int]) -> np.ndarray: image = Image.fromarray(values.astype(np.float32), mode='F') image = image.resize(size, Image.Resampling.BILINEAR) return np.asarray(image, dtype=np.float32) def colorize_confidence(confidence: np.ndarray) -> Image.Image: confidence = confidence.astype(np.float32) valid = np.isfinite(confidence) if not np.any(valid): return Image.fromarray(np.zeros(confidence.shape, dtype=np.uint8), mode='L') lo, hi = np.percentile(confidence[valid], [2, 98]) if hi <= lo: hi = lo + 1.0 vis = np.clip((confidence - lo) / (hi - lo), 0.0, 1.0) vis[~valid] = 0 return Image.fromarray((vis * 255).astype(np.uint8), mode='L') def preprocess_square_tensor(image: Image.Image, target_size: int): import torch width, height = image.size max_dim = max(width, height) left = (max_dim - width) // 2 top = (max_dim - height) // 2 scale = target_size / max_dim crop = ( left * scale, top * scale, (left + width) * scale, (top + height) * scale, ) square = Image.new('RGB', (max_dim, max_dim), (0, 0, 0)) square.paste(image, (left, top)) square = square.resize((target_size, target_size), Image.Resampling.BICUBIC) array = np.asarray(square, dtype=np.float32) / 255.0 tensor = torch.from_numpy(array).permute(2, 0, 1).contiguous() return tensor[None], crop def crop_and_resize_prediction(values: np.ndarray, crop: tuple[float, float, float, float], size: tuple[int, int]) -> np.ndarray: left, top, right, bottom = crop h, w = values.shape[:2] left_i = max(0, min(w - 1, int(np.floor(left)))) top_i = max(0, min(h - 1, int(np.floor(top)))) right_i = max(left_i + 1, min(w, int(np.ceil(right)))) bottom_i = max(top_i + 1, min(h, int(np.ceil(bottom)))) cropped = values[top_i:bottom_i, left_i:right_i] if values.ndim == 2: return resize_float_map(cropped, size) channels = [resize_float_map(cropped[..., channel], size) for channel in range(values.shape[-1])] return np.stack(channels, axis=-1).astype(np.float32) def write_point_cloud_ply( path: Path, points: np.ndarray, colors: np.ndarray, confidence: np.ndarray | None, confidence_threshold: float, max_points: int, ) -> int: valid = np.all(np.isfinite(points), axis=-1) & (points[..., 2] > 0) if confidence is not None: valid &= np.isfinite(confidence) & (confidence >= confidence_threshold) ys, xs = np.where(valid) if ys.size == 0: selected = np.array([], dtype=np.int64) elif max_points > 0 and ys.size > max_points: selected = np.linspace(0, ys.size - 1, max_points, dtype=np.int64) else: selected = np.arange(ys.size, dtype=np.int64) pts = points[ys[selected], xs[selected]] cols = colors[ys[selected], xs[selected]] path.parent.mkdir(parents=True, exist_ok=True) with path.open('w', encoding='ascii') as handle: handle.write('ply\nformat ascii 1.0\n') handle.write(f'element vertex {len(pts)}\n') handle.write('property float x\nproperty float y\nproperty float z\n') handle.write('property uchar red\nproperty uchar green\nproperty uchar blue\n') handle.write('end_header\n') for point, color in zip(pts, cols): handle.write( f'{point[0]:.6f} {point[1]:.6f} {point[2]:.6f} ' f'{int(color[0])} {int(color[1])} {int(color[2])}\n' ) return int(len(pts)) def heuristic_depth(image: Image.Image, near: float, far: float) -> np.ndarray: """Build a deterministic perspective prior from image position and edges.""" rgb = np.asarray(image, dtype=np.float32) / 255.0 h, w = rgb.shape[:2] yy = np.linspace(0.0, 1.0, h, dtype=np.float32)[:, None] depth = far - yy * (far - near) depth = np.repeat(depth, w, axis=1) gray = 0.299 * rgb[..., 0] + 0.587 * rgb[..., 1] + 0.114 * rgb[..., 2] grad_y = np.abs(np.gradient(gray, axis=0)) if np.isfinite(grad_y).any() and float(grad_y.max()) > 0: grad_y = grad_y / float(grad_y.max()) depth -= 0.08 * (far - near) * grad_y.astype(np.float32) return np.clip(depth, min(near, far), max(near, far)).astype(np.float32) def resolve_device(device: str) -> int | str: if device == 'cpu': return -1 if device == 'auto': import torch return 0 if torch.cuda.is_available() else -1 if device.startswith('cuda'): if ':' in device: return int(device.split(':', 1)[1]) return 0 return device def transformers_depth(image: Image.Image, model: str, device: str) -> np.ndarray: try: from transformers import pipeline except ImportError as exc: raise RuntimeError('transformers is not installed; install optional depth dependencies first') from exc estimator = pipeline('depth-estimation', model=model, device=resolve_device(device)) result: dict[str, Any] = estimator(image) if 'predicted_depth' in result: predicted = result['predicted_depth'] if hasattr(predicted, 'detach'): predicted = predicted.detach().cpu().numpy() depth = np.asarray(predicted, dtype=np.float32) if depth.ndim == 3: depth = depth.squeeze() elif 'depth' in result: depth = np.asarray(result['depth'], dtype=np.float32) else: raise RuntimeError(f'Unexpected depth-estimation result keys: {sorted(result)}') if depth.shape[:2] != (image.height, image.width): depth_img = Image.fromarray(depth.astype(np.float32), mode='F') depth_img = depth_img.resize(image.size, Image.Resampling.BILINEAR) depth = np.asarray(depth_img, dtype=np.float32) return depth.astype(np.float32) def resolve_torch_device(device: str) -> str: if device != 'auto': return device import torch return 'cuda' if torch.cuda.is_available() else 'cpu' def depth_anything_v2_depth( image: Image.Image, repo: str | Path, checkpoint: str | Path, encoder: str, device: str, input_size: int, metric: bool, max_depth: float, ) -> np.ndarray: repo = Path(repo).resolve() checkpoint = Path(checkpoint).resolve() if not checkpoint.exists(): raise FileNotFoundError(f'Depth Anything V2 checkpoint not found: {checkpoint}') if not repo.exists(): raise FileNotFoundError(f'Depth Anything V2 repo not found: {repo}') source_root = repo / 'metric_depth' if metric else repo if not source_root.exists(): raise FileNotFoundError(f'Depth Anything V2 source root not found: {source_root}') sys.path.insert(0, str(source_root)) import torch from depth_anything_v2.dpt import DepthAnythingV2 import depth_anything_v2.dinov2_layers.attention as da_attention import depth_anything_v2.dinov2_layers.block as da_block model_configs = { 'vits': {'encoder': 'vits', 'features': 64, 'out_channels': [48, 96, 192, 384]}, 'vitb': {'encoder': 'vitb', 'features': 128, 'out_channels': [96, 192, 384, 768]}, 'vitl': {'encoder': 'vitl', 'features': 256, 'out_channels': [256, 512, 1024, 1024]}, 'vitg': {'encoder': 'vitg', 'features': 384, 'out_channels': [1536, 1536, 1536, 1536]}, } if encoder not in model_configs: raise ValueError(f'Unsupported Depth Anything V2 encoder: {encoder}') config = dict(model_configs[encoder]) if metric: config['max_depth'] = max_depth torch_device = resolve_torch_device(device) # The vendored Depth Anything V2 DINOv2 blocks prefer xFormers when it is # importable, but the available xFormers kernels do not support every # server GPU/dtype combination, including RTX 5090 sm_120 with float32. # Use plain PyTorch attention for reproducible CPU and Slurm inference. da_attention.XFORMERS_AVAILABLE = False da_block.XFORMERS_AVAILABLE = False model = DepthAnythingV2(**config) model.load_state_dict(torch.load(str(checkpoint), map_location='cpu')) model = model.to(torch_device).eval() # ``image`` was EXIF-normalized and optionally downscaled by ``read_rgb``. # Depth Anything's infer_image receives BGR when loaded via OpenCV, so # preserve that channel convention without reopening the raw JPEG. raw = np.asarray(image.convert('RGB'))[:, :, ::-1].copy() with torch.inference_mode(): depth = model.infer_image(raw, input_size) return depth.astype(np.float32) def load_vggt_model( repo: str | Path, model_id: str, checkpoint: str | Path | None, device: str, ): repo = Path(repo).resolve() if not repo.exists(): raise FileNotFoundError(f'VGGT repo not found: {repo}') if str(repo) not in sys.path: sys.path.insert(0, str(repo)) import torch from vggt.models.vggt import VGGT torch_device = resolve_torch_device(device) # We only need camera/depth/point for accessibility geometry. Disabling # track avoids loading the point-tracking branch and lowers memory use. model = VGGT(enable_track=False) if checkpoint: checkpoint = Path(checkpoint).resolve() if not checkpoint.exists(): raise FileNotFoundError(f'VGGT checkpoint not found: {checkpoint}') state = torch.load(str(checkpoint), map_location='cpu') if isinstance(state, dict) and 'model' in state: state = state['model'] else: local_model = Path(model_id) if local_model.exists(): model_path = local_model / 'model.pt' if local_model.is_dir() else local_model if not model_path.exists(): raise FileNotFoundError(f'VGGT local model checkpoint not found: {model_path}') else: from huggingface_hub import hf_hub_download model_path = Path(hf_hub_download(repo_id=model_id, filename='model.pt')) state = torch.load(str(model_path), map_location='cpu') if isinstance(state, dict) and 'model' in state: state = state['model'] missing, unexpected = model.load_state_dict(state, strict=False) if missing: print(f'VGGT checkpoint missing keys: {len(missing)}', file=sys.stderr) if unexpected: print(f'VGGT checkpoint unexpected keys: {len(unexpected)}', file=sys.stderr) model = model.to(torch_device).eval() return model, torch_device def vggt_predict_depth( image: Image.Image, model, device: str, input_size: int, near: float, far: float, invert: bool, keep_raw_depth: bool, ) -> dict[str, np.ndarray]: import torch tensor, crop = preprocess_square_tensor(image, input_size) tensor = tensor.to(device) if device.startswith('cuda'): major = torch.cuda.get_device_capability()[0] dtype = torch.bfloat16 if major >= 8 else torch.float16 autocast = torch.cuda.amp.autocast(dtype=dtype) else: autocast = torch.autocast(device_type='cpu', enabled=False) with torch.inference_mode(): with autocast: predictions = model(tensor) raw_depth = predictions['depth'][0, 0, ..., 0].detach().float().cpu().numpy() depth_conf = predictions['depth_conf'][0, 0].detach().float().cpu().numpy() point_map = predictions['world_points'][0, 0].detach().float().cpu().numpy() point_conf = predictions['world_points_conf'][0, 0].detach().float().cpu().numpy() raw_depth = crop_and_resize_prediction(raw_depth, crop, image.size) depth_conf = crop_and_resize_prediction(depth_conf, crop, image.size) point_map = crop_and_resize_prediction(point_map, crop, image.size) point_conf = crop_and_resize_prediction(point_conf, crop, image.size) depth = raw_depth if keep_raw_depth else normalize_depth(raw_depth, near, far, invert) return { 'depth': depth.astype(np.float32), 'raw_depth': raw_depth.astype(np.float32), 'depth_conf': depth_conf.astype(np.float32), 'world_points': point_map.astype(np.float32), 'world_points_conf': point_conf.astype(np.float32), } def save_depth_vis(path: str | Path, depth: np.ndarray) -> None: valid = np.isfinite(depth) values = depth[valid] if values.size == 0: vis = np.zeros(depth.shape, dtype=np.uint8) else: lo, hi = np.percentile(values, [2, 98]) if hi <= lo: hi = lo + 1.0 vis = np.clip((depth - lo) / (hi - lo), 0.0, 1.0) vis = (vis * 255).astype(np.uint8) Image.fromarray(vis, mode='L').save(path) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description='Generate relative depth for accessibility geometry completion.') parser.add_argument('--image', required=True, help='Input RGB image.') parser.add_argument('--output-depth', required=True, help='Output .npy depth path.') parser.add_argument('--output-vis', default=None, help='Optional grayscale depth visualization path.') parser.add_argument('--manifest', default=None, help='Optional JSON manifest path.') parser.add_argument('--engine', choices=['heuristic', 'transformers', 'depth_anything_v2', 'vggt'], default='heuristic') parser.add_argument('--model', default=DEFAULT_DEPTH_MODEL, help='Transformers depth-estimation model id or local path.') parser.add_argument('--device', default='auto', help='auto, cpu, cuda, cuda:0, or a transformers device string.') parser.add_argument('--depth-anything-repo', default='../diffusion-vas/models/Depth_Anything_V2', help='Local Depth Anything V2 source repo.') parser.add_argument('--checkpoint', default=None, help='Depth Anything V2 .pth checkpoint for --engine depth_anything_v2.') parser.add_argument('--encoder', choices=['vits', 'vitb', 'vitl', 'vitg'], default='vitl') parser.add_argument('--metric-depth', action='store_true', help='Treat the Depth Anything V2 checkpoint as a metric-depth model and do not near/far normalize.') parser.add_argument('--max-depth', type=float, default=20.0, help='Metric Depth Anything max depth in meters.') parser.add_argument('--input-size', type=int, default=518, help='Depth Anything V2 inference input size.') parser.add_argument('--near', type=float, default=1.0, help='Depth value assigned to the near end after normalization.') parser.add_argument('--far', type=float, default=6.0, help='Depth value assigned to the far end after normalization.') parser.add_argument('--invert-depth', action='store_true', help='Invert predicted relative depth before near/far normalization.') parser.add_argument('--max-size', type=int, default=1280, help='Resize longest side before inference. Use 0 to keep original size.') parser.add_argument('--vggt-repo', default='vggt', help='Local facebookresearch/VGGT checkout.') parser.add_argument('--vggt-model', default=DEFAULT_VGGT_MODEL, help='Hugging Face model id or local VGGT model directory.') parser.add_argument('--vggt-checkpoint', default=None, help='Optional local VGGT model.pt checkpoint.') parser.add_argument('--vggt-input-size', type=int, default=518, help='Square VGGT inference resolution.') parser.add_argument('--vggt-keep-raw-depth', action='store_true', help='Do not near/far normalize VGGT depth before saving --output-depth.') parser.add_argument('--output-conf', default=None, help='Optional output .npy confidence map path.') parser.add_argument('--output-conf-vis', default=None, help='Optional output confidence visualization path.') parser.add_argument('--output-raw-depth', default=None, help='Optional output raw VGGT depth .npy path.') parser.add_argument('--output-world-points', default=None, help='Optional output VGGT world point map .npy path.') parser.add_argument('--output-point-conf', default=None, help='Optional output VGGT world point confidence .npy path.') parser.add_argument('--output-point-cloud', default=None, help='Optional output VGGT point cloud .ply path.') parser.add_argument('--point-conf-threshold', type=float, default=1.0, help='VGGT point confidence threshold for --output-point-cloud.') parser.add_argument('--max-point-cloud-points', type=int, default=120000, help='Maximum VGGT point-cloud vertices to export.') return parser def main() -> None: args = build_parser().parse_args() max_size = None if args.max_size == 0 else args.max_size image = read_rgb(args.image, max_size=max_size) if args.engine == 'heuristic': raw_depth = heuristic_depth(image, args.near, args.far) depth = raw_depth elif args.engine == 'transformers': raw_depth = transformers_depth(image, args.model, args.device) depth = normalize_depth(raw_depth, args.near, args.far, args.invert_depth) elif args.engine == 'depth_anything_v2': raw_depth = depth_anything_v2_depth( image, args.depth_anything_repo, args.checkpoint, args.encoder, args.device, args.input_size, args.metric_depth, args.max_depth, ) depth = raw_depth if args.metric_depth else normalize_depth(raw_depth, args.near, args.far, args.invert_depth) else: model, torch_device = load_vggt_model( args.vggt_repo, args.vggt_model, args.vggt_checkpoint, args.device, ) prediction = vggt_predict_depth( image, model, torch_device, args.vggt_input_size, args.near, args.far, args.invert_depth, args.vggt_keep_raw_depth, ) depth = prediction['depth'] raw_depth = prediction['raw_depth'] if args.output_conf: Path(args.output_conf).parent.mkdir(parents=True, exist_ok=True) np.save(Path(args.output_conf), prediction['depth_conf'].astype(np.float32)) if args.output_conf_vis: Path(args.output_conf_vis).parent.mkdir(parents=True, exist_ok=True) colorize_confidence(prediction['depth_conf']).save(args.output_conf_vis) if args.output_raw_depth: Path(args.output_raw_depth).parent.mkdir(parents=True, exist_ok=True) np.save(Path(args.output_raw_depth), prediction['raw_depth'].astype(np.float32)) if args.output_world_points: Path(args.output_world_points).parent.mkdir(parents=True, exist_ok=True) np.save(Path(args.output_world_points), prediction['world_points'].astype(np.float32)) if args.output_point_conf: Path(args.output_point_conf).parent.mkdir(parents=True, exist_ok=True) np.save(Path(args.output_point_conf), prediction['world_points_conf'].astype(np.float32)) if args.output_point_cloud: rgb = np.asarray(image.convert('RGB'), dtype=np.uint8) point_count = write_point_cloud_ply( Path(args.output_point_cloud), prediction['world_points'], rgb, prediction['world_points_conf'], args.point_conf_threshold, args.max_point_cloud_points, ) print(f'Wrote VGGT point cloud to {args.output_point_cloud} ({point_count} points)') output_depth = Path(args.output_depth) output_depth.parent.mkdir(parents=True, exist_ok=True) np.save(output_depth, depth.astype(np.float32)) output_vis = Path(args.output_vis) if args.output_vis else output_depth.with_suffix('.png') output_vis.parent.mkdir(parents=True, exist_ok=True) save_depth_vis(output_vis, depth) note = 'Metric monocular depth estimate; calibrate intrinsics/scale before path-planning use.' if args.engine == 'vggt': note = 'VGGT single-view/few-view geometry estimate; scale is not calibrated metric ground truth.' elif args.engine != 'depth_anything_v2' or not args.metric_depth: note = 'Relative monocular depth; use calibrated metric depth for path-planning truth.' model_identity = depth_model_identity( engine=args.engine, model=args.model, checkpoint=args.checkpoint, encoder=args.encoder, metric_depth=args.metric_depth, vggt_model=args.vggt_model, ) manifest = { 'image': args.image, 'raster_orientation_policy': 'RGB is decoded with PIL ImageOps.exif_transpose before depth inference.', 'display_raster_size': {'width': image.width, 'height': image.height}, 'engine': args.engine, 'model': model_identity['model_id'], 'model_identity': model_identity, 'depth_anything_repo': args.depth_anything_repo if args.engine == 'depth_anything_v2' else None, 'checkpoint': args.checkpoint if args.engine == 'depth_anything_v2' else None, 'encoder': args.encoder if args.engine == 'depth_anything_v2' else None, 'vggt_repo': args.vggt_repo if args.engine == 'vggt' else None, 'vggt_model': args.vggt_model if args.engine == 'vggt' else None, 'vggt_checkpoint': args.vggt_checkpoint if args.engine == 'vggt' else None, 'vggt_input_size': args.vggt_input_size if args.engine == 'vggt' else None, 'vggt_keep_raw_depth': args.vggt_keep_raw_depth if args.engine == 'vggt' else None, 'metric_depth': args.metric_depth if args.engine == 'depth_anything_v2' else None, 'max_depth': args.max_depth if args.metric_depth else None, 'device': args.device if args.engine in {'transformers', 'depth_anything_v2', 'vggt'} else None, 'output_depth': str(output_depth), 'output_vis': str(output_vis), 'output_conf': args.output_conf if args.engine == 'vggt' else None, 'output_raw_depth': args.output_raw_depth if args.engine == 'vggt' else None, 'output_world_points': args.output_world_points if args.engine == 'vggt' else None, 'output_point_cloud': args.output_point_cloud if args.engine == 'vggt' else None, 'near': args.near, 'far': args.far, 'invert_depth': args.invert_depth, 'shape': list(depth.shape), 'note': note, } manifest_path = Path(args.manifest) if args.manifest else output_depth.with_name('depth_manifest.json') manifest_path.write_text(json.dumps(manifest, indent=2), encoding='utf-8') print(f'Wrote depth to {output_depth}') print(f'Wrote depth visualization to {output_vis}') if __name__ == '__main__': main()