src/edurag_math_bot/speech_to_text.py

from __future__ import annotations

import platform
import shutil
import subprocess
import tempfile
from pathlib import Path


TRANSCRIBE_SWIFT = """
import Foundation
import Speech
import AVFoundation

func fail(_ message: String) -> Never {
    fputs(message + "\\n", stderr)
    fflush(stderr)
    exit(1)
}

guard CommandLine.arguments.count > 1 else {
    fail("Missing audio path.")
}

let audioURL = URL(fileURLWithPath: CommandLine.arguments[1])
let localeIdentifier = CommandLine.arguments.count > 2 ? CommandLine.arguments[2] : "en-IN"

let semaphore = DispatchSemaphore(value: 0)

var finalText: String?
var finalError: String?
var didFinish = false

func complete() {
    guard !didFinish else {
        return
    }
    didFinish = true
    semaphore.signal()
}

func startRecognition() {
    guard let recognizer = SFSpeechRecognizer(locale: Locale(identifier: localeIdentifier)) else {
        finalError = "Speech recognizer unavailable for selected language."
        complete()
        return
    }

    guard recognizer.isAvailable else {
        finalError = "Speech recognizer is currently unavailable."
        complete()
        return
    }

    let request = SFSpeechURLRecognitionRequest(url: audioURL)
    request.shouldReportPartialResults = false

    if #available(macOS 13.0, *) {
        request.addsPunctuation = true
    }

    recognizer.recognitionTask(with: request) { result, error in
        if let error {
            finalError = error.localizedDescription
            complete()
            return
        }

        guard let result else {
            return
        }

        if result.isFinal {
            finalText = result.bestTranscription.formattedString
            complete()
        }
    }
}

SFSpeechRecognizer.requestAuthorization { status in
    guard status == .authorized else {
        finalError = "Speech recognition permission denied."
        complete()
        return
    }

    startRecognition()
}

if semaphore.wait(timeout: .now() + 90) == .timedOut {
    fail("Speech transcription timed out.")
}

if let finalError {
    fail(finalError)
}

guard let finalText, !finalText.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty else {
    fail("No speech detected.")
}

print(finalText)
"""

INFO_PLIST = """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>NSSpeechRecognitionUsageDescription</key>
    <string>MathSutra needs speech recognition to transcribe your voice questions.</string>
    <key>NSMicrophoneUsageDescription</key>
    <string>MathSutra needs microphone access to transcribe your voice questions.</string>
</dict>
</plist>
"""


def check_onment() -> None:
    if platform.system() != "Darwin":
        raise RuntimeError(
            "Voice transcription currently supports only macOS."
        )

    if shutil.which("swiftc") is None:
        raise RuntimeError(
            "Swift compiler is not installed. Install Xcode Command Line Tools first."
        )


def transcribe_audio_file(
    audio_path: Path,
    timeout: int = 120,
    locale_identifier: str = "en-IN",
) -> str:
    check_onment()

    with tempfile.TemporaryDirectory() as tmp_dir:
        tmp_path = Path(tmp_dir)
        script_path = tmp_path / "transcribe.swift"
        script_path.write_text(TRANSCRIBE_SWIFT, encoding="utf-8")

        plist_path = tmp_path / "Info.plist"
        plist_path.write_text(INFO_PLIST, encoding="utf-8")

        binary_path = tmp_path / "transcribe"

        compile_cmd = [
            "swiftc", str(script_path),
            "-o", str(binary_path),
            "-Xlinker", "-sectcreate",
            "-Xlinker", "__TEXT",
            "-Xlinker", "__info_plist",
            "-Xlinker", str(plist_path),
        ]
        compile_result = subprocess.run(compile_cmd, capture_output=True, text=True)
        if compile_result.returncode != 0:
            raise RuntimeError(f"Failed to compile transcription script: {compile_result.stderr}")

        command = [
            str(binary_path),
            str(audio_path),
            locale_identifier,
        ]

        result = subprocess.run(
            command,
            capture_output=True,
            text=True,
            timeout=timeout,
        )

    if result.returncode != 0:
        crash_reason = " (Process killed by macOS. Go to System Settings > Privacy & Security and allow Microphone / Speech Recognition)" if result.returncode < 0 else ""
        error_message = (
            result.stderr.strip()
            or result.stdout.strip()
            or f"Unknown transcription error{crash_reason}"
        )

        if "permission" in error_message.lower() or "authorization" in error_message.lower():
            error_message += (
                "\n\nEnable permissions from:\n"
                "System Settings > Privacy & Security > Speech Recognition"
            )

        raise RuntimeError(error_message)

    transcript = result.stdout.strip()

    if not transcript:
        raise RuntimeError("No transcript generated from audio.")

    return transcript



def transcribe_audio_bytes(
    audio_bytes: bytes,
    suffix: str = ".wav",
    timeout: int = 120,
    locale_identifier: str = "en-IN",
) -> str:
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_file:
        tmp_file.write(audio_bytes)
        audio_path = Path(tmp_file.name)

    try:
        return transcribe_audio_file(
            audio_path=audio_path,
            timeout=timeout,
            locale_identifier=locale_identifier,
        )
    finally:
        audio_path.unlink(missing_ok=True)