################################################################################
# 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.
################################################################################
"""Audio information, validation, and filter operators.
These operators inspect audio objects and return scalar values or row values.
They do not modify waveform content. Encoded inputs are accepted only at the
pipeline boundary; waveform-only operators require an explicit ``audio_decode``
step so that pipeline definitions stay predictable.
"""
from pyflink.dataframe import DataType, udf
from pyflink.table.udf import ScalarFunction
from pyflink.multimodal.codec.audio import (
audio_encoded_size,
decode_audio_input,
detect_speech_ranges,
detect_silence_ranges,
ensure_supported_audio_input,
is_audio_data_error,
probe_audio_metadata,
)
from pyflink.multimodal.types.audio import AUDIO_METADATA_TYPE
from pyflink.multimodal.utils import _build_or_apply_udf, _udf_runtime_kwargs
__all__ = [
"is_valid_audio",
"audio_metadata",
"audio_duration",
"audio_duration_filter",
"audio_size_filter",
"audio_silence_detection",
"audio_detect_speech",
]
_AUDIO_TIME_RANGE_LIST_TYPE = DataType.list(
DataType.struct([
("start_ms", DataType.int64()),
("end_ms", DataType.int64()),
("duration_ms", DataType.int64()),
])
)
_AUDIO_VALIDATION_MODES = {"metadata", "decode"}
_WEBRTC_VAD_FRAME_MS = (10, 20, 30)
def _bridge_client_or_none(function_context):
getter = getattr(function_context, "_get_file_system_bridge_client", None)
if getter is None:
return None
return getter()
def _within_optional_bounds(value, lower_bound, upper_bound):
if lower_bound is not None and value < lower_bound:
return False
return upper_bound is None or value <= upper_bound
def _validate_at_least_one_bound(operator_name, *bounds):
if all(bound is None for bound in bounds):
raise ValueError(f"{operator_name} requires at least one bound.")
def _optional_non_negative_float(name, value):
if value is None:
return None
if not isinstance(value, (int, float)) or isinstance(value, bool) or value < 0:
raise ValueError(f"{name} must be a non-negative number, got {value!r}")
return float(value)
def _optional_non_negative_int(name, value):
if value is None:
return None
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise ValueError(f"{name} must be a non-negative integer, got {value!r}")
return value
def _positive_int(name, value):
if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
raise ValueError(f"{name} must be a positive integer, got {value!r}")
return value
def _non_negative_int(name, value):
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
raise ValueError(f"{name} must be a non-negative integer, got {value!r}")
return value
def _int_range(name, value, minimum, maximum):
if not isinstance(value, int) or isinstance(value, bool):
raise ValueError(f"{name} must be an integer, got {value!r}")
if value < minimum or value > maximum:
raise ValueError(f"{name} must be between {minimum} and {maximum}, got {value!r}")
return value
def _int_choice(name, value, choices):
if not isinstance(value, int) or isinstance(value, bool):
raise ValueError(f"{name} must be an integer, got {value!r}")
if value not in choices:
allowed = ", ".join(str(choice) for choice in choices)
raise ValueError(f"{name} must be one of [{allowed}], got {value!r}")
return value
def _validate_ordered_bounds(name, lower_bound, upper_bound):
if lower_bound is not None and upper_bound is not None and upper_bound < lower_bound:
raise ValueError(
f"{name} upper bound must be >= lower bound, "
f"got lower={lower_bound!r}, upper={upper_bound!r}"
)
def _normalize_audio_validation_mode(validation):
if validation not in _AUDIO_VALIDATION_MODES:
raise ValueError(
"is_valid_audio validation must be one of "
f"{sorted(_AUDIO_VALIDATION_MODES)}, got {validation!r}"
)
return validation
class _AudioValidityFilter(ScalarFunction):
"""Check whether BYTES, URI, or AUDIO_CLIP_REF input can be read as audio."""
def __init__(self, validation="metadata"):
super().__init__()
self.validation = _normalize_audio_validation_mode(validation)
self._bridge_client = None
def open(self, function_context):
self._bridge_client = _bridge_client_or_none(function_context)
def eval(self, audio_input):
if audio_input is None:
return False
ensure_supported_audio_input(
"is_valid_audio",
("BYTES", "URI", "AUDIO_CLIP_REF"),
audio_input,
)
try:
if self.validation == "metadata":
probe_audio_metadata(
audio_input,
bridge_client=self._bridge_client,
operator_name="is_valid_audio",
)
return True
if self.validation == "decode":
decode_audio_input(
audio_input,
bridge_client=self._bridge_client,
)
return True
except Exception as e:
if is_audio_data_error(e):
return False
raise
[docs]def is_valid_audio(*columns, validation="metadata", concurrency=None):
"""
Check whether an audio input looks like valid audio.
The default ``validation="metadata"`` only reads enough container/header
data to probe metadata. This is the cheap path for filtering large tables
before decode. ``validation="decode"`` skips the metadata pre-probe and
fully decodes the stream; it is more expensive but catches truncated or
corrupt payloads that have a readable header.
Args:
*columns: Optional audio columns. When provided, the UDF is applied
directly.
validation: ``"metadata"`` for header/container validation, or
``"decode"`` for full decode validation.
concurrency: UDF concurrency. ``None`` uses the framework default.
Returns:
A UDF producing ``BOOLEAN``. Corrupt supported inputs return ``False``.
URI access errors still fail fast.
Raises:
ValueError: If ``validation`` is unsupported, or if a row value has an
unsupported input kind. Supported inputs are ``BYTES``, URI
``STRING``, and ``AUDIO_CLIP_REF``.
Examples::
>>> # Usage 1: create a reusable UDF and apply it to a column.
>>> valid = is_valid_audio(validation="metadata")
>>> df = df.filter(valid(col("audio_bytes")))
>>>
>>> # Usage 2: pass the column directly when building the expression.
>>> df = df.filter(
... is_valid_audio(col("audio_bytes"), validation="metadata")
... )
"""
wrapper = udf(
_AudioValidityFilter(validation=validation),
return_dtype=DataType.boolean(),
**_udf_runtime_kwargs(concurrency=concurrency),
)
return _build_or_apply_udf(wrapper, *columns)
class _AudioMetadata(ScalarFunction):
"""
Extract audio metadata from encoded inputs, references, or waveform rows.
``codec`` is a best-effort media codec name. For WAV/PCM inputs it follows
FFmpeg/PyAV codec short-name convention, e.g. ``pcm_s16le`` for signed
16-bit little-endian PCM.
"""
def __init__(self):
super().__init__()
self._bridge_client = None
def open(self, function_context):
self._bridge_client = _bridge_client_or_none(function_context)
def eval(self, audio_input):
if audio_input is None:
return None
ensure_supported_audio_input(
"audio_metadata",
("BYTES", "URI", "AUDIO_CLIP_REF", "AUDIO_WAVEFORM"),
audio_input,
)
try:
return probe_audio_metadata(
audio_input,
bridge_client=self._bridge_client,
operator_name="audio_metadata",
)
except Exception as e:
if is_audio_data_error(e):
return None
raise
class _AudioDuration(ScalarFunction):
"""Return audio duration in seconds."""
def __init__(self):
super().__init__()
self._bridge_client = None
def open(self, function_context):
self._bridge_client = _bridge_client_or_none(function_context)
def eval(self, audio_input):
if audio_input is None:
return None
ensure_supported_audio_input(
"audio_duration",
("BYTES", "URI", "AUDIO_CLIP_REF", "AUDIO_WAVEFORM"),
audio_input,
)
try:
metadata = probe_audio_metadata(
audio_input,
bridge_client=self._bridge_client,
operator_name="audio_duration",
)
except Exception as e:
if is_audio_data_error(e):
return None
raise
if metadata.duration_ms is None:
return None
return float(metadata.duration_ms) / 1000.0
[docs]def audio_duration(*columns, concurrency=None):
"""
Return audio duration in seconds.
Args:
*columns: Optional audio columns. Supported inputs are ``BYTES``, URI
``STRING``, ``AUDIO_CLIP_REF``, and ``AUDIO_WAVEFORM``.
concurrency: UDF concurrency. ``None`` uses the framework default.
Returns:
A UDF producing ``DOUBLE`` seconds, or ``NULL`` when duration is
unavailable. Corrupt supported encoded inputs also return ``NULL`` as
row-level dirty data. URI access errors, dependency errors, unsupported
input kinds, and unexpected internal errors still fail fast.
Raises:
ValueError: If a row value is not ``BYTES``, URI ``STRING``,
``AUDIO_CLIP_REF``, or ``AUDIO_WAVEFORM``.
Examples::
>>> # Usage 1: create a reusable UDF and apply it to a column.
>>> duration = audio_duration()
>>> df = df.with_column("seconds", duration(col("audio")))
>>>
>>> # Usage 2: pass the column directly when building the expression.
>>> df = df.with_column(
... "seconds",
... audio_duration(col("audio")),
... )
"""
wrapper = udf(
_AudioDuration(),
return_dtype=DataType.float64(),
**_udf_runtime_kwargs(concurrency=concurrency),
)
return _build_or_apply_udf(wrapper, *columns)
class _AudioDurationFilter(ScalarFunction):
"""Filter audio by duration in seconds."""
def __init__(self, min_seconds=None, max_seconds=None):
super().__init__()
_validate_at_least_one_bound(
"audio_duration_filter", min_seconds, max_seconds
)
self.min_seconds = _optional_non_negative_float("min_seconds", min_seconds)
self.max_seconds = _optional_non_negative_float("max_seconds", max_seconds)
_validate_ordered_bounds(
"audio_duration_filter", self.min_seconds, self.max_seconds
)
self._bridge_client = None
def open(self, function_context):
self._bridge_client = _bridge_client_or_none(function_context)
def eval(self, audio_input):
if audio_input is None:
return False
ensure_supported_audio_input(
"audio_duration_filter",
("BYTES", "URI", "AUDIO_CLIP_REF", "AUDIO_WAVEFORM"),
audio_input,
)
try:
metadata = probe_audio_metadata(
audio_input,
bridge_client=self._bridge_client,
operator_name="audio_duration_filter",
)
except Exception as e:
if is_audio_data_error(e):
return False
raise
if metadata.duration_ms is None:
return False
duration_seconds = float(metadata.duration_ms) / 1000.0
return _within_optional_bounds(
duration_seconds, self.min_seconds, self.max_seconds
)
[docs]def audio_duration_filter(
*columns,
min_seconds=None,
max_seconds=None,
concurrency=None,
):
"""
Check whether audio duration is within optional second bounds.
Args:
*columns: Optional audio columns. Supported inputs are ``BYTES``, URI
``STRING``, ``AUDIO_CLIP_REF``, and ``AUDIO_WAVEFORM``.
min_seconds: Inclusive lower duration bound. ``None`` disables it.
max_seconds: Inclusive upper duration bound. ``None`` disables it.
concurrency: UDF concurrency. ``None`` uses the framework default.
Returns:
A UDF producing ``BOOLEAN``. ``NULL`` or unknown-duration rows return
``False``. Corrupt supported encoded inputs also return ``False`` as
row-level dirty data. URI access errors, dependency errors, unsupported
input kinds, and unexpected internal errors still fail fast.
Raises:
ValueError: If neither bound is set, a bound is negative,
``max_seconds < min_seconds``, or if a row value has an unsupported
input kind.
Examples::
>>> # Usage 1: create a reusable UDF and apply it to a column.
>>> duration_filter = audio_duration_filter(
... min_seconds=0.5,
... )
>>> df = df.filter(duration_filter(col("audio")))
>>>
>>> # Usage 2: pass the column directly when building the expression.
>>> df = df.filter(
... audio_duration_filter(
... col("audio"),
... min_seconds=0.5,
... )
... )
"""
wrapper = udf(
_AudioDurationFilter(
min_seconds=min_seconds,
max_seconds=max_seconds,
),
return_dtype=DataType.boolean(),
**_udf_runtime_kwargs(concurrency=concurrency),
)
return _build_or_apply_udf(wrapper, *columns)
class _AudioSizeFilter(ScalarFunction):
"""Filter encoded BYTES or URI audio by byte size."""
def __init__(self, min_bytes=None, max_bytes=None):
super().__init__()
_validate_at_least_one_bound("audio_size_filter", min_bytes, max_bytes)
self.min_bytes = _optional_non_negative_int("min_bytes", min_bytes)
self.max_bytes = _optional_non_negative_int("max_bytes", max_bytes)
_validate_ordered_bounds("audio_size_filter", self.min_bytes, self.max_bytes)
self._bridge_client = None
def open(self, function_context):
self._bridge_client = _bridge_client_or_none(function_context)
def eval(self, audio_input):
if audio_input is None:
return False
ensure_supported_audio_input("audio_size_filter", ("BYTES", "URI"), audio_input)
size = audio_encoded_size(
audio_input,
bridge_client=self._bridge_client,
operator_name="audio_size_filter",
)
return _within_optional_bounds(size, self.min_bytes, self.max_bytes)
[docs]def audio_size_filter(
*columns,
min_bytes=None,
max_bytes=None,
concurrency=None,
):
"""
Check whether encoded audio size is within optional byte bounds.
Args:
*columns: Optional audio columns. Supported inputs are ``BYTES`` and
URI ``STRING``.
min_bytes: Inclusive lower byte bound. ``None`` disables it.
max_bytes: Inclusive upper byte bound. ``None`` disables it.
concurrency: UDF concurrency. ``None`` uses the framework default.
Returns:
A UDF producing ``BOOLEAN``. ``NULL`` inputs return ``False``; URI
access errors still fail fast.
Raises:
ValueError: If neither bound is set, a bound is negative,
``max_bytes < min_bytes``, or if a row value has an unsupported
input kind.
Examples::
>>> # Usage 1: create a reusable UDF and apply it to a column.
>>> size_filter = audio_size_filter(
... max_bytes=10 * 1024 * 1024,
... )
>>> df = df.filter(size_filter(col("audio")))
>>>
>>> # Usage 2: pass the column directly when building the expression.
>>> df = df.filter(
... audio_size_filter(
... col("audio"),
... max_bytes=10 * 1024 * 1024,
... )
... )
"""
wrapper = udf(
_AudioSizeFilter(
min_bytes=min_bytes,
max_bytes=max_bytes,
),
return_dtype=DataType.boolean(),
**_udf_runtime_kwargs(concurrency=concurrency),
)
return _build_or_apply_udf(wrapper, *columns)
class _AudioSilenceDetection(ScalarFunction):
"""Detect continuous low-amplitude ranges in a waveform row."""
def __init__(self, threshold_db, min_silence_ms=0):
super().__init__()
if not isinstance(threshold_db, (int, float)) or isinstance(threshold_db, bool):
raise ValueError(f"threshold_db must be a number, got {threshold_db!r}")
if threshold_db > 0:
raise ValueError(f"threshold_db must be <= 0, got {threshold_db!r}")
if not isinstance(min_silence_ms, int) or isinstance(min_silence_ms, bool):
raise ValueError(
f"min_silence_ms must be an integer, got {min_silence_ms!r}"
)
if min_silence_ms < 0:
raise ValueError(
f"min_silence_ms must be non-negative, got {min_silence_ms!r}"
)
self.threshold_db = float(threshold_db)
self.min_silence_ms = min_silence_ms
def eval(self, audio_input):
if audio_input is None:
return None
return detect_silence_ranges(
audio_input,
threshold_db=self.threshold_db,
min_silence_ms=self.min_silence_ms,
)
[docs]def audio_silence_detection(
*columns,
threshold_db,
min_silence_ms=0,
concurrency=None,
):
"""
Detect low-amplitude silence ranges in an ``AUDIO_WAVEFORM``.
This is an energy-based silence detector, not a speech activity detector.
It downmixes multi-channel waveform data to mono, computes frame RMS,
converts RMS to decibels relative to full-scale amplitude, and returns
ranges whose energy stays at or below ``threshold_db`` for at least
``min_silence_ms``. No ASR or VAD model is used.
``threshold_db`` is required because silence is corpus dependent: a noisy
call-center recording and a studio recording need different thresholds.
``min_silence_ms`` defaults to ``0`` so the operator does not hide a
second corpus-specific heuristic. Tune both values for the input domain.
Use ``audio_detect_speech`` for speech activity detection; silence ranges
are not equivalent to speech ranges.
Args:
*columns: Optional audio columns. Supported input is
``AUDIO_WAVEFORM`` only; call ``audio_decode`` first for encoded
inputs.
threshold_db: Required silence threshold in dB relative to full-scale float
waveform amplitude. It must be ``<= 0``; for example ``-40.0`` means
frames at or below -40 dBFS are treated as silence. ``0`` means
0 dBFS/full scale, so almost all ordinary audio will be treated as
silence.
min_silence_ms: Minimum range length in milliseconds. Shorter quiet
ranges are dropped.
concurrency: UDF concurrency. ``None`` uses the framework default.
Returns:
A UDF producing ``ARRAY<ROW<start_ms BIGINT, end_ms BIGINT,
duration_ms BIGINT>>``. For waveform rows decoded from ``AUDIO_CLIP_REF``
segments, ``start_ms`` and ``end_ms`` are reported in the source audio
timeline so they can be fed into timestamp-based split operators.
Raises:
ValueError: If the input is not ``AUDIO_WAVEFORM`` or parameters are
invalid.
Examples::
>>> # Prepare a decoded waveform before applying silence detection.
>>> decode = audio_decode()
>>> waveform = decode(col("audio_bytes"))
>>> # Usage 1: create a reusable UDF and apply it to a column.
>>> silence = audio_silence_detection(threshold_db=-45.0)
>>> df = df.with_column("silence_ranges", silence(waveform))
>>>
>>> # Usage 2: pass the column directly when building the expression.
>>> df = df.with_column(
... "silence_ranges",
... audio_silence_detection(waveform, threshold_db=-45.0),
... )
"""
wrapper = udf(
_AudioSilenceDetection(
threshold_db=threshold_db,
min_silence_ms=min_silence_ms,
),
return_dtype=_AUDIO_TIME_RANGE_LIST_TYPE,
**_udf_runtime_kwargs(concurrency=concurrency),
)
return _build_or_apply_udf(wrapper, *columns)
class _AudioDetectSpeech(ScalarFunction):
"""Detect speech activity ranges in a mono waveform row using WebRTC VAD."""
def __init__(self,
aggressiveness=0,
frame_ms=30,
min_speech_ms=0,
merge_gap_ms=0,
max_segments=1024):
super().__init__()
self.aggressiveness = _int_range(
"audio_detect_speech.aggressiveness",
aggressiveness,
0,
3,
)
self.frame_ms = _int_choice(
"audio_detect_speech.frame_ms",
frame_ms,
_WEBRTC_VAD_FRAME_MS,
)
self.min_speech_ms = _non_negative_int(
"audio_detect_speech.min_speech_ms",
min_speech_ms,
)
self.merge_gap_ms = _non_negative_int(
"audio_detect_speech.merge_gap_ms",
merge_gap_ms,
)
self.max_segments = _positive_int(
"audio_detect_speech.max_segments",
max_segments,
)
def eval(self, audio_input):
if audio_input is None:
return None
return detect_speech_ranges(
audio_input,
aggressiveness=self.aggressiveness,
frame_ms=self.frame_ms,
min_speech_ms=self.min_speech_ms,
merge_gap_ms=self.merge_gap_ms,
max_segments=self.max_segments,
)
[docs]def audio_detect_speech(
*columns,
aggressiveness=0,
frame_ms=30,
min_speech_ms=0,
merge_gap_ms=0,
max_segments=1024,
concurrency=None,
):
"""
Detect speech activity ranges in a mono waveform with WebRTC VAD.
This operator is the speech-range producer for ``audio_split_by_speech``.
It uses the third-party WebRTC VAD implementation, not an energy threshold
or an ASR model. The input must be mono and have a WebRTC-supported sample
rate: 8000, 16000, 32000, or 48000 Hz. For most ASR pipelines, call
``audio_standardize(sample_rate=16000, channels=1)`` first.
Args:
*columns: Optional audio columns. Supported input is a row with fields
``data TENSOR('float32'), sample_rate INT, channels INT,
frames BIGINT, sample_format STRING, layout STRING,
duration_ms BIGINT, source_uri STRING, start_time_ms BIGINT,
end_time_ms BIGINT``.
aggressiveness: WebRTC VAD aggressiveness mode, from ``0`` (least
aggressive) to ``3`` (most aggressive). The default ``0`` follows
WebRTC VAD's least aggressive mode rather than a PyFlink threshold.
frame_ms: WebRTC frame size in milliseconds. Must be ``10``, ``20``,
or ``30``.
min_speech_ms: Drop detected speech ranges shorter than this duration.
The default ``0`` keeps raw WebRTC VAD ranges.
merge_gap_ms: Merge speech ranges separated by this gap or less. The
default ``0`` only merges adjacent speech frames.
max_segments: Maximum number of ranges to return.
concurrency: UDF concurrency. ``None`` uses the framework default.
Returns:
A UDF producing ``ARRAY<ROW<start_ms BIGINT, end_ms BIGINT,
duration_ms BIGINT>>`` sorted by time. Pass this array as the second
argument to ``audio_split_by_speech``. For waveform rows decoded from
``AUDIO_CLIP_REF`` segments, ``start_ms`` and ``end_ms`` are reported in the
source audio timeline so they can be fed back into split operators.
Raises:
ValueError: If the input is not a supported mono waveform, the sample
rate is unsupported by WebRTC VAD, or parameters are invalid.
Examples::
>>> # Prepare a mono waveform at a WebRTC-supported sample rate.
>>> decode = audio_decode()
>>> standardize = audio_standardize(sample_rate=16000, channels=1)
>>> waveform = standardize(decode(col("audio_bytes")))
>>> # Usage 1: create a reusable UDF and apply it to a column.
>>> detect_speech = audio_detect_speech()
>>> df = df.with_column("speech_ranges", detect_speech(waveform))
>>>
>>> # Usage 2: pass the column directly when building the expression.
>>> df = df.with_column(
... "speech_ranges",
... audio_detect_speech(waveform),
... )
"""
wrapper = udf(
_AudioDetectSpeech(
aggressiveness=aggressiveness,
frame_ms=frame_ms,
min_speech_ms=min_speech_ms,
merge_gap_ms=merge_gap_ms,
max_segments=max_segments,
),
return_dtype=_AUDIO_TIME_RANGE_LIST_TYPE,
**_udf_runtime_kwargs(concurrency=concurrency),
)
return _build_or_apply_udf(wrapper, *columns)