Spaces:
Sleeping
Sleeping
File size: 5,627 Bytes
a86cdfa |
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 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 |
"""
Caching system for generated audio.
Supports local and Hugging Face Hub storage.
"""
import hashlib
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
from loguru import logger
@dataclass
class CacheConfig:
"""Configuration for audio caching."""
enabled: bool = True
local_cache_dir: Optional[str] = None # Local cache directory
hf_repo_id: Optional[str] = None # Hugging Face Hub repo for remote cache
max_duration_seconds: float = 30.0 # Only cache audio shorter than this
class AudioCache:
"""
Cache for generated TTS audio.
Supports both local filesystem and Hugging Face Hub storage.
"""
def __init__(self, config: Optional[CacheConfig] = None):
self.config = config or CacheConfig()
self._hf_fs = None
def _get_cache_key(self, text: str, voice_id: str, backend: str) -> str:
"""Generate a unique cache key for the given parameters."""
content = f"{backend}:{voice_id}:{text}"
return hashlib.md5(content.encode()).hexdigest()
def _get_hf_fs(self):
"""Get HuggingFace filesystem (lazy initialization)."""
if self._hf_fs is None and self.config.hf_repo_id:
try:
from huggingface_hub import HfFileSystem
self._hf_fs = HfFileSystem(token=os.environ.get("HF_TOKEN"))
except Exception as e:
logger.warning(f"Could not initialize HF filesystem: {e}")
return self._hf_fs
def get(self, text: str, voice_id: str, backend: str) -> Optional[bytes]:
"""
Retrieve cached audio if it exists.
Args:
text: Original text that was synthesized
voice_id: Voice identifier used
backend: Backend name used for synthesis
Returns:
Cached audio bytes or None if not found
"""
if not self.config.enabled:
return None
cache_key = self._get_cache_key(text, voice_id, backend)
# Try local cache first
if self.config.local_cache_dir:
local_path = Path(self.config.local_cache_dir) / f"{cache_key}.mp3"
if local_path.exists():
logger.debug(f"Cache hit (local): {cache_key}")
return local_path.read_bytes()
# Try HF Hub cache
if self.config.hf_repo_id:
fs = self._get_hf_fs()
if fs:
hf_path = f"{self.config.hf_repo_id}/{voice_id}/{cache_key}.mp3"
try:
if fs.exists(hf_path):
with fs.open(hf_path, "rb") as f:
logger.debug(f"Cache hit (HF Hub): {cache_key}")
return f.read()
except Exception as e:
logger.debug(f"HF cache lookup failed: {e}")
return None
def set(
self,
text: str,
voice_id: str,
backend: str,
audio_data: bytes,
duration_seconds: Optional[float] = None,
) -> bool:
"""
Store audio in cache.
Args:
text: Original text that was synthesized
voice_id: Voice identifier used
backend: Backend name used for synthesis
audio_data: Audio bytes to cache
duration_seconds: Duration of the audio (for max duration check)
Returns:
True if cached successfully, False otherwise
"""
if not self.config.enabled:
return False
# Check duration limit
if duration_seconds and duration_seconds > self.config.max_duration_seconds:
logger.debug(
f"Audio too long to cache: {duration_seconds}s > {self.config.max_duration_seconds}s"
)
return False
cache_key = self._get_cache_key(text, voice_id, backend)
success = False
# Save to local cache
if self.config.local_cache_dir:
try:
cache_dir = Path(self.config.local_cache_dir)
cache_dir.mkdir(parents=True, exist_ok=True)
local_path = cache_dir / f"{cache_key}.mp3"
local_path.write_bytes(audio_data)
logger.debug(f"Cached locally: {cache_key}")
success = True
except Exception as e:
logger.warning(f"Failed to cache locally: {e}")
# Save to HF Hub
if self.config.hf_repo_id:
fs = self._get_hf_fs()
if fs:
try:
voice_dir = f"{self.config.hf_repo_id}/{voice_id}"
if not fs.exists(voice_dir):
fs.makedirs(voice_dir, exist_ok=True)
hf_path = f"{voice_dir}/{cache_key}.mp3"
with fs.open(hf_path, "wb") as f:
f.write(audio_data)
logger.debug(f"Cached to HF Hub: {cache_key}")
success = True
except Exception as e:
logger.warning(f"Failed to cache to HF Hub: {e}")
return success
def clear_local(self) -> int:
"""Clear local cache. Returns number of files deleted."""
if not self.config.local_cache_dir:
return 0
cache_dir = Path(self.config.local_cache_dir)
if not cache_dir.exists():
return 0
count = 0
for file in cache_dir.glob("*.mp3"):
file.unlink()
count += 1
logger.info(f"Cleared {count} files from local cache")
return count
|