################################################################################
# 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.
################################################################################
"""Multimodal expression namespaces."""
from importlib import import_module
from typing import (
Dict,
Literal,
List,
Mapping,
Optional,
Protocol,
Sequence,
Tuple,
Union,
)
from pyflink.dataframe.udf import DataFrameUDTFCall
from pyflink.table.expression import Expression
from pyflink.multimodal.codec.audio import DEFAULT_MAX_DECODED_AUDIO_BYTES
_DEFAULT_AESTHETIC_SCORER_MODEL = (
"shunk031/aesthetics-predictor-v2-sac-logos-ava1-l14-linearMSE"
)
_DEFAULT_IMAGE_ENCODE_QUALITY = 85
_DEFAULT_NSFW_MODEL = "Falconsai/nsfw_image_detection"
_DEFAULT_WATERMARK_MODEL = "amrul-hzz/watermark_detector"
_IMAGE_DETECT = "pyflink.multimodal.operators.image_detect"
_IMAGE_EMBED = "pyflink.multimodal.operators.image_embed"
_IMAGE_FACE = "pyflink.multimodal.operators.image_face"
_IMAGE_INFO = "pyflink.multimodal.operators.image_info"
_IMAGE_QUALITY = "pyflink.multimodal.operators.image_quality"
_IMAGE_TRANSFORM = "pyflink.multimodal.operators.image_transform"
_AUDIO_INFO = "pyflink.multimodal.operators.audio_info"
_AUDIO_SPEECH = "pyflink.multimodal.operators.audio_speech"
_AUDIO_TRANSFORM = "pyflink.multimodal.operators.audio_transform"
_VIDEO_FRAMES = "pyflink.multimodal.operators.video_frames"
_DEFAULT_AUDIO_MODEL = "openai/whisper-tiny"
_Color = Union[List[int], Tuple[int, int, int, int]]
_CropBox = Sequence[int]
_FloatPair = Sequence[float]
_Languages = Union[List[str], Tuple[str, ...]]
_Size = Union[List[int], Tuple[int, int]]
class _TimestampRow(Protocol):
start_ms: int
end_ms: int
_TimestampRange = Union[Dict[str, int], Sequence[int], _TimestampRow]
_TimestampRanges = Union[List[_TimestampRange], Tuple[_TimestampRange, ...]]
class _ExpressionAccessor:
"""Base class for expression-bound scalar operator namespaces."""
def __init__(self, expression):
self._expression = expression
def _call(self, module_name, function_name, *columns, **kwargs):
function = getattr(import_module(module_name), function_name)
return function(self._expression, *columns, **kwargs)
[docs]class ImageExpressionAccessor(_ExpressionAccessor):
"""
Image operations available on a DataFrame expression.
These methods use the expression on the left side as the image input. For
example, ``col("img").image.resize(width=224, height=224)`` is the
expression-accessor form of ``image_resize(col("img"), width=224,
height=224)``.
Examples::
>>> from pyflink.dataframe import col
>>> df = df.with_column(
... "small", col("img").image.resize(width=224, height=224))
>>> df = df.with_column(
... "jpeg", col("small").image.encode(format="JPEG"))
"""
[docs] def decode(
self,
*,
on_error: Literal["raise", "null"] = "raise",
mode: Optional[str] = None,
pixel_limit: Optional[int] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Decode image bytes into image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_decode`.
See that function for full parameter details.
Args:
on_error: ``"raise"`` to propagate decode failures, or ``"null"``
to return null for unreadable inputs.
mode: Optional output image mode, such as ``"RGB"``.
pixel_limit: Optional maximum decoded pixel count.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces decoded ``IMAGE`` values.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_decode",
on_error=on_error,
mode=mode,
pixel_limit=pixel_limit,
concurrency=concurrency,
)
[docs] def encode(
self,
*,
format: Optional[str] = None,
quality: int = _DEFAULT_IMAGE_ENCODE_QUALITY,
output: Literal["bytes", "data_url"] = "bytes",
concurrency: Optional[int] = None,
) -> Expression:
"""
Encode image values into bytes or data URLs.
Equivalent to :func:`~pyflink.multimodal.operators.image_encode`.
See that function for full parameter details.
Args:
format: Optional output image format, such as ``"JPEG"`` or
``"PNG"``.
quality: Output quality for lossy formats.
output: Output representation, such as ``"bytes"`` or
``"data_url"``.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``BINARY`` values, or ``STRING``
values when ``output="data_url"``.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_encode",
format=format,
quality=quality,
output=output,
concurrency=concurrency,
)
[docs] def compress(
self,
*,
quality: int = _DEFAULT_IMAGE_ENCODE_QUALITY,
format: str = "JPEG",
concurrency: Optional[int] = None,
) -> Expression:
"""
Compress image values using the requested format and quality.
Equivalent to :func:`~pyflink.multimodal.operators.image_compress`.
See that function for full parameter details.
Args:
quality: Output quality for lossy compression.
format: Output image format used for compression.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces compressed ``BINARY`` image bytes.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_compress",
quality=quality,
format=format,
concurrency=concurrency,
)
[docs] def to_tensor(
self,
*,
width: int,
height: int,
mode: str = "RGB",
layout: Literal["CHW", "HWC"] = "CHW",
concurrency: Optional[int] = None,
) -> Expression:
"""
Convert already prepared image values to fixed-shape tensors.
Equivalent to :func:`~pyflink.multimodal.operators.image_to_tensor`.
See that function for full parameter details.
This operation does not resize or convert image mode. Use
``.image.resize(...)`` or ``.image.convert_mode(...)`` upstream when
needed.
Args:
width: Expected input image width in pixels.
height: Expected input image height in pixels.
mode: Expected input image mode before tensor conversion.
layout: Tensor layout, such as ``"CHW"`` or ``"HWC"``.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces a ``FLOAT32`` tensor with
the requested shape and layout.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_to_tensor",
width=width,
height=height,
mode=mode,
layout=layout,
concurrency=concurrency,
)
[docs] def convert_mode(
self,
*,
mode: str,
concurrency: Optional[int] = None,
) -> Expression:
"""
Convert image values to another image mode.
Equivalent to :func:`~pyflink.multimodal.operators.image_convert_mode`.
See that function for full parameter details.
Args:
mode: Target image mode, such as ``"RGB"`` or ``"L"``.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``IMAGE`` values in the
requested mode.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_convert_mode",
mode=mode,
concurrency=concurrency,
)
[docs] def resize(
self,
*,
width: int,
height: int,
method: str = "lanczos",
concurrency: Optional[int] = None,
) -> Expression:
"""
Resize image values to a fixed width and height.
Equivalent to :func:`~pyflink.multimodal.operators.image_resize`.
See that function for full parameter details.
Args:
width: Target width in pixels.
height: Target height in pixels.
method: Resize interpolation method.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces resized ``IMAGE`` values.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_resize",
width=width,
height=height,
method=method,
concurrency=concurrency,
)
[docs] def rescale(
self,
*,
scale: float,
method: str = "lanczos",
concurrency: Optional[int] = None,
) -> Expression:
"""
Scale image values by a ratio.
Equivalent to :func:`~pyflink.multimodal.operators.image_rescale`.
See that function for full parameter details.
Args:
scale: Scale factor applied to both width and height.
method: Resize interpolation method.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces rescaled ``IMAGE`` values.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_rescale",
scale=scale,
method=method,
concurrency=concurrency,
)
[docs] def crop(
self,
*,
crop_coords: Optional[_CropBox] = None,
crop_type: Literal["center", "ratio", "coordinate"] = "center",
crop_ratio: _FloatPair = (0.8, 0.8),
crop_size: Optional[_Size] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Crop image values by coordinates, crop type, ratio, or size.
Equivalent to :func:`~pyflink.multimodal.operators.image_crop`.
See that function for full parameter details.
Args:
crop_coords: Explicit crop box coordinates.
crop_type: Preset crop mode, such as center crop.
crop_ratio: Crop ratio used by ratio-based modes.
crop_size: Target crop size used by size-based modes.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces cropped ``IMAGE`` values.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_crop",
crop_coords=crop_coords,
crop_type=crop_type,
crop_ratio=crop_ratio,
crop_size=crop_size,
concurrency=concurrency,
)
[docs] def crop_black_border(
self,
*,
threshold: Optional[float] = None,
detect_algorithm: Literal["auto", "threshold", "histogram", "edge"] = "auto",
black_threshold: Optional[int] = None,
edge_sensitivity: float = 1.0,
min_border_size: int = 1,
concurrency: Optional[int] = None,
) -> Expression:
"""
Remove black borders from image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_crop_black_border`.
See that function for full parameter details.
Args:
threshold: Optional border detection threshold.
detect_algorithm: Border detection algorithm.
black_threshold: Pixel threshold used to classify black borders.
edge_sensitivity: Sensitivity for detecting border edges.
min_border_size: Minimum border size in pixels.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``IMAGE`` values with black
borders removed.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_crop_black_border",
threshold=threshold,
detect_algorithm=detect_algorithm,
black_threshold=black_threshold,
edge_sensitivity=edge_sensitivity,
min_border_size=min_border_size,
concurrency=concurrency,
)
[docs] def flip(
self,
*,
mode: Literal["horizontal", "vertical", "rotate180"] = "horizontal",
concurrency: Optional[int] = None,
) -> Expression:
"""
Flip image values horizontally or vertically.
Equivalent to :func:`~pyflink.multimodal.operators.image_flip`.
See that function for full parameter details.
Args:
mode: Flip mode, such as ``"horizontal"`` or ``"vertical"``.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces flipped ``IMAGE`` values.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_flip",
mode=mode,
concurrency=concurrency,
)
[docs] def blur(
self,
*,
radius: float = 2,
blur_type: Literal["gaussian", "box", "mean"] = "gaussian",
concurrency: Optional[int] = None,
) -> Expression:
"""
Blur image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_blur`.
See that function for full parameter details.
Args:
radius: Blur radius.
blur_type: Blur algorithm, such as ``"gaussian"``.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces blurred ``IMAGE`` values.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_blur",
radius=radius,
blur_type=blur_type,
concurrency=concurrency,
)
[docs] def adjust_color(
self,
*,
brightness: float = 1.0,
contrast: float = 1.0,
saturation: float = 1.0,
concurrency: Optional[int] = None,
) -> Expression:
"""
Adjust image brightness, contrast, and saturation.
Equivalent to :func:`~pyflink.multimodal.operators.image_adjust_color`.
See that function for full parameter details.
Args:
brightness: Brightness multiplier.
contrast: Contrast multiplier.
saturation: Saturation multiplier.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces color-adjusted ``IMAGE``
values.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_adjust_color",
brightness=brightness,
contrast=contrast,
saturation=saturation,
concurrency=concurrency,
)
[docs] def remove_background(
self,
*,
alpha_matting: bool = False,
alpha_matting_foreground_threshold: int = 240,
alpha_matting_background_threshold: int = 10,
alpha_matting_erode_size: int = 10,
bgcolor: Optional[_Color] = None,
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Remove image backgrounds and optionally composite a new background.
Equivalent to :func:`~pyflink.multimodal.operators.image_remove_background`.
See that function for full parameter details.
Args:
alpha_matting: Whether to refine the foreground mask with alpha
matting.
alpha_matting_foreground_threshold: Foreground threshold for alpha
matting.
alpha_matting_background_threshold: Background threshold for alpha
matting.
alpha_matting_erode_size: Erosion size for alpha matting.
bgcolor: Optional replacement background color.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: GPU type requested for model inference. The DataFrame UDF
runtime requires this when ``num_gpus`` is set.
Returns:
A DataFrame expression that produces foreground ``IMAGE`` values,
optionally composited with the requested background.
"""
return self._call(
_IMAGE_TRANSFORM,
"image_remove_background",
alpha_matting=alpha_matting,
alpha_matting_foreground_threshold=alpha_matting_foreground_threshold,
alpha_matting_background_threshold=alpha_matting_background_threshold,
alpha_matting_erode_size=alpha_matting_erode_size,
bgcolor=bgcolor,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs] def aspect_ratio(
self,
*,
concurrency: Optional[int] = None,
) -> Expression:
"""
Compute image aspect ratio.
Equivalent to :func:`~pyflink.multimodal.operators.image_aspect_ratio`.
See that function for full parameter details.
Args:
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces a ``DOUBLE`` width-to-height
ratio.
"""
return self._call(_IMAGE_INFO, "image_aspect_ratio", concurrency=concurrency)
[docs] def sharpness(
self,
*,
max_edge: Optional[int] = None,
allow_upscale: bool = False,
concurrency: Optional[int] = None,
) -> Expression:
"""
Estimate image sharpness.
Equivalent to :func:`~pyflink.multimodal.operators.image_sharpness`.
See that function for full parameter details.
Args:
max_edge: Optional maximum edge size used before scoring.
allow_upscale: Whether resizing may upscale smaller images.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces a ``DOUBLE`` sharpness
score.
"""
return self._call(
_IMAGE_INFO,
"image_sharpness",
max_edge=max_edge,
allow_upscale=allow_upscale,
concurrency=concurrency,
)
[docs] def hash(
self,
*,
method: str = "phash",
concurrency: Optional[int] = None,
) -> Expression:
"""
Compute an image perceptual hash.
Equivalent to :func:`~pyflink.multimodal.operators.image_hash`.
See that function for full parameter details.
Args:
method: Hash algorithm name.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces a ``STRING`` perceptual hash.
"""
return self._call(
_IMAGE_INFO,
"image_hash",
method=method,
concurrency=concurrency,
)
[docs] def is_valid(
self,
*,
mode: Optional[str] = None,
pixel_limit: Optional[int] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Check whether image inputs can be decoded as valid images.
Equivalent to :func:`~pyflink.multimodal.operators.is_valid_image`.
See that function for full parameter details.
Args:
mode: Optional expected image mode.
pixel_limit: Optional maximum decoded pixel count.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``BOOLEAN`` validity flags.
"""
return self._call(
_IMAGE_INFO,
"is_valid_image",
mode=mode,
pixel_limit=pixel_limit,
concurrency=concurrency,
)
[docs] def size_filter(
self,
*,
min_w: Optional[int] = None,
min_h: Optional[int] = None,
max_w: Optional[int] = None,
max_h: Optional[int] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Filter images by width and height bounds.
Equivalent to :func:`~pyflink.multimodal.operators.image_size_filter`.
See that function for full parameter details.
Args:
min_w: Optional minimum width in pixels.
min_h: Optional minimum height in pixels.
max_w: Optional maximum width in pixels.
max_h: Optional maximum height in pixels.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``BOOLEAN`` filter results.
"""
return self._call(
_IMAGE_INFO,
"image_size_filter",
min_w=min_w,
min_h=min_h,
max_w=max_w,
max_h=max_h,
concurrency=concurrency,
)
[docs] def shape_filter(
self,
*,
min_ratio: Optional[float] = None,
max_ratio: Optional[float] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Filter images by aspect ratio.
Equivalent to :func:`~pyflink.multimodal.operators.image_shape_filter`.
See that function for full parameter details.
Args:
min_ratio: Optional minimum width-to-height ratio.
max_ratio: Optional maximum width-to-height ratio.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``BOOLEAN`` filter results.
"""
return self._call(
_IMAGE_INFO,
"image_shape_filter",
min_ratio=min_ratio,
max_ratio=max_ratio,
concurrency=concurrency,
)
[docs] def file_size_filter(
self,
*,
min_bytes: Optional[int] = None,
max_bytes: Optional[int] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Filter image files by byte size.
Equivalent to :func:`~pyflink.multimodal.operators.image_file_size_filter`.
See that function for full parameter details.
Args:
min_bytes: Optional minimum encoded file size in bytes.
max_bytes: Optional maximum encoded file size in bytes.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``BOOLEAN`` filter results.
"""
return self._call(
_IMAGE_INFO,
"image_file_size_filter",
min_bytes=min_bytes,
max_bytes=max_bytes,
concurrency=concurrency,
)
[docs] def nsfw_score(
self,
*,
hf_nsfw_model: str = _DEFAULT_NSFW_MODEL,
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Score images for NSFW content.
Equivalent to :func:`~pyflink.multimodal.operators.image_nsfw_score`.
See that function for full parameter details.
Args:
hf_nsfw_model: Hugging Face model used for NSFW scoring.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: Optional GPU type requested for model inference.
Returns:
A DataFrame expression that produces a ``DOUBLE`` NSFW score.
"""
return self._call(
_IMAGE_QUALITY,
"image_nsfw_score",
hf_nsfw_model=hf_nsfw_model,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs] def aesthetic_score(
self,
*,
hf_scorer_model: str = _DEFAULT_AESTHETIC_SCORER_MODEL,
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Score images for aesthetic quality.
Equivalent to :func:`~pyflink.multimodal.operators.image_aesthetic_score`.
See that function for full parameter details.
Args:
hf_scorer_model: Hugging Face model used for aesthetic scoring.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: Optional GPU type requested for model inference.
Returns:
A DataFrame expression that produces a ``DOUBLE`` aesthetic score.
"""
return self._call(
_IMAGE_QUALITY,
"image_aesthetic_score",
hf_scorer_model=hf_scorer_model,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs] def quality_score(
self,
*,
concurrency: Optional[int] = None,
) -> Expression:
"""
Score general image quality.
Equivalent to :func:`~pyflink.multimodal.operators.image_quality_score`.
See that function for full parameter details.
Args:
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces a ``DOUBLE`` quality score.
"""
return self._call(
_IMAGE_QUALITY,
"image_quality_score",
concurrency=concurrency,
)
[docs] def watermark_score(
self,
*,
hf_watermark_model: str = _DEFAULT_WATERMARK_MODEL,
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Score images for watermark likelihood.
Equivalent to :func:`~pyflink.multimodal.operators.image_watermark_score`.
See that function for full parameter details.
Args:
hf_watermark_model: Hugging Face model used for watermark scoring.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: Optional GPU type requested for model inference.
Returns:
A DataFrame expression that produces a ``DOUBLE`` watermark
likelihood score.
"""
return self._call(
_IMAGE_QUALITY,
"image_watermark_score",
hf_watermark_model=hf_watermark_model,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs] def face_detect(
self,
*,
cv_classifier: str = "haarcascade_frontalface_alt.xml",
min_neighbors: int = 3,
scale_factor: float = 1.1,
min_size: Optional[_Size] = None,
max_size: Optional[_Size] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Detect faces in image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_face_detect`.
See that function for full parameter details.
Args:
cv_classifier: OpenCV cascade classifier file name.
min_neighbors: Minimum neighbor count for face detection.
scale_factor: Scale factor used by the detector pyramid.
min_size: Optional minimum face size.
max_size: Optional maximum face size.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``ARRAY<ROW>`` face
detections with ``x``, ``y``, ``w``, ``h``, and ``confidence``
fields.
"""
return self._call(
_IMAGE_FACE,
"image_face_detect",
cv_classifier=cv_classifier,
min_neighbors=min_neighbors,
scale_factor=scale_factor,
min_size=min_size,
max_size=max_size,
concurrency=concurrency,
)
[docs] def face_count(
self,
*,
cv_classifier: str = "haarcascade_frontalface_alt.xml",
min_neighbors: int = 3,
scale_factor: float = 1.1,
min_size: Optional[_Size] = None,
max_size: Optional[_Size] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Count faces in image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_face_count`.
See that function for full parameter details.
Args:
cv_classifier: OpenCV cascade classifier file name.
min_neighbors: Minimum neighbor count for face detection.
scale_factor: Scale factor used by the detector pyramid.
min_size: Optional minimum face size.
max_size: Optional maximum face size.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces an ``INT`` face count.
"""
return self._call(
_IMAGE_FACE,
"image_face_count",
cv_classifier=cv_classifier,
min_neighbors=min_neighbors,
scale_factor=scale_factor,
min_size=min_size,
max_size=max_size,
concurrency=concurrency,
)
[docs] def face_blur(
self,
*,
cv_classifier: str = "haarcascade_frontalface_alt.xml",
blur_type: Literal["gaussian", "box"] = "gaussian",
radius: float = 2,
min_neighbors: int = 3,
scale_factor: float = 1.1,
min_size: Optional[_Size] = None,
max_size: Optional[_Size] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Blur detected faces in image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_face_blur`.
See that function for full parameter details.
Args:
cv_classifier: OpenCV cascade classifier file name.
blur_type: Blur algorithm used on detected face regions.
radius: Blur radius.
min_neighbors: Minimum neighbor count for face detection.
scale_factor: Scale factor used by the detector pyramid.
min_size: Optional minimum face size.
max_size: Optional maximum face size.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``IMAGE`` values with
detected face regions blurred.
"""
return self._call(
_IMAGE_FACE,
"image_face_blur",
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,
concurrency=concurrency,
)
[docs] def detect_objects(
self,
*,
model: str = "yolov8n",
confidence: float = 0.05,
imgsz: int = 640,
iou: float = 0.5,
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Detect objects in image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_detect_objects`.
See that function for full parameter details.
Args:
model: Object detection model name or path.
confidence: Minimum detection confidence.
imgsz: Inference image size.
iou: IoU threshold used for non-maximum suppression.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: Optional GPU type requested for model inference.
Returns:
A DataFrame expression that produces ``ARRAY<ROW>`` object
detections with ``label``, ``x``, ``y``, ``w``, ``h``, and
``confidence`` fields.
"""
return self._call(
_IMAGE_DETECT,
"image_detect_objects",
model=model,
confidence=confidence,
imgsz=imgsz,
iou=iou,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs] def segment(
self,
*,
model: str = "FastSAM-x.pt",
confidence: float = 0.05,
imgsz: int = 1024,
iou: float = 0.5,
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Segment objects or regions in image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_segment`.
See that function for full parameter details.
Args:
model: Segmentation model name or path.
confidence: Minimum segmentation confidence.
imgsz: Inference image size.
iou: IoU threshold used for mask filtering.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: Optional GPU type requested for model inference.
Returns:
A DataFrame expression that produces ``BINARY`` indexed PNG mask
bytes.
"""
return self._call(
_IMAGE_DETECT,
"image_segment",
model=model,
confidence=confidence,
imgsz=imgsz,
iou=iou,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs] def ocr(
self,
*,
lang: Optional[_Languages] = None,
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Run OCR on image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_ocr`.
See that function for full parameter details.
Args:
lang: Optional OCR language-code sequence.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: Optional GPU type requested for model inference.
Returns:
A DataFrame expression that produces ``ARRAY<ROW>`` OCR results
with ``text``, ``confidence``, and ``bbox`` fields.
"""
return self._call(
_IMAGE_DETECT,
"image_ocr",
lang=lang,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs] def detect_subplot(
self,
*,
threshold: float = 0.5,
concurrency: Optional[int] = None,
) -> Expression:
"""
Detect subplot regions in image values.
Equivalent to :func:`~pyflink.multimodal.operators.image_detect_subplot`.
See that function for full parameter details.
Args:
threshold: Minimum subplot detection confidence.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces a subplot metadata ``ROW``
with ``is_subplot`` and ``count`` fields.
"""
return self._call(
_IMAGE_DETECT,
"image_detect_subplot",
threshold=threshold,
concurrency=concurrency,
)
[docs] def embedding(
self,
*,
model: str = "ViT-B/32",
pretrained: str = "openai",
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Compute image embeddings.
Equivalent to :func:`~pyflink.multimodal.operators.image_embedding`.
See that function for full parameter details.
Args:
model: Embedding model name.
pretrained: Pretrained weights identifier.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: Optional GPU type requested for model inference.
Returns:
A DataFrame expression that produces ``ARRAY<FLOAT>`` embedding
vectors.
"""
return self._call(
_IMAGE_EMBED,
"image_embedding",
model=model,
pretrained=pretrained,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs] def text_similarity(
self,
*,
text_column: Optional[Expression] = None,
text: Optional[str] = None,
model: str = "ViT-B/32",
pretrained: str = "openai",
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Compute similarity between image values and text.
Equivalent to :func:`~pyflink.multimodal.operators.image_text_similarity`.
See that function for full parameter details.
Exactly one of ``text_column`` or ``text`` must be provided.
Args:
text_column: Text expression compared with this image expression.
text: Constant text compared with this image expression.
model: CLIP-style model name.
pretrained: Pretrained weights identifier.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: Optional GPU type requested for model inference.
Returns:
A DataFrame expression that produces a ``DOUBLE`` similarity
score.
"""
if text_column is None and text is None:
raise ValueError("Either text_column or text must be provided.")
if text_column is not None and text is not None:
raise ValueError("text_column and text must not be provided together.")
columns = () if text_column is None else (text_column,)
return self._call(
_IMAGE_EMBED,
"image_text_similarity",
*columns,
text=text,
model=model,
pretrained=pretrained,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs]class AudioExpressionAccessor(_ExpressionAccessor):
"""
Audio operations available on a DataFrame expression.
These methods use the expression on the left side as the audio input. For
example, ``col("audio_bytes").audio.decode()`` is the expression-accessor
form of ``audio_decode(col("audio_bytes"))``.
Examples::
>>> from pyflink.dataframe import col
>>> df = df.with_column("waveform", col("audio_bytes").audio.decode())
>>> df = df.with_column(
... "speech_ready",
... col("waveform").audio.standardize(sample_rate=16000, channels=1))
"""
[docs] def is_valid(
self,
*,
validation: Literal["metadata", "decode"] = "metadata",
concurrency: Optional[int] = None,
) -> Expression:
"""
Check whether this expression contains valid audio.
Equivalent to :func:`~pyflink.multimodal.operators.is_valid_audio`.
See that function for full parameter details.
Args:
validation: ``"metadata"`` for header-level validation or
``"decode"`` for full decode validation.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``BOOLEAN`` validity flags.
"""
return self._call(
_AUDIO_INFO,
"is_valid_audio",
validation=validation,
concurrency=concurrency,
)
[docs] def duration(
self,
*,
concurrency: Optional[int] = None,
) -> Expression:
"""
Return audio duration in seconds for this expression.
Equivalent to :func:`~pyflink.multimodal.operators.audio_duration`.
See that function for full parameter details.
Args:
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``DOUBLE`` seconds, or ``NULL``
when duration is unavailable.
"""
return self._call(_AUDIO_INFO, "audio_duration", concurrency=concurrency)
[docs] def duration_filter(
self,
*,
min_seconds: Optional[float] = None,
max_seconds: Optional[float] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Check whether audio duration is within second bounds.
Equivalent to :func:`~pyflink.multimodal.operators.audio_duration_filter`.
See that function for full parameter details.
Args:
min_seconds: Optional inclusive lower duration bound.
max_seconds: Optional inclusive upper duration bound.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``BOOLEAN`` filter results.
"""
return self._call(
_AUDIO_INFO,
"audio_duration_filter",
min_seconds=min_seconds,
max_seconds=max_seconds,
concurrency=concurrency,
)
[docs] def size_filter(
self,
*,
min_bytes: Optional[int] = None,
max_bytes: Optional[int] = None,
concurrency: Optional[int] = None,
) -> Expression:
"""
Check whether encoded audio size is within byte bounds.
Equivalent to :func:`~pyflink.multimodal.operators.audio_size_filter`.
See that function for full parameter details.
Args:
min_bytes: Optional inclusive lower encoded byte-size bound.
max_bytes: Optional inclusive upper encoded byte-size bound.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``BOOLEAN`` filter results.
"""
return self._call(
_AUDIO_INFO,
"audio_size_filter",
min_bytes=min_bytes,
max_bytes=max_bytes,
concurrency=concurrency,
)
[docs] def silence_detection(
self,
*,
threshold_db: float,
min_silence_ms: int = 0,
concurrency: Optional[int] = None,
) -> Expression:
"""
Detect low-amplitude silence ranges in waveform values.
Equivalent to :func:`~pyflink.multimodal.operators.audio_silence_detection`.
See that function for full parameter details.
Args:
threshold_db: Silence threshold in dBFS. Must be ``<= 0``.
min_silence_ms: Minimum silence range length in milliseconds.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``ARRAY<ROW<start_ms,
end_ms, duration_ms>>`` silence ranges.
"""
return self._call(
_AUDIO_INFO,
"audio_silence_detection",
threshold_db=threshold_db,
min_silence_ms=min_silence_ms,
concurrency=concurrency,
)
[docs] def detect_speech(
self,
*,
aggressiveness: int = 0,
frame_ms: int = 30,
min_speech_ms: int = 0,
merge_gap_ms: int = 0,
max_segments: int = 1024,
concurrency: Optional[int] = None,
) -> Expression:
"""
Detect speech activity ranges in waveform values.
Equivalent to :func:`~pyflink.multimodal.operators.audio_detect_speech`.
See that function for full parameter details.
Args:
aggressiveness: WebRTC VAD aggressiveness, from ``0`` to ``3``.
frame_ms: WebRTC frame size in milliseconds.
min_speech_ms: Drop detected speech ranges shorter than this value.
merge_gap_ms: Merge ranges separated by this gap or less.
max_segments: Maximum number of ranges to return.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces ``ARRAY<ROW<start_ms,
end_ms, duration_ms>>`` speech ranges.
"""
return self._call(
_AUDIO_INFO,
"audio_detect_speech",
aggressiveness=aggressiveness,
frame_ms=frame_ms,
min_speech_ms=min_speech_ms,
merge_gap_ms=merge_gap_ms,
max_segments=max_segments,
concurrency=concurrency,
)
[docs] def decode(
self,
*,
on_error: Literal["raise", "null"] = "raise",
max_decoded_bytes: int = DEFAULT_MAX_DECODED_AUDIO_BYTES,
concurrency: Optional[int] = None,
) -> Expression:
"""
Decode encoded audio into waveform rows.
Equivalent to :func:`~pyflink.multimodal.operators.audio_decode`.
See that function for full parameter details.
Args:
on_error: ``"raise"`` to propagate corrupt audio failures, or
``"null"`` to return null for rows that cannot be decoded.
max_decoded_bytes: Maximum decoded PCM payload allowed per row.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces decoded audio waveform ``ROW``
values.
"""
return self._call(
_AUDIO_TRANSFORM,
"audio_decode",
on_error=on_error,
max_decoded_bytes=max_decoded_bytes,
concurrency=concurrency,
)
[docs] def encode(
self,
*,
format: str = "wav",
concurrency: Optional[int] = None,
) -> Expression:
"""
Encode waveform rows to audio bytes.
Equivalent to :func:`~pyflink.multimodal.operators.audio_encode`.
See that function for full parameter details.
Args:
format: Encoded output format, such as ``"wav"`` or ``"flac"``.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces encoded ``BINARY`` audio bytes.
"""
return self._call(
_AUDIO_TRANSFORM,
"audio_encode",
format=format,
concurrency=concurrency,
)
[docs] def standardize(
self,
*,
sample_rate: int = 16000,
channels: int = 1,
concurrency: Optional[int] = None,
) -> Expression:
"""
Standardize waveform sample rate and channel count.
Equivalent to :func:`~pyflink.multimodal.operators.audio_standardize`.
See that function for full parameter details.
Args:
sample_rate: Target sample rate in samples per second.
channels: Target channel count.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces standardized waveform ``ROW``
values.
"""
return self._call(
_AUDIO_TRANSFORM,
"audio_standardize",
sample_rate=sample_rate,
channels=channels,
concurrency=concurrency,
)
[docs] def resample(
self,
*,
sample_rate: int,
concurrency: Optional[int] = None,
) -> Expression:
"""
Resample waveform values to a target sample rate.
Equivalent to :func:`~pyflink.multimodal.operators.audio_resample`.
See that function for full parameter details.
Args:
sample_rate: Target sample rate in samples per second.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces resampled waveform ``ROW``
values.
"""
return self._call(
_AUDIO_TRANSFORM,
"audio_resample",
sample_rate=sample_rate,
concurrency=concurrency,
)
[docs] def split_by_duration(
self,
*,
segment_duration_ms: int,
segment_type: Literal["audio", "ref"] = "audio",
max_segments: int = 1024,
concurrency: Optional[int] = None,
) -> DataFrameUDTFCall:
"""
Split audio by fixed duration.
Equivalent to :func:`~pyflink.multimodal.operators.audio_split_by_duration`.
See that function for full parameter details.
Args:
segment_duration_ms: Segment duration in milliseconds.
segment_type: ``"audio"`` for waveform segments or ``"ref"`` for
lazy reference segments.
max_segments: Maximum allowed number of returned segments.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame UDTF call for ``join_lateral``. Alias the output as one
segment column, for example ``df.join_lateral(
col("waveform").audio.split_by_duration(
segment_duration_ms=1000).alias("segment"))``.
"""
return self._call(
_AUDIO_TRANSFORM,
"audio_split_by_duration",
segment_duration_ms=segment_duration_ms,
segment_type=segment_type,
max_segments=max_segments,
concurrency=concurrency,
)
[docs] def split_by_timestamp(
self,
timestamps: Union[Expression, _TimestampRanges],
*,
segment_type: Literal["audio", "ref"] = "audio",
concurrency: Optional[int] = None,
) -> DataFrameUDTFCall:
"""
Split audio by explicit timestamp ranges.
Equivalent to :func:`~pyflink.multimodal.operators.audio_split_by_timestamp`.
See that function for full parameter details.
Args:
timestamps: Timestamp ranges in milliseconds, either a Python
literal sequence or a DataFrame column expression.
segment_type: ``"audio"`` for waveform segments or ``"ref"`` for
lazy reference segments.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame UDTF call for ``join_lateral``. Alias the output as one
segment column before consuming it downstream.
Examples::
>>> ranges = [{"start_ms": 100, "end_ms": 350}]
>>> segments = df.join_lateral(
... col("waveform").audio.split_by_timestamp(
... timestamps=ranges,
... ).alias("segment")
... )
>>> segments = df.join_lateral(
... col("waveform").audio.split_by_timestamp(
... col("ranges"),
... ).alias("segment")
... )
"""
if timestamps is None:
return self._call(
_AUDIO_TRANSFORM,
"audio_split_by_timestamp",
segment_type=segment_type,
concurrency=concurrency,
)
if not isinstance(timestamps, Expression):
return self._call(
_AUDIO_TRANSFORM,
"audio_split_by_timestamp",
timestamps=timestamps,
segment_type=segment_type,
concurrency=concurrency,
)
return self._call(
_AUDIO_TRANSFORM,
"audio_split_by_timestamp",
timestamps,
segment_type=segment_type,
concurrency=concurrency,
)
[docs] def split_by_speech(
self,
speech_activity: Expression,
*,
segment_type: Literal["audio", "ref"] = "audio",
pre_padding_ms: int = 0,
post_padding_ms: int = 0,
merge_gap_ms: int = 0,
min_segment_ms: int = 0,
max_segment_ms: Optional[int] = None,
max_segments: int = 1024,
concurrency: Optional[int] = None,
) -> DataFrameUDTFCall:
"""
Split audio using caller-provided speech activity ranges.
Equivalent to :func:`~pyflink.multimodal.operators.audio_split_by_speech`.
See that function for full parameter details.
Args:
speech_activity: Expression containing speech ranges, such as output
from ``audio_detect_speech``.
segment_type: ``"audio"`` for waveform segments or ``"ref"`` for
lazy reference segments.
pre_padding_ms: Milliseconds to extend before each speech range.
post_padding_ms: Milliseconds to extend after each speech range.
merge_gap_ms: Merge adjacent ranges separated by this gap or less.
min_segment_ms: Drop segments shorter than this value.
max_segment_ms: Optional maximum segment length.
max_segments: Maximum allowed number of returned segments.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame UDTF call for ``join_lateral``. Alias the output as one
segment column before consuming it downstream.
"""
return self._call(
_AUDIO_TRANSFORM,
"audio_split_by_speech",
speech_activity,
segment_type=segment_type,
pre_padding_ms=pre_padding_ms,
post_padding_ms=post_padding_ms,
merge_gap_ms=merge_gap_ms,
min_segment_ms=min_segment_ms,
max_segment_ms=max_segment_ms,
max_segments=max_segments,
concurrency=concurrency,
)
[docs] def concat(
self,
*,
concurrency: Optional[int] = None,
) -> Expression:
"""
Concatenate an array of waveform rows.
Equivalent to :func:`~pyflink.multimodal.operators.audio_concat`.
See that function for full parameter details.
Args:
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
Returns:
A DataFrame expression that produces one concatenated waveform row.
"""
return self._call(_AUDIO_TRANSFORM, "audio_concat", concurrency=concurrency)
[docs] def asr_whisper(
self,
*,
language: Optional[str] = None,
task: Literal["transcribe", "translate"] = "transcribe",
model: str = _DEFAULT_AUDIO_MODEL,
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Run Whisper ASR over waveform values.
Equivalent to :func:`~pyflink.multimodal.operators.audio_asr_whisper`.
See that function for full parameter details.
Args:
language: Optional source language hint.
task: ``"transcribe"`` or ``"translate"``.
model: Whisper or Whisper-compatible model id.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: Optional GPU type requested for model inference.
Returns:
A DataFrame expression that produces an ASR result ``ROW`` with
``asr_result``, ``timestamps``, and ``segments`` fields.
"""
return self._call(
_AUDIO_SPEECH,
"audio_asr_whisper",
language=language,
task=task,
model=model,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs] def detect_language(
self,
*,
top_k: int = 1,
model: str = _DEFAULT_AUDIO_MODEL,
model_sharing: Optional[Literal["process", "shared"]] = None,
concurrency: Optional[int] = None,
batch_size: Optional[int] = None,
num_gpus: Optional[float] = None,
gpu_type: Optional[str] = None,
) -> Expression:
"""
Detect likely spoken language candidates for waveform values.
Equivalent to :func:`~pyflink.multimodal.operators.audio_detect_language`.
See that function for full parameter details.
Args:
top_k: Number of language candidates to return per row.
model: Whisper or Whisper-compatible model id.
model_sharing: Optional model sharing mode.
concurrency: Optional execution concurrency for this operation.
``None`` uses the framework default.
batch_size: Optional inference batch size.
num_gpus: Optional number of GPUs requested for model inference.
gpu_type: GPU type requested for model inference. The DataFrame UDF
runtime requires this when ``num_gpus`` is set.
Returns:
A DataFrame expression that produces ``ARRAY<ROW<language_tag,
language_name, confidence>>`` language candidates.
"""
return self._call(
_AUDIO_SPEECH,
"audio_detect_language",
top_k=top_k,
model=model,
model_sharing=model_sharing,
concurrency=concurrency,
batch_size=batch_size,
num_gpus=num_gpus,
gpu_type=gpu_type,
)
[docs]class VideoExpressionAccessor(_ExpressionAccessor):
"""
Video operations available on a DataFrame expression.
These methods use the expression on the left side as the video URI input.
For example, ``col("uri").video.metadata()`` is the expression-accessor
form of ``video_metadata(col("uri"))``.
Examples::
>>> from pyflink.dataframe import col
>>> df = df.with_column("metadata", col("uri").video.metadata())
>>> df = df.with_column(
... "frames", col("uri").video.extract_frames(max_frames=16))
"""