Use a Raspberry Pi 5 to transcribe phone calls etc locally

Local Call Transcriber — Raspberry Pi 5 / DietPi

This is the conservative, CPU-only Raspberry Pi 5 installer for the local call transcription and speaker-diarization pipeline.

It is deliberately separate from the universal Intel/AMD/GPU installer. The Pi version stays close to the configuration already proven on the Raspberry Pi 5.

Target

  • Raspberry Pi 5
  • 64-bit ARM (aarch64)
  • DietPi / Debian-based Linux
  • CPU-only Whisper.cpp
  • Sherpa-ONNX 1.13.7
  • Two-speaker diarization

The installer refuses to run unless it detects a Raspberry Pi 5 and aarch64.

Known-good diarization configuration

The installer uses the corrected Sherpa-ONNX configuration:

  • segmentation model: model.int8.onnx
  • embedding model: 3D-Speaker ...16k.onnx
  • CPU provider
  • 4 threads
  • 2 speaker clusters
  • clustering threshold 0.5
  • min_duration_on=0.3
  • min_duration_off=0.5

The segmentation and embedding provider/thread settings are applied to their respective model configurations, rather than to the top-level diarization configuration. Sherpa-ONNX’s official Python example structures offline diarization as separate segmentation, embedding and clustering configurations. citeturn0search0

Install

On the Pi:

bash install.sh

The script re-executes itself with sudo if necessary. It does not run apt-get update and does not install python3-pip; the Sherpa packages are installed inside a Python virtual environment.

At startup it shows the actual OS, architecture, kernel, hardware model and CPU thread count, then asks for confirmation.

Run

After installation:

transcribe

You will be asked:

Include speaker diarization? [Y/n]:

Press Enter/Y for Speaker 1/Speaker 2 diarization, or N for transcription only.

Processing behaviour

  • Original recordings are never modified or deleted.
  • Any directory named optimized is skipped, case-insensitively.
  • Existing completed .txt + .json results are skipped.
  • Intermediate files are stored under .work so interrupted jobs can resume.
  • Per-file locks prevent duplicate concurrent processing.
  • A failed recording does not stop the batch.
  • Whisper segments are assigned to the diarization speaker with the greatest temporal overlap.
  • A Whisper sentence is not split when the speaker changes inside it.

Installed paths

/opt/whisper.cpp
/opt/sherpa-onnx-env
/opt/sherpa-onnx-models
/opt/call-pipeline
/usr/local/bin/transcribe
/etc/call-pipeline.conf

GPU

GPU support is intentionally disabled in this Pi 5 package. NVIDIA/AMD GPU experimentation belongs in the separate universal Linux installer so it cannot disturb the known-good Pi deployment.

Models

The installer uses Whisper.cpp’s model download helper and the Sherpa-ONNX speaker-diarization models documented by the project. citeturn0search0turn0search2

Review applicable project/model licenses before redistributing the installer.

INSTALL SCRIPT.

#!/usr/bin/env bash
set -euo pipefail

APP_DIR=/opt/call-pipeline
WHISPER_DIR=/opt/whisper.cpp
WHISPER_MODEL=medium.en
SHERPA_ENV=/opt/sherpa-onnx-env
SHERPA_MODELS=/opt/sherpa-onnx-models
CONFIG=/etc/call-pipeline.conf

if [[ ${EUID} -ne 0 ]]; then
    exec sudo -E bash "$0" "$@"
fi

ARCH=$(uname -m)
KERNEL=$(uname -r)
MODEL="Unknown hardware"
OS_NAME="Unknown Linux"
OS_ID=""

if [[ -r /etc/os-release ]]; then
    # shellcheck disable=SC1091
    . /etc/os-release
    OS_NAME="${PRETTY_NAME:-${NAME:-Unknown Linux}}"
    OS_ID="${ID:-}"
fi

if [[ -r /proc/device-tree/model ]]; then
    MODEL=$(tr -d '0' < /proc/device-tree/model)
fi

IS_PI5=0
if [[ "$MODEL" == *"Raspberry Pi 5"* || "$KERNEL" == *"rpi-2712"* ]]; then
    IS_PI5=1
fi

IS_DIETPI=0
if [[ "$OS_ID" == "dietpi" ||
    -f /boot/dietpi/.version ||
    -f /DietPi/dietpi/.version ]]; then
    IS_DIETPI=1
fi

CPU_THREADS=$(nproc 2>/dev/null || echo 1)

if [[ "$ARCH" != "aarch64" ]]; then
    echo "ERROR: This installer is for Raspberry Pi 5 64-bit ARM (aarch64)." >&2
    echo "Detected architecture: $ARCH" >&2
    exit 1
fi

if (( IS_PI5 == 0 )); then
    echo "ERROR: Raspberry Pi 5 was not detected." >&2
    echo "Detected hardware: $MODEL" >&2
    echo "Detected kernel: $KERNEL" >&2
    exit 1
fi

echo
echo 'Call Transcriber — Raspberry Pi 5 Installer'
echo '============================================'
echo
echo 'Detected system:'
echo "  OS:            $OS_NAME"
echo "  Architecture:  $ARCH"
echo "  Kernel:        $KERNEL"
echo "  Hardware:      $MODEL"
echo "  CPU threads:   $CPU_THREADS"

if (( IS_DIETPI )); then
    echo '  Distribution:  DietPi'
else
    echo '  Distribution:  Raspberry Pi 5 / non-DietPi Linux'
    echo '  WARNING: DietPi was not detected.'
fi

echo
echo 'Installation:'
echo '  Whisper.cpp:             medium.en'
echo '  Speaker diarization:    Sherpa-ONNX 1.13.7'
echo '  Processing:              CPU only'
echo '  GPU acceleration:        disabled'
echo '  Original recordings:    never modified'
echo

read -r -p 'Continue? [Y/n]: ' confirm

case "${confirm:-Y}" in
    n|N|no|NO)
        echo 'Installation cancelled.'
        exit 0
        ;;
esac

echo

export DEBIAN_FRONTEND=noninteractive

apt-get install -y \
    ca-certificates \
    curl \
    ffmpeg \
    g++ \
    git \
    make \
    cmake \
    python3 \
    python3-venv \
    python3-dev \
    pkg-config

mkdir -p /opt

if [[ ! -d "$WHISPER_DIR/.git" ]]; then
    git clone https://github.com/ggml-org/whisper.cpp.git "$WHISPER_DIR"
else
    echo "Using existing $WHISPER_DIR"
fi

cmake \
    -S "$WHISPER_DIR" \
    -B "$WHISPER_DIR/build" \
    -DCMAKE_BUILD_TYPE=Release

cmake \
    --build "$WHISPER_DIR/build" \
    --config Release \
    -j"$CPU_THREADS"

WHISPER_BIN="$WHISPER_DIR/build/bin/whisper-cli"
test -x "$WHISPER_BIN"

mkdir -p "$WHISPER_DIR/models"

if [[ ! -f "$WHISPER_DIR/models/ggml-${WHISPER_MODEL}.bin" ]]; then
    (
        cd "$WHISPER_DIR"
        bash ./models/download-ggml-model.sh "$WHISPER_MODEL"
    )
fi

test -f "$WHISPER_DIR/models/ggml-${WHISPER_MODEL}.bin"

python3 -m venv "$SHERPA_ENV"

"$SHERPA_ENV/bin/pip" install --upgrade pip
"$SHERPA_ENV/bin/pip" install \
    'sherpa-onnx==1.13.7' \
    'numpy'

mkdir -p "$SHERPA_MODELS"

SEG_TARBALL=/tmp/sherpa-segmentation.tar.bz2

SEG_URL='https://github.com/k2-fsa/sherpa-onnx/releases/download/speaker-segmentation-models/sherpa-onnx-pyannote-segmentation-3-0.tar.bz2'

EMB_URL='https://github.com/k2-fsa/sherpa-onnx/releases/download/speaker-recongition-models/3dspeaker_speech_eres2net_base_sv_zh-cn_3dspeaker_16k.onnx'

if [[ ! -f "$SHERPA_MODELS/sherpa-onnx-pyannote-segmentation-3-0/model.int8.onnx" ]]; then
    curl -fL --retry 3 -o "$SEG_TARBALL" "$SEG_URL"
    tar -xjf "$SEG_TARBALL" -C "$SHERPA_MODELS"
fi

if [[ ! -f "$SHERPA_MODELS/3dspeaker_speech_eres2net_base_sv_zh-cn_3dspeaker_16k.onnx" ]]; then
    curl \
        -fL \
        --retry 3 \
        -o "$SHERPA_MODELS/3dspeaker_speech_eres2net_base_sv_zh-cn_3dspeaker_16k.onnx" \
        "$EMB_URL"
fi

test -f "$SHERPA_MODELS/sherpa-onnx-pyannote-segmentation-3-0/model.int8.onnx"
test -f "$SHERPA_MODELS/3dspeaker_speech_eres2net_base_sv_zh-cn_3dspeaker_16k.onnx"

mkdir -p "$APP_DIR"

cat > "$APP_DIR/diarize.py" <<'PY'
#!/usr/bin/env python3

import json
import os
import sys
import wave
from pathlib import Path

import numpy as np
import sherpa_onnx


SEG = Path(
    "/opt/sherpa-onnx-models/"
    "sherpa-onnx-pyannote-segmentation-3-0/model.int8.onnx"
)

EMB = Path(
    "/opt/sherpa-onnx-models/"
    "3dspeaker_speech_eres2net_base_sv_zh-cn_3dspeaker_16k.onnx"
)


def read_wav(path):
    with wave.open(str(path), "rb") as w:
        rate = w.getframerate()
        ch = w.getnchannels()
        sw = w.getsampwidth()
        n = w.getnframes()

        if rate != 16000 or ch != 1 or sw != 2:
            raise RuntimeError(
                f"Expected mono 16-bit 16kHz WAV, "
                f"got {rate}Hz {ch}ch {sw * 8}bit"
            )

        audio = (
            np.frombuffer(w.readframes(n), dtype=np.int16)
            .astype(np.float32)
            / 32768.0
        )

    return rate, audio


def main():
    if len(sys.argv) != 3:
        print(
            f"Usage: {sys.argv[0]} INPUT.wav OUTPUT.json",
            file=sys.stderr,
        )
        raise SystemExit(2)

    src = Path(sys.argv[1])
    out = Path(sys.argv[2])

    rate, audio = read_wav(src)

    print(f"Rate: {rate} Hz")
    print(f"Duration: {len(audio) / rate:.2f}s")

    seg_cfg = sherpa_onnx.OfflineSpeakerSegmentationModelConfig(
        pyannote=sherpa_onnx.OfflineSpeakerSegmentationPyannoteModelConfig(
            model=str(SEG),
            num_threads=4,
            provider="cpu",
        )
    )

    emb_cfg = sherpa_onnx.SpeakerEmbeddingExtractorConfig(
        model=str(EMB),
        num_threads=4,
        provider="cpu",
    )

    cluster_cfg = sherpa_onnx.FastClusteringConfig(
        num_clusters=2,
        threshold=0.5,
    )

    cfg = sherpa_onnx.OfflineSpeakerDiarizationConfig(
        segmentation=seg_cfg,
        embedding=emb_cfg,
        clustering=cluster_cfg,
        min_duration_on=0.3,
        min_duration_off=0.5,
    )

    if not cfg.validate():
        raise RuntimeError(
            "Sherpa-ONNX diarization configuration is invalid"
        )

    print("Starting diarization...")

    d = sherpa_onnx.OfflineSpeakerDiarization(cfg)
    result = d.process(audio)

    segments = []

    for s in result.sort_by_start_time():
        segments.append(
            {
                "start": float(s.start),
                "end": float(s.end),
                "duration": float(s.duration),
                "speaker": int(s.speaker),
            }
        )

    payload = {
        "num_speakers": int(result.num_speakers),
        "num_segments": int(result.num_segments),
        "segments": segments,
    }

    out.parent.mkdir(parents=True, exist_ok=True)

    out.write_text(
        json.dumps(payload, indent=2),
        encoding="utf-8",
    )

    print(f"Speakers: {result.num_speakers}")
    print(f"Segments: {result.num_segments}")
    print(f"Saved: {out}")


if __name__ == "__main__":
    main()
PY

chmod 755 "$APP_DIR/diarize.py"

cat > "$APP_DIR/process_calls.py" <<'PY'
#!/usr/bin/env python3

import argparse
import datetime
import fcntl
import json
import os
import subprocess
import sys
from pathlib import Path


WHISPER = "/opt/whisper.cpp/build/bin/whisper-cli"
WHISPER_MODEL = "/opt/whisper.cpp/models/ggml-medium.en.bin"
DIARIZER = "/opt/call-pipeline/diarize.py"
PYTHON = "/opt/sherpa-onnx-env/bin/python3"


def now():
    return datetime.datetime.now(datetime.timezone.utc).isoformat()


def atomic_json(path, data):
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(
        json.dumps(data, indent=2),
        encoding="utf-8",
    )
    os.replace(tmp, path)


def load_json(p):
    return json.loads(p.read_text(encoding="utf-8"))


def update_status(p, **kw):
    d = load_json(p) if p.exists() else {}
    d.update(kw)
    atomic_json(p, d)


def discover(root):
    for p in sorted(root.rglob("*")):
        if (
            p.is_file()
            and p.suffix.lower() == ".wav"
            and not any(
                x.lower() == "optimized"
                for x in p.relative_to(root).parts
            )
        ):
            yield p


def overlap(a, b, c, d):
    return max(0.0, min(b, d) - max(a, c))


def run(cmd):
    print("$ " + " ".join(map(str, cmd)))
    subprocess.run(
        [str(x) for x in cmd],
        check=True,
    )


def convert(src, dst):
    dst.parent.mkdir(
        parents=True,
        exist_ok=True,
    )

    run(
        [
            "ffmpeg",
            "-hide_banner",
            "-loglevel",
            "error",
            "-y",
            "-i",
            src,
            "-ac",
            "1",
            "-ar",
            "16000",
            "-sample_fmt",
            "s16",
            dst,
        ]
    )


def whisper(audio, outbase):
    run(
        [
            WHISPER,
            "-m",
            WHISPER_MODEL,
            "-f",
            audio,
            "-oj",
            "-of",
            outbase,
            "-l",
            "en",
            "-t",
            str(os.cpu_count() or 1),
        ]
    )

    return Path(str(outbase) + ".json")


def merge(wj, dj):
    dia = load_json(dj).get("segments", [])
    out = []

    for s in wj.get("transcription", []):
        a = float(s["offsets"]["from"]) / 1000
        b = float(s["offsets"]["to"]) / 1000

        best = None
        score = 0

        for d in dia:
            sc = overlap(
                a,
                b,
                float(d["start"]),
                float(d["end"]),
            )

            if sc > score:
                score = sc
                best = d["speaker"]

        out.append(
            {
                "start": a,
                "end": b,
                "text": s.get("text", "").strip(),
                "speaker": (
                    int(best) + 1
                    if best is not None
                    else None
                ),
            }
        )

    return out


def whisper_only(wj):
    return [
        {
            "start": float(s["offsets"]["from"]) / 1000,
            "end": float(s["offsets"]["to"]) / 1000,
            "text": s.get("text", "").strip(),
            "speaker": None,
        }
        for s in wj.get("transcription", [])
    ]


def ts(x):
    ms = round(x * 1000)
    h, ms = divmod(ms, 3600000)
    m, ms = divmod(ms, 60000)
    s, ms = divmod(ms, 1000)

    return f"{h:02d}:{m:02d}:{s:02d}.{ms:03d}"


def write_txt(p, segs):
    with p.open("w", encoding="utf-8") as f:
        for s in segs:
            label = (
                f'Speaker {s["speaker"]} | '
                if s.get("speaker")
                else ""
            )

            f.write(
                f'[{ts(s["start"])} - {ts(s["end"])}] '
                f'{label}{s["text"]}\n'
            )


def process(src, root, outroot, use_diarization):
    rel = src.relative_to(root)
    stem = rel.with_suffix("")
    outdir = outroot / stem.parent

    outdir.mkdir(
        parents=True,
        exist_ok=True,
    )

    txt = outdir / (stem.name + ".txt")
    js = outdir / (stem.name + ".json")

    if txt.exists() and js.exists():
        print(f"SKIP complete: {src}")
        return "skip"

    work = outroot / ".work" / stem

    work.mkdir(
        parents=True,
        exist_ok=True,
    )

    lockp = work / "process.lock"
    status = work / "status.json"

    with lockp.open("w") as lf:
        try:
            fcntl.flock(
                lf,
                fcntl.LOCK_EX | fcntl.LOCK_NB,
            )
        except BlockingIOError:
            print(f"SKIP locked: {src}")
            return "skip"

        update_status(
            status,
            source=str(src),
            started=now(),
            stage="starting",
            diarization=use_diarization,
        )

        try:
            audio = work / "audio-16k-mono.wav"
            wbase = work / "whisper"
            wjson = work / "whisper.json"
            djson = work / "diarization.json"

            if not audio.exists():
                update_status(
                    status,
                    stage="convert",
                )
                convert(src, audio)

            if not wjson.exists():
                update_status(
                    status,
                    stage="whisper",
                )
                whisper(audio, wbase)

            wj = load_json(wjson)

            if use_diarization:
                if not djson.exists():
                    update_status(
                        status,
                        stage="diarization",
                    )
                    run(
                        [
                            PYTHON,
                            DIARIZER,
                            audio,
                            djson,
                        ]
                    )

                segs = merge(wj, djson)
                dmeta = load_json(djson)
            else:
                segs = whisper_only(wj)
                dmeta = None

            update_status(
                status,
                stage="writing",
            )

            write_txt(txt, segs)

            payload = {
                "source": str(src),
                "duration_seconds": (
                    segs[-1]["end"]
                    if segs
                    else 0
                ),
                "diarization": dmeta,
                "segments": segs,
            }

            atomic_json(js, payload)

            update_status(
                status,
                stage="complete",
                finished=now(),
            )

            print(f"DONE: {src}")
            return "done"

        except Exception as e:
            update_status(
                status,
                stage="error",
                error=str(e),
                finished=now(),
            )

            print(
                f"ERROR: {src}: {e}",
                file=sys.stderr,
            )

            return "error"


def main():
    ap = argparse.ArgumentParser()

    ap.add_argument(
        "--source-root",
        required=True,
    )

    ap.add_argument(
        "--output-root",
        required=True,
    )

    ap.add_argument(
        "--no-diarization",
        action="store_true",
    )

    args = ap.parse_args()

    root = (
        Path(args.source_root)
        .expanduser()
        .resolve()
    )

    out = (
        Path(args.output_root)
        .expanduser()
        .resolve()
    )

    root.mkdir(
        parents=True,
        exist_ok=True,
    )

    out.mkdir(
        parents=True,
        exist_ok=True,
    )

    counts = {
        "done": 0,
        "skip": 0,
        "error": 0,
    }

    files = list(discover(root))

    print(f"Found {len(files)} WAV file(s)")

    for p in files:
        counts[
            process(
                p,
                root,
                out,
                not args.no_diarization,
            )
        ] += 1

    print(
        f"Finished: {counts['done']} done, "
        f"{counts['skip']} skipped, "
        f"{counts['error']} errors"
    )


if __name__ == "__main__":
    main()
PY

chmod 755 "$APP_DIR/process_calls.py"

cat > "$APP_DIR/run-calls.sh" <<'EOF'
#!/usr/bin/env bash
set -euo pipefail

source /etc/call-pipeline.conf

args=(
    --source-root "$SOURCE_ROOT"
    --output-root "$OUTPUT_ROOT"
)

if [[ "${1:-}" == "--no-diarization" ]]; then
    args+=(--no-diarization)
fi

exec \
    /opt/sherpa-onnx-env/bin/python3 \
    /opt/call-pipeline/process_calls.py \
    "${args[@]}"
EOF

chmod 755 "$APP_DIR/run-calls.sh"

ORIGINAL_USER=${SUDO_USER:-}

if [[ -z "$ORIGINAL_USER" || "$ORIGINAL_USER" == root ]]; then
    DEFAULT_SOURCE=/mnt/calls
else
    DEFAULT_SOURCE=/home/$ORIGINAL_USER/calls
fi

read -r -p \
    "Source recordings directory [$DEFAULT_SOURCE]: " \
    SOURCE_ROOT

SOURCE_ROOT=${SOURCE_ROOT:-$DEFAULT_SOURCE}

read -r -p \
    "Transcript output directory [/mnt/call-transcripts]: " \
    OUTPUT_ROOT

OUTPUT_ROOT=${OUTPUT_ROOT:-/mnt/call-transcripts}

mkdir -p "$SOURCE_ROOT" "$OUTPUT_ROOT"

cat > "$CONFIG" <<EOF2
SOURCE_ROOT=$(printf '%q' "$SOURCE_ROOT")
OUTPUT_ROOT=$(printf '%q' "$OUTPUT_ROOT")
EOF2

cat > /usr/local/bin/transcribe <<'EOF2'
#!/usr/bin/env bash
set -euo pipefail

source /etc/call-pipeline.conf

read -r -p \
    "Include speaker diarization? [Y/n]: " \
    answer

case "${answer:-Y}" in
    n|N|no|NO)
        exec /opt/call-pipeline/run-calls.sh --no-diarization
        ;;
    *)
        exec /opt/call-pipeline/run-calls.sh
        ;;
esac
EOF2

chmod 755 /usr/local/bin/transcribe

"$SHERPA_ENV/bin/python3" - <<'PY2'
import sherpa_onnx
import numpy

print("sherpa-onnx: import OK")
print("numpy:", numpy.__version__)
PY2

python3 -m py_compile \
    "$APP_DIR/process_calls.py" \
    "$APP_DIR/diarize.py"

"$WHISPER_BIN" --help >/dev/null

echo
echo 'Installation complete.'
echo "Run: transcribe"
echo "Config: $CONFIG"
echo "Pipeline: $APP_DIR"
echo "GPU acceleration: disabled in this Pi 5 installer"
1 Like