lesavka/scripts/manual/run_local_audio_sanity.sh

#!/usr/bin/env bash
# scripts/manual/run_local_audio_sanity.sh
#
# Play a real Lesavka-style speaker tone, record the real local microphone
# path, and fail if the recorded signal does not show strong energy at the
# expected tone frequency.

set -euo pipefail

SCRIPT_DIR="$(cd -- "$(dirname "${BASH_SOURCE[0]}")" >/dev/null 2>&1 && pwd)"
REPO_ROOT="$(cd -- "${SCRIPT_DIR}/../.." >/dev/null 2>&1 && pwd)"

LOCAL_OUTPUT_DIR=${LOCAL_OUTPUT_DIR:-"${REPO_ROOT}/tmp"}
LOCAL_AUDIO_SOURCE=${LOCAL_AUDIO_SOURCE:-}
LOCAL_AUDIO_SINK=${LOCAL_AUDIO_SINK:-}
LOCAL_AUDIO_SANITY_SECONDS=${LOCAL_AUDIO_SANITY_SECONDS:-6}
LOCAL_AUDIO_TONE_DELAY_SECONDS=${LOCAL_AUDIO_TONE_DELAY_SECONDS:-1}
LOCAL_AUDIO_TONE_SECONDS=${LOCAL_AUDIO_TONE_SECONDS:-3}
LOCAL_AUDIO_TONE_FREQUENCY_HZ=${LOCAL_AUDIO_TONE_FREQUENCY_HZ:-880}
LOCAL_AUDIO_TONE_VOLUME=${LOCAL_AUDIO_TONE_VOLUME:-0.25}

resolve_default_source() {
  pactl get-default-source 2>/dev/null
}

resolve_source_mute_state() {
  local source=$1
  pactl list sources 2>/dev/null | awk -v source="${source}" '
    $1 == "Name:" && $2 == source { seen=1; next }
    seen && $1 == "Mute:" { print $2; exit }
    seen && /^$/ { exit }
  '
}

mkdir -p "${LOCAL_OUTPUT_DIR}"
STAMP="$(date +%Y%m%d-%H%M%S)"
CAPTURE_WAV="${LOCAL_OUTPUT_DIR}/lesavka-local-audio-sanity-${STAMP}.wav"

if [[ -z "${LOCAL_AUDIO_SOURCE}" ]]; then
  if ! LOCAL_AUDIO_SOURCE="$(resolve_default_source)"; then
    echo "could not determine a default Pulse/PipeWire source" >&2
    exit 64
  fi
fi

SOURCE_MUTE_STATE="$(resolve_source_mute_state "${LOCAL_AUDIO_SOURCE}")"
if [[ "${SOURCE_MUTE_STATE}" == "yes" ]]; then
  echo "warning: source ${LOCAL_AUDIO_SOURCE} is currently muted; unmuting for sanity test" >&2
  pactl set-source-mute "${LOCAL_AUDIO_SOURCE}" 0
fi

capture_pid=""
cleanup() {
  if [[ -n "${capture_pid}" ]] && kill -0 "${capture_pid}" >/dev/null 2>&1; then
    kill "${capture_pid}" >/dev/null 2>&1 || true
    wait "${capture_pid}" >/dev/null 2>&1 || true
  fi
}
trap cleanup EXIT

echo "==> capturing local microphone source: ${LOCAL_AUDIO_SOURCE}"
ffmpeg -y -hide_banner -loglevel error \
  -f pulse \
  -i "${LOCAL_AUDIO_SOURCE}" \
  -ac 1 \
  -ar 16000 \
  -t "${LOCAL_AUDIO_SANITY_SECONDS}" \
  "${CAPTURE_WAV}" &
capture_pid=$!

sleep "${LOCAL_AUDIO_TONE_DELAY_SECONDS}"

echo "==> playing local speaker tone at ${LOCAL_AUDIO_TONE_FREQUENCY_HZ} Hz"
tone_sink_args=()
if [[ -n "${LOCAL_AUDIO_SINK}" ]]; then
  tone_sink_args=(device="${LOCAL_AUDIO_SINK}")
fi
tone_status=0
timeout "${LOCAL_AUDIO_TONE_SECONDS}" \
  gst-launch-1.0 -q \
    audiotestsrc is-live=true wave=sine \
    freq="${LOCAL_AUDIO_TONE_FREQUENCY_HZ}" \
    volume="${LOCAL_AUDIO_TONE_VOLUME}" \
    ! audioconvert ! audioresample ! queue ! pulsesink "${tone_sink_args[@]}" \
    >/dev/null 2>&1 || tone_status=$?
if [[ "${tone_status}" -ne 0 && "${tone_status}" -ne 124 ]]; then
  echo "local speaker tone failed with status ${tone_status}" >&2
  exit "${tone_status}"
fi

wait "${capture_pid}"
capture_pid=""

python - "${CAPTURE_WAV}" \
  "${LOCAL_AUDIO_TONE_FREQUENCY_HZ}" \
  "${LOCAL_AUDIO_TONE_DELAY_SECONDS}" \
  "${LOCAL_AUDIO_TONE_SECONDS}" <<'PY'
import math
import statistics
import struct
import sys
import wave

path = sys.argv[1]
target_hz = float(sys.argv[2])
tone_start_s = float(sys.argv[3])
tone_duration_s = float(sys.argv[4])
tone_end_s = tone_start_s + tone_duration_s
window_s = 0.1

with wave.open(path, "rb") as wav:
    frame_count = wav.getnframes()
    sample_rate = wav.getframerate()
    sample_width = wav.getsampwidth()
    channel_count = wav.getnchannels()
    raw = wav.readframes(frame_count)

if sample_width != 2:
    raise SystemExit(f"unsupported sample width {sample_width * 8} bits in {path}")

samples = [sample[0] for sample in struct.iter_unpack("<h", raw)]
window_frames = max(1, int(sample_rate * window_s))
omega = 2.0 * math.pi * target_hz / sample_rate
coeff = 2.0 * math.cos(omega)

def goertzel_power(values: list[int]) -> tuple[float, float]:
    s0 = 0.0
    s1 = 0.0
    s2 = 0.0
    for value in values:
        s0 = value + coeff * s1 - s2
        s2 = s1
        s1 = s0
    power = s1 * s1 + s2 * s2 - coeff * s1 * s2
    rms = math.sqrt(sum(value * value for value in values) / len(values)) if values else 0.0
    return power, rms

rows: list[tuple[float, float, float]] = []
for offset in range(0, len(samples), window_frames):
    window = samples[offset:offset + window_frames]
    if len(window) < window_frames:
        break
    power, rms = goertzel_power(window)
    rows.append((offset / sample_rate, power, rms))

baseline_rows = [row for row in rows if row[0] < max(0.0, tone_start_s - 0.2)]
tone_rows = [
    row for row in rows
    if row[0] >= tone_start_s + 0.2 and row[0] <= max(tone_start_s + 0.2, tone_end_s - 0.2)
]
if not baseline_rows or not tone_rows:
    raise SystemExit("sanity analysis did not get enough baseline/tone windows")

baseline_power = statistics.median(row[1] for row in baseline_rows)
baseline_rms = statistics.median(row[2] for row in baseline_rows)
tone_power = statistics.median(row[1] for row in tone_rows)
tone_rms = statistics.median(row[2] for row in tone_rows)
power_ratio = tone_power / max(baseline_power, 1.0)
rms_ratio = tone_rms / max(baseline_rms, 1.0)

print(
    {
        "capture": path,
        "tone_hz": target_hz,
        "baseline_power": round(baseline_power, 1),
        "tone_power": round(tone_power, 1),
        "power_ratio": round(power_ratio, 2),
        "baseline_rms": round(baseline_rms, 1),
        "tone_rms": round(tone_rms, 1),
        "rms_ratio": round(rms_ratio, 2),
    }
)

if tone_power < max(1.0e9, baseline_power * 50.0):
    raise SystemExit(
        f"tone energy at {target_hz:.0f} Hz was not strong enough "
        f"(baseline={baseline_power:.1f}, tone={tone_power:.1f})"
    )
print("local speaker-to-mic sanity passed")
PY

echo "==> local audio sanity capture: ${CAPTURE_WAV}"