DA3-GIANT / da3_input_processor.py
bigshanedogg's picture
Upload DA3-GIANT (transformers port, original code)
e330fba verified
Raw History Blame Contribute Delete
26.8 kB
# ruff: noqa
# fmt: off
"""Depth Anything 3, vendored verbatim from ByteDance-Seed/depth-anything-3 @ 3d835ec1a5802d64a8b8b15f817a1ab54809bfe4: the files below concatenated,
package-internal imports replaced by relative imports of the other vendored files (generated by scripts/model_ports/common/vendor_flat.py; do not edit
by hand). Code license: Apache-2.0 (ByteDance)."""
from __future__ import annotations
# ==============================================================================================================
# depth_anything_3/utils/logger.py
# ==============================================================================================================
# Copyright (c) 2025 ByteDance Ltd. and/or its affiliates
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import sys
class Color:
RED = "\033[91m"
YELLOW = "\033[93m"
WHITE = "\033[97m"
GREEN = "\033[92m"
RESET = "\033[0m"
LOG_LEVELS = {"ERROR": 0, "WARN": 1, "INFO": 2, "DEBUG": 3}
COLOR_MAP = {"ERROR": Color.RED, "WARN": Color.YELLOW, "INFO": Color.WHITE, "DEBUG": Color.GREEN}
def get_env_log_level():
level = os.environ.get("DA3_LOG_LEVEL", "INFO").upper()
return LOG_LEVELS.get(level, LOG_LEVELS["INFO"])
class Logger:
def __init__(self):
self.level = get_env_log_level()
def log(self, level_str, *args, **kwargs):
level_key = level_str.split(":")[0].strip()
level_val = LOG_LEVELS.get(level_key)
if level_val is None:
raise ValueError(f"Unknown log level: {level_str}")
if self.level >= level_val:
color = COLOR_MAP[level_key]
msg = " ".join(str(arg) for arg in args)
# Align log level output in square brackets
# ERROR and DEBUG are 5 characters, INFO and WARN have an extra space for alignment
tag = level_key
if tag in ("INFO", "WARN"):
tag += " "
print(
f"{color}[{tag}] {msg}{Color.RESET}",
file=sys.stderr if level_key == "ERROR" else sys.stdout,
**kwargs,
)
def error(self, *args, **kwargs):
self.log("ERROR:", *args, **kwargs)
def warn(self, *args, **kwargs):
self.log("WARN:", *args, **kwargs)
def info(self, *args, **kwargs):
self.log("INFO:", *args, **kwargs)
def debug(self, *args, **kwargs):
self.log("DEBUG:", *args, **kwargs)
logger = Logger()
__all__ = ["logger"]
if __name__ == "__main__":
logger.info("This is an info message")
logger.warn("This is a warning message")
logger.error("This is an error message")
logger.debug("This is a debug message")
# ==============================================================================================================
# depth_anything_3/utils/parallel_utils.py
# ==============================================================================================================
# Copyright (c) 2025 ByteDance Ltd. and/or its affiliates
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import os
from functools import wraps
from multiprocessing.pool import ThreadPool
from threading import Thread
from typing import Callable, Dict, List
import imageio
from tqdm import tqdm
def async_call_func(func):
@wraps(func)
async def wrapper(*args, **kwargs):
loop = asyncio.get_event_loop()
# Use run_in_executor to run the blocking function in a separate thread
return await loop.run_in_executor(None, func, *args, **kwargs)
return wrapper
slice_func = lambda chunk_index, chunk_dim, chunk_size: [slice(None)] * chunk_dim + [
slice(chunk_index, chunk_index + chunk_size)
]
def async_call(fn):
def wrapper(*args, **kwargs):
Thread(target=fn, args=args, kwargs=kwargs).start()
return wrapper
def _save_image_impl(save_img, save_path):
"""Common implementation for saving images synchronously or asynchronously"""
os.makedirs(os.path.dirname(save_path), exist_ok=True)
imageio.imwrite(save_path, save_img)
@async_call
def save_image_async(save_img, save_path):
"""Save image asynchronously"""
_save_image_impl(save_img, save_path)
def save_image(save_img, save_path):
"""Save image synchronously"""
_save_image_impl(save_img, save_path)
def parallel_execution(
*args,
action: Callable,
num_processes=32,
print_progress=False,
sequential=False,
async_return=False,
desc=None,
**kwargs,
):
# Partially copy from EasyVolumetricVideo (parallel_execution)
# NOTE: we expect first arg / or kwargs to be distributed
# NOTE: print_progress arg is reserved.
# `*args` packs all positional arguments passed to the function into a tuple
args = list(args)
def get_length(args: List, kwargs: Dict):
for a in args:
if isinstance(a, list):
return len(a)
for v in kwargs.values():
if isinstance(v, list):
return len(v)
raise NotImplementedError
def get_action_args(length: int, args: List, kwargs: Dict, i: int):
action_args = [
(arg[i] if isinstance(arg, list) and len(arg) == length else arg) for arg in args
]
# TODO: Support all types of iterable
action_kwargs = {
key: (
kwargs[key][i]
if isinstance(kwargs[key], list) and len(kwargs[key]) == length
else kwargs[key]
)
for key in kwargs
}
return action_args, action_kwargs
if not sequential:
# Create ThreadPool
pool = ThreadPool(processes=num_processes)
# Spawn threads
results = []
asyncs = []
length = get_length(args, kwargs)
for i in range(length):
action_args, action_kwargs = get_action_args(length, args, kwargs, i)
async_result = pool.apply_async(action, action_args, action_kwargs)
asyncs.append(async_result)
# Join threads and get return values
if not async_return:
for async_result in tqdm(asyncs, desc=desc, disable=not print_progress):
results.append(async_result.get()) # will sync the corresponding thread
pool.close()
pool.join()
return results
else:
return pool
else:
results = []
length = get_length(args, kwargs)
for i in tqdm(range(length), desc=desc, disable=not print_progress):
action_args, action_kwargs = get_action_args(length, args, kwargs, i)
async_result = action(*action_args, **action_kwargs)
results.append(async_result)
return results
# ==============================================================================================================
# depth_anything_3/utils/io/input_processor.py
# ==============================================================================================================
# Copyright (c) 2025 ByteDance Ltd. and/or its affiliates
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Input processor for Depth Anything 3 (parallelized).
This version removes the square center-crop step for "*crop" methods (same as your note).
In addition, it parallelizes per-image preprocessing using the provided `parallel_execution`.
"""
from typing import Sequence
import cv2
import numpy as np
import torch
import torchvision.transforms as T
from PIL import Image
class InputProcessor:
"""Prepares a batch of images for model inference.
This processor converts a list of image file paths into a single, model-ready
tensor. The processing pipeline is executed in parallel across multiple workers
for efficiency.
Pipeline:
1) Load image and convert to RGB
2) Boundary resize (upper/lower bound, preserving aspect ratio)
3) Enforce divisibility by PATCH_SIZE:
- "*resize" methods: each dimension is rounded to nearest multiple
(may up/downscale a few px)
- "*crop" methods: each dimension is floored to nearest multiple via center crop
4) Convert to tensor and apply ImageNet normalization
5) Stack into (1, N, 3, H, W)
Parallelization:
- Each image is processed independently in a worker.
- Order of outputs matches the input order.
"""
NORMALIZE = T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
PATCH_SIZE = 14
def __init__(self):
pass
# -----------------------------
# Public API
# -----------------------------
def __call__(
self,
image: list[np.ndarray | Image.Image | str],
extrinsics: np.ndarray | None = None,
intrinsics: np.ndarray | None = None,
process_res: int = 504,
process_res_method: str = "upper_bound_resize",
*,
num_workers: int = 8,
print_progress: bool = False,
sequential: bool | None = None,
desc: str | None = "Preprocess",
) -> tuple[torch.Tensor, torch.Tensor | None, torch.Tensor | None]:
"""
Returns:
(tensor, extrinsics_list, intrinsics_list)
tensor shape: (1, N, 3, H, W)
"""
sequential = self._resolve_sequential(sequential, num_workers)
exts_list, ixts_list = self._validate_and_pack_meta(image, extrinsics, intrinsics)
results = self._run_parallel(
image=image,
exts_list=exts_list,
ixts_list=ixts_list,
process_res=process_res,
process_res_method=process_res_method,
num_workers=num_workers,
print_progress=print_progress,
sequential=sequential,
desc=desc,
)
proc_imgs, out_sizes, out_ixts, out_exts = self._unpack_results(results)
proc_imgs, out_sizes, out_ixts = self._unify_batch_shapes(proc_imgs, out_sizes, out_ixts)
batch_tensor = self._stack_batch(proc_imgs)
out_exts = (
torch.from_numpy(np.asarray(out_exts)).float()
if out_exts is not None and out_exts[0] is not None
else None
)
out_ixts = (
torch.from_numpy(np.asarray(out_ixts)).float()
if out_ixts is not None and out_ixts[0] is not None
else None
)
return (batch_tensor, out_exts, out_ixts)
# -----------------------------
# __call__ helpers
# -----------------------------
def _resolve_sequential(self, sequential: bool | None, num_workers: int) -> bool:
return (num_workers <= 1) if sequential is None else sequential
def _validate_and_pack_meta(
self,
images: list[np.ndarray | Image.Image | str],
extrinsics: np.ndarray | None,
intrinsics: np.ndarray | None,
) -> tuple[list[np.ndarray | None] | None, list[np.ndarray | None] | None]:
if extrinsics is not None and len(extrinsics) != len(images):
raise ValueError("Length of extrinsics must match images when provided.")
if intrinsics is not None and len(intrinsics) != len(images):
raise ValueError("Length of intrinsics must match images when provided.")
exts_list = [e for e in extrinsics] if extrinsics is not None else None
ixts_list = [k for k in intrinsics] if intrinsics is not None else None
return exts_list, ixts_list
def _run_parallel(
self,
*,
image: list[np.ndarray | Image.Image | str],
exts_list: list[np.ndarray | None] | None,
ixts_list: list[np.ndarray | None] | None,
process_res: int,
process_res_method: str,
num_workers: int,
print_progress: bool,
sequential: bool,
desc: str | None,
):
results = parallel_execution(
image,
exts_list,
ixts_list,
action=self._process_one, # (img, extrinsic, intrinsic, ...)
num_processes=num_workers,
print_progress=print_progress,
sequential=sequential,
desc=desc,
process_res=process_res,
process_res_method=process_res_method,
)
if not results:
raise RuntimeError(
"No preprocessing results returned. Check inputs and parallel_execution."
)
return results
def _unpack_results(self, results):
"""
results: List[Tuple[torch.Tensor, Tuple[H, W], Optional[np.ndarray], Optional[np.ndarray]]]
-> processed_images, out_sizes, out_intrinsics, out_extrinsics
"""
try:
processed_images, out_sizes, out_intrinsics, out_extrinsics = zip(*results)
except Exception as e:
raise RuntimeError(
"Unexpected results structure from parallel_execution: "
f"{type(results)} / sample: {results[0]}"
) from e
return list(processed_images), list(out_sizes), list(out_intrinsics), list(out_extrinsics)
def _unify_batch_shapes(
self,
processed_images: list[torch.Tensor],
out_sizes: list[tuple[int, int]],
out_intrinsics: list[np.ndarray | None],
) -> tuple[list[torch.Tensor], list[tuple[int, int]], list[np.ndarray | None]]:
"""Center-crop all tensors to the smallest H, W; adjust intrinsics' cx, cy accordingly."""
if len(set(out_sizes)) <= 1:
return processed_images, out_sizes, out_intrinsics
min_h = min(h for h, _ in out_sizes)
min_w = min(w for _, w in out_sizes)
logger.warn(
f"Images in batch have different sizes {out_sizes}; "
f"center-cropping all to smallest ({min_h},{min_w})"
)
center_crop = T.CenterCrop((min_h, min_w))
new_imgs, new_sizes, new_ixts = [], [], []
for img_t, (H, W), K in zip(processed_images, out_sizes, out_intrinsics):
crop_top = max(0, (H - min_h) // 2)
crop_left = max(0, (W - min_w) // 2)
new_imgs.append(center_crop(img_t))
new_sizes.append((min_h, min_w))
if K is None:
new_ixts.append(None)
else:
K_adj = K.copy()
K_adj[0, 2] -= crop_left
K_adj[1, 2] -= crop_top
new_ixts.append(K_adj)
return new_imgs, new_sizes, new_ixts
def _stack_batch(self, processed_images: list[torch.Tensor]) -> torch.Tensor:
return torch.stack(processed_images)
# -----------------------------
# Per-item worker
# -----------------------------
def _process_one(
self,
img: np.ndarray | Image.Image | str,
extrinsic: np.ndarray | None = None,
intrinsic: np.ndarray | None = None,
*,
process_res: int,
process_res_method: str,
) -> tuple[torch.Tensor, tuple[int, int], np.ndarray | None, np.ndarray | None]:
# Load & remember original size
pil_img = self._load_image(img)
orig_w, orig_h = pil_img.size
# Boundary resize
pil_img = self._resize_image(pil_img, process_res, process_res_method)
w, h = pil_img.size
intrinsic = self._resize_ixt(intrinsic, orig_w, orig_h, w, h)
# Enforce divisibility by PATCH_SIZE
if process_res_method.endswith("resize"):
pil_img = self._make_divisible_by_resize(pil_img, self.PATCH_SIZE)
new_w, new_h = pil_img.size
intrinsic = self._resize_ixt(intrinsic, w, h, new_w, new_h)
w, h = new_w, new_h
elif process_res_method.endswith("crop"):
pil_img = self._make_divisible_by_crop(pil_img, self.PATCH_SIZE)
new_w, new_h = pil_img.size
intrinsic = self._crop_ixt(intrinsic, w, h, new_w, new_h)
w, h = new_w, new_h
else:
raise ValueError(f"Unsupported process_res_method: {process_res_method}")
# Convert to tensor & normalize
img_tensor = self._normalize_image(pil_img)
_, H, W = img_tensor.shape
assert (W, H) == (w, h), "Tensor size mismatch with PIL image size after processing."
# Return: (img_tensor, (H, W), intrinsic, extrinsic)
return img_tensor, (H, W), intrinsic, extrinsic
# -----------------------------
# Intrinsics transforms
# -----------------------------
def _resize_ixt(
self,
intrinsic: np.ndarray | None,
orig_w: int,
orig_h: int,
w: int,
h: int,
) -> np.ndarray | None:
if intrinsic is None:
return None
K = intrinsic.copy()
# scale fx, cx by w ratio; fy, cy by h ratio
K[:1] *= w / float(orig_w)
K[1:2] *= h / float(orig_h)
return K
def _crop_ixt(
self,
intrinsic: np.ndarray | None,
orig_w: int,
orig_h: int,
w: int,
h: int,
) -> np.ndarray | None:
if intrinsic is None:
return None
K = intrinsic.copy()
crop_h = (orig_h - h) // 2
crop_w = (orig_w - w) // 2
K[0, 2] -= crop_w
K[1, 2] -= crop_h
return K
# -----------------------------
# I/O & normalization
# -----------------------------
def _load_image(self, img: np.ndarray | Image.Image | str) -> Image.Image:
if isinstance(img, str):
return Image.open(img).convert("RGB")
elif isinstance(img, np.ndarray):
# Assume HxWxC uint8/RGB
return Image.fromarray(img).convert("RGB")
elif isinstance(img, Image.Image):
return img.convert("RGB")
else:
raise ValueError(f"Unsupported image type: {type(img)}")
def _normalize_image(self, img: Image.Image) -> torch.Tensor:
img_tensor = T.ToTensor()(img)
return self.NORMALIZE(img_tensor)
# -----------------------------
# Boundary resizing
# -----------------------------
def _resize_image(self, img: Image.Image, target_size: int, method: str) -> Image.Image:
if method in ("upper_bound_resize", "upper_bound_crop"):
return self._resize_longest_side(img, target_size)
elif method in ("lower_bound_resize", "lower_bound_crop"):
return self._resize_shortest_side(img, target_size)
else:
raise ValueError(f"Unsupported resize method: {method}")
def _resize_longest_side(self, img: Image.Image, target_size: int) -> Image.Image:
w, h = img.size
longest = max(w, h)
if longest == target_size:
return img
scale = target_size / float(longest)
new_w = max(1, int(round(w * scale)))
new_h = max(1, int(round(h * scale)))
interpolation = cv2.INTER_CUBIC if scale > 1.0 else cv2.INTER_AREA
arr = cv2.resize(np.asarray(img), (new_w, new_h), interpolation=interpolation)
return Image.fromarray(arr)
def _resize_shortest_side(self, img: Image.Image, target_size: int) -> Image.Image:
w, h = img.size
shortest = min(w, h)
if shortest == target_size:
return img
scale = target_size / float(shortest)
new_w = max(1, int(round(w * scale)))
new_h = max(1, int(round(h * scale)))
interpolation = cv2.INTER_CUBIC if scale > 1.0 else cv2.INTER_AREA
arr = cv2.resize(np.asarray(img), (new_w, new_h), interpolation=interpolation)
return Image.fromarray(arr)
# -----------------------------
# Make divisible by PATCH_SIZE
# -----------------------------
def _make_divisible_by_crop(self, img: Image.Image, patch: int) -> Image.Image:
"""
Floor each dimension to the nearest multiple of PATCH_SIZE via center crop.
Example: 504x377 -> 504x364
"""
w, h = img.size
new_w = (w // patch) * patch
new_h = (h // patch) * patch
if new_w == w and new_h == h:
return img
left = (w - new_w) // 2
top = (h - new_h) // 2
return img.crop((left, top, left + new_w, top + new_h))
def _make_divisible_by_resize(self, img: Image.Image, patch: int) -> Image.Image:
"""
Round each dimension to nearest multiple of PATCH_SIZE via small resize.
"""
w, h = img.size
def nearest_multiple(x: int, p: int) -> int:
down = (x // p) * p
up = down + p
return up if abs(up - x) <= abs(x - down) else down
new_w = max(1, nearest_multiple(w, patch))
new_h = max(1, nearest_multiple(h, patch))
if new_w == w and new_h == h:
return img
upscale = (new_w > w) or (new_h > h)
interpolation = cv2.INTER_CUBIC if upscale else cv2.INTER_AREA
arr = cv2.resize(np.asarray(img), (new_w, new_h), interpolation=interpolation)
return Image.fromarray(arr)
# Backward compatibility alias
InputAdapter = InputProcessor
# ===========================
# Minimal test runner (parallel execution)
# ===========================
if __name__ == "__main__":
"""
Minimal test suite:
- Creates pairs of images so batch shapes match.
- Tests all four process_res_methods.
- Prints fx fy cx cy IN->OUT per image.
- Includes cases with K/E provided and with None.
"""
def fmt_k_line(K: np.ndarray | None) -> str:
if K is None:
return "None"
fx, fy, cx, cy = float(K[0, 0]), float(K[1, 1]), float(K[0, 2]), float(K[1, 2])
return f"fx={fx:.3f} fy={fy:.3f} cx={cx:.3f} cy={cy:.3f}"
def show_result(
tag: str,
tensor: torch.Tensor,
Ks_in: Sequence[np.ndarray | None] | None = None,
Ks_out: Sequence[np.ndarray | None] | None = None,
):
B, N, C, H, W = tensor.shape
print(f"[{tag}] shape={tuple(tensor.shape)} HxW=({H},{W}) div14=({H%14==0},{W%14==0})")
assert H % 14 == 0 and W % 14 == 0, f"{tag}: output size not divisible by 14!"
if Ks_in is not None or Ks_out is not None:
Ks_in = Ks_in or [None] * N
Ks_out = Ks_out or [None] * N
for i in range(N):
print(f" K[{i}]: {fmt_k_line(Ks_in[i])} -> {fmt_k_line(Ks_out[i])}")
proc = InputProcessor()
process_res = 504
methods = ["upper_bound_resize", "upper_bound_crop", "lower_bound_resize", "lower_bound_crop"]
# Example sizes (two orientations)
small_sizes = [(680, 1208), (1208, 680)]
large_sizes = [(1208, 680), (680, 1208)]
def make_K(w, h, fx=1200.0, fy=1100.0):
cx, cy = w / 2.0, h / 2.0
K = np.array([[fx, 0, cx], [0, fy, cy], [0, 0, 1]], dtype=np.float32)
return K
def run_suite(suite_name: str, sizes: list[tuple[int, int]]):
print(f"\n===== {suite_name} =====")
for w, h in sizes:
img = Image.new("RGB", (w, h), color=(123, 222, 100))
batch_imgs = [img, img]
# intrinsics / extrinsics examples
Ks_in = [make_K(w, h), make_K(w, h)]
Es_in = [np.eye(4, dtype=np.float32), np.eye(4, dtype=np.float32)]
for m in methods:
tensor, Es_out, Ks_out = proc(
image=batch_imgs,
process_res=process_res,
process_res_method=m,
num_workers=8,
print_progress=False,
intrinsics=Ks_in, # test with non-None
extrinsics=Es_in,
)
show_result(f"{suite_name} size=({w},{h}) | {m}", tensor, Ks_in, Ks_out)
# Also test None path
tensor2, Es_out2, Ks_out2 = proc(
image=batch_imgs,
process_res=process_res,
process_res_method="upper_bound_resize",
num_workers=8,
intrinsics=None,
extrinsics=None,
)
show_result(
f"{suite_name} size=({w},{h}) | upper_bound_resize | no K/E",
tensor2,
None,
Ks_out2,
)
run_suite("SMALL", small_sizes)
run_suite("LARGE", large_sizes)
# Extra sanity for 504x376
print("\n===== EXTRA sanity for 504x376 =====")
img_example = Image.new("RGB", (504, 376), color=(10, 20, 30))
Ks_in_extra = [make_K(504, 376, fx=900.0, fy=900.0), make_K(504, 376, fx=900.0, fy=900.0)]
out_r, _, Ks_out_r = proc(
image=[img_example, img_example],
process_res=504,
process_res_method="upper_bound_resize",
num_workers=8,
intrinsics=Ks_in_extra,
)
out_c, _, Ks_out_c = proc(
image=[img_example, img_example],
process_res=504,
process_res_method="upper_bound_crop",
num_workers=8,
intrinsics=Ks_in_extra,
)
_, _, _, Hr, Wr = out_r.shape
_, _, _, Hc, Wc = out_c.shape
print(f"upper_bound_resize -> ({Hr},{Wr}) (rounded to nearest multiple of 14)")
show_result("Ks after upper_bound_resize", out_r, Ks_in_extra, Ks_out_r)
print(f"upper_bound_crop -> ({Hc},{Wc}) (floored to multiple of 14)")
show_result("Ks after upper_bound_crop", out_c, Ks_in_extra, Ks_out_c)