114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""Shared audio-silence sentence boundary detection.
|
|
|
|
This module intentionally has no GUI or web-service dependencies so the
|
|
desktop player, the pre-generation CLI, and the API worker use the same
|
|
algorithm.
|
|
"""
|
|
|
|
SILENCE_FLOOR_PERCENTILE = 10
|
|
SILENCE_THRESHOLD_FACTOR = 1.5
|
|
SILENCE_BRIDGE_GAP = 0.06
|
|
|
|
|
|
def detect_sentence_boundaries(path, min_silence=0.30, min_sentence=0.35):
|
|
"""Return sentence start times in seconds, or None when unavailable.
|
|
|
|
Audio is resampled to mono 16 kHz, divided into 30 ms windows, and
|
|
silence runs are converted into boundaries. This matches the original
|
|
desktop player's behavior.
|
|
"""
|
|
try:
|
|
import av
|
|
import numpy as np
|
|
except ImportError:
|
|
return None
|
|
|
|
container = None
|
|
try:
|
|
container = av.open(str(path))
|
|
audio = next((stream for stream in container.streams if stream.type == "audio"), None)
|
|
if audio is None:
|
|
return None
|
|
|
|
sample_rate = 16_000
|
|
resampler = av.AudioResampler(format="fltp", layout="mono", rate=sample_rate)
|
|
window = 480
|
|
frame_seconds = window / sample_rate
|
|
energies = []
|
|
buffer = []
|
|
|
|
def push(data):
|
|
buffer.append(data)
|
|
total = sum(len(chunk) for chunk in buffer)
|
|
if total < window:
|
|
return
|
|
array = np.concatenate(buffer)
|
|
buffer.clear()
|
|
complete = len(array) // window * window
|
|
windows = array[:complete].reshape(-1, window)
|
|
energies.extend((np.mean(windows * windows, axis=1) ** 0.5).tolist())
|
|
if len(array) > complete:
|
|
buffer.append(array[complete:])
|
|
|
|
for packet in container.demux(audio):
|
|
for frame in packet.decode():
|
|
for output in resampler.resample(frame):
|
|
push(output.to_ndarray()[0])
|
|
for output in resampler.resample(None):
|
|
push(output.to_ndarray()[0])
|
|
if buffer:
|
|
push(np.zeros(window, dtype=np.float32))
|
|
|
|
duration = container.duration
|
|
if len(energies) < 3:
|
|
return None
|
|
|
|
energies = np.asarray(energies, dtype=np.float64)
|
|
signal_floor = float(np.percentile(energies, 95))
|
|
if signal_floor <= 0.0:
|
|
return None
|
|
noise_floor = float(np.percentile(energies, SILENCE_FLOOR_PERCENTILE))
|
|
threshold = max(0.008, SILENCE_THRESHOLD_FACTOR * noise_floor)
|
|
silence = energies < threshold
|
|
|
|
bridge_frames = int(round(SILENCE_BRIDGE_GAP / frame_seconds))
|
|
if bridge_frames > 0:
|
|
bridged = silence.copy()
|
|
run_start = None
|
|
for index, is_silent in enumerate(silence):
|
|
if not is_silent and run_start is None:
|
|
run_start = index
|
|
elif is_silent and run_start is not None:
|
|
if run_start > 0 and index - run_start <= bridge_frames:
|
|
bridged[run_start:index] = True
|
|
run_start = None
|
|
silence = bridged
|
|
|
|
boundaries = [0.0]
|
|
run_start = None
|
|
for index, is_silent in enumerate(silence):
|
|
if is_silent and run_start is None:
|
|
run_start = index
|
|
elif not is_silent and run_start is not None:
|
|
if (index - run_start) * frame_seconds >= min_silence:
|
|
end_seconds = index * frame_seconds
|
|
if end_seconds - boundaries[-1] >= min_sentence:
|
|
boundaries.append(end_seconds)
|
|
run_start = None
|
|
if run_start is not None and (len(silence) - run_start) * frame_seconds >= min_silence:
|
|
end_seconds = len(silence) * frame_seconds
|
|
if end_seconds - boundaries[-1] >= min_sentence:
|
|
boundaries.append(end_seconds)
|
|
|
|
if duration and duration > 0:
|
|
boundaries = [boundary for boundary in boundaries if boundary < duration / 1e6 - 0.1]
|
|
return boundaries
|
|
except Exception:
|
|
return None
|
|
finally:
|
|
if container is not None:
|
|
container.close()
|
|
|
|
|
|
__all__ = ["detect_sentence_boundaries"]
|