Spaces:
Running
Running
File size: 2,365 Bytes
0157ac7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """Platform-neutral voice note helpers."""
from __future__ import annotations
import asyncio
from pathlib import Path
class PendingVoiceRegistry:
"""Track voice notes that are still waiting on transcription."""
def __init__(self) -> None:
self._pending: dict[tuple[str, str], tuple[str, str]] = {}
self._lock = asyncio.Lock()
async def register(
self, chat_id: str, voice_msg_id: str, status_msg_id: str
) -> None:
async with self._lock:
entry = (voice_msg_id, status_msg_id)
self._pending[(chat_id, voice_msg_id)] = entry
self._pending[(chat_id, status_msg_id)] = entry
async def cancel(self, chat_id: str, reply_id: str) -> tuple[str, str] | None:
async with self._lock:
entry = self._pending.pop((chat_id, reply_id), None)
if entry is None:
return None
voice_msg_id, status_msg_id = entry
self._pending.pop((chat_id, voice_msg_id), None)
self._pending.pop((chat_id, status_msg_id), None)
return entry
async def is_pending(self, chat_id: str, voice_msg_id: str) -> bool:
async with self._lock:
return (chat_id, voice_msg_id) in self._pending
async def complete(
self, chat_id: str, voice_msg_id: str, status_msg_id: str
) -> None:
async with self._lock:
self._pending.pop((chat_id, voice_msg_id), None)
self._pending.pop((chat_id, status_msg_id), None)
class VoiceTranscriptionService:
"""Run configured transcription backends off the event loop."""
def __init__(
self,
*,
hf_token: str = "",
nvidia_nim_api_key: str = "",
) -> None:
self._hf_token = hf_token
self._nvidia_nim_api_key = nvidia_nim_api_key
async def transcribe(
self,
file_path: Path,
mime_type: str,
*,
whisper_model: str,
whisper_device: str,
) -> str:
from .transcription import transcribe_audio
return await asyncio.to_thread(
transcribe_audio,
file_path,
mime_type,
whisper_model=whisper_model,
whisper_device=whisper_device,
hf_token=self._hf_token,
nvidia_nim_api_key=self._nvidia_nim_api_key,
)
|