Bugfix in MotionBlur (#2529)

* Bugfix in MotionBlur

* Bugfix in MotionBlur

Author: ternaus

albumentations/augmentations/blur/functional.py

@@ -263,6 +263,9 @@ def create_motion_kernel(
         np.ndarray: Motion blur kernel
 
     """
+    # Validate direction range to prevent unexpected interpolation results
+    direction = np.clip(direction, -1.0, 1.0)
+
     kernel = np.zeros((kernel_size, kernel_size), dtype=np.float32)
     center = kernel_size // 2
 
@@ -275,20 +278,34 @@ def create_motion_kernel(
 
     # Create line points with direction bias
     line_length = kernel_size // 2
-    t = np.linspace(-line_length, line_length, kernel_size * 2)
 
-    # Apply direction bias
-    if direction != 0:
-        t = t * (1 + abs(direction))
-        if direction < 0:
-            t = t * -1
+    # Apply direction bias to control the distribution of blur
+    if direction < 0:
+        # Backward bias: interpolate between symmetric and backward-only
+        # direction = -1: only backward, direction = 0: symmetric
+        bias_factor = abs(direction)
+        t_start = float(-line_length)
+        t_end = line_length * (1 - bias_factor)
+    elif direction > 0:
+        # Forward bias: interpolate between symmetric and forward-only
+        # direction = 1: only forward, direction = 0: symmetric
+        bias_factor = direction
+        t_start = -line_length * (1 - bias_factor)
+        t_end = float(line_length)
+    else:
+        # Symmetric case (direction = 0)
+        t_start = float(-line_length)
+        t_end = float(line_length)
+
+    # Generate points along the biased line
+    t = np.linspace(t_start, t_end, kernel_size)
 
     # Generate line coordinates
     x = center + dx * t
     y = center + dy * t
 
     # Apply random shift if allowed
-    if allow_shifted and random_state is not None:
+    if allow_shifted:
         shift_x = random_state.uniform(-1, 1) * line_length / 2
         shift_y = random_state.uniform(-1, 1) * line_length / 2
         x += shift_x

tests/functional/test_blur.py

@@ -144,3 +144,307 @@ def test_kernel_visual_comparison():
     pil_kernel = create_pil_kernel(sigma)
 
     np.testing.assert_allclose(our_kernel, pil_kernel, rtol=1e-5)
+
+
+# === Motion Kernel Tests ===
+
+@pytest.mark.parametrize(
+    "kernel_size, expected_shape",
+    [
+        (3, (3, 3)),
+        (5, (5, 5)),
+        (7, (7, 7)),
+        (9, (9, 9)),
+    ]
+)
+def test_create_motion_kernel_shape(kernel_size, expected_shape):
+    """Test that motion kernel has the correct shape."""
+    random_state = Random(42)
+    kernel = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=0.0,
+        allow_shifted=False,
+        random_state=random_state
+    )
+    assert kernel.shape == expected_shape
+
+
+@pytest.mark.parametrize("kernel_size", [3, 5, 7, 9])
+def test_create_motion_kernel_normalization(kernel_size):
+    """Test that motion kernel has equal weight distribution along motion line."""
+    random_state = Random(42)
+    kernel = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=0.0,
+        allow_shifted=False,
+        random_state=random_state
+    )
+    # For symmetric horizontal motion, the kernel sum should equal the kernel size
+    # (each pixel along the horizontal line gets weight 1.0)
+    np.testing.assert_allclose(kernel.sum(), kernel_size, atol=1e-6)
+
+
+@pytest.mark.parametrize(
+    "direction, expected_behavior",
+    [
+        (-1.0, "backward"),   # Should have more weight toward start of line
+        (-0.5, "backward"),   # Moderate backward bias
+        (0.0, "symmetric"),   # Should be symmetric
+        (0.5, "forward"),     # Moderate forward bias
+        (1.0, "forward"),     # Should have more weight toward end of line
+    ]
+)
+def test_create_motion_kernel_direction_bias(direction, expected_behavior):
+    """Test that direction parameter controls blur direction correctly."""
+    random_state = Random(42)
+    kernel_size = 7
+    kernel = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,  # Horizontal line for easier testing
+        direction=direction,
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    # For horizontal motion, check the middle row
+    middle_row = kernel_size // 2
+    row = kernel[middle_row, :]
+
+    # Find center and calculate weights
+    center_col = kernel_size // 2
+    left_weight = np.sum(row[:center_col])
+    right_weight = np.sum(row[center_col + 1:])
+
+    if expected_behavior == "backward":
+        assert left_weight >= right_weight, f"direction={direction} should have more weight on the left"
+        assert left_weight > 0, f"direction={direction} should have some weight on the left"
+    elif expected_behavior == "symmetric":
+        assert abs(left_weight - right_weight) < 1e-6, f"direction={direction} should be symmetric"
+        assert left_weight > 0 and right_weight > 0, f"direction={direction} should have weight on both sides"
+    elif expected_behavior == "forward":
+        assert right_weight >= left_weight, f"direction={direction} should have more weight on the right"
+        assert right_weight > 0, f"direction={direction} should have some weight on the right"
+
+
+@pytest.mark.parametrize(
+    "angle, expected_orientation",
+    [
+        (0.0, "horizontal"),
+        (90.0, "vertical"),
+        (45.0, "diagonal"),
+        (135.0, "diagonal"),
+        (180.0, "horizontal"),
+        (270.0, "vertical"),
+    ]
+)
+def test_create_motion_kernel_angles(angle, expected_orientation):
+    """Test that angle parameter controls motion direction correctly."""
+    random_state = Random(42)
+    kernel_size = 7
+    kernel = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=angle,
+        direction=0.0,  # Symmetric for easier testing
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    # Check that kernel has non-zero values
+    assert kernel.sum() > 0, f"Kernel should have non-zero values for angle={angle}"
+
+    # For horizontal angles (0, 180), expect motion in middle row
+    if expected_orientation == "horizontal":
+        middle_row = kernel_size // 2
+        row_sum = np.sum(kernel[middle_row, :])
+        assert np.isclose(row_sum, kernel_size, atol=1.5) or row_sum > kernel_size * 0.8, (
+            f"Horizontal motion should be concentrated in middle row for angle={angle} "
+            f"(row_sum={row_sum}, expected≈{kernel_size})"
+        )
+
+    # For vertical angles (90, 270), expect motion in middle column
+    elif expected_orientation == "vertical":
+        middle_col = kernel_size // 2
+        col_sum = np.sum(kernel[:, middle_col])
+        assert np.isclose(col_sum, kernel_size, atol=1.5) or col_sum > kernel_size * 0.8, (
+            f"Vertical motion should be concentrated in middle column for angle={angle} "
+            f"(col_sum={col_sum}, expected≈{kernel_size})"
+        )
+
+
+@pytest.mark.parametrize("allow_shifted", [True, False])
+def test_create_motion_kernel_allow_shifted(allow_shifted):
+    """Test that allow_shifted parameter works correctly."""
+    random_state = Random(42)
+    kernel_size = 7
+
+    # Generate multiple kernels to test shifting behavior
+    kernels = []
+    for _ in range(10):
+        kernel = fblur.create_motion_kernel(
+            kernel_size=kernel_size,
+            angle=0.0,
+            direction=0.0,
+            allow_shifted=allow_shifted,
+            random_state=random_state
+        )
+        kernels.append(kernel)
+
+    if not allow_shifted:
+        # All kernels should be identical when shifting is disabled
+        for i in range(1, len(kernels)):
+            np.testing.assert_array_equal(kernels[0], kernels[i],
+                                        "Kernels should be identical when allow_shifted=False")
+    # Note: With shifting enabled, kernels might be different
+    # (This is harder to test deterministically due to randomness)
+
+
+@pytest.mark.parametrize(
+    "direction1, direction2",
+    [
+        (-1.0, 0.0),
+        (-1.0, 1.0),
+        (0.0, 1.0),
+        (-0.5, 0.5),
+    ]
+)
+def test_create_motion_kernel_different_directions_produce_different_kernels(direction1, direction2):
+    """Test that different direction values produce different kernels."""
+    random_state = Random(42)
+    kernel_size = 5
+
+    kernel1 = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=direction1,
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    # Reset random state to ensure same other parameters
+    random_state = Random(42)
+    kernel2 = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=direction2,
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    assert not np.array_equal(kernel1, kernel2), \
+        f"direction={direction1} and direction={direction2} should produce different kernels"
+
+
+def test_create_motion_kernel_extreme_directions():
+    """Test motion kernel with extreme direction values."""
+    random_state = Random(42)
+    kernel_size = 7
+
+    # Test direction = -1 (fully backward)
+    kernel_backward = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=-1.0,
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    # Test direction = 1 (fully forward)
+    random_state = Random(42)  # Reset for consistency
+    kernel_forward = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=1.0,
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    # For horizontal motion, check middle row
+    middle_row = kernel_size // 2
+    center_col = kernel_size // 2
+
+    # Backward should have no weight strictly to the right of center
+    backward_row = kernel_backward[middle_row, :]
+    assert np.sum(backward_row[center_col+1:]) == 0, "Fully backward motion should have no weight strictly to the right of center"
+    assert np.sum(backward_row[:center_col+1]) > 0, "Fully backward motion should have weight on left side and center"
+
+    # Forward should have no weight strictly to the left of center
+    forward_row = kernel_forward[middle_row, :]
+    assert np.sum(forward_row[:center_col]) == 0, "Fully forward motion should have no weight strictly to the left of center"
+    assert np.sum(forward_row[center_col:]) > 0, "Fully forward motion should have weight on center and right side"
+
+
+def test_create_motion_kernel_center_pixel_behavior():
+    """Test that center pixel is handled correctly for symmetric motion."""
+    random_state = Random(42)
+    kernel_size = 5  # Use odd size for clear center
+    kernel = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=0.0,  # Symmetric
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    # For symmetric horizontal motion, center pixel should have weight
+    center = kernel_size // 2
+    assert kernel[center, center] > 0, "Center pixel should have weight for symmetric motion"
+
+
+@pytest.mark.parametrize("kernel_size", [3, 5, 7])
+def test_create_motion_kernel_non_empty(kernel_size):
+    """Test that kernel always has at least one non-zero pixel."""
+    random_state = Random(42)
+
+    # Test with extreme parameters that might create empty kernels
+    kernel = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=0.0,
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    assert kernel.sum() > 0, "Kernel should never be completely empty"
+    assert np.any(kernel > 0), "Kernel should have at least one non-zero pixel"
+
+
+@pytest.mark.parametrize(
+    "direction_input, expected_clamped",
+    [
+        (-2.0, -1.0),  # Should be clamped to -1.0
+        (-1.5, -1.0),  # Should be clamped to -1.0
+        (1.5, 1.0),    # Should be clamped to 1.0
+        (2.0, 1.0),    # Should be clamped to 1.0
+        (0.5, 0.5),    # Should remain unchanged
+        (-0.5, -0.5),  # Should remain unchanged
+    ]
+)
+def test_create_motion_kernel_direction_validation(direction_input, expected_clamped):
+    """Test that direction values are properly clamped to [-1, 1] range."""
+    random_state = Random(42)
+    kernel_size = 5
+
+    # Create kernel with extreme direction value
+    kernel = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=direction_input,
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    # Create reference kernel with expected clamped value
+    random_state = Random(42)  # Reset for consistency
+    reference_kernel = fblur.create_motion_kernel(
+        kernel_size=kernel_size,
+        angle=0.0,
+        direction=expected_clamped,
+        allow_shifted=False,
+        random_state=random_state
+    )
+
+    # Kernels should be identical (direction was clamped to valid range)
+    np.testing.assert_array_equal(kernel, reference_kernel,
+                                  f"direction={direction_input} should be clamped to {expected_clamped}")