diff --git a/pyproject.toml b/pyproject.toml index 3094dd3..aafff0e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", ] diff --git a/sermon_clean/cli.py b/sermon_clean/cli.py index 8494d76..44f09bd 100644 --- a/sermon_clean/cli.py +++ b/sermon_clean/cli.py @@ -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") diff --git a/sermon_clean/transcribe.py b/sermon_clean/transcribe.py new file mode 100644 index 0000000..ba7ea61 --- /dev/null +++ b/sermon_clean/transcribe.py @@ -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 diff --git a/sermon_clean/waveform.py b/sermon_clean/waveform.py new file mode 100644 index 0000000..1a9f3ec --- /dev/null +++ b/sermon_clean/waveform.py @@ -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 + ) \ No newline at end of file diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py new file mode 100644 index 0000000..d0c3407 --- /dev/null +++ b/tests/test_transcribe.py @@ -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 \ No newline at end of file diff --git a/tests/test_waveform.py b/tests/test_waveform.py new file mode 100644 index 0000000..a6527a2 --- /dev/null +++ b/tests/test_waveform.py @@ -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) \ No newline at end of file