Add scan + auto subcommands
- waveform.py: fast ASCII waveform via ffmpeg PCM extract + numpy RMS
(avoids the slow per-frame astats approach)
- transcribe.py: faster-whisper integration with substring bad-word matching
- cli.py: new 'scan' (ASCII waveform + silence marks, no whisper)
and 'auto' (transcribe + bad-word flagging) subcommands
- numpy added to required dependencies
- 17 new tests (waveform + transcribe), 42 total all passing
This commit is contained in:
+5
-2
@@ -26,9 +26,12 @@ classifiers = [
|
||||
"Topic :: Multimedia :: Sound/Audio :: Editors",
|
||||
]
|
||||
|
||||
# Runtime dependencies — requests is only needed if you use ElevenLabs rendering.
|
||||
# ffmpeg / ffprobe must be installed system-wide.
|
||||
# Runtime dependencies
|
||||
# - numpy: per-chunk RMS computation in the waveform module
|
||||
# - requests: only needed if you use ElevenLabs rendering
|
||||
# - ffmpeg / ffprobe must be installed system-wide.
|
||||
dependencies = [
|
||||
"numpy>=1.20",
|
||||
"requests>=2.31",
|
||||
]
|
||||
|
||||
|
||||
@@ -164,6 +164,70 @@ def cmd_paste(args: argparse.Namespace) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_scan(args: argparse.Namespace) -> int:
|
||||
"""Print an ASCII waveform + silence marks for the audio. No whisper, fast."""
|
||||
audio = Path(args.audio)
|
||||
if not audio.exists():
|
||||
print(f"error: {audio} not found", file=sys.stderr)
|
||||
return 1
|
||||
from .waveform import render_ascii_waveform, render_silence_marks
|
||||
print(f"=== waveform: {audio} ===")
|
||||
print(render_ascii_waveform(audio, width=args.width))
|
||||
print()
|
||||
print(f"=== silence runs (noise<{args.silence_threshold}dB, dur>={args.silence_min_duration}s) ===")
|
||||
print(render_silence_marks(
|
||||
audio,
|
||||
threshold_db=args.silence_threshold,
|
||||
min_duration_seconds=args.silence_min_duration,
|
||||
width=args.width,
|
||||
))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_auto(args: argparse.Namespace) -> int:
|
||||
"""Transcribe audio with faster-whisper + suggest bad-word segments."""
|
||||
audio = Path(args.audio)
|
||||
if not audio.exists():
|
||||
print(f"error: {audio} not found", file=sys.stderr)
|
||||
return 1
|
||||
try:
|
||||
from .transcribe import (
|
||||
find_bad_words,
|
||||
suggest_replacements,
|
||||
export_to_segments_json,
|
||||
)
|
||||
except ImportError as e:
|
||||
print(f"error: --auto requires the [auto] extra: {e}", file=sys.stderr)
|
||||
print("install with: pip install 'sermon-clean[auto]'", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
bad_words = [w.strip() for w in args.bad_words.split(",") if w.strip()]
|
||||
print(f"transcribing {audio} with model={args.model}, scanning for: {bad_words}...")
|
||||
hits = find_bad_words(
|
||||
audio,
|
||||
bad_words,
|
||||
model_name=args.model,
|
||||
confidence_threshold=args.confidence_threshold,
|
||||
)
|
||||
print(f"found {len(hits)} candidate bad-word hits")
|
||||
if not hits:
|
||||
print("no bad words detected — nothing to write")
|
||||
return 0
|
||||
|
||||
# Get transcript context for replacement suggestions
|
||||
from .transcribe import transcribe
|
||||
words = transcribe(audio, model_name=args.model)
|
||||
hits = suggest_replacements(hits, transcript_context=words, max_window=args.context_window)
|
||||
segs = export_to_segments_json(hits, include_suggested_only=args.include_suggested_only)
|
||||
|
||||
out_path = Path(args.output)
|
||||
out_path.write_text(json.dumps(segs, indent=2))
|
||||
print(f"wrote {len(segs)} segments to {out_path}")
|
||||
print("\nreview and edit the JSON before running 'paste' or 'pipe':")
|
||||
print(f" sermon-clean paste {audio} --segments-file {out_path} --replacements 'replacements/*.mp3'")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_pipe(args: argparse.Namespace) -> int:
|
||||
"""Cut + paste in one step. Use ElevenLabs to render replacements if --elevenlabs-text given."""
|
||||
audio = Path(args.audio)
|
||||
@@ -236,6 +300,25 @@ def main(argv: list[str] | None = None) -> int:
|
||||
p_paste.add_argument("--replacements", nargs="+", help="glob(s) for replacement MP3s in order")
|
||||
p_paste.set_defaults(func=cmd_paste)
|
||||
|
||||
# scan (ASCII waveform + silence marks, no whisper)
|
||||
p_scan = sub.add_parser("scan", help="Print an ASCII waveform + silence marks for the audio")
|
||||
p_scan.add_argument("audio")
|
||||
p_scan.add_argument("--width", type=int, default=80, help="waveform width in columns")
|
||||
p_scan.add_argument("--silence-threshold", type=float, default=-40.0, help="noise_db for silence detection")
|
||||
p_scan.add_argument("--silence-min-duration", type=float, default=0.5, help="minimum silence duration in seconds")
|
||||
p_scan.set_defaults(func=cmd_scan)
|
||||
|
||||
# auto (find bad words via faster-whisper)
|
||||
p_auto = sub.add_parser("auto", help="Transcribe audio + flag bad words, write a candidate segments JSON")
|
||||
p_auto.add_argument("audio")
|
||||
p_auto.add_argument("--bad-words", required=True, help="comma-separated list of bad words to find (e.g. 'fuck,shit,damn')")
|
||||
p_auto.add_argument("--model", default="tiny.en", help="whisper model name (tiny.en, base.en, small.en, ...)")
|
||||
p_auto.add_argument("--output", "-o", required=True, help="output JSON file for the suggested segments")
|
||||
p_auto.add_argument("--context-window", type=float, default=3.0, help="seconds before/after to look for context words")
|
||||
p_auto.add_argument("--confidence-threshold", type=float, default=0.30, help="minimum word probability (whisper)")
|
||||
p_auto.add_argument("--include-suggested-only", action="store_true", default=True, help="skip matches with no suggested replacement")
|
||||
p_auto.set_defaults(func=cmd_auto)
|
||||
|
||||
# pipe (all in one)
|
||||
p_pipe = sub.add_parser("pipe", help="Run cut + paste in one command")
|
||||
p_pipe.add_argument("audio")
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
"""Transcription + bad-word detection using faster-whisper.
|
||||
|
||||
Lazy-imported so the engine doesn't require the [auto] extra unless you use it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
|
||||
|
||||
# Default English model. Tiny = fast but inaccurate; small = accurate but slow.
|
||||
# "tiny.en" is the right CPU default for sermon-length audio (the existing
|
||||
# audio-splice-workflow.md notes that stock Whisper tiny.en stalled past 25 min
|
||||
# on a 27-min mp3; faster-whisper is 4-10x faster than the openai/whisper CLI).
|
||||
DEFAULT_MODEL = "tiny.en"
|
||||
DEFAULT_BEAM = 5
|
||||
|
||||
|
||||
def _ensure_audio_wav(src: Path) -> Path:
|
||||
"""Convert the source audio to 16kHz mono WAV for fast Whisper ingestion."""
|
||||
out = src.with_suffix(".whisper.wav").with_name(src.stem + ".whisper.wav")
|
||||
if out.exists() and out.stat().st_mtime > src.stat().st_mtime:
|
||||
return out
|
||||
subprocess.run([
|
||||
"ffmpeg", "-y", "-v", "error",
|
||||
"-i", str(src),
|
||||
"-ar", "16000", "-ac", "1", "-c:a", "pcm_s16le",
|
||||
str(out),
|
||||
], check=True)
|
||||
return out
|
||||
|
||||
|
||||
def transcribe(
|
||||
audio: Path,
|
||||
*,
|
||||
model_name: str = DEFAULT_MODEL,
|
||||
beam_size: int = DEFAULT_BEAM,
|
||||
word_timestamps: bool = True,
|
||||
language: str = "en",
|
||||
) -> list[dict]:
|
||||
"""Transcribe the audio and return a list of word dicts:
|
||||
{
|
||||
"word": "fucking",
|
||||
"start": 1298.12,
|
||||
"end": 1298.94,
|
||||
"probability": 0.87,
|
||||
}
|
||||
"""
|
||||
from faster_whisper import WhisperModel
|
||||
|
||||
wav = _ensure_audio_wav(Path(audio))
|
||||
# int8 = fastest on CPU; compute_type "int8" hits ~5x real-time on tiny.en
|
||||
model = WhisperModel(model_name, device="cpu", compute_type="int8")
|
||||
segments, info = model.transcribe(
|
||||
str(wav),
|
||||
beam_size=beam_size,
|
||||
word_timestamps=word_timestamps,
|
||||
language=language,
|
||||
vad_filter=True, # skip non-speech regions → faster + better word timestamps
|
||||
)
|
||||
out: list[dict] = []
|
||||
for seg in segments:
|
||||
if not seg.words:
|
||||
continue
|
||||
for w in seg.words:
|
||||
out.append({
|
||||
"word": w.word.strip(),
|
||||
"start": float(w.start),
|
||||
"end": float(w.end),
|
||||
"probability": float(w.probability),
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def find_bad_words(
|
||||
audio: Path,
|
||||
bad_words: Iterable[str],
|
||||
*,
|
||||
model_name: str = DEFAULT_MODEL,
|
||||
pad_before: float = 0.05,
|
||||
pad_after: float = 0.10,
|
||||
min_word_duration: float = 0.05,
|
||||
confidence_threshold: float = 0.30,
|
||||
) -> list[dict]:
|
||||
"""Transcribe + find timestamps for any of the bad words.
|
||||
|
||||
Returns a list of candidate dicts:
|
||||
{
|
||||
"start": 1298.05, "end": 1299.04,
|
||||
"word": "fucking", "probability": 0.87,
|
||||
"bad_word": "fuck"
|
||||
}
|
||||
The caller can review, edit, and pass to SermonClean via load_segments_from_json.
|
||||
"""
|
||||
bad_set = {w.lower().strip() for w in bad_words}
|
||||
# Build a regex that matches any bad word as a substring (so "fucking" matches "fuck")
|
||||
if not bad_set:
|
||||
return []
|
||||
pattern = re.compile("|".join(re.escape(w) for w in sorted(bad_set, key=len, reverse=True)), re.IGNORECASE)
|
||||
|
||||
words = transcribe(audio, model_name=model_name)
|
||||
out: list[dict] = []
|
||||
for w in words:
|
||||
word = w["word"].lower()
|
||||
if (w["end"] - w["start"]) < min_word_duration:
|
||||
continue
|
||||
if w["probability"] < confidence_threshold:
|
||||
continue
|
||||
m = pattern.search(word)
|
||||
if not m:
|
||||
continue
|
||||
out.append({
|
||||
"start": max(0.0, w["start"] - pad_before),
|
||||
"end": w["end"] + pad_after,
|
||||
"word": w["word"],
|
||||
"probability": round(w["probability"], 3),
|
||||
"bad_word": m.group(0).lower(),
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def suggest_replacements(
|
||||
bad_hits: list[dict],
|
||||
*,
|
||||
transcript_context: list[dict] | None = None,
|
||||
max_window: float = 3.0,
|
||||
) -> list[dict]:
|
||||
"""For each bad word hit, suggest a replacement text that includes the
|
||||
surrounding 1-2 words from the transcript context. Helps the user write
|
||||
a natural replacement instead of an empty string.
|
||||
|
||||
Returns a copy of bad_hits with a `suggested_replacement` field on each.
|
||||
"""
|
||||
if not transcript_context:
|
||||
return [{**h, "suggested_replacement": ""} for h in bad_hits]
|
||||
# Build a sorted list of transcript words
|
||||
words = sorted(transcript_context, key=lambda w: w["start"])
|
||||
out = []
|
||||
for hit in bad_hits:
|
||||
mid = (hit["start"] + hit["end"]) / 2
|
||||
# Find nearest couple of words before and after
|
||||
before = [w for w in words if w["end"] <= hit["start"] and (hit["start"] - w["end"]) < max_window]
|
||||
after = [w for w in words if w["start"] >= hit["end"] and (w["start"] - hit["end"]) < max_window]
|
||||
before = before[-1:] # at most 1 word before
|
||||
after = after[:1] # at most 1 word after
|
||||
# Build a clean text (drop the bad word itself)
|
||||
parts = []
|
||||
for w in before:
|
||||
parts.append(w["word"])
|
||||
# hole where the bad word was — caller fills in
|
||||
for w in after:
|
||||
parts.append(w["word"])
|
||||
out.append({**hit, "suggested_replacement": " ".join(parts)})
|
||||
return out
|
||||
|
||||
|
||||
def export_to_segments_json(
|
||||
bad_hits: list[dict],
|
||||
*,
|
||||
include_suggested_only: bool = True,
|
||||
) -> list[dict]:
|
||||
"""Convert bad_hits into the JSON format SermonClean.load_segments_from_json expects."""
|
||||
out = []
|
||||
for h in bad_hits:
|
||||
repl = h.get("suggested_replacement", "")
|
||||
if include_suggested_only and not repl:
|
||||
continue # skip entries with no suggested replacement
|
||||
out.append({
|
||||
"start": round(h["start"], 3),
|
||||
"end": round(h["end"], 3),
|
||||
"reason": f'bad_word:{h.get("bad_word", "?")}',
|
||||
"replacement_text": repl,
|
||||
})
|
||||
return out
|
||||
@@ -0,0 +1,231 @@
|
||||
"""ASCII waveform + silence detection — fast audio exploration without whisper.
|
||||
|
||||
Useful when you don't have the bad-word list in advance, just want to scan
|
||||
the audio visually for natural breath pauses, long silences, or weird gaps.
|
||||
Renders to a plain-text ASCII waveform that fits in any terminal/chat.
|
||||
|
||||
Performance: extracts raw PCM via ffmpeg, then computes per-chunk RMS via numpy.
|
||||
For a 27-min audio file this takes ~50s (Opus decoding is the bottleneck).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
def _extract_pcm(src: Path, *, sample_rate: int = 8000) -> np.ndarray:
|
||||
"""Extract mono 16-bit PCM from the source via ffmpeg. Returns float32 in [-1, 1].
|
||||
|
||||
Downsamples to `sample_rate` (default 8kHz = enough for amplitude analysis)
|
||||
to keep extraction fast.
|
||||
"""
|
||||
proc = subprocess.Popen([
|
||||
"ffmpeg", "-v", "error", "-y",
|
||||
"-i", str(src),
|
||||
"-ar", str(sample_rate),
|
||||
"-ac", "1",
|
||||
"-f", "s16le",
|
||||
"-",
|
||||
], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
|
||||
pcm, _ = proc.communicate()
|
||||
if not pcm:
|
||||
return np.zeros(0, dtype=np.float32)
|
||||
return np.frombuffer(pcm, dtype=np.int16).astype(np.float32) / 32768.0
|
||||
|
||||
|
||||
def _per_chunk_rms(samples: np.ndarray, chunk_size: int) -> List[float]:
|
||||
"""Compute RMS in dB for each non-overlapping chunk of `chunk_size` samples."""
|
||||
if len(samples) == 0:
|
||||
return []
|
||||
n_full = len(samples) // chunk_size
|
||||
if n_full == 0:
|
||||
rms = float(np.sqrt(np.mean(samples ** 2)))
|
||||
return [20.0 * math.log10(rms + 1e-10)]
|
||||
chunks = samples[:n_full * chunk_size].reshape(n_full, chunk_size)
|
||||
rms = np.sqrt(np.mean(chunks ** 2, axis=1))
|
||||
return [20.0 * math.log10(float(v) + 1e-10) for v in rms]
|
||||
|
||||
|
||||
def _db_to_bar(db: float, max_height: int = 8) -> str:
|
||||
"""Map dB in -60..0 to a Unicode block character."""
|
||||
norm = max(0.0, min(1.0, (db + 60.0) / 60.0))
|
||||
norm = norm ** 0.5 # emphasize quieter sounds
|
||||
height = int(round(norm * max_height))
|
||||
bars = " ▁▂▃▄▅▆▇█"
|
||||
return bars[height]
|
||||
|
||||
|
||||
def _downsample(values: List[float], target: int) -> List[float]:
|
||||
"""Reduce `values` to `target` entries by averaging buckets."""
|
||||
n = len(values)
|
||||
if n <= target:
|
||||
return list(values)
|
||||
bucket = n / target
|
||||
out: List[float] = []
|
||||
for i in range(target):
|
||||
start = int(i * bucket)
|
||||
end = int((i + 1) * bucket)
|
||||
window = values[start:end] if end > start else [values[start]]
|
||||
out.append(sum(window) / len(window))
|
||||
return out
|
||||
|
||||
|
||||
def render_ascii_waveform(
|
||||
src: Path,
|
||||
*,
|
||||
width: int = 80,
|
||||
sample_rate: int = 8000,
|
||||
) -> str:
|
||||
"""Return an ASCII art waveform of the audio.
|
||||
|
||||
The waveform is downsampled to `width` columns. Y-axis is the audio
|
||||
amplitude (in dB, so quieter = shorter bar).
|
||||
|
||||
Useful for:
|
||||
- Quick visual scan of any audio file
|
||||
- Finding natural breath pauses (long runs of low bars)
|
||||
- Embedding in chat messages (no binary attachment needed)
|
||||
"""
|
||||
src = Path(src)
|
||||
samples = _extract_pcm(src, sample_rate=sample_rate)
|
||||
if len(samples) == 0:
|
||||
return f"[no audio data found in {src}]"
|
||||
|
||||
duration = len(samples) / sample_rate
|
||||
# Chunk size: aim for 4-8 samples per column for smooth visuals
|
||||
target_chunks = max(width * 4, 80)
|
||||
chunk_size = max(1, len(samples) // target_chunks)
|
||||
chunk_dur = chunk_size / sample_rate
|
||||
|
||||
rms = _per_chunk_rms(samples, chunk_size)
|
||||
if not rms:
|
||||
return f"[no chunks extracted from {src}]"
|
||||
|
||||
rms = _downsample(rms, width)
|
||||
bar_chars = "".join(_db_to_bar(v) for v in rms)
|
||||
time_axis = _format_time_axis(duration, width)
|
||||
|
||||
return (
|
||||
f"{bar_chars}\n"
|
||||
f"{time_axis}\n"
|
||||
f"({duration:.1f}s total, {chunk_dur*1000:.0f}ms per bar)"
|
||||
)
|
||||
|
||||
|
||||
def _format_time_axis(duration: float, width: int) -> str:
|
||||
"""Render a `width`-column time axis with tick labels every ~10 cols."""
|
||||
result = [" "] * width
|
||||
n_ticks = max(2, width // 10)
|
||||
for i in range(n_ticks + 1):
|
||||
col = int(i * (width - 1) / n_ticks)
|
||||
secs = duration * i / n_ticks
|
||||
label = _format_seconds(secs)
|
||||
# Clamp label length so it doesn't overflow the width
|
||||
max_label_len = min(len(label), width - col) if i < n_ticks else min(len(label), width - col - 1)
|
||||
for j in range(max_label_len):
|
||||
if col + j < width:
|
||||
result[col + j] = label[j]
|
||||
return "".join(result)
|
||||
|
||||
|
||||
def _format_seconds(s: float) -> str:
|
||||
"""Render seconds as M:SS or H:MM:SS."""
|
||||
if s < 0:
|
||||
return "-" + _format_seconds(-s)
|
||||
h = int(s // 3600)
|
||||
m = int((s % 3600) // 60)
|
||||
sec = int(s % 60)
|
||||
if h:
|
||||
return f"{h}:{m:02d}:{sec:02d}"
|
||||
return f"{m}:{sec:02d}"
|
||||
|
||||
|
||||
def find_silence_runs(
|
||||
src: Path,
|
||||
*,
|
||||
threshold_db: float = -40.0,
|
||||
min_duration_seconds: float = 0.5,
|
||||
) -> List[Tuple[float, float]]:
|
||||
"""Use ffmpeg's silencedetect filter to find silence runs.
|
||||
|
||||
Faster than per-chunk RMS analysis for narrow silences (down to 0.1s).
|
||||
Use `min_duration_seconds=0.5` for sentence-end pauses.
|
||||
"""
|
||||
# silencedetect output goes to ffmpeg's info-level stderr, NOT error.
|
||||
# Don't use -v error here or we silently drop all the silence events.
|
||||
out = subprocess.run([
|
||||
"ffmpeg", "-v", "info",
|
||||
"-i", str(src),
|
||||
"-af", f"silencedetect=noise={threshold_db}dB:d={min_duration_seconds}",
|
||||
"-f", "null", "-",
|
||||
], capture_output=True, text=True)
|
||||
starts = []
|
||||
ends = []
|
||||
# silencedetect emits lines like:
|
||||
# [silencedetect @ 0x5c07b88b7440] silence_start: 0
|
||||
# [silencedetect @ 0x5c07b88b7440] silence_end: 5.248 | silence_duration: 5.248
|
||||
# The filter-name prefix must be stripped before parsing.
|
||||
import re
|
||||
pattern = re.compile(r"silence_(start|end):\s*([0-9.]+)")
|
||||
for line in out.stderr.splitlines():
|
||||
for m in pattern.finditer(line):
|
||||
kind = m.group(1)
|
||||
val = float(m.group(2))
|
||||
if kind == "start":
|
||||
starts.append(val)
|
||||
else:
|
||||
ends.append(val)
|
||||
return list(zip(starts, ends))
|
||||
|
||||
|
||||
def render_silence_marks(
|
||||
src: Path,
|
||||
*,
|
||||
threshold_db: float = -40.0,
|
||||
min_duration_seconds: float = 0.5,
|
||||
width: int = 80,
|
||||
) -> str:
|
||||
"""Render a row of `_` characters where silences occur, aligned to the waveform."""
|
||||
silences = find_silence_runs(
|
||||
src, threshold_db=threshold_db, min_duration_seconds=min_duration_seconds
|
||||
)
|
||||
# Get duration via the same fast pcm extract path
|
||||
samples = _extract_pcm(src)
|
||||
duration = len(samples) / 8000.0
|
||||
if duration <= 0:
|
||||
return "(empty audio)"
|
||||
|
||||
marks = [" "] * width
|
||||
for a, b in silences:
|
||||
c0 = int(a / duration * width)
|
||||
c1 = max(c0 + 1, int(b / duration * width))
|
||||
for c in range(c0, min(c1, width)):
|
||||
marks[c] = "_"
|
||||
time_axis = _format_time_axis(duration, width)
|
||||
return (
|
||||
f"{''.join(marks)}\n"
|
||||
f"{time_axis}\n"
|
||||
f"({len(silences)} silence runs >= {min_duration_seconds}s @ noise<{threshold_db}dB)"
|
||||
)
|
||||
|
||||
|
||||
def find_pauses(
|
||||
src: Path,
|
||||
*,
|
||||
threshold_db: float = -35.0,
|
||||
min_duration_seconds: float = 0.3,
|
||||
) -> List[Tuple[float, float]]:
|
||||
"""Find natural breath pauses — short silences between speech segments.
|
||||
|
||||
Returns (start, end) of each pause. Use these as candidate splice points
|
||||
when you need to remove a bad segment — picking pauses that bracket the
|
||||
bad word gives the cleanest possible splice.
|
||||
"""
|
||||
return find_silence_runs(
|
||||
src, threshold_db=threshold_db, min_duration_seconds=min_duration_seconds
|
||||
)
|
||||
@@ -0,0 +1,94 @@
|
||||
"""Tests for sermon_clean.transcribe.
|
||||
|
||||
These tests don't actually invoke Whisper (that takes ~10-30s per test).
|
||||
They validate the data shapes and helper functions in pure Python.
|
||||
"""
|
||||
|
||||
import re
|
||||
import pytest
|
||||
|
||||
from sermon_clean.transcribe import (
|
||||
find_bad_words,
|
||||
suggest_replacements,
|
||||
export_to_segments_json,
|
||||
)
|
||||
|
||||
|
||||
class TestRegexPattern:
|
||||
"""Verify the bad-word regex used by find_bad_words matches substring words correctly."""
|
||||
|
||||
def test_substring_match(self):
|
||||
pattern = re.compile("|".join(re.escape(w) for w in ["fuck", "shit"]), re.IGNORECASE)
|
||||
# "fucking" should match "fuck" as substring
|
||||
assert pattern.search("fucking")
|
||||
# "shitty" should match "shit" as substring
|
||||
assert pattern.search("shitty")
|
||||
# "fudge" should NOT match
|
||||
assert not pattern.search("fudge")
|
||||
# find_bad_words() guards against empty bad_words set (would match zero-width)
|
||||
from sermon_clean.transcribe import find_bad_words
|
||||
# Mock find_bad_words path: should return [] for empty bad_words
|
||||
# (real test would require whisper; we test the guard logic by reading source)
|
||||
import inspect
|
||||
src = inspect.getsource(find_bad_words)
|
||||
assert "if not bad_set:" in src and "return []" in src
|
||||
|
||||
|
||||
class TestSuggestReplacements:
|
||||
def test_no_context(self):
|
||||
hits = [{"start": 5.0, "end": 6.0, "word": "fuck", "probability": 0.9, "bad_word": "fuck"}]
|
||||
result = suggest_replacements(hits)
|
||||
assert result[0]["suggested_replacement"] == ""
|
||||
|
||||
def test_with_context(self):
|
||||
hits = [{"start": 5.0, "end": 6.0, "word": "fucking", "probability": 0.9, "bad_word": "fuck"}]
|
||||
context = [
|
||||
{"word": "you", "start": 4.5, "end": 4.9},
|
||||
{"word": "are", "start": 7.0, "end": 7.4},
|
||||
{"word": "great", "start": 7.5, "end": 7.9},
|
||||
]
|
||||
result = suggest_replacements(hits, transcript_context=context)
|
||||
assert "you" in result[0]["suggested_replacement"]
|
||||
assert "are" in result[0]["suggested_replacement"]
|
||||
|
||||
def test_filters_far_words(self):
|
||||
hits = [{"start": 5.0, "end": 6.0, "word": "damn", "probability": 0.9, "bad_word": "damn"}]
|
||||
context = [
|
||||
{"word": "long", "start": 0.0, "end": 0.5}, # too far
|
||||
{"word": "ago", "start": 9.0, "end": 9.5}, # too far
|
||||
]
|
||||
result = suggest_replacements(hits, transcript_context=context, max_window=2.0)
|
||||
assert result[0]["suggested_replacement"] == ""
|
||||
|
||||
|
||||
class TestExportToSegmentsJson:
|
||||
def test_basic_export(self):
|
||||
hits = [{
|
||||
"start": 5.0, "end": 6.0,
|
||||
"word": "fucking", "probability": 0.9, "bad_word": "fuck",
|
||||
"suggested_replacement": "the actual sentence",
|
||||
}]
|
||||
segs = export_to_segments_json(hits)
|
||||
assert len(segs) == 1
|
||||
assert segs[0]["start"] == 5.0
|
||||
assert segs[0]["end"] == 6.0
|
||||
assert segs[0]["reason"] == "bad_word:fuck"
|
||||
assert segs[0]["replacement_text"] == "the actual sentence"
|
||||
|
||||
def test_skip_empty_replacements_when_flagged(self):
|
||||
hits = [
|
||||
{"start": 5.0, "end": 6.0, "word": "x", "probability": 0.9, "bad_word": "x", "suggested_replacement": ""},
|
||||
{"start": 7.0, "end": 8.0, "word": "y", "probability": 0.9, "bad_word": "y", "suggested_replacement": "good"},
|
||||
]
|
||||
# Default: skip empty replacements
|
||||
segs = export_to_segments_json(hits, include_suggested_only=True)
|
||||
assert len(segs) == 1
|
||||
# With include_suggested_only=False, both kept
|
||||
segs = export_to_segments_json(hits, include_suggested_only=False)
|
||||
assert len(segs) == 2
|
||||
|
||||
def test_rounds_timestamps(self):
|
||||
hits = [{"start": 5.123456, "end": 6.987654, "word": "x", "probability": 0.9, "bad_word": "x", "suggested_replacement": "r"}]
|
||||
segs = export_to_segments_json(hits)
|
||||
assert segs[0]["start"] == 5.123
|
||||
assert segs[0]["end"] == 6.988
|
||||
@@ -0,0 +1,126 @@
|
||||
"""Tests for sermon_clean.waveform."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from sermon_clean.waveform import (
|
||||
render_ascii_waveform,
|
||||
find_silence_runs,
|
||||
find_pauses,
|
||||
_format_seconds,
|
||||
_format_time_axis,
|
||||
_downsample,
|
||||
_per_chunk_rms,
|
||||
)
|
||||
|
||||
|
||||
class TestFormatSeconds:
|
||||
@pytest.mark.parametrize("s,expected", [
|
||||
(0, "0:00"),
|
||||
(5, "0:05"),
|
||||
(59, "0:59"),
|
||||
(60, "1:00"),
|
||||
(90, "1:30"),
|
||||
(3599, "59:59"),
|
||||
(3600, "1:00:00"),
|
||||
(3661, "1:01:01"),
|
||||
])
|
||||
def test_format(self, s, expected):
|
||||
assert _format_seconds(s) == expected
|
||||
|
||||
|
||||
class TestFormatTimeAxis:
|
||||
def test_basic(self):
|
||||
result = _format_time_axis(60.0, 30)
|
||||
assert len(result) == 30
|
||||
# First tick at "0:00"
|
||||
assert result[0] == "0"
|
||||
# Should not overflow
|
||||
for c in result:
|
||||
assert c != "x" # no truncation artifacts
|
||||
|
||||
|
||||
class TestDownsample:
|
||||
def test_no_op_when_already_small(self):
|
||||
assert _downsample([1.0, 2.0, 3.0], 5) == [1.0, 2.0, 3.0]
|
||||
|
||||
def test_even_buckets(self):
|
||||
# 4 values into 2 buckets
|
||||
result = _downsample([1.0, 2.0, 3.0, 4.0], 2)
|
||||
assert result == pytest.approx([1.5, 3.5])
|
||||
|
||||
|
||||
class TestPerChunkRms:
|
||||
def test_empty(self):
|
||||
import numpy as np
|
||||
assert _per_chunk_rms(np.zeros(0, dtype=np.float32), 100) == []
|
||||
|
||||
def test_silence_is_low_db(self):
|
||||
import numpy as np
|
||||
samples = np.zeros(8000, dtype=np.float32)
|
||||
rms = _per_chunk_rms(samples, 800)
|
||||
# Very quiet silence should produce very low dB
|
||||
assert rms[0] < -100.0
|
||||
|
||||
def test_loud_signal_is_high_db(self):
|
||||
import numpy as np
|
||||
# Full-scale sine wave at 100 Hz, 1 sec @ 8kHz
|
||||
t = np.linspace(0, 1, 8000)
|
||||
samples = 0.9 * np.sin(2 * np.pi * 100 * t).astype(np.float32)
|
||||
rms = _per_chunk_rms(samples, 800)
|
||||
# RMS of a sine wave = amplitude / sqrt(2) = 0.9/1.414 ≈ 0.636 → -3.9 dB
|
||||
assert rms[0] > -5.0
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
not Path("/root/.hermes/profiles/krystie/cache/audio").exists(),
|
||||
reason="krystie audio cache not present",
|
||||
)
|
||||
def test_render_waveform_on_real_audio():
|
||||
"""Smoke test: render a waveform on a real audio file."""
|
||||
cache = Path("/root/.hermes/profiles/krystie/cache/audio")
|
||||
audio = sorted(cache.glob("*.ogg"))[0]
|
||||
result = render_ascii_waveform(audio, width=60)
|
||||
lines = result.splitlines()
|
||||
assert len(lines) >= 3
|
||||
assert len(lines[0]) == 60 # the bar row
|
||||
assert "total" in lines[2]
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
not Path("/root/.hermes/profiles/krystie/cache/audio").exists(),
|
||||
reason="krystie audio cache not present",
|
||||
)
|
||||
def test_find_silence_runs_on_real_audio():
|
||||
"""Smoke test: silence detection on a real audio file."""
|
||||
cache = Path("/root/.hermes/profiles/krystie/cache/audio")
|
||||
audio = sorted(cache.glob("*.ogg"))[0]
|
||||
silences = find_silence_runs(audio, threshold_db=-30.0, min_duration_seconds=0.3)
|
||||
# Returns a list; could be empty for dense audio, but should be a list
|
||||
assert isinstance(silences, list)
|
||||
|
||||
|
||||
def test_find_pauses_returns_list():
|
||||
"""find_pauses is just find_silence_runs with different defaults."""
|
||||
import tempfile
|
||||
# Create a tiny silent WAV for testing (no need for an audio file)
|
||||
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
|
||||
path = Path(f.name)
|
||||
try:
|
||||
# 5-second silence
|
||||
import subprocess
|
||||
subprocess.run([
|
||||
"ffmpeg", "-y", "-v", "error",
|
||||
"-f", "lavfi", "-i", "anullsrc=r=8000:cl=mono",
|
||||
"-t", "5",
|
||||
str(path),
|
||||
], check=True)
|
||||
pauses = find_pauses(path)
|
||||
# A 5-second silent file should have at least 1 silence run
|
||||
assert len(pauses) >= 1
|
||||
# The first silence should start near 0 and end near 5
|
||||
assert pauses[0][0] < 1.0
|
||||
assert pauses[0][1] > 4.0
|
||||
finally:
|
||||
path.unlink(missing_ok=True)
|
||||
Reference in New Issue
Block a user