Adolfo Reyna
114 lines
4.9 KiB
Python
114 lines
4.9 KiB
Python
"""Direct Apple SpeechTranscriber file execution for Reyna CLI."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
_SWIFT_SOURCE = r'''import Speech
|
|
import AVFoundation
|
|
import Foundation
|
|
|
|
@main
|
|
struct ReynaTranscriber {
|
|
static func main() async {
|
|
let args = CommandLine.arguments
|
|
guard args.count >= 2 else { fail("audio path is required", code: 2) }
|
|
let audioPath = args[1]
|
|
let localeID = args.count >= 3 ? args[2] : "en-US"
|
|
guard FileManager.default.fileExists(atPath: audioPath) else { fail("audio file not found", code: 3) }
|
|
guard SpeechTranscriber.isAvailable else { fail("SpeechTranscriber is unavailable", code: 4) }
|
|
let requested = Locale(identifier: localeID)
|
|
guard let locale = await SpeechTranscriber.supportedLocale(equivalentTo: requested) else { fail("unsupported locale", code: 5) }
|
|
let transcriber = SpeechTranscriber(locale: locale, preset: .transcription)
|
|
do {
|
|
if let request = try await AssetInventory.assetInstallationRequest(supporting: [transcriber]) {
|
|
try await request.downloadAndInstall()
|
|
}
|
|
let file = try AVAudioFile(forReading: URL(fileURLWithPath: audioPath))
|
|
let analyzer = try await SpeechAnalyzer(inputAudioFile: file, modules: [transcriber], finishAfterFile: true)
|
|
_ = analyzer
|
|
var segments: [String] = []
|
|
for try await result in transcriber.results {
|
|
let text = String(result.text.characters).trimmingCharacters(in: .whitespacesAndNewlines)
|
|
if !text.isEmpty { segments.append(text) }
|
|
}
|
|
let payload: [String: Any] = [
|
|
"ok": true,
|
|
"engine": "SpeechAnalyzer+SpeechTranscriber",
|
|
"locale": locale.identifier,
|
|
"requestedLocale": localeID,
|
|
"transcript": segments.joined(separator: " "),
|
|
"segments": segments,
|
|
"audioPath": audioPath,
|
|
"isAvailable": SpeechTranscriber.isAvailable
|
|
]
|
|
let data = try JSONSerialization.data(withJSONObject: payload, options: [.sortedKeys])
|
|
FileHandle.standardOutput.write(data)
|
|
} catch {
|
|
fail(String(describing: error), code: 6)
|
|
}
|
|
}
|
|
static func fail(_ message: String, code: Int32) -> Never {
|
|
let payload: [String: Any] = ["ok": false, "error": message]
|
|
if let data = try? JSONSerialization.data(withJSONObject: payload, options: []) { FileHandle.standardOutput.write(data) }
|
|
exit(code)
|
|
}
|
|
}
|
|
'''
|
|
|
|
|
|
class SpeechExecutionError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _cache_dir() -> Path:
|
|
return Path.home() / "Library" / "Application Support" / "reyna-cli" / "speech"
|
|
|
|
|
|
def _binary_path() -> Path:
|
|
digest = hashlib.sha256(_SWIFT_SOURCE.encode()).hexdigest()[:16]
|
|
return _cache_dir() / f"transcriber-{digest}"
|
|
|
|
|
|
def _ensure_binary() -> Path:
|
|
cached = _binary_path()
|
|
if cached.exists() and os.access(cached, os.X_OK):
|
|
return cached
|
|
swiftc = shutil.which("swiftc") or "/usr/bin/swiftc"
|
|
if not Path(swiftc).exists():
|
|
raise SpeechExecutionError("swiftc is required for SpeechTranscriber file execution")
|
|
_cache_dir().mkdir(parents=True, exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix="reyna-speech-build-") as tmp:
|
|
source = Path(tmp) / "Transcriber.swift"
|
|
binary = Path(tmp) / "transcriber"
|
|
source.write_text(_SWIFT_SOURCE, encoding="utf-8")
|
|
result = subprocess.run([swiftc, "-O", "-parse-as-library", str(source), "-o", str(binary), "-framework", "AVFoundation", "-framework", "Speech"], capture_output=True, text=True, timeout=120, check=False)
|
|
if result.returncode != 0:
|
|
raise SpeechExecutionError(f"swiftc failed: {(result.stderr or result.stdout).strip()[:2000]}")
|
|
shutil.copyfile(binary, cached)
|
|
cached.chmod(0o700)
|
|
return cached
|
|
|
|
|
|
def transcribe_file(audio_path: Path, *, locale: str = "en-US", timeout: int = 300) -> Dict[str, Any]:
|
|
audio = audio_path.expanduser().resolve()
|
|
if not audio.is_file():
|
|
raise SpeechExecutionError(f"audio file not found: {audio}")
|
|
binary = _ensure_binary()
|
|
result = subprocess.run([str(binary), str(audio), locale], capture_output=True, text=True, timeout=timeout, check=False)
|
|
raw = (result.stdout or "").strip()
|
|
try:
|
|
payload = json.loads(raw) if raw else {}
|
|
except json.JSONDecodeError as exc:
|
|
raise SpeechExecutionError(f"SpeechTranscriber returned invalid JSON: {raw[:500]}") from exc
|
|
if result.returncode != 0 or not payload.get("ok"):
|
|
raise SpeechExecutionError(str(payload.get("error") or result.stderr.strip() or "SpeechTranscriber failed"))
|
|
return {**payload, "source": "reyna_cli_direct", "binary": str(binary)}
|