tests/test_video_processing_common.py - main - githubmirror/transformers

tests/test_video_processing_common.py
523 строки24 KB
Defalt
[Tests] Fix slow video tensor creation from list of numpy arrays in SmolVLM (#44731)
16 июл 2026, 15:31
16 июл 2026, 15:31fe6ff6c
# Copyright 2025 HuggingFace Inc.
#
# 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 inspect
import json
import os
import tempfile
import warnings
from copy import deepcopy

import numpy as np
import pytest

from transformers import AutoVideoProcessor
from transformers.testing_utils import (
    check_json_file_has_correct_format,
    require_torch,
    require_torch_accelerator,
    require_vision,
    slow,
    torch_device,
)
from transformers.utils import is_torch_available, is_vision_available
from transformers.video_utils import VideoMetadata


if is_torch_available():
    import torch

if is_vision_available():
    from PIL import Image


def prepare_video(num_frames, num_channels, width=10, height=10, return_tensors="pil"):
    """This function prepares a video as a list of PIL images/NumPy arrays/PyTorch tensors."""

    video = []
    for frame_idx in range(num_frames):
        video.append(np.random.randint(255, size=(width, height, num_channels), dtype=np.uint8))

    if return_tensors == "pil":
        # PIL expects the channel dimension as last dimension
        video = [Image.fromarray(frame) for frame in video]
    elif return_tensors == "torch":
        # Torch images are typically in channels first format
        video = torch.from_numpy(np.array(video)).permute(0, 3, 1, 2)
    elif return_tensors == "np":
        # Numpy images are typically in channels last format
        video = np.array(video)

    return video


def prepare_video_inputs(
    batch_size,
    num_frames,
    num_channels,
    min_resolution,
    max_resolution,
    equal_resolution=False,
    return_tensors="pil",
):
    """This function prepares a batch of videos: a list of list of PIL images, or a list of list of numpy arrays if
    one specifies return_tensors="np", or a list of list of PyTorch tensors if one specifies return_tensors="torch".

    One can specify whether the videos are of the same resolution or not.
    """

    video_inputs = []
    for i in range(batch_size):
        if equal_resolution:
            width = height = max_resolution
        else:
            width, height = np.random.choice(np.arange(min_resolution, max_resolution), 2)
        video = prepare_video(
            num_frames=num_frames,
            num_channels=num_channels,
            width=width,
            height=height,
            return_tensors=return_tensors,
        )
        video_inputs.append(video)

    return video_inputs


class VideoProcessingTestMixin:
    test_cast_dtype = None
    fast_video_processing_class = None
    video_processor_list = None
    input_name = "pixel_values_videos"

    def setUp(self):
        video_processor_list = []

        if self.fast_video_processing_class:
            video_processor_list.append(self.fast_video_processing_class)

        self.video_processor_list = video_processor_list

    def test_video_processor_to_json_string(self):
        for video_processing_class in self.video_processor_list:
            video_processor = video_processing_class(**self.video_processor_dict)
            obj = json.loads(video_processor.to_json_string())
            for key, value in self.video_processor_dict.items():
                self.assertEqual(obj[key], value)

    def test_video_processor_to_json_file(self):
        for video_processing_class in self.video_processor_list:
            video_processor_first = video_processing_class(**self.video_processor_dict)

            with tempfile.TemporaryDirectory() as tmpdirname:
                json_file_path = os.path.join(tmpdirname, "video_processor.json")
                video_processor_first.to_json_file(json_file_path)
                video_processor_second = video_processing_class.from_json_file(json_file_path)

            self.assertEqual(video_processor_second.to_dict(), video_processor_first.to_dict())

    def test_video_processor_from_dict_with_kwargs(self):
        video_processor = self.fast_video_processing_class.from_dict(self.video_processor_dict)
        self.assertEqual(video_processor.size, {"shortest_edge": 20})
        self.assertEqual(video_processor.crop_size, {"height": 18, "width": 18})

        video_processor = self.fast_video_processing_class.from_dict(self.video_processor_dict, size=42, crop_size=84)
        self.assertEqual(video_processor.size, {"shortest_edge": 42})
        self.assertEqual(video_processor.crop_size, {"height": 84, "width": 84})

    def test_video_processor_from_and_save_pretrained(self):
        for video_processing_class in self.video_processor_list:
            video_processor_first = video_processing_class(**self.video_processor_dict)

            with tempfile.TemporaryDirectory() as tmpdirname:
                saved_file = video_processor_first.save_pretrained(tmpdirname)[0]
                check_json_file_has_correct_format(saved_file)
                video_processor_second = video_processing_class.from_pretrained(tmpdirname)

            self.assertEqual(video_processor_second.to_dict(), video_processor_first.to_dict())

    def test_video_processor_save_load_with_autovideoprocessor(self):
        for video_processing_class in self.video_processor_list:
            video_processor_first = video_processing_class(**self.video_processor_dict)

            with tempfile.TemporaryDirectory() as tmpdirname:
                saved_file = video_processor_first.save_pretrained(tmpdirname)[0]
                check_json_file_has_correct_format(saved_file)

                use_fast = video_processing_class.__name__.endswith("Fast")
                video_processor_second = AutoVideoProcessor.from_pretrained(tmpdirname, use_fast=use_fast)

            self.assertEqual(video_processor_second.to_dict(), video_processor_first.to_dict())

    def test_init_without_params(self):
        for video_processing_class in self.video_processor_list:
            video_processor = video_processing_class()
            self.assertIsNotNone(video_processor)

    def test_video_processor_explicit_none_preserved(self):
        """Test that explicitly setting an attribute to None is preserved through save/load."""

        # Find an attribute with a non-None class default to test explicit None override
        test_attr = None
        for attr in ["do_resize", "do_rescale", "do_normalize"]:
            if getattr(self.fast_video_processing_class, attr, None) is not None:
                test_attr = attr
                break

        if test_attr is None:
            self.skipTest("Could not find a suitable attribute to test")

        # Create processor with explicit None (override the attribute)
        kwargs = self.video_processor_dict.copy()
        kwargs[test_attr] = None
        video_processor = self.fast_video_processing_class(**kwargs)

        # Verify it's in to_dict() as None (not filtered out)
        self.assertIn(test_attr, video_processor.to_dict())
        self.assertIsNone(video_processor.to_dict()[test_attr])

        # Verify explicit None survives save/load cycle
        with tempfile.TemporaryDirectory() as tmpdirname:
            video_processor.save_pretrained(tmpdirname)
            reloaded = self.fast_video_processing_class.from_pretrained(tmpdirname)

        self.assertIsNone(getattr(reloaded, test_attr), f"Explicit None for {test_attr} was lost after reload")

    @slow
    @require_torch_accelerator
    @require_vision
    @pytest.mark.torch_compile_test
    def test_can_compile_fast_video_processor(self):
        if self.fast_video_processing_class is None:
            self.skipTest("Skipping compilation test as fast video processor is not defined")

        torch.compiler.reset()
        video_inputs = self.video_processor_tester.prepare_video_inputs(equal_resolution=False, return_tensors="torch")
        video_processor = self.fast_video_processing_class(**self.video_processor_dict)
        output_eager = video_processor(video_inputs, device=torch_device, do_sample_frames=False, return_tensors="pt")

        video_processor = torch.compile(video_processor, mode="reduce-overhead")
        output_compiled = video_processor(
            video_inputs, device=torch_device, do_sample_frames=False, return_tensors="pt"
        )

        torch.testing.assert_close(
            output_eager[self.input_name], output_compiled[self.input_name], rtol=1e-4, atol=1e-4
        )

    @require_torch
    @require_vision
    def test_cast_dtype_device(self):
        for video_processing_class in self.video_processor_list:
            if self.test_cast_dtype is not None:
                # Initialize video_processor
                video_processor = video_processing_class(**self.video_processor_dict)

                # create random PyTorch tensors
                video_inputs = self.video_processor_tester.prepare_video_inputs(
                    equal_resolution=False, return_tensors="torch"
                )

                encoding = video_processor(video_inputs, return_tensors="pt")

                self.assertEqual(encoding[self.input_name].device, torch.device("cpu"))
                self.assertEqual(encoding[self.input_name].dtype, torch.float32)

                encoding = video_processor(video_inputs, return_tensors="pt").to(torch.float16)
                self.assertEqual(encoding[self.input_name].device, torch.device("cpu"))
                self.assertEqual(encoding[self.input_name].dtype, torch.float16)

                encoding = video_processor(video_inputs, return_tensors="pt").to("cpu", torch.bfloat16)
                self.assertEqual(encoding[self.input_name].device, torch.device("cpu"))
                self.assertEqual(encoding[self.input_name].dtype, torch.bfloat16)

                with self.assertRaises(TypeError):
                    _ = video_processor(video_inputs, return_tensors="pt").to(torch.bfloat16, "cpu")

                # Try with text + video feature
                encoding = video_processor(video_inputs, return_tensors="pt")
                encoding.update({"input_ids": torch.LongTensor([[1, 2, 3], [4, 5, 6]])})
                encoding = encoding.to(torch.float16)

                self.assertEqual(encoding[self.input_name].device, torch.device("cpu"))
                self.assertEqual(encoding[self.input_name].dtype, torch.float16)
                self.assertEqual(encoding.input_ids.dtype, torch.long)

    def test_call_pil(self):
        for video_processing_class in self.video_processor_list:
            # Initialize video_processing
            video_processing = video_processing_class(**self.video_processor_dict)
            video_inputs = self.video_processor_tester.prepare_video_inputs(equal_resolution=False)

            # Each video is a list of PIL Images
            for video in video_inputs:
                self.assertIsInstance(video[0], Image.Image)

            # Test not batched input
            encoded_videos = video_processing(video_inputs[0], return_tensors="pt")[self.input_name]
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape([video_inputs[0]])
            self.assertEqual(tuple(encoded_videos.shape), (1, *expected_output_video_shape))

            # Test batched
            encoded_videos = video_processing(video_inputs, return_tensors="pt")[self.input_name]
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape(video_inputs)
            self.assertEqual(
                tuple(encoded_videos.shape), (self.video_processor_tester.batch_size, *expected_output_video_shape)
            )

    def test_call_numpy(self):
        for video_processing_class in self.video_processor_list:
            # Initialize video_processing
            video_processing = video_processing_class(**self.video_processor_dict)
            # create random numpy tensors
            video_inputs = self.video_processor_tester.prepare_video_inputs(
                equal_resolution=False, return_tensors="np"
            )
            for video in video_inputs:
                self.assertIsInstance(video, np.ndarray)

            # Test not batched input
            encoded_videos = video_processing(video_inputs[0], return_tensors="pt")[self.input_name]
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape([video_inputs[0]])
            self.assertEqual(tuple(encoded_videos.shape), (1, *expected_output_video_shape))

            # Test batched
            encoded_videos = video_processing(video_inputs, return_tensors="pt")[self.input_name]
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape(video_inputs)
            self.assertEqual(
                tuple(encoded_videos.shape), (self.video_processor_tester.batch_size, *expected_output_video_shape)
            )

    def test_call_pytorch(self):
        for video_processing_class in self.video_processor_list:
            # Initialize video_processing
            video_processing = video_processing_class(**self.video_processor_dict)
            # create random PyTorch tensors
            video_inputs = self.video_processor_tester.prepare_video_inputs(
                equal_resolution=False, return_tensors="torch"
            )

            for video in video_inputs:
                self.assertIsInstance(video, torch.Tensor)

            # Test not batched input
            encoded_videos = video_processing(video_inputs[0], return_tensors="pt")[self.input_name]
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape([video_inputs[0]])
            self.assertEqual(tuple(encoded_videos.shape), (1, *expected_output_video_shape))

            # Test batched
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape(video_inputs)
            encoded_videos = video_processing(video_inputs, return_tensors="pt")[self.input_name]
            self.assertEqual(
                tuple(encoded_videos.shape),
                (self.video_processor_tester.batch_size, *expected_output_video_shape),
            )

    def test_call_sample_frames(self):
        for video_processing_class in self.video_processor_list:
            video_processing = video_processing_class(**self.video_processor_dict)

            prev_num_frames = self.video_processor_tester.num_frames
            self.video_processor_tester.num_frames = 8
            video_inputs = self.video_processor_tester.prepare_video_inputs(
                equal_resolution=False,
                return_tensors="torch",
            )

            # Force set sampling to False. No sampling is expected even when `num_frames` exists
            video_processing.do_sample_frames = False

            encoded_videos = video_processing(video_inputs[0], return_tensors="pt", num_frames=3)[self.input_name]
            encoded_videos_batched = video_processing(video_inputs, return_tensors="pt", num_frames=3)[self.input_name]
            self.assertEqual(encoded_videos.shape[1], 8)
            self.assertEqual(encoded_videos_batched.shape[1], 8)

            # Set sampling to True. Video frames should be sampled with `num_frames` in the output
            video_processing.do_sample_frames = True

            encoded_videos = video_processing(video_inputs[0], return_tensors="pt", num_frames=3)[self.input_name]
            encoded_videos_batched = video_processing(video_inputs, return_tensors="pt", num_frames=3)[self.input_name]
            self.assertEqual(encoded_videos.shape[1], 3)
            self.assertEqual(encoded_videos_batched.shape[1], 3)

            # Sample with `fps` requires metadata to infer number of frames from total duration
            with self.assertRaises(ValueError):
                metadata = VideoMetadata(**{"total_num_frames": 8})
                video_processing.sample_frames(metadata=metadata, fps=3)

            metadata = [[{"duration": 2.0, "total_num_frames": 8, "fps": 4}]]
            batched_metadata = metadata * len(video_inputs)
            encoded_videos = video_processing(video_inputs[0], return_tensors="pt", fps=3, video_metadata=metadata)[
                self.input_name
            ]
            encoded_videos_batched = video_processing(
                video_inputs, return_tensors="pt", fps=3, video_metadata=batched_metadata
            )[self.input_name]
            self.assertEqual(encoded_videos.shape[1], 6)
            self.assertEqual(encoded_videos_batched.shape[1], 6)

            # The same as above but uses a `VideoMetadata` object in the input
            metadata = [[VideoMetadata(duration=2.0, total_num_frames=8, fps=4)]]
            batched_metadata = metadata * len(video_inputs)
            encoded_videos = video_processing(video_inputs[0], return_tensors="pt", fps=3, video_metadata=metadata)[
                self.input_name
            ]

            # We should raise error when asked to sample more frames than there are in input video
            with self.assertRaises(ValueError):
                encoded_videos = video_processing(video_inputs[0], return_tensors="pt", num_frames=10)[self.input_name]
                encoded_videos_batched = video_processing(video_inputs, return_tensors="pt", num_frames=10)[
                    self.input_name
                ]

            # Assign back the actual num frames in tester
            self.video_processor_tester.num_frames = prev_num_frames

    def test_nested_input(self):
        """Tests that the processor can work with nested list where each video is a list of arrays"""
        for video_processing_class in self.video_processor_list:
            video_processing = video_processing_class(**self.video_processor_dict)
            video_inputs = self.video_processor_tester.prepare_video_inputs(
                equal_resolution=False, return_tensors="np"
            )

            # Test not batched input
            video_inputs = [list(video) for video in video_inputs]
            encoded_videos = video_processing(video_inputs[0], return_tensors="pt")[self.input_name]
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape([video_inputs[0]])
            self.assertEqual(tuple(encoded_videos.shape), (1, *expected_output_video_shape))

            # Test batched
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape(video_inputs)
            encoded_videos = video_processing(video_inputs, return_tensors="pt")[self.input_name]
            self.assertEqual(
                tuple(encoded_videos.shape),
                (self.video_processor_tester.batch_size, *expected_output_video_shape),
            )

    def test_call_numpy_4_channels(self):
        for video_processing_class in self.video_processor_list:
            # Test that can process videos which have an arbitrary number of channels
            # Initialize video_processing
            video_processor = video_processing_class(**self.video_processor_dict)

            # create random numpy tensors
            self.video_processor_tester.num_channels = 4
            video_inputs = self.video_processor_tester.prepare_video_inputs(
                equal_resolution=False, return_tensors="pil"
            )

            # Test not batched input
            encoded_videos = video_processor(
                video_inputs[0],
                return_tensors="pt",
                input_data_format="channels_last",
                image_mean=0.0,
                image_std=1.0,
            )[self.input_name]
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape([video_inputs[0]])
            if video_processor.do_convert_rgb:
                expected_output_video_shape = list(expected_output_video_shape)
                expected_output_video_shape[1] = 3
            self.assertEqual(tuple(encoded_videos.shape), (1, *expected_output_video_shape))

            # Test batched
            encoded_videos = video_processor(
                video_inputs,
                return_tensors="pt",
                input_data_format="channels_last",
                image_mean=0.0,
                image_std=1.0,
            )[self.input_name]
            expected_output_video_shape = self.video_processor_tester.expected_output_video_shape(video_inputs)
            if video_processor.do_convert_rgb:
                expected_output_video_shape = list(expected_output_video_shape)
                expected_output_video_shape[1] = 3
            self.assertEqual(
                tuple(encoded_videos.shape), (self.video_processor_tester.batch_size, *expected_output_video_shape)
            )

    def test_video_processor_preprocess_arguments(self):
        is_tested = False

        for video_processing_class in self.video_processor_list:
            video_processor = video_processing_class(**self.video_processor_dict)

            # validation done by _valid_processor_keys attribute
            if hasattr(video_processor, "_valid_processor_keys") and hasattr(video_processor, "preprocess"):
                preprocess_parameter_names = inspect.getfullargspec(video_processor.preprocess).args
                preprocess_parameter_names.remove("self")
                preprocess_parameter_names.sort()
                valid_processor_keys = video_processor._valid_processor_keys
                valid_processor_keys.sort()
                self.assertEqual(preprocess_parameter_names, valid_processor_keys)
                is_tested = True

            # validation done by @filter_out_non_signature_kwargs decorator
            if hasattr(video_processor.preprocess, "_filter_out_non_signature_kwargs"):
                if hasattr(self.video_processor_tester, "prepare_video_inputs"):
                    inputs = self.video_processor_tester.prepare_video_inputs()
                elif hasattr(self.video_processor_tester, "prepare_video_inputs"):
                    inputs = self.video_processor_tester.prepare_video_inputs()
                else:
                    self.skipTest(reason="No valid input preparation method found")

                with warnings.catch_warnings(record=True) as raised_warnings:
                    warnings.simplefilter("always")
                    video_processor(inputs, extra_argument=True)

                messages = " ".join([str(w.message) for w in raised_warnings])
                self.assertGreaterEqual(len(raised_warnings), 1)
                self.assertIn("extra_argument", messages)
                is_tested = True

        if not is_tested:
            self.skipTest(reason="No validation found for `preprocess` method")

    def test_override_instance_attributes_does_not_affect_other_instances(self):
        if self.fast_video_processing_class is None:
            self.skipTest(
                "Only testing fast video processor, as most slow processors break this test and are to be deprecated"
            )

        video_processing_class = self.fast_video_processing_class
        video_processor_1 = video_processing_class()
        video_processor_2 = video_processing_class()
        if not (hasattr(video_processor_1, "size") and isinstance(video_processor_1.size, dict)) or not (
            hasattr(video_processor_1, "image_mean") and isinstance(video_processor_1.image_mean, list)
        ):
            self.skipTest(
                reason="Skipping test as the image processor does not have dict size or list image_mean attributes"
            )

        original_size_2 = deepcopy(video_processor_2.size)
        for key in video_processor_1.size:
            video_processor_1.size[key] = -1
        modified_copied_size_1 = deepcopy(video_processor_1.size)

        original_image_mean_2 = deepcopy(video_processor_2.image_mean)
        video_processor_1.image_mean[0] = -1
        modified_copied_image_mean_1 = deepcopy(video_processor_1.image_mean)

        # check that the original attributes of the second instance are not affected
        self.assertEqual(video_processor_2.size, original_size_2)
        self.assertEqual(video_processor_2.image_mean, original_image_mean_2)

        for key in video_processor_2.size:
            video_processor_2.size[key] = -2
        video_processor_2.image_mean[0] = -2

        # check that the modified attributes of the first instance are not affected by the second instance
        self.assertEqual(video_processor_1.size, modified_copied_size_1)
        self.assertEqual(video_processor_1.image_mean, modified_copied_image_mean_1)