diff --git a/src/supervision/_cv2/__init__.py b/src/supervision/_cv2/__init__.py index 8a845339..7d103259 100644 --- a/src/supervision/_cv2/__init__.py +++ b/src/supervision/_cv2/__init__.py @@ -2,12 +2,49 @@ from __future__ import annotations -from typing import NoReturn - - -class BackendUnavailableError(RuntimeError): - """Raised when an OpenCV operation is used without an available backend.""" - +from supervision._cv2._color import _cvt_color, _merge, _split +from supervision._cv2._common import BackendUnavailableError, _unavailable +from supervision._cv2._image import ( + _add_weighted, + _convert_scale_abs, + _copy_make_border, + _flip, + _imread, + _imwrite, + _mean, + _resize, +) +from supervision._cv2._transform import ( + _blur, + _distance_transform, + _get_rotation_matrix_2d, + _warp_affine, +) +from supervision._cv2.constants import ( + _BORDER_CONSTANT, + _CAP_PROP_FPS, + _CAP_PROP_FRAME_COUNT, + _CAP_PROP_FRAME_HEIGHT, + _CAP_PROP_FRAME_WIDTH, + _CAP_PROP_POS_FRAMES, + _CC_STAT_AREA, + _CHAIN_APPROX_SIMPLE, + _COLOR_BGR2GRAY, + _COLOR_BGR2RGB, + _COLOR_GRAY2BGR, + _COLOR_HSV2BGR, + _COLOR_RGB2BGR, + _DIST_L2, + _FONT_HERSHEY_SIMPLEX, + _IMREAD_COLOR, + _IMREAD_UNCHANGED, + _INTER_LINEAR, + _INTER_NEAREST, + _LINE_4, + _LINE_AA, + _RETR_CCOMP, + _RETR_TREE, +) try: import cv2 @@ -16,30 +53,6 @@ except (ImportError, OSError): else: _IS_CV2_AVAILABLE = True -_BORDER_CONSTANT = 0 -_CAP_PROP_FPS = 5 -_CAP_PROP_FRAME_COUNT = 7 -_CAP_PROP_FRAME_HEIGHT = 4 -_CAP_PROP_FRAME_WIDTH = 3 -_CAP_PROP_POS_FRAMES = 1 -_CC_STAT_AREA = 4 -_CHAIN_APPROX_SIMPLE = 2 -_COLOR_BGR2GRAY = 6 -_COLOR_BGR2RGB = 4 -_COLOR_GRAY2BGR = 8 -_COLOR_HSV2BGR = 54 -_COLOR_RGB2BGR = 4 -_DIST_L2 = 2 -_FONT_HERSHEY_SIMPLEX = 0 -_IMREAD_COLOR = 1 -_IMREAD_UNCHANGED = -1 -_INTER_LINEAR = 1 -_INTER_NEAREST = 0 -_LINE_4 = 4 -_LINE_AA = 16 -_RETR_CCOMP = 2 -_RETR_TREE = 3 - if _IS_CV2_AVAILABLE: from cv2 import ( BORDER_CONSTANT, @@ -128,46 +141,39 @@ else: RETR_CCOMP = _RETR_CCOMP RETR_TREE = _RETR_TREE - def _unavailable(*args: object, **kwargs: object) -> NoReturn: - """Fail clearly until the corresponding fallback is implemented.""" - del args, kwargs - raise BackendUnavailableError( - "OpenCV is not installed and this operation has no fallback yet." - ) - VideoCapture = _unavailable VideoWriter = _unavailable VideoWriter_fourcc = _unavailable - addWeighted = _unavailable + addWeighted = _add_weighted approxPolyDP = _unavailable - blur = _unavailable + blur = _blur circle = _unavailable connectedComponents = _unavailable connectedComponentsWithStats = _unavailable contourArea = _unavailable - convertScaleAbs = _unavailable - copyMakeBorder = _unavailable - cvtColor = _unavailable - distanceTransform = _unavailable + convertScaleAbs = _convert_scale_abs + copyMakeBorder = _copy_make_border + cvtColor = _cvt_color + distanceTransform = _distance_transform drawContours = _unavailable ellipse = _unavailable fillPoly = _unavailable findContours = _unavailable - flip = _unavailable - getRotationMatrix2D = _unavailable + flip = _flip + getRotationMatrix2D = _get_rotation_matrix_2d getTextSize = _unavailable - imread = _unavailable - imwrite = _unavailable + imread = _imread + imwrite = _imwrite intersectConvexConvex = _unavailable line = _unavailable - mean = _unavailable - merge = _unavailable + mean = _mean + merge = _merge polylines = _unavailable putText = _unavailable rectangle = _unavailable - resize = _unavailable - split = _unavailable - warpAffine = _unavailable + resize = _resize + split = _split + warpAffine = _warp_affine __all__ = [ diff --git a/src/supervision/_cv2/_color.py b/src/supervision/_cv2/_color.py new file mode 100644 index 00000000..993723f2 --- /dev/null +++ b/src/supervision/_cv2/_color.py @@ -0,0 +1,85 @@ +"""Private color and channel-operation fallbacks.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +import numpy as np +import numpy.typing as npt + +from supervision._cv2._common import _cast_array_like_opencv +from supervision._cv2.constants import ( + _COLOR_BGR2GRAY, + _COLOR_BGR2RGB, + _COLOR_GRAY2BGR, + _COLOR_HSV2BGR, + _COLOR_RGB2BGR, +) + + +def _cvt_color(image: npt.NDArray[Any], code: int) -> npt.NDArray[Any]: + """Convert the BGR, RGB, grayscale, and 8-bit HSV formats used by Supervision.""" + if code in (_COLOR_BGR2RGB, _COLOR_RGB2BGR): + if image.ndim != 3 or image.shape[2] != 3: + raise ValueError("BGR/RGB conversion requires a three-channel image") + return np.ascontiguousarray(image[..., ::-1]) + + if code == _COLOR_GRAY2BGR: + if image.ndim != 2: + raise ValueError("GRAY2BGR conversion requires a two-dimensional image") + return np.repeat(image[..., np.newaxis], 3, axis=2) + + if code == _COLOR_BGR2GRAY: + if image.ndim != 3 or image.shape[2] != 3: + raise ValueError("BGR2GRAY conversion requires a three-channel image") + values = ( + image[..., 0].astype(np.float64) * 0.114 + + image[..., 1].astype(np.float64) * 0.587 + + image[..., 2].astype(np.float64) * 0.299 + ) + return _cast_array_like_opencv(values, image.dtype) + + if code == _COLOR_HSV2BGR: + if image.ndim != 3 or image.shape[2] != 3: + raise ValueError("HSV2BGR conversion requires a three-channel image") + return _hsv_to_bgr(image) + + raise ValueError(f"Unsupported color conversion code: {code}") + + +def _hsv_to_bgr(image: npt.NDArray[Any]) -> npt.NDArray[Any]: + """Convert OpenCV's 8-bit HSV representation to BGR.""" + values = image.astype(np.float64) + hue = values[..., 0] / 30.0 + saturation = values[..., 1] / 255.0 + value = values[..., 2] / 255.0 + + chroma = value * saturation + sector_index = np.floor(hue).astype(np.int64) % 6 + sector = hue - np.floor(hue) + x = chroma * (1 - np.abs(((sector_index + sector) % 2) - 1)) + match = value - chroma + zeros = np.zeros_like(chroma) + + red = np.choose(sector_index, (chroma, x, zeros, zeros, x, chroma)) + green = np.choose(sector_index, (x, chroma, chroma, x, zeros, zeros)) + blue = np.choose(sector_index, (zeros, zeros, x, chroma, chroma, x)) + bgr = np.stack((blue + match, green + match, red + match), axis=-1) * 255 + return _cast_array_like_opencv(bgr, image.dtype) + + +def _split(image: npt.NDArray[Any]) -> tuple[npt.NDArray[Any], ...]: + """Split an image into contiguous single-channel arrays.""" + if image.ndim == 2: + return (np.ascontiguousarray(image),) + return tuple( + np.ascontiguousarray(image[..., index]) for index in range(image.shape[2]) + ) + + +def _merge(channels: Sequence[npt.NDArray[Any]]) -> npt.NDArray[Any]: + """Merge single-channel arrays along their final axis.""" + if not channels: + raise ValueError("At least one channel is required") + return np.ascontiguousarray(np.stack(channels, axis=-1)) diff --git a/src/supervision/_cv2/_common.py b/src/supervision/_cv2/_common.py new file mode 100644 index 00000000..9d480d11 --- /dev/null +++ b/src/supervision/_cv2/_common.py @@ -0,0 +1,31 @@ +"""Private helpers shared by OpenCV fallback implementations.""" + +from __future__ import annotations + +from typing import Any, NoReturn + +import numpy as np +import numpy.typing as npt + + +class BackendUnavailableError(RuntimeError): + """Raised when an OpenCV operation is used without an available backend.""" + + +def _unavailable(*args: object, **kwargs: object) -> NoReturn: + """Fail clearly until a later domain PR provides the fallback operation.""" + del args, kwargs + raise BackendUnavailableError( + "OpenCV is not installed and this operation has no fallback yet." + ) + + +def _cast_array_like_opencv( + values: npt.NDArray[Any], dtype: np.dtype[Any] +) -> npt.NDArray[Any]: + """Round integer results using OpenCV's saturating conversion convention.""" + if np.issubdtype(dtype, np.integer): + info = np.iinfo(dtype) + values = np.rint(values) + values = np.clip(values, info.min, info.max) + return values.astype(dtype, copy=False) diff --git a/src/supervision/_cv2/_image.py b/src/supervision/_cv2/_image.py new file mode 100644 index 00000000..8bdb917d --- /dev/null +++ b/src/supervision/_cv2/_image.py @@ -0,0 +1,226 @@ +"""Private image-operation and image-I/O fallbacks.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any, cast + +import numpy as np +import numpy.typing as npt + +from supervision._cv2._common import _cast_array_like_opencv +from supervision._cv2.constants import ( + _BORDER_CONSTANT, + _IMREAD_COLOR, + _IMREAD_UNCHANGED, + _INTER_LINEAR, + _INTER_NEAREST, +) + + +def _flip(image: npt.NDArray[Any], flip_code: int) -> npt.NDArray[Any]: + """Flip an image vertically, horizontally, or along both axes.""" + if flip_code == 0: + axes: tuple[int, ...] = (0,) + elif flip_code == 1: + axes = (1,) + elif flip_code == -1: + axes = (0, 1) + else: + raise ValueError(f"Unsupported flip code: {flip_code}") + return np.ascontiguousarray(np.flip(image, axis=axes)) + + +def _copy_make_border( + image: npt.NDArray[Any], + top: int, + bottom: int, + left: int, + right: int, + border_type: int, + value: int | float | Sequence[int | float] = 0, +) -> npt.NDArray[Any]: + """Add a constant border around an image.""" + if border_type != _BORDER_CONSTANT: + raise ValueError("Only BORDER_CONSTANT is supported by the fallback") + if min(top, bottom, left, right) < 0: + raise ValueError("Border sizes must be non-negative") + + height, width = image.shape[:2] + shape = (height + top + bottom, width + left + right, *image.shape[2:]) + + fill_value: Any = value + if isinstance(value, Sequence): + if image.ndim == 2: + raise ValueError("Sequence border value requires a multi-channel image") + fill = np.asarray(value, dtype=image.dtype) + if fill.shape != (image.shape[2],): + raise ValueError("Border value must match the number of channels") + fill_value = fill.reshape((1, 1, -1)) + + result = np.full(shape, fill_value, dtype=image.dtype) + result[top : top + height, left : left + width] = image + return result + + +def _add_weighted( + source1: npt.NDArray[Any], + alpha: float, + source2: npt.NDArray[Any], + beta: float, + gamma: float, + dst: npt.NDArray[Any] | None = None, +) -> npt.NDArray[Any]: + """Blend two arrays with OpenCV-compatible saturation and optional mutation.""" + if source1.shape != source2.shape: + raise ValueError("addWeighted inputs must have equal shapes") + result = _cast_array_like_opencv( + source1.astype(np.float64) * alpha + source2.astype(np.float64) * beta + gamma, + source1.dtype, + ) + if dst is not None: + dst[...] = result + return dst + return result + + +def _convert_scale_abs( + image: npt.NDArray[Any], alpha: float = 1, beta: float = 0 +) -> npt.NDArray[np.uint8]: + """Scale, offset, take the absolute value, and saturate to uint8.""" + values = np.abs(image.astype(np.float64) * alpha + beta) + return _cast_array_like_opencv(values, np.dtype(np.uint8)) + + +def _mean( + image: npt.NDArray[Any], mask: npt.NDArray[Any] | None = None +) -> tuple[float, float, float, float]: + """Return per-channel means using OpenCV's four-value result contract.""" + if mask is None: + selected = ( + image.reshape(-1, 1) + if image.ndim == 2 + else image.reshape(-1, image.shape[2]) + ) + else: + if mask.shape != image.shape[:2]: + raise ValueError("Mean mask must match the image height and width") + selected = image[mask != 0] + if image.ndim == 2: + selected = selected.reshape(-1, 1) + if selected.size == 0: + means = np.zeros(4, dtype=np.float64) + else: + means = np.zeros(4, dtype=np.float64) + means[: selected.shape[1]] = np.mean(selected, axis=0) + return cast( + tuple[float, float, float, float], + tuple(float(value) for value in means), + ) + + +def _resize( + image: npt.NDArray[Any], + dsize: tuple[int, int], + fx: float = 0, + fy: float = 0, + interpolation: int = _INTER_LINEAR, +) -> npt.NDArray[Any]: + """Resize an array using OpenCV-compatible nearest or half-pixel linear sampling.""" + source_height, source_width = image.shape[:2] + width, height = dsize + if width == 0 or height == 0: + width = round(source_width * fx) + height = round(source_height * fy) + if min(width, height, source_width, source_height) <= 0: + raise ValueError("Resize dimensions must be positive") + + if interpolation == _INTER_NEAREST: + y_indices = np.minimum( + (np.arange(height) * source_height // height), source_height - 1 + ) + x_indices = np.minimum( + (np.arange(width) * source_width // width), source_width - 1 + ) + return np.ascontiguousarray(image[y_indices[:, np.newaxis], x_indices]) + + if interpolation != _INTER_LINEAR: + raise ValueError(f"Unsupported interpolation mode: {interpolation}") + + y = (np.arange(height) + 0.5) * source_height / height - 0.5 + x = (np.arange(width) + 0.5) * source_width / width - 0.5 + y_floor = np.floor(y).astype(np.int64) + x_floor = np.floor(x).astype(np.int64) + y0 = np.clip(y_floor, 0, source_height - 1) + y1 = np.clip(y_floor + 1, 0, source_height - 1) + x0 = np.clip(x_floor, 0, source_width - 1) + x1 = np.clip(x_floor + 1, 0, source_width - 1) + wy = y - y_floor + wx = x - x_floor + + source = image.astype(np.float64) + top = source[y0[:, np.newaxis], x0] + top_right = source[y0[:, np.newaxis], x1] + bottom = source[y1[:, np.newaxis], x0] + bottom_right = source[y1[:, np.newaxis], x1] + if image.ndim == 3: + wy = wy[:, np.newaxis, np.newaxis] + wx = wx[np.newaxis, :, np.newaxis] + else: + wy = wy[:, np.newaxis] + wx = wx[np.newaxis, :] + result = ( + top * (1 - wx) * (1 - wy) + + top_right * wx * (1 - wy) + + bottom * (1 - wx) * wy + + bottom_right * wx * wy + ) + return np.ascontiguousarray(_cast_array_like_opencv(result, image.dtype)) + + +def _imread(filename: str, flags: int = _IMREAD_COLOR) -> npt.NDArray[Any] | None: + """Read an image with Pillow while returning BGR or BGRA arrays.""" + from PIL import Image + + try: + with Image.open(filename) as image: + if flags == _IMREAD_UNCHANGED: + if image.mode == "P": + image = image.convert( + "RGBA" if "transparency" in image.info else "RGB" + ) + values = np.asarray(image) + elif image.mode in {"I", "I;16", "I;16B", "I;16L"}: + values = np.asarray(image).astype(np.float64) + values = np.clip(np.rint(values / 256), 0, 255).astype(np.uint8) + if values.ndim == 2: + values = np.repeat(values[..., np.newaxis], 3, axis=2) + else: + values = np.asarray(image.convert("RGB")) + except (FileNotFoundError, OSError): + return None + + if values.ndim == 3 and values.shape[2] == 3: + values = values[..., ::-1] + elif values.ndim == 3 and values.shape[2] == 4: + values = values[..., [2, 1, 0, 3]] + return np.ascontiguousarray(values) + + +def _imwrite( + filename: str, image: npt.NDArray[Any], params: Sequence[int] | None = None +) -> bool: + """Write a BGR or BGRA array with Pillow and return OpenCV's boolean status.""" + from PIL import Image + + del params + values = np.asarray(image) + if values.ndim == 3 and values.shape[2] == 3: + values = values[..., ::-1] + elif values.ndim == 3 and values.shape[2] == 4: + values = values[..., [2, 1, 0, 3]] + try: + Image.fromarray(np.ascontiguousarray(values)).save(filename) + except (OSError, ValueError): + return False + return True diff --git a/src/supervision/_cv2/_transform.py b/src/supervision/_cv2/_transform.py new file mode 100644 index 00000000..1f0430e6 --- /dev/null +++ b/src/supervision/_cv2/_transform.py @@ -0,0 +1,127 @@ +"""Private transform and filter fallbacks.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any, cast + +import numpy as np +import numpy.typing as npt + +from supervision._cv2._common import _cast_array_like_opencv +from supervision._cv2.constants import ( + _BORDER_CONSTANT, + _DIST_L2, + _INTER_LINEAR, + _INTER_NEAREST, +) + + +def _get_rotation_matrix_2d( + center: tuple[float, float], angle: float, scale: float +) -> npt.NDArray[np.float64]: + """Build OpenCV's two-dimensional rotation matrix.""" + radians = np.deg2rad(angle) + alpha = scale * np.cos(radians) + beta = scale * np.sin(radians) + center_x, center_y = center + return np.array( + [ + [alpha, beta, (1 - alpha) * center_x - beta * center_y], + [-beta, alpha, beta * center_x + (1 - alpha) * center_y], + ], + dtype=np.float64, + ) + + +def _warp_affine( + image: npt.NDArray[Any], + matrix: npt.NDArray[Any], + dsize: tuple[int, int], + flags: int = _INTER_LINEAR, + border_mode: int = _BORDER_CONSTANT, + border_value: float | Sequence[float] = 0, +) -> npt.NDArray[Any]: + """Warp an image through an affine matrix using SciPy's inverse sampler.""" + if flags not in (_INTER_NEAREST, _INTER_LINEAR): + raise ValueError(f"Unsupported interpolation mode: {flags}") + if border_mode != _BORDER_CONSTANT: + raise ValueError("Only BORDER_CONSTANT is supported by the fallback") + + from scipy import ndimage + + width, height = dsize + linear = np.asarray(matrix, dtype=np.float64)[:, :2] + translation = np.asarray(matrix, dtype=np.float64)[:, 2] + inverse = np.linalg.inv(linear) + offset_xy = -inverse @ translation + transform = inverse[[1, 0]][:, [1, 0]] + offset = offset_xy[[1, 0]] + order = 0 if flags == _INTER_NEAREST else 1 + values = np.asarray(image) + + def transform_channel(channel: npt.NDArray[Any], cval: float) -> npt.NDArray[Any]: + """Apply the shared affine mapping to one channel with padded borders.""" + padded = np.pad(channel, 1, mode="constant", constant_values=cval) + return cast( + npt.NDArray[Any], + ndimage.affine_transform( + padded, + transform, + offset=offset + 1, + output_shape=(height, width), + order=order, + mode="constant", + cval=cval, + prefilter=False, + ), + ) + + if values.ndim == 2: + cval = ( + float(border_value[0]) + if isinstance(border_value, Sequence) + else float(border_value) + ) + return transform_channel(values, cval) + + channels = [] + for channel in range(values.shape[2]): + cval = ( + float(border_value[channel]) + if isinstance(border_value, Sequence) + else float(border_value) + ) + channels.append(transform_channel(values[..., channel], cval)) + return np.stack(channels, axis=-1).astype(image.dtype, copy=False) + + +def _blur( + image: npt.NDArray[Any], ksize: tuple[int, int], border_type: int = 4 +) -> npt.NDArray[Any]: + """Apply a box filter with OpenCV's default reflect-101 boundary behavior.""" + if min(ksize) <= 0: + raise ValueError("Blur kernel dimensions must be positive") + if border_type != 4: + raise ValueError("Only OpenCV's default blur border is supported") + + from scipy import ndimage + + size = (*ksize[::-1], 1) if image.ndim == 3 else ksize[::-1] + values = ndimage.uniform_filter(image.astype(np.float64), size=size, mode="mirror") + return np.ascontiguousarray(_cast_array_like_opencv(values, image.dtype)) + + +def _distance_transform( + image: npt.NDArray[Any], distance_type: int, mask_size: int, dst_type: int = 5 +) -> npt.NDArray[np.float32]: + """Compute the L2 distance to the nearest zero pixel.""" + if distance_type != _DIST_L2: + raise ValueError("Only DIST_L2 is supported by the fallback") + del mask_size, dst_type + from scipy import ndimage + + return cast( + npt.NDArray[np.float32], + ndimage.distance_transform_edt(image != 0).astype(np.float32), + ) diff --git a/src/supervision/_cv2/constants.py b/src/supervision/_cv2/constants.py new file mode 100644 index 00000000..434a2223 --- /dev/null +++ b/src/supervision/_cv2/constants.py @@ -0,0 +1,25 @@ +"""Private numeric constants used by the OpenCV compatibility modules.""" + +_BORDER_CONSTANT = 0 +_CAP_PROP_FPS = 5 +_CAP_PROP_FRAME_COUNT = 7 +_CAP_PROP_FRAME_HEIGHT = 4 +_CAP_PROP_FRAME_WIDTH = 3 +_CAP_PROP_POS_FRAMES = 1 +_CC_STAT_AREA = 4 +_CHAIN_APPROX_SIMPLE = 2 +_COLOR_BGR2GRAY = 6 +_COLOR_BGR2RGB = 4 +_COLOR_GRAY2BGR = 8 +_COLOR_HSV2BGR = 54 +_COLOR_RGB2BGR = 4 +_DIST_L2 = 2 +_FONT_HERSHEY_SIMPLEX = 0 +_IMREAD_COLOR = 1 +_IMREAD_UNCHANGED = -1 +_INTER_LINEAR = 1 +_INTER_NEAREST = 0 +_LINE_4 = 4 +_LINE_AA = 16 +_RETR_CCOMP = 2 +_RETR_TREE = 3 diff --git a/tests/cv2/test_color.py b/tests/cv2/test_color.py new file mode 100644 index 00000000..40144791 --- /dev/null +++ b/tests/cv2/test_color.py @@ -0,0 +1,102 @@ +"""Tests for private color and channel fallbacks.""" + +from __future__ import annotations + +import importlib + +import numpy as np +import pytest + +from supervision._cv2._color import _cvt_color, _merge, _split +from supervision._cv2.constants import ( + _COLOR_BGR2GRAY, + _COLOR_BGR2RGB, + _COLOR_GRAY2BGR, + _COLOR_HSV2BGR, +) + +try: + cv2 = importlib.import_module("cv2") +except (ImportError, OSError): + pytest.skip( + "OpenCV is required as the reference implementation for this test module", + allow_module_level=True, + ) + + +@pytest.mark.parametrize( + ("source", "fallback_code", "opencv_code", "atol"), + [ + pytest.param( + np.array( + [[[10, 20, 30], [40, 50, 60]], [[70, 80, 90], [100, 110, 120]]], + dtype=np.uint8, + ), + _COLOR_BGR2RGB, + cv2.COLOR_BGR2RGB, + 0, + id="bgr-to-rgb", + ), + pytest.param( + np.array( + [[[10, 20, 30], [40, 50, 60]], [[70, 80, 90], [100, 110, 120]]], + dtype=np.uint8, + ), + _COLOR_BGR2GRAY, + cv2.COLOR_BGR2GRAY, + 0, + id="bgr-to-gray", + ), + pytest.param( + np.array([[0, 64], [128, 255]], dtype=np.uint8), + _COLOR_GRAY2BGR, + cv2.COLOR_GRAY2BGR, + 0, + id="gray-to-bgr", + ), + pytest.param( + np.array( + [[[0, 255, 255], [30, 255, 255]], [[60, 255, 255], [150, 255, 255]]], + dtype=np.uint8, + ), + _COLOR_HSV2BGR, + cv2.COLOR_HSV2BGR, + 1, + id="hsv-to-bgr", + ), + ], +) +def test_fallback_color_operations_match_opencv( + source: np.ndarray, fallback_code: int, opencv_code: int, atol: int +) -> None: + """Match OpenCV for each supported color conversion.""" + actual = _cvt_color(source, fallback_code) + expected = cv2.cvtColor(source, opencv_code) + + np.testing.assert_allclose(actual, expected, atol=atol, rtol=0) + + +def test_fallback_split_matches_opencv() -> None: + """Match OpenCV channel splitting.""" + bgr = np.array( + [[[10, 20, 30], [40, 50, 60]], [[70, 80, 90], [100, 110, 120]]], + dtype=np.uint8, + ) + + actual = _split(bgr) + expected = cv2.split(bgr) + + assert len(actual) == len(expected) + for actual_channel, expected_channel in zip(actual, expected): + np.testing.assert_array_equal(actual_channel, expected_channel) + + +def test_fallback_merge_matches_opencv() -> None: + """Match OpenCV channel merging.""" + bgr = np.array( + [[[10, 20, 30], [40, 50, 60]], [[70, 80, 90], [100, 110, 120]]], + dtype=np.uint8, + ) + channels = cv2.split(bgr) + + np.testing.assert_array_equal(_merge(channels), cv2.merge(channels)) diff --git a/tests/cv2/test_common.py b/tests/cv2/test_common.py new file mode 100644 index 00000000..49f1cb4c --- /dev/null +++ b/tests/cv2/test_common.py @@ -0,0 +1,13 @@ +"""Tests for shared private OpenCV fallback helpers.""" + +from __future__ import annotations + +import pytest + +from supervision._cv2._common import BackendUnavailableError, _unavailable + + +def test_unavailable_operation_raises_actionable_error() -> None: + """Explain which backend is missing when an operation is not implemented.""" + with pytest.raises(BackendUnavailableError, match="OpenCV is not installed"): + _unavailable() diff --git a/tests/cv2/test_constants.py b/tests/cv2/test_constants.py new file mode 100644 index 00000000..926a3843 --- /dev/null +++ b/tests/cv2/test_constants.py @@ -0,0 +1,216 @@ +"""Tests for private OpenCV compatibility constants.""" + +from __future__ import annotations + +import importlib +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from supervision import _cv2 + +try: + cv2 = importlib.import_module("cv2") +except (ImportError, OSError): + pytest.skip( + "OpenCV is required as the reference implementation for this test module", + allow_module_level=True, + ) + + +OPENCV_CONSTANTS = [ + "BORDER_CONSTANT", + "CAP_PROP_FPS", + "CAP_PROP_FRAME_COUNT", + "CAP_PROP_FRAME_HEIGHT", + "CAP_PROP_FRAME_WIDTH", + "CAP_PROP_POS_FRAMES", + "CC_STAT_AREA", + "CHAIN_APPROX_SIMPLE", + "COLOR_BGR2GRAY", + "COLOR_BGR2RGB", + "COLOR_GRAY2BGR", + "COLOR_HSV2BGR", + "COLOR_RGB2BGR", + "DIST_L2", + "FONT_HERSHEY_SIMPLEX", + "IMREAD_COLOR", + "IMREAD_UNCHANGED", + "INTER_LINEAR", + "INTER_NEAREST", + "LINE_4", + "LINE_AA", + "RETR_CCOMP", + "RETR_TREE", +] + + +def _run_without_opencv(source: str) -> None: + """Run a Python snippet with cv2 imports blocked.""" + env = os.environ.copy() + source_path = str(Path(__file__).resolve().parents[2] / "src") + env["PYTHONPATH"] = os.pathsep.join( + filter(None, (source_path, env.get("PYTHONPATH"))) + ) + subprocess.run( # noqa: S603 + [sys.executable, "-c", source], + check=True, + env=env, + ) + + +@pytest.mark.parametrize("name", OPENCV_CONSTANTS) +def test_fallback_constant_matches_opencv(name: str) -> None: + """Keep each private fallback constant aligned with the OpenCV reference.""" + actual = getattr(_cv2, f"_{name}") + expected = getattr(cv2, name) + + assert actual == expected + + +def test_facade_reports_fallback_backend_without_opencv() -> None: + """Report the fallback backend when cv2 is unavailable.""" + _run_without_opencv( + """ +import sys + + +class BlockCv2: + def find_spec(self, fullname, path=None, target=None): + if fullname == "cv2": + raise ModuleNotFoundError("blocked for test") + return None + + +sys.meta_path.insert(0, BlockCv2()) +from supervision import _cv2 + +assert _cv2._IS_CV2_AVAILABLE is False +assert _cv2.BACKEND_NAME == "fallback" +""" + ) + + +def test_facade_preserves_constants_without_opencv() -> None: + """Preserve the OpenCV constant values when cv2 is unavailable.""" + expected_constants = {name: getattr(cv2, name) for name in OPENCV_CONSTANTS} + _run_without_opencv( + f""" +import sys + + +class BlockCv2: + def find_spec(self, fullname, path=None, target=None): + if fullname == "cv2": + raise ModuleNotFoundError("blocked for test") + return None + + +sys.meta_path.insert(0, BlockCv2()) +from supervision import _cv2 + +assert _cv2._IS_CV2_AVAILABLE is False +expected = {expected_constants!r} +actual = {{name: getattr(_cv2, name) for name in expected}} +fallback = {{name: getattr(_cv2, f"_{{name}}") for name in expected}} +assert fallback == expected +assert actual == expected +""" + ) + + +def test_facade_routes_color_calls_without_opencv() -> None: + """Route color conversion calls to the fallback without cv2.""" + _run_without_opencv( + """ +import sys + + +class BlockCv2: + def find_spec(self, fullname, path=None, target=None): + if fullname == "cv2": + raise ModuleNotFoundError("blocked for test") + return None + + +sys.meta_path.insert(0, BlockCv2()) +from supervision import _cv2 + +image = __import__("numpy").array([[[10, 20, 30]]], dtype="uint8") +assert _cv2.cvtColor(image, _cv2.COLOR_BGR2RGB).tolist() == [[[30, 20, 10]]] +""" + ) + + +def test_facade_routes_resize_calls_without_opencv() -> None: + """Route resize calls to the fallback without cv2.""" + _run_without_opencv( + """ +import sys + + +class BlockCv2: + def find_spec(self, fullname, path=None, target=None): + if fullname == "cv2": + raise ModuleNotFoundError("blocked for test") + return None + + +sys.meta_path.insert(0, BlockCv2()) +from supervision import _cv2 + +image = __import__("numpy").array([[[10, 20, 30]]], dtype="uint8") +assert _cv2.resize(image, (2, 1), interpolation=_cv2.INTER_NEAREST).shape == (1, 2, 3) +""" + ) + + +@pytest.mark.parametrize( + ("public_name", "private_name"), + [ + pytest.param("addWeighted", "_add_weighted", id="addWeighted"), + pytest.param("blur", "_blur", id="blur"), + pytest.param("convertScaleAbs", "_convert_scale_abs", id="convertScaleAbs"), + pytest.param("copyMakeBorder", "_copy_make_border", id="copyMakeBorder"), + pytest.param("cvtColor", "_cvt_color", id="cvtColor"), + pytest.param( + "distanceTransform", "_distance_transform", id="distanceTransform" + ), + pytest.param("flip", "_flip", id="flip"), + pytest.param( + "getRotationMatrix2D", "_get_rotation_matrix_2d", id="getRotationMatrix2D" + ), + pytest.param("imread", "_imread", id="imread"), + pytest.param("imwrite", "_imwrite", id="imwrite"), + pytest.param("mean", "_mean", id="mean"), + pytest.param("merge", "_merge", id="merge"), + pytest.param("resize", "_resize", id="resize"), + pytest.param("split", "_split", id="split"), + pytest.param("warpAffine", "_warp_affine", id="warpAffine"), + ], +) +def test_facade_binds_fallback_operation_without_opencv( + public_name: str, private_name: str +) -> None: + """Bind each public fallback operation to its private implementation.""" + _run_without_opencv( + f""" +import sys + + +class BlockCv2: + def find_spec(self, fullname, path=None, target=None): + if fullname == "cv2": + raise ModuleNotFoundError("blocked for test") + return None + + +sys.meta_path.insert(0, BlockCv2()) +from supervision import _cv2 + +assert getattr(_cv2, {public_name!r}) is getattr(_cv2, {private_name!r}) +""" + ) diff --git a/tests/cv2/test_cv2.py b/tests/cv2/test_cv2.py index a172fcc8..65fd0da3 100644 --- a/tests/cv2/test_cv2.py +++ b/tests/cv2/test_cv2.py @@ -1,19 +1,16 @@ -"""Tests for the private OpenCV compatibility surface.""" +"""Tests for the private OpenCV facade.""" from __future__ import annotations import importlib -import os import subprocess import sys -from pathlib import Path import numpy as np import pytest from supervision import _cv2 -# Use the real OpenCV module as the oracle for compatibility comparisons. try: cv2 = importlib.import_module("cv2") except (ImportError, OSError): @@ -23,32 +20,6 @@ except (ImportError, OSError): ) -OPENCV_CONSTANTS = [ - "BORDER_CONSTANT", - "CAP_PROP_FPS", - "CAP_PROP_FRAME_COUNT", - "CAP_PROP_FRAME_HEIGHT", - "CAP_PROP_FRAME_WIDTH", - "CAP_PROP_POS_FRAMES", - "CC_STAT_AREA", - "CHAIN_APPROX_SIMPLE", - "COLOR_BGR2GRAY", - "COLOR_BGR2RGB", - "COLOR_GRAY2BGR", - "COLOR_HSV2BGR", - "COLOR_RGB2BGR", - "DIST_L2", - "FONT_HERSHEY_SIMPLEX", - "IMREAD_COLOR", - "IMREAD_UNCHANGED", - "INTER_LINEAR", - "INTER_NEAREST", - "LINE_4", - "LINE_AA", - "RETR_CCOMP", - "RETR_TREE", -] - REQUIRED_SYMBOLS = { "VideoCapture", "VideoWriter", @@ -83,89 +54,49 @@ REQUIRED_SYMBOLS = { "resize", "split", "warpAffine", -} | set(OPENCV_CONSTANTS) +} -def test_facade_exports_the_required_opencv_surface() -> None: - """Expose every OpenCV symbol used by production call sites.""" - assert REQUIRED_SYMBOLS <= set(_cv2.__all__) - assert all(hasattr(_cv2, symbol) for symbol in REQUIRED_SYMBOLS) +@pytest.mark.parametrize( + "symbol", + [ + pytest.param(symbol, id=symbol.lower().replace("_", "-")) + for symbol in sorted(REQUIRED_SYMBOLS) + ], +) +def test_facade_exports_required_opencv_symbol(symbol: str) -> None: + """Expose each OpenCV symbol used by production call sites.""" + assert symbol in _cv2.__all__ + assert hasattr(_cv2, symbol) + + +def test_facade_reports_opencv_backend() -> None: + """Report OpenCV when the native backend is available.""" assert _cv2.BACKEND_NAME == "opencv" -@pytest.mark.parametrize("name", OPENCV_CONSTANTS) -def test_fallback_constant_matches_opencv(name: str) -> None: - """Keep each private fallback constant aligned with the OpenCV reference.""" - actual = getattr(_cv2, f"_{name}") - expected = getattr(cv2, name) - - assert actual == expected - - -def test_facade_calls_the_package_imported_surface() -> None: - """Route production-style calls through the Supervision facade.""" +def test_facade_routes_color_calls_to_opencv() -> None: + """Route color conversion calls through the Supervision facade.""" image = np.array([[[10, 20, 30], [40, 50, 60]]], dtype=np.uint8) np.testing.assert_array_equal( _cv2.cvtColor(image, _cv2.COLOR_BGR2RGB), cv2.cvtColor(image, cv2.COLOR_BGR2RGB), ) + + +def test_facade_routes_resize_calls_to_opencv() -> None: + """Route resize calls through the Supervision facade.""" + image = np.array([[[10, 20, 30], [40, 50, 60]]], dtype=np.uint8) + np.testing.assert_array_equal( _cv2.resize(image, (4, 2), interpolation=_cv2.INTER_NEAREST), cv2.resize(image, (4, 2), interpolation=cv2.INTER_NEAREST), ) -def test_facade_imports_without_opencv() -> None: - """Keep fallback imports and constants valid when cv2 is unavailable.""" - env = os.environ.copy() - source_path = str(Path(__file__).resolve().parents[2] / "src") - env["PYTHONPATH"] = os.pathsep.join( - filter(None, (source_path, env.get("PYTHONPATH"))) - ) - expected_constants = {name: getattr(cv2, name) for name in OPENCV_CONSTANTS} - code = f""" -import sys - - -class BlockCv2: - def find_spec(self, fullname, path=None, target=None): - if fullname == "cv2": - raise ModuleNotFoundError("blocked for test") - return None - - -sys.meta_path.insert(0, BlockCv2()) -from supervision import _cv2 - -assert _cv2._IS_CV2_AVAILABLE is False -assert _cv2.BACKEND_NAME == "fallback" -expected = {expected_constants!r} -actual = {{name: getattr(_cv2, name) for name in expected}} -fallback = {{name: getattr(_cv2, f"_{{name}}") for name in expected}} -assert fallback == expected -assert actual == expected -try: - _cv2.resize(None, (1, 1), interpolation=_cv2.INTER_NEAREST) -except _cv2.BackendUnavailableError: - pass -else: - raise AssertionError("missing OpenCV must fail explicitly") -""" - subprocess.run( # noqa: S603 - [sys.executable, "-c", code], - check=True, - env=env, - ) - - def test_facade_does_not_hide_a_breaking_opencv_import() -> None: """Raise when cv2 imports but no longer exposes a required symbol.""" - env = os.environ.copy() - source_path = str(Path(__file__).resolve().parents[2] / "src") - env["PYTHONPATH"] = os.pathsep.join( - filter(None, (source_path, env.get("PYTHONPATH"))) - ) code = """ import sys import types @@ -177,7 +108,6 @@ from supervision import _cv2 [sys.executable, "-c", code], capture_output=True, text=True, - env=env, ) assert result.returncode != 0 diff --git a/tests/cv2/test_image.py b/tests/cv2/test_image.py new file mode 100644 index 00000000..9590996b --- /dev/null +++ b/tests/cv2/test_image.py @@ -0,0 +1,172 @@ +"""Tests for private image-operation and I/O fallbacks.""" + +from __future__ import annotations + +import importlib +from pathlib import Path + +import numpy as np +import pytest + +from supervision._cv2._image import ( + _add_weighted, + _convert_scale_abs, + _copy_make_border, + _flip, + _imread, + _imwrite, + _mean, + _resize, +) +from supervision._cv2.constants import ( + _BORDER_CONSTANT, + _IMREAD_COLOR, + _IMREAD_UNCHANGED, +) + +try: + cv2 = importlib.import_module("cv2") +except (ImportError, OSError): + pytest.skip( + "OpenCV is required as the reference implementation for this test module", + allow_module_level=True, + ) + + +@pytest.mark.parametrize( + ("flip_code", "expected"), + [ + pytest.param(0, np.array([[3, 4], [1, 2]], dtype=np.uint8), id="vertical"), + pytest.param(1, np.array([[2, 1], [4, 3]], dtype=np.uint8), id="horizontal"), + pytest.param(-1, np.array([[4, 3], [2, 1]], dtype=np.uint8), id="both"), + ], +) +def test_fallback_flip_matches_opencv(flip_code: int, expected: np.ndarray) -> None: + """Match OpenCV flip direction and return a contiguous array.""" + source = np.array([[1, 2], [3, 4]], dtype=np.uint8) + + np.testing.assert_array_equal(_flip(source, flip_code), expected) + np.testing.assert_array_equal(_flip(source, flip_code), cv2.flip(source, flip_code)) + + +def test_fallback_copy_make_border_matches_opencv() -> None: + """Match OpenCV constant-border padding.""" + source = np.array([[0, 100], [200, 255]], dtype=np.uint8) + + np.testing.assert_array_equal( + _copy_make_border(source, 1, 1, 2, 2, _BORDER_CONSTANT, 7), + cv2.copyMakeBorder(source, 1, 1, 2, 2, cv2.BORDER_CONSTANT, value=7), + ) + + +def test_fallback_add_weighted_matches_opencv() -> None: + """Match OpenCV weighted image blending.""" + source = np.array([[0, 100], [200, 255]], dtype=np.uint8) + other = np.full_like(source, 50) + + np.testing.assert_array_equal( + _add_weighted(source, 0.5, other, 0.5, 10), + cv2.addWeighted(source, 0.5, other, 0.5, 10), + ) + + +def test_fallback_add_weighted_supports_destination() -> None: + """Write weighted image blending results into the provided destination.""" + source = np.array([[0, 100], [200, 255]], dtype=np.uint8) + other = np.full_like(source, 50) + destination = np.empty_like(source) + + actual = _add_weighted(source, 0.5, other, 0.5, 10, dst=destination) + + assert actual is destination + np.testing.assert_array_equal(actual, cv2.addWeighted(source, 0.5, other, 0.5, 10)) + + +def test_fallback_convert_scale_abs_matches_opencv() -> None: + """Match OpenCV absolute scale-and-convert semantics.""" + source = np.array([[0, 100], [200, 255]], dtype=np.uint8) + + np.testing.assert_array_equal( + _convert_scale_abs(source, 1.5, -20), + cv2.convertScaleAbs(source, alpha=1.5, beta=-20), + ) + + +def test_fallback_mean_matches_opencv() -> None: + """Match OpenCV masked mean semantics.""" + source = np.array([[0, 100], [200, 255]], dtype=np.uint8) + mask = np.array([[255, 0], [0, 255]], dtype=np.uint8) + + assert _mean(source, mask) == cv2.mean(source, mask) + + +@pytest.mark.parametrize( + ("interpolation", "atol"), + [ + pytest.param(cv2.INTER_NEAREST, 0, id="nearest"), + pytest.param(cv2.INTER_LINEAR, 1, id="linear"), + ], +) +def test_fallback_resize_matches_opencv(interpolation: int, atol: int) -> None: + """Match OpenCV resize shape and pixel values within the interpolation budget.""" + source = np.arange(20, dtype=np.uint8).reshape(4, 5) + + actual = _resize(source, (9, 7), interpolation=interpolation) + expected = cv2.resize(source, (9, 7), interpolation=interpolation) + + assert actual.shape == expected.shape + np.testing.assert_allclose(actual, expected, atol=atol, rtol=0) + + +def test_fallback_image_io_preserves_bgr(tmp_path: Path) -> None: + """Preserve BGR channel order when writing and reading an image.""" + image = np.array([[[10, 20, 30], [40, 50, 60]]], dtype=np.uint8) + image_path = tmp_path / "image.png" + + assert _imwrite(str(image_path), image) + actual = _imread(str(image_path), _IMREAD_COLOR) + assert actual is not None + np.testing.assert_array_equal(actual, image) + + +def test_fallback_image_io_returns_none_for_missing_file(tmp_path: Path) -> None: + """Return None when reading a missing image file.""" + assert _imread(str(tmp_path / "missing.png"), _IMREAD_COLOR) is None + + +def test_fallback_image_io_preserves_alpha(tmp_path: Path) -> None: + """Preserve alpha channels when reading unchanged images.""" + alpha = np.array([[[10, 20, 30, 40], [50, 60, 70, 80]]], dtype=np.uint8) + alpha_path = tmp_path / "alpha.png" + + assert _imwrite(str(alpha_path), alpha) + np.testing.assert_array_equal( + _imread(str(alpha_path), _IMREAD_UNCHANGED), + cv2.imread(str(alpha_path), cv2.IMREAD_UNCHANGED), + ) + + +def test_fallback_image_io_preserves_sixteen_bit_unchanged(tmp_path: Path) -> None: + """Preserve sixteen-bit pixel values when reading unchanged images.""" + sixteen_bit = np.array([[0, 12345], [54321, 65535]], dtype=np.uint16) + sixteen_bit_path = tmp_path / "sixteen-bit.png" + + assert _imwrite(str(sixteen_bit_path), sixteen_bit) + np.testing.assert_array_equal( + _imread(str(sixteen_bit_path), _IMREAD_UNCHANGED), + cv2.imread(str(sixteen_bit_path), cv2.IMREAD_UNCHANGED), + ) + + +def test_fallback_image_io_matches_opencv_color_conversion_for_sixteen_bit( + tmp_path: Path, +) -> None: + """Match OpenCV color conversion when reading a sixteen-bit image.""" + sixteen_bit = np.array([[0, 12345], [54321, 65535]], dtype=np.uint16) + sixteen_bit_path = tmp_path / "sixteen-bit.png" + + assert _imwrite(str(sixteen_bit_path), sixteen_bit) + np.testing.assert_array_equal( + _imread(str(sixteen_bit_path), _IMREAD_COLOR), + cv2.imread(str(sixteen_bit_path), cv2.IMREAD_COLOR), + ) diff --git a/tests/cv2/test_transform.py b/tests/cv2/test_transform.py new file mode 100644 index 00000000..5ee9b927 --- /dev/null +++ b/tests/cv2/test_transform.py @@ -0,0 +1,74 @@ +"""Tests for private transform and filter fallbacks.""" + +from __future__ import annotations + +import importlib + +import numpy as np +import pytest + +from supervision._cv2._transform import ( + _blur, + _distance_transform, + _get_rotation_matrix_2d, + _warp_affine, +) +from supervision._cv2.constants import _DIST_L2 + +try: + cv2 = importlib.import_module("cv2") +except (ImportError, OSError): + pytest.skip( + "OpenCV is required as the reference implementation for this test module", + allow_module_level=True, + ) + + +def test_fallback_identity_affine_matches_opencv() -> None: + """Match OpenCV for an identity affine transform.""" + source = np.arange(25, dtype=np.uint8).reshape(5, 5) + matrix = _get_rotation_matrix_2d((2, 2), 0, 1) + + actual = _warp_affine(source, matrix, (5, 5)) + expected = cv2.warpAffine(source, matrix, (5, 5)) + np.testing.assert_array_equal(actual, expected) + + +def test_fallback_rotated_affine_matches_opencv() -> None: + """Match OpenCV for a rotated affine transform within its pixel budget.""" + source = np.arange(25, dtype=np.uint8).reshape(5, 5) + rotated_matrix = _get_rotation_matrix_2d((2, 2), 17, 1) + rotated = _warp_affine(source, rotated_matrix, (5, 5)) + expected_rotated = cv2.warpAffine(source, rotated_matrix, (5, 5)) + np.testing.assert_allclose(rotated, expected_rotated, atol=3, rtol=0) + + +def test_fallback_blur_preserves_shape_and_dtype() -> None: + """Preserve source shape and dtype during blurring.""" + source = np.arange(25, dtype=np.uint8).reshape(5, 5) + blurred = _blur(source, (3, 3)) + + assert blurred.shape == source.shape + assert blurred.dtype == source.dtype + + +def test_fallback_distance_transform_preserves_shape_and_dtype() -> None: + """Preserve source shape and expose float32 distance values.""" + source = np.ones((7, 7), dtype=np.uint8) + source[3, 3] = 0 + + actual = _distance_transform(source, _DIST_L2, 3) + + assert actual.shape == source.shape + assert actual.dtype == np.float32 + + +def test_fallback_distance_transform_preserves_distance_order() -> None: + """Preserve zero locations and monotonic distances for the L2 transform.""" + source = np.ones((7, 7), dtype=np.uint8) + source[3, 3] = 0 + + actual = _distance_transform(source, _DIST_L2, 3) + + assert actual[3, 3] == 0 + assert actual[3, 2] < actual[3, 1] < actual[3, 0]