feat(cv2): add image fallback backend (#2431)
- Organize facade constants and implementations into thematic private modules. - Add NumPy/Pillow/SciPy fallbacks with OpenCV parity coverage. - Mirror package modules in cv2 tests and verify blocked imports. - Split compound operation tests into isolated cases. - Parameterize color parity and fallback bindings for targeted failures. - Inline color conversion cases at their only use site. - Inline fallback binding cases while preserving reusable manifests. --------- Co-authored-by: Codex <codex@openai.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
parent
8ecd9a6680
commit
3d669f1ab4
|
|
@ -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__ = [
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
@ -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)
|
||||
|
|
@ -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
|
||||
|
|
@ -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),
|
||||
)
|
||||
|
|
@ -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
|
||||
|
|
@ -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))
|
||||
|
|
@ -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()
|
||||
|
|
@ -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})
|
||||
"""
|
||||
)
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
)
|
||||
|
|
@ -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]
|
||||
Loading…
Reference in New Issue