################################################################################
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
"""
Face processing operators.
Operators in this module detect, count, or blur faces in image inputs with
OpenCV Haar cascade classifiers.
Requires: ``pip install opencv-python-headless``. ``image_face_blur`` also
requires ``pip install Pillow``.
Runtime args:
``concurrency`` is forwarded to the DataFrame UDF runtime.
Usage::
>>> from pyflink.multimodal.operators.image_face import (
... image_face_detect, image_face_count, image_face_blur,
... )
>>> from pyflink.dataframe import col
>>> detect = image_face_detect(min_neighbors=3)
>>> df = df.with_column("faces", detect(col("img")))
>>> count = image_face_count(min_neighbors=3)
>>> df = df.with_column("face_count", count(col("img")))
>>> blur = image_face_blur(radius=15, min_neighbors=3)
>>> df = df.with_column("blurred", blur(col("img")))
"""
import math
from pyflink.dataframe import udf, DataType
from pyflink.table.udf import ScalarFunction
from pyflink.multimodal.codec import (
convert_image_array_to_mode,
decode_image_input,
image_array_to_pil_compatible,
pil_to_image,
)
from pyflink.multimodal.utils import (
_build_or_apply_udf,
_setup_cv2_threads,
_udf_runtime_kwargs,
)
__all__ = [
"image_face_detect",
"image_face_count",
"image_face_blur",
]
# "mean" intentionally excluded: it mapped to BoxBlur (identical to "box")
# but Data-Juicer maps "mean" to ImageFilter.BLUR which differs.
_VALID_BLUR_TYPES = {"gaussian", "box"}
_IMAGE_FILTER_MODES = {"L", "LA", "RGB", "RGBA"}
# Some PIL modes are representable but not accepted by ImageFilter. For face
# blur, prefer a filter-compatible 8-bit mode over preserving high precision.
_IMAGE_FILTER_MODE_BY_CHANNELS = {1: "L", 2: "LA", 3: "RGB", 4: "RGBA"}
def _validate_cv_classifier_name(cv_classifier):
if (
not isinstance(cv_classifier, str)
or not cv_classifier
or "/" in cv_classifier
or "\\" in cv_classifier
or cv_classifier in (".", "..")
):
raise ValueError(
"cv_classifier must be an OpenCV Haar cascade XML filename, "
f"got {cv_classifier!r}"
)
return cv_classifier
def _validate_face_size(name, size):
if size is None:
return None
if not isinstance(size, (list, tuple)) or len(size) != 2:
raise ValueError(f"{name} must be a (width, height) pair, got {size!r}")
width, height = size
if (
not isinstance(width, int)
or isinstance(width, bool)
or not isinstance(height, int)
or isinstance(height, bool)
or width <= 0
or height <= 0
):
raise ValueError(
f"{name} must contain positive integer pixels, got {size!r}"
)
return (width, height)
def _validate_face_detection_params(min_neighbors, scale_factor, min_size, max_size):
if not isinstance(min_neighbors, int) or isinstance(min_neighbors, bool):
raise ValueError(
"min_neighbors must be a non-negative integer, "
f"got {min_neighbors!r}"
)
if min_neighbors < 0:
raise ValueError(
"min_neighbors must be a non-negative integer, "
f"got {min_neighbors!r}"
)
if (
not isinstance(scale_factor, (int, float))
or isinstance(scale_factor, bool)
or not math.isfinite(scale_factor)
or scale_factor <= 1.0
):
raise ValueError(f"scale_factor must be > 1.0, got {scale_factor}")
min_size = _validate_face_size("min_size", min_size)
max_size = _validate_face_size("max_size", max_size)
if (
min_size is not None
and max_size is not None
and (max_size[0] < min_size[0] or max_size[1] < min_size[1])
):
raise ValueError(
"max_size must be greater than or equal to min_size, "
f"got min_size={min_size!r}, max_size={max_size!r}"
)
return min_size, max_size
def _image_array_to_filterable_pil(pixel_array):
"""Convert pixels to a PIL image that supports Pillow ImageFilter."""
pil_image = image_array_to_pil_compatible(pixel_array)
if pil_image.mode in _IMAGE_FILTER_MODES:
return pil_image
channels = 1 if pixel_array.ndim == 2 else pixel_array.shape[2]
target_mode = _IMAGE_FILTER_MODE_BY_CHANNELS[channels]
return image_array_to_pil_compatible(
convert_image_array_to_mode(pixel_array, target_mode)
)
def _detect_faces_cv(
cv2, np, cascade, rgb_array, scale_factor, min_neighbors,
min_size=None, max_size=None,
):
"""Run face detection on a decoded image ndarray.
Args:
cv2: OpenCV module reference.
np: NumPy module reference.
cascade: Loaded Haar cascade classifier.
rgb_array: Decoded image as ``(H, W)``, ``(H, W, 1)``,
``(H, W, 3)``, or ``(H, W, 4)`` ndarray.
scale_factor: Image pyramid scale factor.
min_neighbors: minNeighbors parameter for detectMultiScale.
min_size: Minimum face size ``(width, height)`` or ``None``.
max_size: Maximum face size ``(width, height)`` or ``None``.
Returns:
faces: ndarray of shape ``(N, 4)`` with detected face rectangles.
Always an ndarray -- empty ``(0, 4)`` when no faces are found.
"""
if rgb_array.ndim not in (2, 3):
raise ValueError(
"Face detection expects a 2D or 3D image array, "
f"got shape {getattr(rgb_array, 'shape', None)!r}"
)
if rgb_array.ndim == 2:
gray = rgb_array
else:
channels = rgb_array.shape[2]
if channels == 1:
gray = rgb_array[:, :, 0]
elif channels == 3:
gray = cv2.cvtColor(rgb_array, cv2.COLOR_RGB2GRAY)
elif channels == 4:
gray = cv2.cvtColor(rgb_array, cv2.COLOR_RGBA2GRAY)
else:
raise ValueError(
"Face detection expects 1, 3, or 4 channels for 3D arrays, "
f"got shape {rgb_array.shape!r}"
)
kwargs = dict(scaleFactor=scale_factor, minNeighbors=min_neighbors)
if min_size is not None:
# OpenCV expects (width, height) tuple for minSize.
kwargs["minSize"] = tuple(min_size)
if max_size is not None:
# OpenCV expects (width, height) tuple for maxSize.
kwargs["maxSize"] = tuple(max_size)
faces = cascade.detectMultiScale(gray, **kwargs)
# detectMultiScale returns () when no faces found; normalize to ndarray
# so callers can always use len() / iterate without type-checking.
if not isinstance(faces, np.ndarray):
faces = np.empty((0, 4), dtype=int)
return faces
class _ImageFaceBase(ScalarFunction):
"""Shared OpenCV Haar cascade setup for scalar face operators."""
def __init__(
self,
cv_classifier="haarcascade_frontalface_alt.xml",
min_neighbors=3,
scale_factor=1.1,
min_size=None,
max_size=None,
):
super().__init__()
self.cv_classifier = _validate_cv_classifier_name(cv_classifier)
min_size, max_size = _validate_face_detection_params(
min_neighbors, scale_factor, min_size, max_size
)
# OpenCV minNeighbors: higher values reduce false positives but may
# miss weak faces. Default 3 matches OpenCV's common Haar examples.
self.min_neighbors = min_neighbors
self.scale_factor = scale_factor
self.min_size = min_size
self.max_size = max_size
def open(self, function_context):
from pyflink.model.cache_manager import check_dependencies
check_dependencies("cv2", "numpy")
import cv2
import numpy as np
_setup_cv2_threads(cv2)
self._cv2 = cv2
self._np = np
path = cv2.data.haarcascades + self.cv_classifier
self._cascade = cv2.CascadeClassifier(path)
if self._cascade.empty():
raise ValueError(
f"Failed to load Haar cascade classifier "
f"'{self.cv_classifier}' from '{path}'"
)
def close(self):
self._cv2 = None
self._np = None
self._cascade = None
def _detect_faces(self, pixel_array):
return _detect_faces_cv(
self._cv2, self._np, self._cascade, pixel_array,
self.scale_factor, self.min_neighbors,
min_size=self.min_size, max_size=self.max_size,
)
# ImageFaceDetect - detect faces and return bounding boxes
class _ImageFaceDetect(_ImageFaceBase):
"""Detect faces in an image and return bounding boxes."""
def eval(self, image_input):
if image_input is None:
return None
# Haar Cascade detection needs an OpenCV-compatible image. Normalize
# decoded image modes such as LA/RGBA to RGB before detection so
# two-channel alpha images do not reach cv2.cvtColor directly.
pixel_array = decode_image_input(image_input, mode="RGB")
faces = self._detect_faces(pixel_array)
result = []
for (x, y, w, h) in faces:
# Haar Cascade does not expose per-detection probability. Preserve
# the output schema but mark confidence unknown instead of
# reporting a synthetic perfect score.
result.append((int(x), int(y), int(w), int(h), None))
return result
[docs]def image_face_detect(
*columns,
cv_classifier="haarcascade_frontalface_alt.xml",
min_neighbors=3,
scale_factor=1.1,
min_size=None,
max_size=None,
concurrency=None,
):
"""
Create a face detection UDF (OpenCV Haar Cascade-based).
Requires ``pip install opencv-python-headless``. This is a scalar UDF.
Args:
*columns: Optional image column(s). When provided, the UDF is
applied directly instead of returning a factory.
cv_classifier: OpenCV Haar cascade XML filename. Must be a bare
filename without path separators. Default
``"haarcascade_frontalface_alt.xml"``.
min_neighbors: OpenCV ``detectMultiScale`` ``minNeighbors`` value.
Higher values reduce false positives but may miss faces.
Default ``3``.
scale_factor: Image pyramid scale factor for multi-scale detection. Default ``1.1``.
min_size: Minimum face size as ``(width, height)`` tuple in pixels.
Detections smaller than this are ignored. ``None`` (default) uses
OpenCV's built-in minimum.
max_size: Maximum face size as ``(width, height)`` tuple in pixels.
Detections larger than this are ignored. ``None`` (default)
imposes no upper limit.
concurrency: UDF concurrency. ``None`` uses the framework default.
Returns:
A UDF that returns a list of face bounding boxes as
``(x, y, w, h, confidence)`` structs, or ``None`` for null input.
The ``confidence`` field is always ``None`` for Haar Cascade
detection because OpenCV does not expose per-detection probability.
Raises:
ValueError: If detection parameters are invalid.
Usage::
>>> # As a reusable variable
>>> detect = image_face_detect(min_neighbors=3)
>>> df = df.with_column("faces", detect(col("img")))
>>>
>>> # Inline
>>> df = df.with_column("faces", image_face_detect(col("img")))
"""
wrapper = udf(
_ImageFaceDetect(
cv_classifier=cv_classifier,
min_neighbors=min_neighbors,
scale_factor=scale_factor,
min_size=min_size,
max_size=max_size,
),
return_dtype=DataType.list(
DataType.struct(
{
"x": DataType.int32(),
"y": DataType.int32(),
"w": DataType.int32(),
"h": DataType.int32(),
"confidence": DataType.float64(),
}
)
),
**_udf_runtime_kwargs(concurrency=concurrency),
)
return _build_or_apply_udf(wrapper, *columns)
# ImageFaceCount - count number of faces
class _ImageFaceCount(_ImageFaceBase):
"""Count the number of faces in an image."""
def eval(self, image_input):
# None input -> None output (unknown), distinct from 0 (no faces found).
if image_input is None:
return None
# Haar Cascade detection needs an OpenCV-compatible image. Normalize
# decoded image modes such as LA/RGBA to RGB before detection so
# two-channel alpha images do not reach cv2.cvtColor directly.
pixel_array = decode_image_input(image_input, mode="RGB")
faces = self._detect_faces(pixel_array)
# _detect_faces_cv always returns ndarray; len() gives count directly.
return len(faces)
[docs]def image_face_count(
*columns,
cv_classifier="haarcascade_frontalface_alt.xml",
min_neighbors=3,
scale_factor=1.1,
min_size=None,
max_size=None,
concurrency=None,
):
"""
Create a face count UDF (OpenCV Haar Cascade-based).
Requires ``pip install opencv-python-headless``. This is a scalar UDF that
returns the number of detected faces as an integer.
Args:
*columns: Optional image column(s). When provided, the UDF is
applied directly instead of returning a factory.
cv_classifier: OpenCV Haar cascade XML filename. Must be a bare
filename without path separators. Default
``"haarcascade_frontalface_alt.xml"``.
min_neighbors: OpenCV ``detectMultiScale`` ``minNeighbors`` value.
Higher values reduce false positives but may miss faces.
Default ``3``.
scale_factor: Image pyramid scale factor for multi-scale detection. Default ``1.1``.
min_size: Minimum face size as ``(width, height)`` tuple in pixels.
Detections smaller than this are ignored. ``None`` (default) uses
OpenCV's built-in minimum.
max_size: Maximum face size as ``(width, height)`` tuple in pixels.
Detections larger than this are ignored. ``None`` (default)
imposes no upper limit.
concurrency: UDF concurrency. ``None`` uses the framework default.
Returns:
A UDF that returns the number of detected faces as an integer, or
``None`` for null input. ``None`` means the input itself is null;
``0`` means a valid image was processed and no faces were detected.
Raises:
ValueError: If detection parameters are invalid.
Usage::
>>> # As a reusable variable
>>> count = image_face_count(min_neighbors=3)
>>> df = df.with_column("face_count", count(col("img")))
>>>
>>> # Inline
>>> df = df.with_column("face_count", image_face_count(col("img")))
"""
wrapper = udf(
_ImageFaceCount(
cv_classifier=cv_classifier,
min_neighbors=min_neighbors,
scale_factor=scale_factor,
min_size=min_size,
max_size=max_size,
),
return_dtype=DataType.int32(),
**_udf_runtime_kwargs(concurrency=concurrency),
)
return _build_or_apply_udf(wrapper, *columns)
# ImageFaceBlur - detect and blur faces
class _ImageFaceBlur(_ImageFaceBase):
"""Detect and blur faces in an image."""
def __init__(
self,
cv_classifier="haarcascade_frontalface_alt.xml",
blur_type="gaussian",
radius=2,
min_neighbors=3,
scale_factor=1.1,
min_size=None,
max_size=None,
):
super().__init__(
cv_classifier=cv_classifier,
min_neighbors=min_neighbors,
scale_factor=scale_factor,
min_size=min_size,
max_size=max_size,
)
if (
not isinstance(radius, (int, float))
or isinstance(radius, bool)
or not math.isfinite(radius)
or radius < 0
):
raise ValueError(f"radius must be a finite number >= 0, got {radius!r}")
if blur_type not in _VALID_BLUR_TYPES:
raise ValueError(
f"blur_type must be one of {sorted(_VALID_BLUR_TYPES)}, "
f"got {blur_type!r}"
)
self.blur_type = blur_type
self.radius = radius
self._blur_filter = None
def _make_blur_filter(self):
if self.blur_type == "box":
return self._ImageFilter.BoxBlur(self.radius)
return self._ImageFilter.GaussianBlur(self.radius)
def open(self, function_context):
super().open(function_context)
from pyflink.model.cache_manager import check_dependencies
check_dependencies("PIL")
from PIL import ImageFilter
self._ImageFilter = ImageFilter
self._blur_filter = self._make_blur_filter()
def eval(self, image_input):
if image_input is None:
return None
pixel_array = decode_image_input(image_input)
# Use a normalized RGB view/copy for detection only. The blur operation
# below still edits the original decoded image, preserving LA/RGBA
# output modes and alpha channels instead of forcing RGB output.
detection_array = convert_image_array_to_mode(pixel_array, "RGB")
faces = self._detect_faces(detection_array)
pil_image = _image_array_to_filterable_pil(pixel_array)
# _detect_faces_cv always returns ndarray; check length directly.
if len(faces) > 0:
blur_filter = self._blur_filter
for (x, y, w, h) in faces:
region = pil_image.crop((int(x), int(y), int(x + w), int(y + h)))
region = region.filter(blur_filter)
pil_image.paste(region, (int(x), int(y)))
return pil_to_image(pil_image)
def close(self):
super().close()
self._ImageFilter = None
self._blur_filter = None
[docs]def image_face_blur(
*columns,
cv_classifier="haarcascade_frontalface_alt.xml",
blur_type="gaussian",
radius=2,
min_neighbors=3,
scale_factor=1.1,
min_size=None,
max_size=None,
concurrency=None,
):
"""
Create a face blurring UDF (OpenCV Haar Cascade + PIL-based).
Requires ``pip install opencv-python Pillow``. This is a scalar UDF that
returns a decoded image.
Args:
*columns: Optional image column(s). When provided, the UDF is
applied directly instead of returning a factory.
cv_classifier: OpenCV Haar cascade XML filename. Must be a bare
filename without path separators. Default
``"haarcascade_frontalface_alt.xml"``.
blur_type: Blur filter type. One of ``"gaussian"``, ``"box"``. Default ``"gaussian"``.
radius: Blur kernel radius in pixels. Default ``2`` is a light visual
blur; use a larger value for privacy/anonymization.
min_neighbors: OpenCV ``detectMultiScale`` ``minNeighbors`` value.
Higher values reduce false positives but may miss faces.
Default ``3``.
scale_factor: Image pyramid scale factor for multi-scale detection. Default ``1.1``.
min_size: Minimum face size as ``(width, height)`` tuple in pixels.
Detections smaller than this are ignored. ``None`` (default) uses
OpenCV's built-in minimum.
max_size: Maximum face size as ``(width, height)`` tuple in pixels.
Detections larger than this are ignored. ``None`` (default)
imposes no upper limit.
concurrency: UDF concurrency. ``None`` uses the framework default.
Returns:
A UDF that returns a decoded image with faces blurred,
or ``None`` for null input. Unsupported decoded image outputs
fail fast at the codec boundary.
Raises:
ValueError: If blur or detection parameters are invalid.
Usage::
>>> # As a reusable variable
>>> blur = image_face_blur(blur_type="gaussian", radius=3)
>>> df = df.with_column("blurred", blur(col("img")))
>>>
>>> # Inline
>>> df = df.with_column("blurred", image_face_blur(col("img")))
"""
wrapper = udf(
_ImageFaceBlur(
cv_classifier=cv_classifier,
blur_type=blur_type,
radius=radius,
min_neighbors=min_neighbors,
scale_factor=scale_factor,
min_size=min_size,
max_size=max_size,
),
return_dtype=DataType.image(),
**_udf_runtime_kwargs(concurrency=concurrency),
)
return _build_or_apply_udf(wrapper, *columns)