Upload Video-Depth-Anything-Small (transformers port, original code)

fb8632e verified 1 day ago

25.6 kB

# ruff: noqa
# fmt: off
"""Video Depth Anything, vendored verbatim from DepthAnything/Video-Depth-Anything @ 4f5ae23172ba60fd7bc11ef671cca678842c7072: 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."""
from .vda_motion_module import TemporalModule


# ==============================================================================================================
# Video-Depth-Anything/video_depth_anything/util/blocks.py
# ==============================================================================================================
import torch.nn as nn


def _make_scratch(in_shape, out_shape, groups=1, expand=False):
    scratch = nn.Module()

    out_shape1 = out_shape
    out_shape2 = out_shape
    out_shape3 = out_shape
    if len(in_shape) >= 4:
        out_shape4 = out_shape

    if expand:
        out_shape1 = out_shape
        out_shape2 = out_shape * 2
        out_shape3 = out_shape * 4
        if len(in_shape) >= 4:
            out_shape4 = out_shape * 8

    scratch.layer1_rn = nn.Conv2d(
        in_shape[0], out_shape1, kernel_size=3, stride=1, padding=1, bias=False, groups=groups
    )
    scratch.layer2_rn = nn.Conv2d(
        in_shape[1], out_shape2, kernel_size=3, stride=1, padding=1, bias=False, groups=groups
    )
    scratch.layer3_rn = nn.Conv2d(
        in_shape[2], out_shape3, kernel_size=3, stride=1, padding=1, bias=False, groups=groups
    )
    if len(in_shape) >= 4:
        scratch.layer4_rn = nn.Conv2d(
            in_shape[3], out_shape4, kernel_size=3, stride=1, padding=1, bias=False, groups=groups
        )

    return scratch


class ResidualConvUnit(nn.Module):
    """Residual convolution module."""

    def __init__(self, features, activation, bn):
        """Init.
        Args:
            features (int): number of features
        """
        super().__init__()

        self.bn = bn

        self.groups = 1

        self.conv1 = nn.Conv2d(
            features, features, kernel_size=3, stride=1, padding=1, bias=True, groups=self.groups
        )

        self.conv2 = nn.Conv2d(
            features, features, kernel_size=3, stride=1, padding=1, bias=True, groups=self.groups
        )

        if self.bn is True:
            self.bn1 = nn.BatchNorm2d(features)
            self.bn2 = nn.BatchNorm2d(features)

        self.activation = activation

        self.skip_add = nn.quantized.FloatFunctional()

    def forward(self, x):
        """Forward pass.
        Args:
            x (tensor): input
        Returns:
            tensor: output
        """

        out = self.activation(x)
        out = self.conv1(out)
        if self.bn is True:
            out = self.bn1(out)

        out = self.activation(out)
        out = self.conv2(out)
        if self.bn is True:
            out = self.bn2(out)

        if self.groups > 1:
            out = self.conv_merge(out)

        return self.skip_add.add(out, x)


class FeatureFusionBlock(nn.Module):
    """Feature fusion block."""

    def __init__(
        self,
        features,
        activation,
        deconv=False,
        bn=False,
        expand=False,
        align_corners=True,
        size=None,
    ):
        """Init.
        Args:
            features (int): number of features
        """
        super().__init__()

        self.deconv = deconv
        self.align_corners = align_corners

        self.groups = 1

        self.expand = expand
        out_features = features
        if self.expand is True:
            out_features = features // 2

        self.out_conv = nn.Conv2d(
            features, out_features, kernel_size=1, stride=1, padding=0, bias=True, groups=1
        )

        self.resConfUnit1 = ResidualConvUnit(features, activation, bn)
        self.resConfUnit2 = ResidualConvUnit(features, activation, bn)

        self.skip_add = nn.quantized.FloatFunctional()

        self.size = size

    def forward(self, *xs, size=None):
        """Forward pass.
        Returns:
            tensor: output
        """
        output = xs[0]

        if len(xs) == 2:
            res = self.resConfUnit1(xs[1])
            output = self.skip_add.add(output, res)

        output = self.resConfUnit2(output)

        if (size is None) and (self.size is None):
            modifier = {"scale_factor": 2}
        elif size is None:
            modifier = {"size": self.size}
        else:
            modifier = {"size": size}

        output = nn.functional.interpolate(
            output.contiguous(), **modifier, mode="bilinear", align_corners=self.align_corners
        )

        output = self.out_conv(output)

        return output


# ==============================================================================================================
# Video-Depth-Anything/video_depth_anything/util/transform.py
# ==============================================================================================================
import numpy as np
import cv2


class Resize(object):
    """Resize sample to given size (width, height).
    """

    def __init__(
        self,
        width,
        height,
        resize_target=True,
        keep_aspect_ratio=False,
        ensure_multiple_of=1,
        resize_method="lower_bound",
        image_interpolation_method=cv2.INTER_AREA,
    ):
        """Init.
        Args:
            width (int): desired output width
            height (int): desired output height
            resize_target (bool, optional):
                True: Resize the full sample (image, mask, target).
                False: Resize image only.
                Defaults to True.
            keep_aspect_ratio (bool, optional):
                True: Keep the aspect ratio of the input sample.
                Output sample might not have the given width and height, and
                resize behaviour depends on the parameter 'resize_method'.
                Defaults to False.
            ensure_multiple_of (int, optional):
                Output width and height is constrained to be multiple of this parameter.
                Defaults to 1.
            resize_method (str, optional):
                "lower_bound": Output will be at least as large as the given size.
                "upper_bound": Output will be at max as large as the given size. (Output size might be smaller than given size.)
                "minimal": Scale as least as possible.  (Output size might be smaller than given size.)
                Defaults to "lower_bound".
        """
        self.__width = width
        self.__height = height

        self.__resize_target = resize_target
        self.__keep_aspect_ratio = keep_aspect_ratio
        self.__multiple_of = ensure_multiple_of
        self.__resize_method = resize_method
        self.__image_interpolation_method = image_interpolation_method

    def constrain_to_multiple_of(self, x, min_val=0, max_val=None):
        y = (np.round(x / self.__multiple_of) * self.__multiple_of).astype(int)

        if max_val is not None and y > max_val:
            y = (np.floor(x / self.__multiple_of) * self.__multiple_of).astype(int)

        if y < min_val:
            y = (np.ceil(x / self.__multiple_of) * self.__multiple_of).astype(int)

        return y

    def get_size(self, width, height):
        # determine new height and width
        scale_height = self.__height / height
        scale_width = self.__width / width

        if self.__keep_aspect_ratio:
            if self.__resize_method == "lower_bound":
                # scale such that output size is lower bound
                if scale_width > scale_height:
                    # fit width
                    scale_height = scale_width
                else:
                    # fit height
                    scale_width = scale_height
            elif self.__resize_method == "upper_bound":
                # scale such that output size is upper bound
                if scale_width < scale_height:
                    # fit width
                    scale_height = scale_width
                else:
                    # fit height
                    scale_width = scale_height
            elif self.__resize_method == "minimal":
                # scale as least as possbile
                if abs(1 - scale_width) < abs(1 - scale_height):
                    # fit width
                    scale_height = scale_width
                else:
                    # fit height
                    scale_width = scale_height
            else:
                raise ValueError(f"resize_method {self.__resize_method} not implemented")

        if self.__resize_method == "lower_bound":
            new_height = self.constrain_to_multiple_of(scale_height * height, min_val=self.__height)
            new_width = self.constrain_to_multiple_of(scale_width * width, min_val=self.__width)
        elif self.__resize_method == "upper_bound":
            new_height = self.constrain_to_multiple_of(scale_height * height, max_val=self.__height)
            new_width = self.constrain_to_multiple_of(scale_width * width, max_val=self.__width)
        elif self.__resize_method == "minimal":
            new_height = self.constrain_to_multiple_of(scale_height * height)
            new_width = self.constrain_to_multiple_of(scale_width * width)
        else:
            raise ValueError(f"resize_method {self.__resize_method} not implemented")

        return (new_width, new_height)

    def __call__(self, sample):
        width, height = self.get_size(sample["image"].shape[1], sample["image"].shape[0])
        
        # resize sample
        sample["image"] = cv2.resize(sample["image"], (width, height), interpolation=self.__image_interpolation_method)

        if self.__resize_target:
            if "depth" in sample:
                sample["depth"] = cv2.resize(sample["depth"], (width, height), interpolation=cv2.INTER_NEAREST)
                
            if "mask" in sample:
                sample["mask"] = cv2.resize(sample["mask"].astype(np.float32), (width, height), interpolation=cv2.INTER_NEAREST)
        
        return sample


class NormalizeImage(object):
    """Normlize image by given mean and std.
    """

    def __init__(self, mean, std):
        self.__mean = mean
        self.__std = std

    def __call__(self, sample):
        sample["image"] = (sample["image"] - self.__mean) / self.__std

        return sample


class PrepareForNet(object):
    """Prepare sample for usage as network input.
    """

    def __init__(self):
        pass

    def __call__(self, sample):
        image = np.transpose(sample["image"], (2, 0, 1))
        sample["image"] = np.ascontiguousarray(image).astype(np.float32)

        if "depth" in sample:
            depth = sample["depth"].astype(np.float32)
            sample["depth"] = np.ascontiguousarray(depth)
        
        if "mask" in sample:
            sample["mask"] = sample["mask"].astype(np.float32)
            sample["mask"] = np.ascontiguousarray(sample["mask"])
        
        return sample


# ==============================================================================================================
# Video-Depth-Anything/video_depth_anything/dpt.py
# ==============================================================================================================
# Copyright (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 torch
import torch.nn as nn
import torch.nn.functional as F



def _make_fusion_block(features, use_bn, size=None):
    return FeatureFusionBlock(
        features,
        nn.ReLU(False),
        deconv=False,
        bn=use_bn,
        expand=False,
        align_corners=True,
        size=size,
    )


class ConvBlock(nn.Module):
    def __init__(self, in_feature, out_feature):
        super().__init__()
        
        self.conv_block = nn.Sequential(
            nn.Conv2d(in_feature, out_feature, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(out_feature),
            nn.ReLU(True)
        )
    
    def forward(self, x):
        return self.conv_block(x)


class DPTHead(nn.Module):
    def __init__(
        self, 
        in_channels, 
        features=256, 
        use_bn=False, 
        out_channels=[256, 512, 1024, 1024], 
        use_clstoken=False
    ):
        super(DPTHead, self).__init__()
        
        self.use_clstoken = use_clstoken
        
        self.projects = nn.ModuleList([
            nn.Conv2d(
                in_channels=in_channels,
                out_channels=out_channel,
                kernel_size=1,
                stride=1,
                padding=0,
            ) for out_channel in out_channels
        ])
        
        self.resize_layers = nn.ModuleList([
            nn.ConvTranspose2d(
                in_channels=out_channels[0],
                out_channels=out_channels[0],
                kernel_size=4,
                stride=4,
                padding=0),
            nn.ConvTranspose2d(
                in_channels=out_channels[1],
                out_channels=out_channels[1],
                kernel_size=2,
                stride=2,
                padding=0),
            nn.Identity(),
            nn.Conv2d(
                in_channels=out_channels[3],
                out_channels=out_channels[3],
                kernel_size=3,
                stride=2,
                padding=1)
        ])
        
        if use_clstoken:
            self.readout_projects = nn.ModuleList()
            for _ in range(len(self.projects)):
                self.readout_projects.append(
                    nn.Sequential(
                        nn.Linear(2 * in_channels, in_channels),
                        nn.GELU()))
        
        self.scratch = _make_scratch(
            out_channels,
            features,
            groups=1,
            expand=False,
        )
        
        self.scratch.stem_transpose = None
        
        self.scratch.refinenet1 = _make_fusion_block(features, use_bn)
        self.scratch.refinenet2 = _make_fusion_block(features, use_bn)
        self.scratch.refinenet3 = _make_fusion_block(features, use_bn)
        self.scratch.refinenet4 = _make_fusion_block(features, use_bn)
        
        head_features_1 = features
        head_features_2 = 32
        
        self.scratch.output_conv1 = nn.Conv2d(head_features_1, head_features_1 // 2, kernel_size=3, stride=1, padding=1)
        self.scratch.output_conv2 = nn.Sequential(
            nn.Conv2d(head_features_1 // 2, head_features_2, kernel_size=3, stride=1, padding=1),
            nn.ReLU(True),
            nn.Conv2d(head_features_2, 1, kernel_size=1, stride=1, padding=0),
            nn.ReLU(True),
            nn.Identity(),
        )
    
    def forward(self, out_features, patch_h, patch_w):
        out = []
        for i, x in enumerate(out_features):
            if self.use_clstoken:
                x, cls_token = x[0], x[1]
                readout = cls_token.unsqueeze(1).expand_as(x)
                x = self.readout_projects[i](torch.cat((x, readout), -1))
            else:
                x = x[0]
            
            x = x.permute(0, 2, 1).reshape((x.shape[0], x.shape[-1], patch_h, patch_w))
            
            x = self.projects[i](x)
            x = self.resize_layers[i](x)
            
            out.append(x)
        
        layer_1, layer_2, layer_3, layer_4 = out
        
        layer_1_rn = self.scratch.layer1_rn(layer_1)
        layer_2_rn = self.scratch.layer2_rn(layer_2)
        layer_3_rn = self.scratch.layer3_rn(layer_3)
        layer_4_rn = self.scratch.layer4_rn(layer_4)
        
        path_4 = self.scratch.refinenet4(layer_4_rn, size=layer_3_rn.shape[2:])        
        path_3 = self.scratch.refinenet3(path_4, layer_3_rn, size=layer_2_rn.shape[2:])
        path_2 = self.scratch.refinenet2(path_3, layer_2_rn, size=layer_1_rn.shape[2:])
        path_1 = self.scratch.refinenet1(path_2, layer_1_rn)
        
        out = self.scratch.output_conv1(path_1)
        out = F.interpolate(out, (int(patch_h * 14), int(patch_w * 14)), mode="bilinear", align_corners=True)
        out = self.scratch.output_conv2(out)
        
        return out


# ==============================================================================================================
# Video-Depth-Anything/video_depth_anything/dpt_temporal.py
# ==============================================================================================================
# Copyright (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 torch
import torch.nn.functional as F
import torch.nn as nn
from easydict import EasyDict


class DPTHeadTemporal(DPTHead):
    def __init__(self, 
        in_channels, 
        features=256, 
        use_bn=False, 
        out_channels=[256, 512, 1024, 1024], 
        use_clstoken=False,
        num_frames=32,
        pe='ape'
    ):
        super().__init__(in_channels, features, use_bn, out_channels, use_clstoken)

        assert num_frames > 0
        motion_module_kwargs = EasyDict(num_attention_heads                = 8,
                                        num_transformer_block              = 1,
                                        num_attention_blocks               = 2,
                                        temporal_max_len                   = num_frames,
                                        zero_initialize                    = True,
                                        pos_embedding_type                 = pe)

        self.motion_modules = nn.ModuleList([
            TemporalModule(in_channels=out_channels[2], 
                           **motion_module_kwargs),
            TemporalModule(in_channels=out_channels[3],
                           **motion_module_kwargs),
            TemporalModule(in_channels=features,
                           **motion_module_kwargs),
            TemporalModule(in_channels=features,
                           **motion_module_kwargs)
        ])

    def forward(self, out_features, patch_h, patch_w, frame_length, micro_batch_size=4, cached_hidden_state_list=None):
        out = []
        for i, x in enumerate(out_features):
            if self.use_clstoken:
                x, cls_token = x[0], x[1]
                readout = cls_token.unsqueeze(1).expand_as(x)
                x = self.readout_projects[i](torch.cat((x, readout), -1))
            else:
                x = x[0]

            x = x.permute(0, 2, 1).reshape((x.shape[0], x.shape[-1], patch_h, patch_w)).contiguous()

            B, T = x.shape[0] // frame_length, frame_length
            x = self.projects[i](x)
            x = self.resize_layers[i](x)

            out.append(x)

        layer_1, layer_2, layer_3, layer_4 = out

        B, T = layer_1.shape[0] // frame_length, frame_length
        if cached_hidden_state_list is not None:
            N = len(cached_hidden_state_list) // len(self.motion_modules)
        else:
            N = 0

        layer_3, h0 = self.motion_modules[0](layer_3.unflatten(0, (B, T)).permute(0, 2, 1, 3, 4), None, None, cached_hidden_state_list[0:N] if N else None)
        layer_3 = layer_3.permute(0, 2, 1, 3, 4).flatten(0, 1)
        layer_4, h1 = self.motion_modules[1](layer_4.unflatten(0, (B, T)).permute(0, 2, 1, 3, 4), None, None, cached_hidden_state_list[N:2*N] if N else None)
        layer_4 = layer_4.permute(0, 2, 1, 3, 4).flatten(0, 1)

        layer_1_rn = self.scratch.layer1_rn(layer_1)
        layer_2_rn = self.scratch.layer2_rn(layer_2)
        layer_3_rn = self.scratch.layer3_rn(layer_3)
        layer_4_rn = self.scratch.layer4_rn(layer_4)

        path_4 = self.scratch.refinenet4(layer_4_rn, size=layer_3_rn.shape[2:])
        path_4, h2 = self.motion_modules[2](path_4.unflatten(0, (B, T)).permute(0, 2, 1, 3, 4), None, None, cached_hidden_state_list[2*N:3*N] if N else None)
        path_4 = path_4.permute(0, 2, 1, 3, 4).flatten(0, 1)
        path_3 = self.scratch.refinenet3(path_4, layer_3_rn, size=layer_2_rn.shape[2:])
        path_3, h3 = self.motion_modules[3](path_3.unflatten(0, (B, T)).permute(0, 2, 1, 3, 4), None, None, cached_hidden_state_list[3*N:] if N else None)
        path_3 = path_3.permute(0, 2, 1, 3, 4).flatten(0, 1)

        batch_size = layer_1_rn.shape[0]
        if batch_size <= micro_batch_size or batch_size % micro_batch_size != 0:
            path_2 = self.scratch.refinenet2(path_3, layer_2_rn, size=layer_1_rn.shape[2:])
            path_1 = self.scratch.refinenet1(path_2, layer_1_rn)

            out = self.scratch.output_conv1(path_1)
            out = F.interpolate(
                out, (int(patch_h * 14), int(patch_w * 14)), mode="bilinear", align_corners=True
            )
            ori_type = out.dtype
            with torch.autocast(device_type="cuda", enabled=False):
                out = self.scratch.output_conv2(out.float())

            output = out.to(ori_type) 
        else:
            ret = []
            for i in range(0, batch_size, micro_batch_size):
                path_2 = self.scratch.refinenet2(path_3[i:i + micro_batch_size], layer_2_rn[i:i + micro_batch_size], size=layer_1_rn[i:i + micro_batch_size].shape[2:])
                path_1 = self.scratch.refinenet1(path_2, layer_1_rn[i:i + micro_batch_size])
                out = self.scratch.output_conv1(path_1)
                out = F.interpolate(
                    out, (int(patch_h * 14), int(patch_w * 14)), mode="bilinear", align_corners=True
                )
                ori_type = out.dtype
                with torch.autocast(device_type="cuda", enabled=False):
                    out = self.scratch.output_conv2(out.float())
                ret.append(out.to(ori_type))
            output = torch.cat(ret, dim=0)
        
        return output, h0 + h1 + h2 + h3


# ==============================================================================================================
# Video-Depth-Anything/utils/util.py
# ==============================================================================================================
# Copyright (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 numpy as np

def compute_scale_and_shift(prediction, target, mask, scale_only=False):
    if scale_only:
        return compute_scale(prediction, target, mask), 0
    else:
        return compute_scale_and_shift_full(prediction, target, mask)


def compute_scale(prediction, target, mask):
    # system matrix: A = [[a_00, a_01], [a_10, a_11]]
    prediction = prediction.astype(np.float32)
    target = target.astype(np.float32)
    mask = mask.astype(np.float32)

    a_00 = np.sum(mask * prediction * prediction)
    a_01 = np.sum(mask * prediction)
    a_11 = np.sum(mask)

    # right hand side: b = [b_0, b_1]
    b_0 = np.sum(mask * prediction * target)

    x_0 = b_0 / (a_00 + 1e-6)

    return x_0

def compute_scale_and_shift_full(prediction, target, mask):
    # system matrix: A = [[a_00, a_01], [a_10, a_11]]
    prediction = prediction.astype(np.float32)
    target = target.astype(np.float32)
    mask = mask.astype(np.float32)

    a_00 = np.sum(mask * prediction * prediction)
    a_01 = np.sum(mask * prediction)
    a_11 = np.sum(mask)

    b_0 = np.sum(mask * prediction * target)
    b_1 = np.sum(mask * target)

    x_0 = 1
    x_1 = 0

    det = a_00 * a_11 - a_01 * a_01

    if det != 0:
        x_0 = (a_11 * b_0 - a_01 * b_1) / det
        x_1 = (-a_01 * b_0 + a_00 * b_1) / det

    return x_0, x_1


def get_interpolate_frames(frame_list_pre, frame_list_post):
    assert len(frame_list_pre) == len(frame_list_post)
    min_w = 0.0
    max_w = 1.0
    step = (max_w - min_w) / (len(frame_list_pre)-1)
    post_w_list = [min_w] + [i * step for i in range(1,len(frame_list_pre)-1)] + [max_w]
    interpolated_frames = []
    for i in range(len(frame_list_pre)):
        interpolated_frames.append(frame_list_pre[i] * (1-post_w_list[i]) + frame_list_post[i] * post_w_list[i])
    return interpolated_frames

	# ruff: noqa
	# fmt: off
	"""Video Depth Anything, vendored verbatim from DepthAnything/Video-Depth-Anything @ 4f5ae23172ba60fd7bc11ef671cca678842c7072: 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."""
	from .vda_motion_module import TemporalModule


	# ==============================================================================================================
	# Video-Depth-Anything/video_depth_anything/util/blocks.py
	# ==============================================================================================================
	import torch.nn as nn


	def _make_scratch(in_shape, out_shape, groups=1, expand=False):
	scratch = nn.Module()

	out_shape1 = out_shape
	out_shape2 = out_shape
	out_shape3 = out_shape
	if len(in_shape) >= 4:
	out_shape4 = out_shape

	if expand:
	out_shape1 = out_shape
	out_shape2 = out_shape * 2
	out_shape3 = out_shape * 4
	if len(in_shape) >= 4:
	out_shape4 = out_shape * 8

	scratch.layer1_rn = nn.Conv2d(
	in_shape[0], out_shape1, kernel_size=3, stride=1, padding=1, bias=False, groups=groups
	)
	scratch.layer2_rn = nn.Conv2d(
	in_shape[1], out_shape2, kernel_size=3, stride=1, padding=1, bias=False, groups=groups
	)
	scratch.layer3_rn = nn.Conv2d(
	in_shape[2], out_shape3, kernel_size=3, stride=1, padding=1, bias=False, groups=groups
	)
	if len(in_shape) >= 4:
	scratch.layer4_rn = nn.Conv2d(
	in_shape[3], out_shape4, kernel_size=3, stride=1, padding=1, bias=False, groups=groups
	)

	return scratch


	class ResidualConvUnit(nn.Module):
	"""Residual convolution module."""

	def __init__(self, features, activation, bn):
	"""Init.

	Args:
	features (int): number of features
	"""
	super().__init__()

	self.bn = bn

	self.groups = 1

	self.conv1 = nn.Conv2d(
	features, features, kernel_size=3, stride=1, padding=1, bias=True, groups=self.groups
	)

	self.conv2 = nn.Conv2d(
	features, features, kernel_size=3, stride=1, padding=1, bias=True, groups=self.groups
	)

	if self.bn is True:
	self.bn1 = nn.BatchNorm2d(features)
	self.bn2 = nn.BatchNorm2d(features)

	self.activation = activation

	self.skip_add = nn.quantized.FloatFunctional()

	def forward(self, x):
	"""Forward pass.

	Args:
	x (tensor): input

	Returns:
	tensor: output
	"""

	out = self.activation(x)
	out = self.conv1(out)
	if self.bn is True:
	out = self.bn1(out)

	out = self.activation(out)
	out = self.conv2(out)
	if self.bn is True:
	out = self.bn2(out)

	if self.groups > 1:
	out = self.conv_merge(out)

	return self.skip_add.add(out, x)


	class FeatureFusionBlock(nn.Module):
	"""Feature fusion block."""

	def __init__(
	self,
	features,
	activation,
	deconv=False,
	bn=False,
	expand=False,
	align_corners=True,
	size=None,
	):
	"""Init.

	Args:
	features (int): number of features
	"""
	super().__init__()

	self.deconv = deconv
	self.align_corners = align_corners

	self.groups = 1

	self.expand = expand
	out_features = features
	if self.expand is True:
	out_features = features // 2

	self.out_conv = nn.Conv2d(
	features, out_features, kernel_size=1, stride=1, padding=0, bias=True, groups=1
	)

	self.resConfUnit1 = ResidualConvUnit(features, activation, bn)
	self.resConfUnit2 = ResidualConvUnit(features, activation, bn)

	self.skip_add = nn.quantized.FloatFunctional()

	self.size = size

	def forward(self, *xs, size=None):
	"""Forward pass.

	Returns:
	tensor: output
	"""
	output = xs[0]

	if len(xs) == 2:
	res = self.resConfUnit1(xs[1])
	output = self.skip_add.add(output, res)

	output = self.resConfUnit2(output)

	if (size is None) and (self.size is None):
	modifier = {"scale_factor": 2}
	elif size is None:
	modifier = {"size": self.size}
	else:
	modifier = {"size": size}

	output = nn.functional.interpolate(
	output.contiguous(), **modifier, mode="bilinear", align_corners=self.align_corners
	)

	output = self.out_conv(output)

	return output


	# ==============================================================================================================
	# Video-Depth-Anything/video_depth_anything/util/transform.py
	# ==============================================================================================================
	import numpy as np
	import cv2


	class Resize(object):
	"""Resize sample to given size (width, height).
	"""

	def __init__(
	self,
	width,
	height,
	resize_target=True,
	keep_aspect_ratio=False,
	ensure_multiple_of=1,
	resize_method="lower_bound",
	image_interpolation_method=cv2.INTER_AREA,
	):
	"""Init.

	Args:
	width (int): desired output width
	height (int): desired output height
	resize_target (bool, optional):
	True: Resize the full sample (image, mask, target).
	False: Resize image only.
	Defaults to True.
	keep_aspect_ratio (bool, optional):
	True: Keep the aspect ratio of the input sample.
	Output sample might not have the given width and height, and
	resize behaviour depends on the parameter 'resize_method'.
	Defaults to False.
	ensure_multiple_of (int, optional):
	Output width and height is constrained to be multiple of this parameter.
	Defaults to 1.
	resize_method (str, optional):
	"lower_bound": Output will be at least as large as the given size.
	"upper_bound": Output will be at max as large as the given size. (Output size might be smaller than given size.)
	"minimal": Scale as least as possible. (Output size might be smaller than given size.)
	Defaults to "lower_bound".
	"""
	self.__width = width
	self.__height = height

	self.__resize_target = resize_target
	self.__keep_aspect_ratio = keep_aspect_ratio
	self.__multiple_of = ensure_multiple_of
	self.__resize_method = resize_method
	self.__image_interpolation_method = image_interpolation_method

	def constrain_to_multiple_of(self, x, min_val=0, max_val=None):
	y = (np.round(x / self.__multiple_of) * self.__multiple_of).astype(int)

	if max_val is not None and y > max_val:
	y = (np.floor(x / self.__multiple_of) * self.__multiple_of).astype(int)

	if y < min_val:
	y = (np.ceil(x / self.__multiple_of) * self.__multiple_of).astype(int)

	return y

	def get_size(self, width, height):
	# determine new height and width
	scale_height = self.__height / height
	scale_width = self.__width / width

	if self.__keep_aspect_ratio:
	if self.__resize_method == "lower_bound":
	# scale such that output size is lower bound
	if scale_width > scale_height:
	# fit width
	scale_height = scale_width
	else:
	# fit height
	scale_width = scale_height
	elif self.__resize_method == "upper_bound":
	# scale such that output size is upper bound
	if scale_width < scale_height:
	# fit width
	scale_height = scale_width
	else:
	# fit height
	scale_width = scale_height
	elif self.__resize_method == "minimal":
	# scale as least as possbile
	if abs(1 - scale_width) < abs(1 - scale_height):
	# fit width
	scale_height = scale_width
	else:
	# fit height
	scale_width = scale_height
	else:
	raise ValueError(f"resize_method {self.__resize_method} not implemented")

	if self.__resize_method == "lower_bound":
	new_height = self.constrain_to_multiple_of(scale_height * height, min_val=self.__height)
	new_width = self.constrain_to_multiple_of(scale_width * width, min_val=self.__width)
	elif self.__resize_method == "upper_bound":
	new_height = self.constrain_to_multiple_of(scale_height * height, max_val=self.__height)
	new_width = self.constrain_to_multiple_of(scale_width * width, max_val=self.__width)
	elif self.__resize_method == "minimal":
	new_height = self.constrain_to_multiple_of(scale_height * height)
	new_width = self.constrain_to_multiple_of(scale_width * width)
	else:
	raise ValueError(f"resize_method {self.__resize_method} not implemented")

	return (new_width, new_height)

	def __call__(self, sample):
	width, height = self.get_size(sample["image"].shape[1], sample["image"].shape[0])

	# resize sample
	sample["image"] = cv2.resize(sample["image"], (width, height), interpolation=self.__image_interpolation_method)

	if self.__resize_target:
	if "depth" in sample:
	sample["depth"] = cv2.resize(sample["depth"], (width, height), interpolation=cv2.INTER_NEAREST)

	if "mask" in sample:
	sample["mask"] = cv2.resize(sample["mask"].astype(np.float32), (width, height), interpolation=cv2.INTER_NEAREST)

	return sample


	class NormalizeImage(object):
	"""Normlize image by given mean and std.
	"""

	def __init__(self, mean, std):
	self.__mean = mean
	self.__std = std

	def __call__(self, sample):
	sample["image"] = (sample["image"] - self.__mean) / self.__std

	return sample


	class PrepareForNet(object):
	"""Prepare sample for usage as network input.
	"""

	def __init__(self):
	pass

	def __call__(self, sample):
	image = np.transpose(sample["image"], (2, 0, 1))
	sample["image"] = np.ascontiguousarray(image).astype(np.float32)

	if "depth" in sample:
	depth = sample["depth"].astype(np.float32)
	sample["depth"] = np.ascontiguousarray(depth)

	if "mask" in sample:
	sample["mask"] = sample["mask"].astype(np.float32)
	sample["mask"] = np.ascontiguousarray(sample["mask"])

	return sample


	# ==============================================================================================================
	# Video-Depth-Anything/video_depth_anything/dpt.py
	# ==============================================================================================================
	# Copyright (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 torch
	import torch.nn as nn
	import torch.nn.functional as F



	def _make_fusion_block(features, use_bn, size=None):
	return FeatureFusionBlock(
	features,
	nn.ReLU(False),
	deconv=False,
	bn=use_bn,
	expand=False,
	align_corners=True,
	size=size,
	)


	class ConvBlock(nn.Module):
	def __init__(self, in_feature, out_feature):
	super().__init__()

	self.conv_block = nn.Sequential(
	nn.Conv2d(in_feature, out_feature, kernel_size=3, stride=1, padding=1),
	nn.BatchNorm2d(out_feature),
	nn.ReLU(True)
	)

	def forward(self, x):
	return self.conv_block(x)


	class DPTHead(nn.Module):
	def __init__(
	self,
	in_channels,
	features=256,
	use_bn=False,
	out_channels=[256, 512, 1024, 1024],
	use_clstoken=False
	):
	super(DPTHead, self).__init__()

	self.use_clstoken = use_clstoken

	self.projects = nn.ModuleList([
	nn.Conv2d(
	in_channels=in_channels,
	out_channels=out_channel,
	kernel_size=1,
	stride=1,
	padding=0,
	) for out_channel in out_channels
	])

	self.resize_layers = nn.ModuleList([
	nn.ConvTranspose2d(
	in_channels=out_channels[0],
	out_channels=out_channels[0],
	kernel_size=4,
	stride=4,
	padding=0),
	nn.ConvTranspose2d(
	in_channels=out_channels[1],
	out_channels=out_channels[1],
	kernel_size=2,
	stride=2,
	padding=0),
	nn.Identity(),
	nn.Conv2d(
	in_channels=out_channels[3],
	out_channels=out_channels[3],
	kernel_size=3,
	stride=2,
	padding=1)
	])

	if use_clstoken:
	self.readout_projects = nn.ModuleList()
	for _ in range(len(self.projects)):
	self.readout_projects.append(
	nn.Sequential(
	nn.Linear(2 * in_channels, in_channels),
	nn.GELU()))

	self.scratch = _make_scratch(
	out_channels,
	features,
	groups=1,
	expand=False,
	)

	self.scratch.stem_transpose = None

	self.scratch.refinenet1 = _make_fusion_block(features, use_bn)
	self.scratch.refinenet2 = _make_fusion_block(features, use_bn)
	self.scratch.refinenet3 = _make_fusion_block(features, use_bn)
	self.scratch.refinenet4 = _make_fusion_block(features, use_bn)

	head_features_1 = features
	head_features_2 = 32

	self.scratch.output_conv1 = nn.Conv2d(head_features_1, head_features_1 // 2, kernel_size=3, stride=1, padding=1)
	self.scratch.output_conv2 = nn.Sequential(
	nn.Conv2d(head_features_1 // 2, head_features_2, kernel_size=3, stride=1, padding=1),
	nn.ReLU(True),
	nn.Conv2d(head_features_2, 1, kernel_size=1, stride=1, padding=0),
	nn.ReLU(True),
	nn.Identity(),
	)

	def forward(self, out_features, patch_h, patch_w):
	out = []
	for i, x in enumerate(out_features):
	if self.use_clstoken:
	x, cls_token = x[0], x[1]
	readout = cls_token.unsqueeze(1).expand_as(x)
	x = self.readout_projects[i](torch.cat((x, readout), -1))
	else:
	x = x[0]

	x = x.permute(0, 2, 1).reshape((x.shape[0], x.shape[-1], patch_h, patch_w))

	x = self.projects[i](x)
	x = self.resize_layers[i](x)

	out.append(x)

	layer_1, layer_2, layer_3, layer_4 = out

	layer_1_rn = self.scratch.layer1_rn(layer_1)
	layer_2_rn = self.scratch.layer2_rn(layer_2)
	layer_3_rn = self.scratch.layer3_rn(layer_3)
	layer_4_rn = self.scratch.layer4_rn(layer_4)

	path_4 = self.scratch.refinenet4(layer_4_rn, size=layer_3_rn.shape[2:])
	path_3 = self.scratch.refinenet3(path_4, layer_3_rn, size=layer_2_rn.shape[2:])
	path_2 = self.scratch.refinenet2(path_3, layer_2_rn, size=layer_1_rn.shape[2:])
	path_1 = self.scratch.refinenet1(path_2, layer_1_rn)

	out = self.scratch.output_conv1(path_1)
	out = F.interpolate(out, (int(patch_h * 14), int(patch_w * 14)), mode="bilinear", align_corners=True)
	out = self.scratch.output_conv2(out)

	return out


	# ==============================================================================================================
	# Video-Depth-Anything/video_depth_anything/dpt_temporal.py
	# ==============================================================================================================
	# Copyright (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 torch
	import torch.nn.functional as F
	import torch.nn as nn
	from easydict import EasyDict


	class DPTHeadTemporal(DPTHead):
	def __init__(self,
	in_channels,
	features=256,
	use_bn=False,
	out_channels=[256, 512, 1024, 1024],
	use_clstoken=False,
	num_frames=32,
	pe='ape'
	):
	super().__init__(in_channels, features, use_bn, out_channels, use_clstoken)

	assert num_frames > 0
	motion_module_kwargs = EasyDict(num_attention_heads = 8,
	num_transformer_block = 1,
	num_attention_blocks = 2,
	temporal_max_len = num_frames,
	zero_initialize = True,
	pos_embedding_type = pe)

	self.motion_modules = nn.ModuleList([
	TemporalModule(in_channels=out_channels[2],
	**motion_module_kwargs),
	TemporalModule(in_channels=out_channels[3],
	**motion_module_kwargs),
	TemporalModule(in_channels=features,
	**motion_module_kwargs),
	TemporalModule(in_channels=features,
	**motion_module_kwargs)
	])

	def forward(self, out_features, patch_h, patch_w, frame_length, micro_batch_size=4, cached_hidden_state_list=None):
	out = []
	for i, x in enumerate(out_features):
	if self.use_clstoken:
	x, cls_token = x[0], x[1]
	readout = cls_token.unsqueeze(1).expand_as(x)
	x = self.readout_projects[i](torch.cat((x, readout), -1))
	else:
	x = x[0]

	x = x.permute(0, 2, 1).reshape((x.shape[0], x.shape[-1], patch_h, patch_w)).contiguous()

	B, T = x.shape[0] // frame_length, frame_length
	x = self.projects[i](x)
	x = self.resize_layers[i](x)

	out.append(x)

	layer_1, layer_2, layer_3, layer_4 = out

	B, T = layer_1.shape[0] // frame_length, frame_length
	if cached_hidden_state_list is not None:
	N = len(cached_hidden_state_list) // len(self.motion_modules)
	else:
	N = 0

	layer_3, h0 = self.motion_modules[0](layer_3.unflatten(0, (B, T)).permute(0, 2, 1, 3, 4), None, None, cached_hidden_state_list[0:N] if N else None)
	layer_3 = layer_3.permute(0, 2, 1, 3, 4).flatten(0, 1)
	layer_4, h1 = self.motion_modules[1](layer_4.unflatten(0, (B, T)).permute(0, 2, 1, 3, 4), None, None, cached_hidden_state_list[N:2*N] if N else None)
	layer_4 = layer_4.permute(0, 2, 1, 3, 4).flatten(0, 1)

	layer_1_rn = self.scratch.layer1_rn(layer_1)
	layer_2_rn = self.scratch.layer2_rn(layer_2)
	layer_3_rn = self.scratch.layer3_rn(layer_3)
	layer_4_rn = self.scratch.layer4_rn(layer_4)

	path_4 = self.scratch.refinenet4(layer_4_rn, size=layer_3_rn.shape[2:])
	path_4, h2 = self.motion_modules[2](path_4.unflatten(0, (B, T)).permute(0, 2, 1, 3, 4), None, None, cached_hidden_state_list[2N:3N] if N else None)
	path_4 = path_4.permute(0, 2, 1, 3, 4).flatten(0, 1)
	path_3 = self.scratch.refinenet3(path_4, layer_3_rn, size=layer_2_rn.shape[2:])
	path_3, h3 = self.motion_modules[3](path_3.unflatten(0, (B, T)).permute(0, 2, 1, 3, 4), None, None, cached_hidden_state_list[3*N:] if N else None)
	path_3 = path_3.permute(0, 2, 1, 3, 4).flatten(0, 1)

	batch_size = layer_1_rn.shape[0]
	if batch_size <= micro_batch_size or batch_size % micro_batch_size != 0:
	path_2 = self.scratch.refinenet2(path_3, layer_2_rn, size=layer_1_rn.shape[2:])
	path_1 = self.scratch.refinenet1(path_2, layer_1_rn)

	out = self.scratch.output_conv1(path_1)
	out = F.interpolate(
	out, (int(patch_h * 14), int(patch_w * 14)), mode="bilinear", align_corners=True
	)
	ori_type = out.dtype
	with torch.autocast(device_type="cuda", enabled=False):
	out = self.scratch.output_conv2(out.float())

	output = out.to(ori_type)
	else:
	ret = []
	for i in range(0, batch_size, micro_batch_size):
	path_2 = self.scratch.refinenet2(path_3[i:i + micro_batch_size], layer_2_rn[i:i + micro_batch_size], size=layer_1_rn[i:i + micro_batch_size].shape[2:])
	path_1 = self.scratch.refinenet1(path_2, layer_1_rn[i:i + micro_batch_size])
	out = self.scratch.output_conv1(path_1)
	out = F.interpolate(
	out, (int(patch_h * 14), int(patch_w * 14)), mode="bilinear", align_corners=True
	)
	ori_type = out.dtype
	with torch.autocast(device_type="cuda", enabled=False):
	out = self.scratch.output_conv2(out.float())
	ret.append(out.to(ori_type))
	output = torch.cat(ret, dim=0)

	return output, h0 + h1 + h2 + h3


	# ==============================================================================================================
	# Video-Depth-Anything/utils/util.py
	# ==============================================================================================================
	# Copyright (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 numpy as np

	def compute_scale_and_shift(prediction, target, mask, scale_only=False):
	if scale_only:
	return compute_scale(prediction, target, mask), 0
	else:
	return compute_scale_and_shift_full(prediction, target, mask)


	def compute_scale(prediction, target, mask):
	# system matrix: A = [[a_00, a_01], [a_10, a_11]]
	prediction = prediction.astype(np.float32)
	target = target.astype(np.float32)
	mask = mask.astype(np.float32)

	a_00 = np.sum(mask * prediction * prediction)
	a_01 = np.sum(mask * prediction)
	a_11 = np.sum(mask)

	# right hand side: b = [b_0, b_1]
	b_0 = np.sum(mask * prediction * target)

	x_0 = b_0 / (a_00 + 1e-6)

	return x_0

	def compute_scale_and_shift_full(prediction, target, mask):
	# system matrix: A = [[a_00, a_01], [a_10, a_11]]
	prediction = prediction.astype(np.float32)
	target = target.astype(np.float32)
	mask = mask.astype(np.float32)

	a_00 = np.sum(mask * prediction * prediction)
	a_01 = np.sum(mask * prediction)
	a_11 = np.sum(mask)

	b_0 = np.sum(mask * prediction * target)
	b_1 = np.sum(mask * target)

	x_0 = 1
	x_1 = 0

	det = a_00 * a_11 - a_01 * a_01

	if det != 0:
	x_0 = (a_11 * b_0 - a_01 * b_1) / det
	x_1 = (-a_01 * b_0 + a_00 * b_1) / det

	return x_0, x_1


	def get_interpolate_frames(frame_list_pre, frame_list_post):
	assert len(frame_list_pre) == len(frame_list_post)
	min_w = 0.0
	max_w = 1.0
	step = (max_w - min_w) / (len(frame_list_pre)-1)
	post_w_list = [min_w] + [i * step for i in range(1,len(frame_list_pre)-1)] + [max_w]
	interpolated_frames = []
	for i in range(len(frame_list_pre)):
	interpolated_frames.append(frame_list_pre[i] * (1-post_w_list[i]) + frame_list_post[i] * post_w_list[i])
	return interpolated_frames