refactor Datapoint dispatch mechanism (#7747)

Co-authored-by: Nicolas Hug <contact@nicolas-hug.com>

test/common_utils.py

@@ -829,6 +829,10 @@ def make_video(size=DEFAULT_SIZE, *, num_frames=3, batch_dims=(), **kwargs):
     return datapoints.Video(make_image(size, batch_dims=(*batch_dims, num_frames), **kwargs))
 
 
+def make_video_tensor(*args, **kwargs):
+    return make_video(*args, **kwargs).as_subclass(torch.Tensor)
+
+
 def make_video_loader(
     size=DEFAULT_PORTRAIT_SPATIAL_SIZE,
     *,

test/datasets_utils.py

@@ -567,7 +567,7 @@ class DatasetTestCase(unittest.TestCase):
 
     @test_all_configs
     def test_transforms_v2_wrapper(self, config):
-        from torchvision.datapoints._datapoint import Datapoint
+        from torchvision import datapoints
         from torchvision.datasets import wrap_dataset_for_transforms_v2
 
         try:
@@ -588,7 +588,9 @@ class DatasetTestCase(unittest.TestCase):
                     assert len(wrapped_dataset) == info["num_examples"]
 
                     wrapped_sample = wrapped_dataset[0]
-                    assert tree_any(lambda item: isinstance(item, (Datapoint, PIL.Image.Image)), wrapped_sample)
+                    assert tree_any(
+                        lambda item: isinstance(item, (datapoints.Datapoint, PIL.Image.Image)), wrapped_sample
+                    )
         except TypeError as error:
             msg = f"No wrapper exists for dataset class {type(dataset).__name__}"
             if str(error).startswith(msg):

test/test_transforms_v2.py

@@ -1344,12 +1344,12 @@ def test_antialias_warning():
         transforms.RandomResize(10, 20)(tensor_img)
 
     with pytest.warns(UserWarning, match=match):
-        datapoints.Image(tensor_img).resized_crop(0, 0, 10, 10, (20, 20))
+        F.resized_crop(datapoints.Image(tensor_img), 0, 0, 10, 10, (20, 20))
 
     with pytest.warns(UserWarning, match=match):
-        datapoints.Video(tensor_video).resize((20, 20))
+        F.resize(datapoints.Video(tensor_video), (20, 20))
     with pytest.warns(UserWarning, match=match):
-        datapoints.Video(tensor_video).resized_crop(0, 0, 10, 10, (20, 20))
+        F.resized_crop(datapoints.Video(tensor_video), 0, 0, 10, 10, (20, 20))
 
     with warnings.catch_warnings():
         warnings.simplefilter("error")
@@ -1363,8 +1363,8 @@ def test_antialias_warning():
         transforms.RandomShortestSize((20, 20), antialias=True)(tensor_img)
         transforms.RandomResize(10, 20, antialias=True)(tensor_img)
 
-        datapoints.Image(tensor_img).resized_crop(0, 0, 10, 10, (20, 20), antialias=True)
-        datapoints.Video(tensor_video).resized_crop(0, 0, 10, 10, (20, 20), antialias=True)
+        F.resized_crop(datapoints.Image(tensor_img), 0, 0, 10, 10, (20, 20), antialias=True)
+        F.resized_crop(datapoints.Video(tensor_video), 0, 0, 10, 10, (20, 20), antialias=True)
 
 
 @pytest.mark.parametrize("image_type", (PIL.Image, torch.Tensor, datapoints.Image))

test/test_transforms_v2_functional.py

@@ -2,13 +2,11 @@ import inspect
 import math
 import os
 import re
-
-from typing import get_type_hints
+from unittest import mock
 
 import numpy as np
 import PIL.Image
 import pytest
-
 import torch
 
 from common_utils import (
@@ -27,6 +25,7 @@ from torchvision.transforms.functional import _get_perspective_coeffs
 from torchvision.transforms.v2 import functional as F
 from torchvision.transforms.v2.functional._geometry import _center_crop_compute_padding
 from torchvision.transforms.v2.functional._meta import clamp_bounding_boxes, convert_format_bounding_boxes
+from torchvision.transforms.v2.functional._utils import _KERNEL_REGISTRY
 from torchvision.transforms.v2.utils import is_simple_tensor
 from transforms_v2_dispatcher_infos import DISPATCHER_INFOS
 from transforms_v2_kernel_infos import KERNEL_INFOS
@@ -424,11 +423,17 @@ class TestDispatchers:
     def test_dispatch_datapoint(self, info, args_kwargs, spy_on):
         (datapoint, *other_args), kwargs = args_kwargs.load()
 
-        method_name = info.id
-        method = getattr(datapoint, method_name)
-        datapoint_type = type(datapoint)
-        spy = spy_on(method, module=datapoint_type.__module__, name=f"{datapoint_type.__name__}.{method_name}")
+        input_type = type(datapoint)
+
+        wrapped_kernel = _KERNEL_REGISTRY[info.dispatcher][input_type]
 
+        # In case the wrapper was decorated with @functools.wraps, we can make the check more strict and test if the
+        # proper kernel was wrapped
+        if hasattr(wrapped_kernel, "__wrapped__"):
+            assert wrapped_kernel.__wrapped__ is info.kernels[input_type]
+
+        spy = mock.MagicMock(wraps=wrapped_kernel, name=wrapped_kernel.__name__)
+        with mock.patch.dict(_KERNEL_REGISTRY[info.dispatcher], values={input_type: spy}):
             info.dispatcher(datapoint, *other_args, **kwargs)
 
         spy.assert_called_once()
@@ -462,9 +467,12 @@ class TestDispatchers:
         kernel_params = list(kernel_signature.parameters.values())[1:]
 
         # We filter out metadata that is implicitly passed to the dispatcher through the input datapoint, but has to be
-        # explicit passed to the kernel.
-        datapoint_type_metadata = datapoint_type.__annotations__.keys()
-        kernel_params = [param for param in kernel_params if param.name not in datapoint_type_metadata]
+        # explicitly passed to the kernel.
+        input_type = {v: k for k, v in dispatcher_info.kernels.items()}.get(kernel_info.kernel)
+        explicit_metadata = {
+            datapoints.BoundingBoxes: {"format", "canvas_size"},
+        }
+        kernel_params = [param for param in kernel_params if param.name not in explicit_metadata.get(input_type, set())]
 
         dispatcher_params = iter(dispatcher_params)
         for dispatcher_param, kernel_param in zip(dispatcher_params, kernel_params):
@@ -481,28 +489,6 @@ class TestDispatchers:
 
             assert dispatcher_param == kernel_param
 
-    @pytest.mark.parametrize("info", DISPATCHER_INFOS, ids=lambda info: info.id)
-    def test_dispatcher_datapoint_signatures_consistency(self, info):
-        try:
-            datapoint_method = getattr(datapoints._datapoint.Datapoint, info.id)
-        except AttributeError:
-            pytest.skip("Dispatcher doesn't support arbitrary datapoint dispatch.")
-
-        dispatcher_signature = inspect.signature(info.dispatcher)
-        dispatcher_params = list(dispatcher_signature.parameters.values())[1:]
-
-        datapoint_signature = inspect.signature(datapoint_method)
-        datapoint_params = list(datapoint_signature.parameters.values())[1:]
-
-        # Because we use `from __future__ import annotations` inside the module where `datapoints._datapoint` is
-        # defined, the annotations are stored as strings. This makes them concrete again, so they can be compared to the
-        # natively concrete dispatcher annotations.
-        datapoint_annotations = get_type_hints(datapoint_method)
-        for param in datapoint_params:
-            param._annotation = datapoint_annotations[param.name]
-
-        assert dispatcher_params == datapoint_params
-
     @pytest.mark.parametrize("info", DISPATCHER_INFOS, ids=lambda info: info.id)
     def test_unkown_type(self, info):
         unkown_input = object()

test/test_transforms_v2_refactored.py

@@ -3,7 +3,6 @@ import decimal
 import inspect
 import math
 import re
-from typing import get_type_hints
 from unittest import mock
 
 import numpy as np
@@ -26,6 +25,7 @@ from common_utils import (
     make_image_tensor,
     make_segmentation_mask,
     make_video,
+    make_video_tensor,
     needs_cuda,
     set_rng_seed,
 )
@@ -39,6 +39,7 @@ from torchvision import datapoints
 from torchvision.transforms._functional_tensor import _max_value as get_max_value
 from torchvision.transforms.functional import pil_modes_mapping
 from torchvision.transforms.v2 import functional as F
+from torchvision.transforms.v2.functional._utils import _KERNEL_REGISTRY
 
 
 @pytest.fixture(autouse=True)
@@ -176,15 +177,18 @@ def _check_dispatcher_dispatch(dispatcher, kernel, input, *args, **kwargs):
     """Checks if the dispatcher correctly dispatches the input to the corresponding kernel and that the input type is
     preserved in doing so. For bounding boxes also checks that the format is preserved.
     """
-    if isinstance(input, datapoints._datapoint.Datapoint):
-        # Due to our complex dispatch architecture for datapoints, we cannot spy on the kernel directly,
-        # but rather have to patch the `Datapoint.__F` attribute to contain the spied on kernel.
-        spy = mock.MagicMock(wraps=kernel, name=kernel.__name__)
-        with mock.patch.object(F, kernel.__name__, spy):
-            # Due to Python's name mangling, the `Datapoint.__F` attribute is only accessible from inside the class.
-            # Since that is not the case here, we need to prefix f"_{cls.__name__}"
-            # See https://docs.python.org/3/tutorial/classes.html#private-variables for details
-            with mock.patch.object(datapoints._datapoint.Datapoint, "_Datapoint__F", new=F):
+    input_type = type(input)
+
+    if isinstance(input, datapoints.Datapoint):
+        wrapped_kernel = _KERNEL_REGISTRY[dispatcher][input_type]
+
+        # In case the wrapper was decorated with @functools.wraps, we can make the check more strict and test if the
+        # proper kernel was wrapped
+        if hasattr(wrapped_kernel, "__wrapped__"):
+            assert wrapped_kernel.__wrapped__ is kernel
+
+        spy = mock.MagicMock(wraps=wrapped_kernel, name=wrapped_kernel.__name__)
+        with mock.patch.dict(_KERNEL_REGISTRY[dispatcher], values={input_type: spy}):
             output = dispatcher(input, *args, **kwargs)
 
         spy.assert_called_once()
@@ -194,7 +198,7 @@ def _check_dispatcher_dispatch(dispatcher, kernel, input, *args, **kwargs):
 
             spy.assert_called_once()
 
-    assert isinstance(output, type(input))
+    assert isinstance(output, input_type)
 
     if isinstance(input, datapoints.BoundingBoxes):
         assert output.format == input.format
@@ -209,15 +213,13 @@ def check_dispatcher(
     check_dispatch=True,
     **kwargs,
 ):
-    with mock.patch("torch._C._log_api_usage_once", wraps=torch._C._log_api_usage_once) as spy:
-        dispatcher(input, *args, **kwargs)
-
-        spy.assert_any_call(f"{dispatcher.__module__}.{dispatcher.__name__}")
-
     unknown_input = object()
+    with mock.patch("torch._C._log_api_usage_once", wraps=torch._C._log_api_usage_once) as spy:
         with pytest.raises(TypeError, match=re.escape(str(type(unknown_input)))):
             dispatcher(unknown_input, *args, **kwargs)
 
+        spy.assert_any_call(f"{dispatcher.__module__}.{dispatcher.__name__}")
+
     if check_scripted_smoke:
         _check_dispatcher_scripted_smoke(dispatcher, input, *args, **kwargs)
 
@@ -225,18 +227,18 @@ def check_dispatcher(
         _check_dispatcher_dispatch(dispatcher, kernel, input, *args, **kwargs)
 
 
-def _check_dispatcher_kernel_signature_match(dispatcher, *, kernel, input_type):
+def check_dispatcher_kernel_signature_match(dispatcher, *, kernel, input_type):
     """Checks if the signature of the dispatcher matches the kernel signature."""
-    dispatcher_signature = inspect.signature(dispatcher)
-    dispatcher_params = list(dispatcher_signature.parameters.values())[1:]
-
-    kernel_signature = inspect.signature(kernel)
-    kernel_params = list(kernel_signature.parameters.values())[1:]
+    dispatcher_params = list(inspect.signature(dispatcher).parameters.values())[1:]
+    kernel_params = list(inspect.signature(kernel).parameters.values())[1:]
 
-    if issubclass(input_type, datapoints._datapoint.Datapoint):
+    if issubclass(input_type, datapoints.Datapoint):
         # We filter out metadata that is implicitly passed to the dispatcher through the input datapoint, but has to be
         # explicitly passed to the kernel.
-        kernel_params = [param for param in kernel_params if param.name not in input_type.__annotations__.keys()]
+        explicit_metadata = {
+            datapoints.BoundingBoxes: {"format", "canvas_size"},
+        }
+        kernel_params = [param for param in kernel_params if param.name not in explicit_metadata.get(input_type, set())]
 
     dispatcher_params = iter(dispatcher_params)
     for dispatcher_param, kernel_param in zip(dispatcher_params, kernel_params):
@@ -259,30 +261,6 @@ def _check_dispatcher_kernel_signature_match(dispatcher, *, kernel, input_type):
         assert dispatcher_param == kernel_param
 
 
-def _check_dispatcher_datapoint_signature_match(dispatcher):
-    """Checks if the signature of the dispatcher matches the corresponding method signature on the Datapoint class."""
-    dispatcher_signature = inspect.signature(dispatcher)
-    dispatcher_params = list(dispatcher_signature.parameters.values())[1:]
-
-    datapoint_method = getattr(datapoints._datapoint.Datapoint, dispatcher.__name__)
-    datapoint_signature = inspect.signature(datapoint_method)
-    datapoint_params = list(datapoint_signature.parameters.values())[1:]
-
-    # Some annotations in the `datapoints._datapoint` module
-    # are stored as strings. The block below makes them concrete again (non-strings), so they can be compared to the
-    # natively concrete dispatcher annotations.
-    datapoint_annotations = get_type_hints(datapoint_method)
-    for param in datapoint_params:
-        param._annotation = datapoint_annotations[param.name]
-
-    assert dispatcher_params == datapoint_params
-
-
-def check_dispatcher_signatures_match(dispatcher, *, kernel, input_type):
-    _check_dispatcher_kernel_signature_match(dispatcher, kernel=kernel, input_type=input_type)
-    _check_dispatcher_datapoint_signature_match(dispatcher)
-
-
 def _check_transform_v1_compatibility(transform, input):
     """If the transform defines the ``_v1_transform_cls`` attribute, checks if the transform has a public, static
     ``get_params`` method, is scriptable, and the scripted version can be called without error."""
@@ -433,6 +411,33 @@ def reference_affine_bounding_boxes_helper(bounding_boxes, *, format, canvas_siz
     return torch.stack([transform(b) for b in bounding_boxes.reshape(-1, 4).unbind()]).reshape(bounding_boxes.shape)
 
 
+@pytest.mark.parametrize(
+    ("dispatcher", "registered_datapoint_clss"),
+    [(dispatcher, set(registry.keys())) for dispatcher, registry in _KERNEL_REGISTRY.items()],
+)
+def test_exhaustive_kernel_registration(dispatcher, registered_datapoint_clss):
+    missing = {
+        datapoints.Image,
+        datapoints.BoundingBoxes,
+        datapoints.Mask,
+        datapoints.Video,
+    } - registered_datapoint_clss
+    if missing:
+        names = sorted(f"datapoints.{cls.__name__}" for cls in missing)
+        raise AssertionError(
+            "\n".join(
+                [
+                    f"The dispatcher '{dispatcher.__name__}' has no kernel registered for",
+                    "",
+                    *[f"- {name}" for name in names],
+                    "",
+                    f"If available, register the kernels with @_register_kernel_internal({dispatcher.__name__}, ...).",
+                    f"If not, register explicit no-ops with @_register_explicit_noop({', '.join(names)})",
+                ]
+            )
+        )
+
+
 class TestResize:
     INPUT_SIZE = (17, 11)
     OUTPUT_SIZES = [17, [17], (17,), [12, 13], (12, 13)]
@@ -568,7 +573,7 @@ class TestResize:
         ],
     )
     def test_dispatcher_signature(self, kernel, input_type):
-        check_dispatcher_signatures_match(F.resize, kernel=kernel, input_type=input_type)
+        check_dispatcher_kernel_signature_match(F.resize, kernel=kernel, input_type=input_type)
 
     @pytest.mark.parametrize("size", OUTPUT_SIZES)
     @pytest.mark.parametrize("device", cpu_and_cuda())
@@ -766,7 +771,7 @@ class TestResize:
 
         # This identity check is not a requirement. It is here to avoid breaking the behavior by accident. If there
         # is a good reason to break this, feel free to downgrade to an equality check.
-        if isinstance(input, datapoints._datapoint.Datapoint):
+        if isinstance(input, datapoints.Datapoint):
             # We can't test identity directly, since that checks for the identity of the Python object. Since all
             # datapoints unwrap before a kernel and wrap again afterwards, the Python object changes. Thus, we check
             # that the underlying storage is the same
@@ -850,7 +855,7 @@ class TestHorizontalFlip:
         ],
     )
     def test_dispatcher_signature(self, kernel, input_type):
-        check_dispatcher_signatures_match(F.horizontal_flip, kernel=kernel, input_type=input_type)
+        check_dispatcher_kernel_signature_match(F.horizontal_flip, kernel=kernel, input_type=input_type)
 
     @pytest.mark.parametrize(
         "make_input",
@@ -1033,7 +1038,7 @@ class TestAffine:
         ],
     )
     def test_dispatcher_signature(self, kernel, input_type):
-        check_dispatcher_signatures_match(F.affine, kernel=kernel, input_type=input_type)
+        check_dispatcher_kernel_signature_match(F.affine, kernel=kernel, input_type=input_type)
 
     @pytest.mark.parametrize(
         "make_input",
@@ -1329,7 +1334,7 @@ class TestVerticalFlip:
         ],
     )
     def test_dispatcher_signature(self, kernel, input_type):
-        check_dispatcher_signatures_match(F.vertical_flip, kernel=kernel, input_type=input_type)
+        check_dispatcher_kernel_signature_match(F.vertical_flip, kernel=kernel, input_type=input_type)
 
     @pytest.mark.parametrize(
         "make_input",
@@ -1486,7 +1491,7 @@ class TestRotate:
         ],
     )
     def test_dispatcher_signature(self, kernel, input_type):
-        check_dispatcher_signatures_match(F.rotate, kernel=kernel, input_type=input_type)
+        check_dispatcher_kernel_signature_match(F.rotate, kernel=kernel, input_type=input_type)
 
     @pytest.mark.parametrize(
         "make_input",
@@ -1899,6 +1904,56 @@ class TestToDtype:
         assert out["mask"].dtype == mask_dtype
 
 
+class TestAdjustBrightness:
+    _CORRECTNESS_BRIGHTNESS_FACTORS = [0.5, 0.0, 1.0, 5.0]
+    _DEFAULT_BRIGHTNESS_FACTOR = _CORRECTNESS_BRIGHTNESS_FACTORS[0]
+
+    @pytest.mark.parametrize(
+        ("kernel", "make_input"),
+        [
+            (F.adjust_brightness_image_tensor, make_image),
+            (F.adjust_brightness_video, make_video),
+        ],
+    )
+    @pytest.mark.parametrize("dtype", [torch.float32, torch.uint8])
+    @pytest.mark.parametrize("device", cpu_and_cuda())
+    def test_kernel(self, kernel, make_input, dtype, device):
+        check_kernel(kernel, make_input(dtype=dtype, device=device), brightness_factor=self._DEFAULT_BRIGHTNESS_FACTOR)
+
+    @pytest.mark.parametrize(
+        ("kernel", "make_input"),
+        [
+            (F.adjust_brightness_image_tensor, make_image_tensor),
+            (F.adjust_brightness_image_pil, make_image_pil),
+            (F.adjust_brightness_image_tensor, make_image),
+            (F.adjust_brightness_video, make_video),
+        ],
+    )
+    def test_dispatcher(self, kernel, make_input):
+        check_dispatcher(F.adjust_brightness, kernel, make_input(), brightness_factor=self._DEFAULT_BRIGHTNESS_FACTOR)
+
+    @pytest.mark.parametrize(
+        ("kernel", "input_type"),
+        [
+            (F.adjust_brightness_image_tensor, torch.Tensor),
+            (F.adjust_brightness_image_pil, PIL.Image.Image),
+            (F.adjust_brightness_image_tensor, datapoints.Image),
+            (F.adjust_brightness_video, datapoints.Video),
+        ],
+    )
+    def test_dispatcher_signature(self, kernel, input_type):
+        check_dispatcher_kernel_signature_match(F.adjust_brightness, kernel=kernel, input_type=input_type)
+
+    @pytest.mark.parametrize("brightness_factor", _CORRECTNESS_BRIGHTNESS_FACTORS)
+    def test_image_correctness(self, brightness_factor):
+        image = make_image(dtype=torch.uint8, device="cpu")
+
+        actual = F.adjust_brightness(image, brightness_factor=brightness_factor)
+        expected = F.to_image_tensor(F.adjust_brightness(F.to_image_pil(image), brightness_factor=brightness_factor))
+
+        torch.testing.assert_close(actual, expected)
+
+
 class TestCutMixMixUp:
     class DummyDataset:
         def __init__(self, size, num_classes):
@@ -2036,3 +2091,93 @@ def test_labels_getter_default_heuristic(key, sample_type):
         # it takes precedence over other keys which would otherwise be a match
         d = {key: "something_else", "labels": labels}
         assert transforms._utils._find_labels_default_heuristic(d) is labels
+
+
+class TestShapeGetters:
+    @pytest.mark.parametrize(
+        ("kernel", "make_input"),
+        [
+            (F.get_dimensions_image_tensor, make_image_tensor),
+            (F.get_dimensions_image_pil, make_image_pil),
+            (F.get_dimensions_image_tensor, make_image),
+            (F.get_dimensions_video, make_video),
+        ],
+    )
+    def test_get_dimensions(self, kernel, make_input):
+        size = (10, 10)
+        color_space, num_channels = "RGB", 3
+
+        input = make_input(size, color_space=color_space)
+
+        assert kernel(input) == F.get_dimensions(input) == [num_channels, *size]
+
+    @pytest.mark.parametrize(
+        ("kernel", "make_input"),
+        [
+            (F.get_num_channels_image_tensor, make_image_tensor),
+            (F.get_num_channels_image_pil, make_image_pil),
+            (F.get_num_channels_image_tensor, make_image),
+            (F.get_num_channels_video, make_video),
+        ],
+    )
+    def test_get_num_channels(self, kernel, make_input):
+        color_space, num_channels = "RGB", 3
+
+        input = make_input(color_space=color_space)
+
+        assert kernel(input) == F.get_num_channels(input) == num_channels
+
+    @pytest.mark.parametrize(
+        ("kernel", "make_input"),
+        [
+            (F.get_size_image_tensor, make_image_tensor),
+            (F.get_size_image_pil, make_image_pil),
+            (F.get_size_image_tensor, make_image),
+            (F.get_size_bounding_boxes, make_bounding_box),
+            (F.get_size_mask, make_detection_mask),
+            (F.get_size_mask, make_segmentation_mask),
+            (F.get_size_video, make_video),
+        ],
+    )
+    def test_get_size(self, kernel, make_input):
+        size = (10, 10)
+
+        input = make_input(size)
+
+        assert kernel(input) == F.get_size(input) == list(size)
+
+    @pytest.mark.parametrize(
+        ("kernel", "make_input"),
+        [
+            (F.get_num_frames_video, make_video_tensor),
+            (F.get_num_frames_video, make_video),
+        ],
+    )
+    def test_get_num_frames(self, kernel, make_input):
+        num_frames = 4
+
+        input = make_input(num_frames=num_frames)
+
+        assert kernel(input) == F.get_num_frames(input) == num_frames
+
+    @pytest.mark.parametrize(
+        ("dispatcher", "make_input"),
+        [
+            (F.get_dimensions, make_bounding_box),
+            (F.get_dimensions, make_detection_mask),
+            (F.get_dimensions, make_segmentation_mask),
+            (F.get_num_channels, make_bounding_box),
+            (F.get_num_channels, make_detection_mask),
+            (F.get_num_channels, make_segmentation_mask),
+            (F.get_num_frames, make_image_pil),
+            (F.get_num_frames, make_image),
+            (F.get_num_frames, make_bounding_box),
+            (F.get_num_frames, make_detection_mask),
+            (F.get_num_frames, make_segmentation_mask),
+        ],
+    )
+    def test_unsupported_types(self, dispatcher, make_input):
+        input = make_input()
+
+        with pytest.raises(TypeError, match=re.escape(str(type(input)))):
+            dispatcher(input)

test/transforms_v2_dispatcher_infos.py

@@ -69,6 +69,7 @@ class DispatcherInfo(InfoBase):
             import itertools
 
             for args_kwargs in sample_inputs:
+                if hasattr(datapoint_type, "__annotations__"):
                     for name in itertools.chain(
                         datapoint_type.__annotations__.keys(),
                         # FIXME: this seems ok for conversion dispatchers, but we should probably handle this on a
@@ -289,14 +290,6 @@ DISPATCHER_INFOS = [
             skip_dispatch_datapoint,
         ],
     ),
-    DispatcherInfo(
-        F.adjust_brightness,
-        kernels={
-            datapoints.Image: F.adjust_brightness_image_tensor,
-            datapoints.Video: F.adjust_brightness_video,
-        },
-        pil_kernel_info=PILKernelInfo(F.adjust_brightness_image_pil, kernel_name="adjust_brightness_image_pil"),
-    ),
     DispatcherInfo(
         F.adjust_contrast,
         kernels={

test/transforms_v2_kernel_infos.py

@@ -1259,46 +1259,6 @@ KERNEL_INFOS.extend(
     ]
 )
 
-_ADJUST_BRIGHTNESS_FACTORS = [0.1, 0.5]
-
-
-def sample_inputs_adjust_brightness_image_tensor():
-    for image_loader in make_image_loaders(sizes=[DEFAULT_PORTRAIT_SPATIAL_SIZE], color_spaces=("GRAY", "RGB")):
-        yield ArgsKwargs(image_loader, brightness_factor=_ADJUST_BRIGHTNESS_FACTORS[0])
-
-
-def reference_inputs_adjust_brightness_image_tensor():
-    for image_loader, brightness_factor in itertools.product(
-        make_image_loaders(color_spaces=("GRAY", "RGB"), extra_dims=[()], dtypes=[torch.uint8]),
-        _ADJUST_BRIGHTNESS_FACTORS,
-    ):
-        yield ArgsKwargs(image_loader, brightness_factor=brightness_factor)
-
-
-def sample_inputs_adjust_brightness_video():
-    for video_loader in make_video_loaders(sizes=[DEFAULT_PORTRAIT_SPATIAL_SIZE], num_frames=[3]):
-        yield ArgsKwargs(video_loader, brightness_factor=_ADJUST_BRIGHTNESS_FACTORS[0])
-
-
-KERNEL_INFOS.extend(
-    [
-        KernelInfo(
-            F.adjust_brightness_image_tensor,
-            kernel_name="adjust_brightness_image_tensor",
-            sample_inputs_fn=sample_inputs_adjust_brightness_image_tensor,
-            reference_fn=pil_reference_wrapper(F.adjust_brightness_image_pil),
-            reference_inputs_fn=reference_inputs_adjust_brightness_image_tensor,
-            float32_vs_uint8=True,
-            closeness_kwargs=float32_vs_uint8_pixel_difference(),
-        ),
-        KernelInfo(
-            F.adjust_brightness_video,
-            sample_inputs_fn=sample_inputs_adjust_brightness_video,
-        ),
-    ]
-)
-
-
 _ADJUST_CONTRAST_FACTORS = [0.1, 0.5]
 
 

torchvision/datapoints/__init__.py

 from torchvision import _BETA_TRANSFORMS_WARNING, _WARN_ABOUT_BETA_TRANSFORMS
 
 from ._bounding_box import BoundingBoxes, BoundingBoxFormat
-from ._datapoint import _FillType, _FillTypeJIT, _InputType, _InputTypeJIT
+from ._datapoint import _FillType, _FillTypeJIT, _InputType, _InputTypeJIT, Datapoint
 from ._image import _ImageType, _ImageTypeJIT, _TensorImageType, _TensorImageTypeJIT, Image
 from ._mask import Mask
 from ._video import _TensorVideoType, _TensorVideoTypeJIT, _VideoType, _VideoTypeJIT, Video

torchvision/datapoints/_bounding_box.py

 from __future__ import annotations
 
 from enum import Enum
-from typing import Any, List, Optional, Sequence, Tuple, Union
+from typing import Any, Optional, Tuple, Union
 
 import torch
-from torchvision.transforms import InterpolationMode  # TODO: this needs to be moved out of transforms
 
-from ._datapoint import _FillTypeJIT, Datapoint
+from ._datapoint import Datapoint
 
 
 class BoundingBoxFormat(Enum):
@@ -97,141 +96,3 @@ class BoundingBoxes(Datapoint):
 
     def __repr__(self, *, tensor_contents: Any = None) -> str:  # type: ignore[override]
         return self._make_repr(format=self.format, canvas_size=self.canvas_size)
-
-    def horizontal_flip(self) -> BoundingBoxes:
-        output = self._F.horizontal_flip_bounding_boxes(
-            self.as_subclass(torch.Tensor), format=self.format, canvas_size=self.canvas_size
-        )
-        return BoundingBoxes.wrap_like(self, output)
-
-    def vertical_flip(self) -> BoundingBoxes:
-        output = self._F.vertical_flip_bounding_boxes(
-            self.as_subclass(torch.Tensor), format=self.format, canvas_size=self.canvas_size
-        )
-        return BoundingBoxes.wrap_like(self, output)
-
-    def resize(  # type: ignore[override]
-        self,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        max_size: Optional[int] = None,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> BoundingBoxes:
-        output, canvas_size = self._F.resize_bounding_boxes(
-            self.as_subclass(torch.Tensor),
-            canvas_size=self.canvas_size,
-            size=size,
-            max_size=max_size,
-        )
-        return BoundingBoxes.wrap_like(self, output, canvas_size=canvas_size)
-
-    def crop(self, top: int, left: int, height: int, width: int) -> BoundingBoxes:
-        output, canvas_size = self._F.crop_bounding_boxes(
-            self.as_subclass(torch.Tensor), self.format, top=top, left=left, height=height, width=width
-        )
-        return BoundingBoxes.wrap_like(self, output, canvas_size=canvas_size)
-
-    def center_crop(self, output_size: List[int]) -> BoundingBoxes:
-        output, canvas_size = self._F.center_crop_bounding_boxes(
-            self.as_subclass(torch.Tensor), format=self.format, canvas_size=self.canvas_size, output_size=output_size
-        )
-        return BoundingBoxes.wrap_like(self, output, canvas_size=canvas_size)
-
-    def resized_crop(
-        self,
-        top: int,
-        left: int,
-        height: int,
-        width: int,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> BoundingBoxes:
-        output, canvas_size = self._F.resized_crop_bounding_boxes(
-            self.as_subclass(torch.Tensor), self.format, top, left, height, width, size=size
-        )
-        return BoundingBoxes.wrap_like(self, output, canvas_size=canvas_size)
-
-    def pad(
-        self,
-        padding: Union[int, Sequence[int]],
-        fill: Optional[Union[int, float, List[float]]] = None,
-        padding_mode: str = "constant",
-    ) -> BoundingBoxes:
-        output, canvas_size = self._F.pad_bounding_boxes(
-            self.as_subclass(torch.Tensor),
-            format=self.format,
-            canvas_size=self.canvas_size,
-            padding=padding,
-            padding_mode=padding_mode,
-        )
-        return BoundingBoxes.wrap_like(self, output, canvas_size=canvas_size)
-
-    def rotate(
-        self,
-        angle: float,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        expand: bool = False,
-        center: Optional[List[float]] = None,
-        fill: _FillTypeJIT = None,
-    ) -> BoundingBoxes:
-        output, canvas_size = self._F.rotate_bounding_boxes(
-            self.as_subclass(torch.Tensor),
-            format=self.format,
-            canvas_size=self.canvas_size,
-            angle=angle,
-            expand=expand,
-            center=center,
-        )
-        return BoundingBoxes.wrap_like(self, output, canvas_size=canvas_size)
-
-    def affine(
-        self,
-        angle: Union[int, float],
-        translate: List[float],
-        scale: float,
-        shear: List[float],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        fill: _FillTypeJIT = None,
-        center: Optional[List[float]] = None,
-    ) -> BoundingBoxes:
-        output = self._F.affine_bounding_boxes(
-            self.as_subclass(torch.Tensor),
-            self.format,
-            self.canvas_size,
-            angle,
-            translate=translate,
-            scale=scale,
-            shear=shear,
-            center=center,
-        )
-        return BoundingBoxes.wrap_like(self, output)
-
-    def perspective(
-        self,
-        startpoints: Optional[List[List[int]]],
-        endpoints: Optional[List[List[int]]],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        fill: _FillTypeJIT = None,
-        coefficients: Optional[List[float]] = None,
-    ) -> BoundingBoxes:
-        output = self._F.perspective_bounding_boxes(
-            self.as_subclass(torch.Tensor),
-            format=self.format,
-            canvas_size=self.canvas_size,
-            startpoints=startpoints,
-            endpoints=endpoints,
-            coefficients=coefficients,
-        )
-        return BoundingBoxes.wrap_like(self, output)
-
-    def elastic(
-        self,
-        displacement: torch.Tensor,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        fill: _FillTypeJIT = None,
-    ) -> BoundingBoxes:
-        output = self._F.elastic_bounding_boxes(
-            self.as_subclass(torch.Tensor), self.format, self.canvas_size, displacement=displacement
-        )
-        return BoundingBoxes.wrap_like(self, output)

torchvision/datapoints/_datapoint.py

 from __future__ import annotations
 
-from types import ModuleType
 from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Type, TypeVar, Union
 
 import PIL.Image
 import torch
 from torch._C import DisableTorchFunctionSubclass
 from torch.types import _device, _dtype, _size
-from torchvision.transforms import InterpolationMode
 
 
 D = TypeVar("D", bound="Datapoint")
@@ -16,8 +14,6 @@ _FillTypeJIT = Optional[List[float]]
 
 
 class Datapoint(torch.Tensor):
-    __F: Optional[ModuleType] = None
-
     @staticmethod
     def _to_tensor(
         data: Any,
@@ -99,18 +95,6 @@ class Datapoint(torch.Tensor):
         extra_repr = ", ".join(f"{key}={value}" for key, value in kwargs.items())
         return f"{super().__repr__()[:-1]}, {extra_repr})"
 
-    @property
-    def _F(self) -> ModuleType:
-        # This implements a lazy import of the functional to get around the cyclic import. This import is deferred
-        # until the first time we need reference to the functional module and it's shared across all instances of
-        # the class. This approach avoids the DataLoader issue described at
-        # https://github.com/pytorch/vision/pull/6476#discussion_r953588621
-        if Datapoint.__F is None:
-            from ..transforms.v2 import functional
-
-            Datapoint.__F = functional
-        return Datapoint.__F
-
     # Add properties for common attributes like shape, dtype, device, ndim etc
     # this way we return the result without passing into __torch_function__
     @property
@@ -142,128 +126,6 @@ class Datapoint(torch.Tensor):
         # `BoundingBoxes.clone()`.
         return self.detach().clone().requires_grad_(self.requires_grad)  # type: ignore[return-value]
 
-    def horizontal_flip(self) -> Datapoint:
-        return self
-
-    def vertical_flip(self) -> Datapoint:
-        return self
-
-    # TODO: We have to ignore override mypy error as there is torch.Tensor built-in deprecated op: Tensor.resize
-    # https://github.com/pytorch/pytorch/blob/e8727994eb7cdb2ab642749d6549bc497563aa06/torch/_tensor.py#L588-L593
-    def resize(  # type: ignore[override]
-        self,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        max_size: Optional[int] = None,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> Datapoint:
-        return self
-
-    def crop(self, top: int, left: int, height: int, width: int) -> Datapoint:
-        return self
-
-    def center_crop(self, output_size: List[int]) -> Datapoint:
-        return self
-
-    def resized_crop(
-        self,
-        top: int,
-        left: int,
-        height: int,
-        width: int,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> Datapoint:
-        return self
-
-    def pad(
-        self,
-        padding: List[int],
-        fill: Optional[Union[int, float, List[float]]] = None,
-        padding_mode: str = "constant",
-    ) -> Datapoint:
-        return self
-
-    def rotate(
-        self,
-        angle: float,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        expand: bool = False,
-        center: Optional[List[float]] = None,
-        fill: _FillTypeJIT = None,
-    ) -> Datapoint:
-        return self
-
-    def affine(
-        self,
-        angle: Union[int, float],
-        translate: List[float],
-        scale: float,
-        shear: List[float],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        fill: _FillTypeJIT = None,
-        center: Optional[List[float]] = None,
-    ) -> Datapoint:
-        return self
-
-    def perspective(
-        self,
-        startpoints: Optional[List[List[int]]],
-        endpoints: Optional[List[List[int]]],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        fill: _FillTypeJIT = None,
-        coefficients: Optional[List[float]] = None,
-    ) -> Datapoint:
-        return self
-
-    def elastic(
-        self,
-        displacement: torch.Tensor,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        fill: _FillTypeJIT = None,
-    ) -> Datapoint:
-        return self
-
-    def rgb_to_grayscale(self, num_output_channels: int = 1) -> Datapoint:
-        return self
-
-    def adjust_brightness(self, brightness_factor: float) -> Datapoint:
-        return self
-
-    def adjust_saturation(self, saturation_factor: float) -> Datapoint:
-        return self
-
-    def adjust_contrast(self, contrast_factor: float) -> Datapoint:
-        return self
-
-    def adjust_sharpness(self, sharpness_factor: float) -> Datapoint:
-        return self
-
-    def adjust_hue(self, hue_factor: float) -> Datapoint:
-        return self
-
-    def adjust_gamma(self, gamma: float, gain: float = 1) -> Datapoint:
-        return self
-
-    def posterize(self, bits: int) -> Datapoint:
-        return self
-
-    def solarize(self, threshold: float) -> Datapoint:
-        return self
-
-    def autocontrast(self) -> Datapoint:
-        return self
-
-    def equalize(self) -> Datapoint:
-        return self
-
-    def invert(self) -> Datapoint:
-        return self
-
-    def gaussian_blur(self, kernel_size: List[int], sigma: Optional[List[float]] = None) -> Datapoint:
-        return self
-
 
 _InputType = Union[torch.Tensor, PIL.Image.Image, Datapoint]
 _InputTypeJIT = torch.Tensor

torchvision/datapoints/_image.py

 from __future__ import annotations
 
-from typing import Any, List, Optional, Union
+from typing import Any, Optional, Union
 
 import PIL.Image
 import torch
-from torchvision.transforms.functional import InterpolationMode
 
-from ._datapoint import _FillTypeJIT, Datapoint
+from ._datapoint import Datapoint
 
 
 class Image(Datapoint):
@@ -56,195 +55,6 @@ class Image(Datapoint):
     def __repr__(self, *, tensor_contents: Any = None) -> str:  # type: ignore[override]
         return self._make_repr()
 
-    def horizontal_flip(self) -> Image:
-        output = self._F.horizontal_flip_image_tensor(self.as_subclass(torch.Tensor))
-        return Image.wrap_like(self, output)
-
-    def vertical_flip(self) -> Image:
-        output = self._F.vertical_flip_image_tensor(self.as_subclass(torch.Tensor))
-        return Image.wrap_like(self, output)
-
-    def resize(  # type: ignore[override]
-        self,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        max_size: Optional[int] = None,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> Image:
-        output = self._F.resize_image_tensor(
-            self.as_subclass(torch.Tensor), size, interpolation=interpolation, max_size=max_size, antialias=antialias
-        )
-        return Image.wrap_like(self, output)
-
-    def crop(self, top: int, left: int, height: int, width: int) -> Image:
-        output = self._F.crop_image_tensor(self.as_subclass(torch.Tensor), top, left, height, width)
-        return Image.wrap_like(self, output)
-
-    def center_crop(self, output_size: List[int]) -> Image:
-        output = self._F.center_crop_image_tensor(self.as_subclass(torch.Tensor), output_size=output_size)
-        return Image.wrap_like(self, output)
-
-    def resized_crop(
-        self,
-        top: int,
-        left: int,
-        height: int,
-        width: int,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> Image:
-        output = self._F.resized_crop_image_tensor(
-            self.as_subclass(torch.Tensor),
-            top,
-            left,
-            height,
-            width,
-            size=list(size),
-            interpolation=interpolation,
-            antialias=antialias,
-        )
-        return Image.wrap_like(self, output)
-
-    def pad(
-        self,
-        padding: List[int],
-        fill: Optional[Union[int, float, List[float]]] = None,
-        padding_mode: str = "constant",
-    ) -> Image:
-        output = self._F.pad_image_tensor(self.as_subclass(torch.Tensor), padding, fill=fill, padding_mode=padding_mode)
-        return Image.wrap_like(self, output)
-
-    def rotate(
-        self,
-        angle: float,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        expand: bool = False,
-        center: Optional[List[float]] = None,
-        fill: _FillTypeJIT = None,
-    ) -> Image:
-        output = self._F.rotate_image_tensor(
-            self.as_subclass(torch.Tensor), angle, interpolation=interpolation, expand=expand, fill=fill, center=center
-        )
-        return Image.wrap_like(self, output)
-
-    def affine(
-        self,
-        angle: Union[int, float],
-        translate: List[float],
-        scale: float,
-        shear: List[float],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        fill: _FillTypeJIT = None,
-        center: Optional[List[float]] = None,
-    ) -> Image:
-        output = self._F.affine_image_tensor(
-            self.as_subclass(torch.Tensor),
-            angle,
-            translate=translate,
-            scale=scale,
-            shear=shear,
-            interpolation=interpolation,
-            fill=fill,
-            center=center,
-        )
-        return Image.wrap_like(self, output)
-
-    def perspective(
-        self,
-        startpoints: Optional[List[List[int]]],
-        endpoints: Optional[List[List[int]]],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        fill: _FillTypeJIT = None,
-        coefficients: Optional[List[float]] = None,
-    ) -> Image:
-        output = self._F.perspective_image_tensor(
-            self.as_subclass(torch.Tensor),
-            startpoints,
-            endpoints,
-            interpolation=interpolation,
-            fill=fill,
-            coefficients=coefficients,
-        )
-        return Image.wrap_like(self, output)
-
-    def elastic(
-        self,
-        displacement: torch.Tensor,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        fill: _FillTypeJIT = None,
-    ) -> Image:
-        output = self._F.elastic_image_tensor(
-            self.as_subclass(torch.Tensor), displacement, interpolation=interpolation, fill=fill
-        )
-        return Image.wrap_like(self, output)
-
-    def rgb_to_grayscale(self, num_output_channels: int = 1) -> Image:
-        output = self._F.rgb_to_grayscale_image_tensor(
-            self.as_subclass(torch.Tensor), num_output_channels=num_output_channels
-        )
-        return Image.wrap_like(self, output)
-
-    def adjust_brightness(self, brightness_factor: float) -> Image:
-        output = self._F.adjust_brightness_image_tensor(
-            self.as_subclass(torch.Tensor), brightness_factor=brightness_factor
-        )
-        return Image.wrap_like(self, output)
-
-    def adjust_saturation(self, saturation_factor: float) -> Image:
-        output = self._F.adjust_saturation_image_tensor(
-            self.as_subclass(torch.Tensor), saturation_factor=saturation_factor
-        )
-        return Image.wrap_like(self, output)
-
-    def adjust_contrast(self, contrast_factor: float) -> Image:
-        output = self._F.adjust_contrast_image_tensor(self.as_subclass(torch.Tensor), contrast_factor=contrast_factor)
-        return Image.wrap_like(self, output)
-
-    def adjust_sharpness(self, sharpness_factor: float) -> Image:
-        output = self._F.adjust_sharpness_image_tensor(
-            self.as_subclass(torch.Tensor), sharpness_factor=sharpness_factor
-        )
-        return Image.wrap_like(self, output)
-
-    def adjust_hue(self, hue_factor: float) -> Image:
-        output = self._F.adjust_hue_image_tensor(self.as_subclass(torch.Tensor), hue_factor=hue_factor)
-        return Image.wrap_like(self, output)
-
-    def adjust_gamma(self, gamma: float, gain: float = 1) -> Image:
-        output = self._F.adjust_gamma_image_tensor(self.as_subclass(torch.Tensor), gamma=gamma, gain=gain)
-        return Image.wrap_like(self, output)
-
-    def posterize(self, bits: int) -> Image:
-        output = self._F.posterize_image_tensor(self.as_subclass(torch.Tensor), bits=bits)
-        return Image.wrap_like(self, output)
-
-    def solarize(self, threshold: float) -> Image:
-        output = self._F.solarize_image_tensor(self.as_subclass(torch.Tensor), threshold=threshold)
-        return Image.wrap_like(self, output)
-
-    def autocontrast(self) -> Image:
-        output = self._F.autocontrast_image_tensor(self.as_subclass(torch.Tensor))
-        return Image.wrap_like(self, output)
-
-    def equalize(self) -> Image:
-        output = self._F.equalize_image_tensor(self.as_subclass(torch.Tensor))
-        return Image.wrap_like(self, output)
-
-    def invert(self) -> Image:
-        output = self._F.invert_image_tensor(self.as_subclass(torch.Tensor))
-        return Image.wrap_like(self, output)
-
-    def gaussian_blur(self, kernel_size: List[int], sigma: Optional[List[float]] = None) -> Image:
-        output = self._F.gaussian_blur_image_tensor(
-            self.as_subclass(torch.Tensor), kernel_size=kernel_size, sigma=sigma
-        )
-        return Image.wrap_like(self, output)
-
-    def normalize(self, mean: List[float], std: List[float], inplace: bool = False) -> Image:
-        output = self._F.normalize_image_tensor(self.as_subclass(torch.Tensor), mean=mean, std=std, inplace=inplace)
-        return Image.wrap_like(self, output)
-
 
 _ImageType = Union[torch.Tensor, PIL.Image.Image, Image]
 _ImageTypeJIT = torch.Tensor

torchvision/datapoints/_mask.py

 from __future__ import annotations
 
-from typing import Any, List, Optional, Union
+from typing import Any, Optional, Union
 
 import PIL.Image
 import torch
-from torchvision.transforms import InterpolationMode
 
-from ._datapoint import _FillTypeJIT, Datapoint
+from ._datapoint import Datapoint
 
 
 class Mask(Datapoint):
@@ -50,105 +49,3 @@ class Mask(Datapoint):
         tensor: torch.Tensor,
     ) -> Mask:
         return cls._wrap(tensor)
-
-    def horizontal_flip(self) -> Mask:
-        output = self._F.horizontal_flip_mask(self.as_subclass(torch.Tensor))
-        return Mask.wrap_like(self, output)
-
-    def vertical_flip(self) -> Mask:
-        output = self._F.vertical_flip_mask(self.as_subclass(torch.Tensor))
-        return Mask.wrap_like(self, output)
-
-    def resize(  # type: ignore[override]
-        self,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        max_size: Optional[int] = None,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> Mask:
-        output = self._F.resize_mask(self.as_subclass(torch.Tensor), size, max_size=max_size)
-        return Mask.wrap_like(self, output)
-
-    def crop(self, top: int, left: int, height: int, width: int) -> Mask:
-        output = self._F.crop_mask(self.as_subclass(torch.Tensor), top, left, height, width)
-        return Mask.wrap_like(self, output)
-
-    def center_crop(self, output_size: List[int]) -> Mask:
-        output = self._F.center_crop_mask(self.as_subclass(torch.Tensor), output_size=output_size)
-        return Mask.wrap_like(self, output)
-
-    def resized_crop(
-        self,
-        top: int,
-        left: int,
-        height: int,
-        width: int,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> Mask:
-        output = self._F.resized_crop_mask(self.as_subclass(torch.Tensor), top, left, height, width, size=size)
-        return Mask.wrap_like(self, output)
-
-    def pad(
-        self,
-        padding: List[int],
-        fill: Optional[Union[int, float, List[float]]] = None,
-        padding_mode: str = "constant",
-    ) -> Mask:
-        output = self._F.pad_mask(self.as_subclass(torch.Tensor), padding, padding_mode=padding_mode, fill=fill)
-        return Mask.wrap_like(self, output)
-
-    def rotate(
-        self,
-        angle: float,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        expand: bool = False,
-        center: Optional[List[float]] = None,
-        fill: _FillTypeJIT = None,
-    ) -> Mask:
-        output = self._F.rotate_mask(self.as_subclass(torch.Tensor), angle, expand=expand, center=center, fill=fill)
-        return Mask.wrap_like(self, output)
-
-    def affine(
-        self,
-        angle: Union[int, float],
-        translate: List[float],
-        scale: float,
-        shear: List[float],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        fill: _FillTypeJIT = None,
-        center: Optional[List[float]] = None,
-    ) -> Mask:
-        output = self._F.affine_mask(
-            self.as_subclass(torch.Tensor),
-            angle,
-            translate=translate,
-            scale=scale,
-            shear=shear,
-            fill=fill,
-            center=center,
-        )
-        return Mask.wrap_like(self, output)
-
-    def perspective(
-        self,
-        startpoints: Optional[List[List[int]]],
-        endpoints: Optional[List[List[int]]],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        fill: _FillTypeJIT = None,
-        coefficients: Optional[List[float]] = None,
-    ) -> Mask:
-        output = self._F.perspective_mask(
-            self.as_subclass(torch.Tensor), startpoints, endpoints, fill=fill, coefficients=coefficients
-        )
-        return Mask.wrap_like(self, output)
-
-    def elastic(
-        self,
-        displacement: torch.Tensor,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        fill: _FillTypeJIT = None,
-    ) -> Mask:
-        output = self._F.elastic_mask(self.as_subclass(torch.Tensor), displacement, fill=fill)
-        return Mask.wrap_like(self, output)

torchvision/datapoints/_video.py

 from __future__ import annotations
 
-from typing import Any, List, Optional, Union
+from typing import Any, Optional, Union
 
 import torch
-from torchvision.transforms.functional import InterpolationMode
 
-from ._datapoint import _FillTypeJIT, Datapoint
+from ._datapoint import Datapoint
 
 
 class Video(Datapoint):
@@ -46,191 +45,6 @@ class Video(Datapoint):
     def __repr__(self, *, tensor_contents: Any = None) -> str:  # type: ignore[override]
         return self._make_repr()
 
-    def horizontal_flip(self) -> Video:
-        output = self._F.horizontal_flip_video(self.as_subclass(torch.Tensor))
-        return Video.wrap_like(self, output)
-
-    def vertical_flip(self) -> Video:
-        output = self._F.vertical_flip_video(self.as_subclass(torch.Tensor))
-        return Video.wrap_like(self, output)
-
-    def resize(  # type: ignore[override]
-        self,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        max_size: Optional[int] = None,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> Video:
-        output = self._F.resize_video(
-            self.as_subclass(torch.Tensor),
-            size,
-            interpolation=interpolation,
-            max_size=max_size,
-            antialias=antialias,
-        )
-        return Video.wrap_like(self, output)
-
-    def crop(self, top: int, left: int, height: int, width: int) -> Video:
-        output = self._F.crop_video(self.as_subclass(torch.Tensor), top, left, height, width)
-        return Video.wrap_like(self, output)
-
-    def center_crop(self, output_size: List[int]) -> Video:
-        output = self._F.center_crop_video(self.as_subclass(torch.Tensor), output_size=output_size)
-        return Video.wrap_like(self, output)
-
-    def resized_crop(
-        self,
-        top: int,
-        left: int,
-        height: int,
-        width: int,
-        size: List[int],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        antialias: Optional[Union[str, bool]] = "warn",
-    ) -> Video:
-        output = self._F.resized_crop_video(
-            self.as_subclass(torch.Tensor),
-            top,
-            left,
-            height,
-            width,
-            size=list(size),
-            interpolation=interpolation,
-            antialias=antialias,
-        )
-        return Video.wrap_like(self, output)
-
-    def pad(
-        self,
-        padding: List[int],
-        fill: Optional[Union[int, float, List[float]]] = None,
-        padding_mode: str = "constant",
-    ) -> Video:
-        output = self._F.pad_video(self.as_subclass(torch.Tensor), padding, fill=fill, padding_mode=padding_mode)
-        return Video.wrap_like(self, output)
-
-    def rotate(
-        self,
-        angle: float,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        expand: bool = False,
-        center: Optional[List[float]] = None,
-        fill: _FillTypeJIT = None,
-    ) -> Video:
-        output = self._F.rotate_video(
-            self.as_subclass(torch.Tensor), angle, interpolation=interpolation, expand=expand, fill=fill, center=center
-        )
-        return Video.wrap_like(self, output)
-
-    def affine(
-        self,
-        angle: Union[int, float],
-        translate: List[float],
-        scale: float,
-        shear: List[float],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.NEAREST,
-        fill: _FillTypeJIT = None,
-        center: Optional[List[float]] = None,
-    ) -> Video:
-        output = self._F.affine_video(
-            self.as_subclass(torch.Tensor),
-            angle,
-            translate=translate,
-            scale=scale,
-            shear=shear,
-            interpolation=interpolation,
-            fill=fill,
-            center=center,
-        )
-        return Video.wrap_like(self, output)
-
-    def perspective(
-        self,
-        startpoints: Optional[List[List[int]]],
-        endpoints: Optional[List[List[int]]],
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        fill: _FillTypeJIT = None,
-        coefficients: Optional[List[float]] = None,
-    ) -> Video:
-        output = self._F.perspective_video(
-            self.as_subclass(torch.Tensor),
-            startpoints,
-            endpoints,
-            interpolation=interpolation,
-            fill=fill,
-            coefficients=coefficients,
-        )
-        return Video.wrap_like(self, output)
-
-    def elastic(
-        self,
-        displacement: torch.Tensor,
-        interpolation: Union[InterpolationMode, int] = InterpolationMode.BILINEAR,
-        fill: _FillTypeJIT = None,
-    ) -> Video:
-        output = self._F.elastic_video(
-            self.as_subclass(torch.Tensor), displacement, interpolation=interpolation, fill=fill
-        )
-        return Video.wrap_like(self, output)
-
-    def rgb_to_grayscale(self, num_output_channels: int = 1) -> Video:
-        output = self._F.rgb_to_grayscale_image_tensor(
-            self.as_subclass(torch.Tensor), num_output_channels=num_output_channels
-        )
-        return Video.wrap_like(self, output)
-
-    def adjust_brightness(self, brightness_factor: float) -> Video:
-        output = self._F.adjust_brightness_video(self.as_subclass(torch.Tensor), brightness_factor=brightness_factor)
-        return Video.wrap_like(self, output)
-
-    def adjust_saturation(self, saturation_factor: float) -> Video:
-        output = self._F.adjust_saturation_video(self.as_subclass(torch.Tensor), saturation_factor=saturation_factor)
-        return Video.wrap_like(self, output)
-
-    def adjust_contrast(self, contrast_factor: float) -> Video:
-        output = self._F.adjust_contrast_video(self.as_subclass(torch.Tensor), contrast_factor=contrast_factor)
-        return Video.wrap_like(self, output)
-
-    def adjust_sharpness(self, sharpness_factor: float) -> Video:
-        output = self._F.adjust_sharpness_video(self.as_subclass(torch.Tensor), sharpness_factor=sharpness_factor)
-        return Video.wrap_like(self, output)
-
-    def adjust_hue(self, hue_factor: float) -> Video:
-        output = self._F.adjust_hue_video(self.as_subclass(torch.Tensor), hue_factor=hue_factor)
-        return Video.wrap_like(self, output)
-
-    def adjust_gamma(self, gamma: float, gain: float = 1) -> Video:
-        output = self._F.adjust_gamma_video(self.as_subclass(torch.Tensor), gamma=gamma, gain=gain)
-        return Video.wrap_like(self, output)
-
-    def posterize(self, bits: int) -> Video:
-        output = self._F.posterize_video(self.as_subclass(torch.Tensor), bits=bits)
-        return Video.wrap_like(self, output)
-
-    def solarize(self, threshold: float) -> Video:
-        output = self._F.solarize_video(self.as_subclass(torch.Tensor), threshold=threshold)
-        return Video.wrap_like(self, output)
-
-    def autocontrast(self) -> Video:
-        output = self._F.autocontrast_video(self.as_subclass(torch.Tensor))
-        return Video.wrap_like(self, output)
-
-    def equalize(self) -> Video:
-        output = self._F.equalize_video(self.as_subclass(torch.Tensor))
-        return Video.wrap_like(self, output)
-
-    def invert(self) -> Video:
-        output = self._F.invert_video(self.as_subclass(torch.Tensor))
-        return Video.wrap_like(self, output)
-
-    def gaussian_blur(self, kernel_size: List[int], sigma: Optional[List[float]] = None) -> Video:
-        output = self._F.gaussian_blur_video(self.as_subclass(torch.Tensor), kernel_size=kernel_size, sigma=sigma)
-        return Video.wrap_like(self, output)
-
-    def normalize(self, mean: List[float], std: List[float], inplace: bool = False) -> Video:
-        output = self._F.normalize_video(self.as_subclass(torch.Tensor), mean=mean, std=std, inplace=inplace)
-        return Video.wrap_like(self, output)
-
 
 _VideoType = Union[torch.Tensor, Video]
 _VideoTypeJIT = torch.Tensor

torchvision/transforms/v2/_augment.py

 import math
 import numbers
 import warnings
-from typing import Any, Dict, List, Tuple, Union
+from typing import Any, Dict, List, Tuple
 
 import PIL.Image
 import torch
@@ -56,8 +56,6 @@ class RandomErasing(_RandomApplyTransform):
             value="random" if self.value is None else self.value,
         )
 
-    _transformed_types = (is_simple_tensor, datapoints.Image, PIL.Image.Image, datapoints.Video)
-
     def __init__(
         self,
         p: float = 0.5,
@@ -131,9 +129,7 @@ class RandomErasing(_RandomApplyTransform):
 
         return dict(i=i, j=j, h=h, w=w, v=v)
 
-    def _transform(
-        self, inpt: Union[datapoints._ImageType, datapoints._VideoType], params: Dict[str, Any]
-    ) -> Union[datapoints._ImageType, datapoints._VideoType]:
+    def _transform(self, inpt: Any, params: Dict[str, Any]) -> Any:
         if params["v"] is not None:
             inpt = F.erase(inpt, **params, inplace=self.inplace)
 

torchvision/transforms/v2/_geometry.py

@@ -355,20 +355,11 @@ class FiveCrop(Transform):
 
     _v1_transform_cls = _transforms.FiveCrop
 
-    _transformed_types = (
-        datapoints.Image,
-        PIL.Image.Image,
-        is_simple_tensor,
-        datapoints.Video,
-    )
-
     def __init__(self, size: Union[int, Sequence[int]]) -> None:
         super().__init__()
         self.size = _setup_size(size, error_msg="Please provide only two dimensions (h, w) for size.")
 
-    def _transform(
-        self, inpt: ImageOrVideoTypeJIT, params: Dict[str, Any]
-    ) -> Tuple[ImageOrVideoTypeJIT, ImageOrVideoTypeJIT, ImageOrVideoTypeJIT, ImageOrVideoTypeJIT, ImageOrVideoTypeJIT]:
+    def _transform(self, inpt: Any, params: Dict[str, Any]) -> Any:
         return F.five_crop(inpt, self.size)
 
     def _check_inputs(self, flat_inputs: List[Any]) -> None:
@@ -402,13 +393,6 @@ class TenCrop(Transform):
 
     _v1_transform_cls = _transforms.TenCrop
 
-    _transformed_types = (
-        datapoints.Image,
-        PIL.Image.Image,
-        is_simple_tensor,
-        datapoints.Video,
-    )
-
     def __init__(self, size: Union[int, Sequence[int]], vertical_flip: bool = False) -> None:
         super().__init__()
         self.size = _setup_size(size, error_msg="Please provide only two dimensions (h, w) for size.")
@@ -418,20 +402,7 @@ class TenCrop(Transform):
         if has_any(flat_inputs, datapoints.BoundingBoxes, datapoints.Mask):
             raise TypeError(f"BoundingBoxes'es and Mask's are not supported by {type(self).__name__}()")
 
-    def _transform(
-        self, inpt: Union[datapoints._ImageType, datapoints._VideoType], params: Dict[str, Any]
-    ) -> Tuple[
-        ImageOrVideoTypeJIT,
-        ImageOrVideoTypeJIT,
-        ImageOrVideoTypeJIT,
-        ImageOrVideoTypeJIT,
-        ImageOrVideoTypeJIT,
-        ImageOrVideoTypeJIT,
-        ImageOrVideoTypeJIT,
-        ImageOrVideoTypeJIT,
-        ImageOrVideoTypeJIT,
-        ImageOrVideoTypeJIT,
-    ]:
+    def _transform(self, inpt: Any, params: Dict[str, Any]) -> Any:
         return F.ten_crop(inpt, self.size, vertical_flip=self.vertical_flip)
 
 

torchvision/transforms/v2/_temporal.py

 from typing import Any, Dict
 
+import torch
 from torchvision import datapoints
 from torchvision.transforms.v2 import functional as F, Transform
 
-from torchvision.transforms.v2.utils import is_simple_tensor
-
 
 class UniformTemporalSubsample(Transform):
     """[BETA] Uniformly subsample ``num_samples`` indices from the temporal dimension of the video.
@@ -20,7 +19,7 @@ class UniformTemporalSubsample(Transform):
         num_samples (int): The number of equispaced samples to be selected
     """
 
-    _transformed_types = (is_simple_tensor, datapoints.Video)
+    _transformed_types = (torch.Tensor,)
 
     def __init__(self, num_samples: int):
         super().__init__()

torchvision/transforms/v2/functional/__init__.py

 from torchvision.transforms import InterpolationMode  # usort: skip
 
-from ._utils import is_simple_tensor  # usort: skip
+from ._utils import is_simple_tensor, register_kernel  # usort: skip
 
 from ._meta import (
     clamp_bounding_boxes,
     convert_format_bounding_boxes,
     get_dimensions_image_tensor,
     get_dimensions_image_pil,
+    get_dimensions_video,
     get_dimensions,
     get_num_frames_video,
     get_num_frames,

torchvision/transforms/v2/functional/_augment.py

@@ -7,9 +7,37 @@ from torchvision import datapoints
 from torchvision.transforms.functional import pil_to_tensor, to_pil_image
 from torchvision.utils import _log_api_usage_once
 
-from ._utils import is_simple_tensor
+from ._utils import _get_kernel, _register_explicit_noop, _register_kernel_internal, is_simple_tensor
 
 
+@_register_explicit_noop(datapoints.Mask, datapoints.BoundingBoxes, warn_passthrough=True)
+def erase(
+    inpt: Union[datapoints._ImageTypeJIT, datapoints._VideoTypeJIT],
+    i: int,
+    j: int,
+    h: int,
+    w: int,
+    v: torch.Tensor,
+    inplace: bool = False,
+) -> Union[datapoints._ImageTypeJIT, datapoints._VideoTypeJIT]:
+    if not torch.jit.is_scripting():
+        _log_api_usage_once(erase)
+
+    if torch.jit.is_scripting() or is_simple_tensor(inpt):
+        return erase_image_tensor(inpt, i=i, j=j, h=h, w=w, v=v, inplace=inplace)
+    elif isinstance(inpt, datapoints.Datapoint):
+        kernel = _get_kernel(erase, type(inpt))
+        return kernel(inpt, i=i, j=j, h=h, w=w, v=v, inplace=inplace)
+    elif isinstance(inpt, PIL.Image.Image):
+        return erase_image_pil(inpt, i=i, j=j, h=h, w=w, v=v, inplace=inplace)
+    else:
+        raise TypeError(
+            f"Input can either be a plain tensor, any TorchVision datapoint, or a PIL image, "
+            f"but got {type(inpt)} instead."
+        )
+
+
+@_register_kernel_internal(erase, datapoints.Image)
 def erase_image_tensor(
     image: torch.Tensor, i: int, j: int, h: int, w: int, v: torch.Tensor, inplace: bool = False
 ) -> torch.Tensor:
@@ -29,36 +57,8 @@ def erase_image_pil(
     return to_pil_image(output, mode=image.mode)
 
 
+@_register_kernel_internal(erase, datapoints.Video)
 def erase_video(
     video: torch.Tensor, i: int, j: int, h: int, w: int, v: torch.Tensor, inplace: bool = False
 ) -> torch.Tensor:
     return erase_image_tensor(video, i=i, j=j, h=h, w=w, v=v, inplace=inplace)
-
-
-def erase(
-    inpt: Union[datapoints._ImageTypeJIT, datapoints._VideoTypeJIT],
-    i: int,
-    j: int,
-    h: int,
-    w: int,
-    v: torch.Tensor,
-    inplace: bool = False,
-) -> Union[datapoints._ImageTypeJIT, datapoints._VideoTypeJIT]:
-    if not torch.jit.is_scripting():
-        _log_api_usage_once(erase)
-
-    if torch.jit.is_scripting() or is_simple_tensor(inpt):
-        return erase_image_tensor(inpt, i=i, j=j, h=h, w=w, v=v, inplace=inplace)
-    elif isinstance(inpt, datapoints.Image):
-        output = erase_image_tensor(inpt.as_subclass(torch.Tensor), i=i, j=j, h=h, w=w, v=v, inplace=inplace)
-        return datapoints.Image.wrap_like(inpt, output)
-    elif isinstance(inpt, datapoints.Video):
-        output = erase_video(inpt.as_subclass(torch.Tensor), i=i, j=j, h=h, w=w, v=v, inplace=inplace)
-        return datapoints.Video.wrap_like(inpt, output)
-    elif isinstance(inpt, PIL.Image.Image):
-        return erase_image_pil(inpt, i=i, j=j, h=h, w=w, v=v, inplace=inplace)
-    else:
-        raise TypeError(
-            f"Input can either be a plain tensor, an `Image` or `Video` datapoint, or a PIL image, "
-            f"but got {type(inpt)} instead."
-        )