nifty-coder/stemsplit-backend
0
1"""2Cache Manager implementation for the voice control optimization system.3 4This module provides intelligent caching of transcription results using audio5fingerprinting to avoid redundant API calls and improve response times.6"""7 8import asyncio9import hashlib10import logging11import time12from collections import OrderedDict13from typing import Dict, Optional, Any, Set, List, Callable14from datetime import datetime, timedelta15from dataclasses import dataclass, field16 17from .interfaces import CacheManagerInterface18from .models import TranscriptionResult, CacheStats19from .exceptions import CacheError20 21 22logger = logging.getLogger(__name__)23 24 25@dataclass26class CacheEntry:27 """Represents a cached transcription result with metadata."""28 result: TranscriptionResult29 created_at: float30 last_accessed: float31 ttl: int32 access_count: int = 033 size_bytes: int = 034 35 def __post_init__(self):36 """Calculate entry size after initialization."""37 if self.size_bytes == 0:38 self.size_bytes = self._calculate_size()39 40 def _calculate_size(self) -> int:41 """Calculate approximate memory size of the cache entry."""42 # Base size for the entry object43 size = 200 # Approximate overhead44 45 # Size of transcription result46 size += len(self.result.text.encode('utf-8'))47 size += len(self.result.provider.encode('utf-8'))48 size += len(self.result.language.encode('utf-8'))49 50 # Size of alternatives51 for alt in self.result.alternatives:52 size += len(alt.encode('utf-8'))53 54 # Size of word timestamps if present55 if self.result.word_timestamps:56 for word_ts in self.result.word_timestamps:57 size += len(word_ts.word.encode('utf-8')) + 32 # word + timestamps58 59 return size60 61 @property62 def is_expired(self) -> bool:63 """Check if the cache entry has expired."""64 return time.time() - self.created_at > self.ttl65 66 def touch(self) -> None:67 """Update last accessed time and increment access count."""68 self.last_accessed = time.time()69 self.access_count += 170 71 72class CacheManager(CacheManagerInterface):73 """74 Manages caching of transcription results with LRU eviction and TTL expiration.75 76 Features:77 - Audio fingerprint-based cache keys78 - LRU (Least Recently Used) eviction policy79 - TTL (Time To Live) based expiration80 - Configurable size limits81 - Cache statistics tracking82 """83 84 def __init__(85 self,86 max_size_mb: int = 100,87 default_ttl: int = 3600,88 cleanup_interval: int = 600,89 enable_stats: bool = True90 ):91 """92 Initialize the cache manager.93 94 Args:95 max_size_mb: Maximum cache size in megabytes96 default_ttl: Default time-to-live in seconds97 cleanup_interval: Interval for cleanup tasks in seconds98 enable_stats: Whether to track cache statistics99 """100 self.max_size_bytes = max_size_mb * 1024 * 1024101 self.default_ttl = default_ttl102 self.cleanup_interval = cleanup_interval103 self.enable_stats = enable_stats104 105 # Cache storage using OrderedDict for LRU behavior106 self._cache: OrderedDict[str, CacheEntry] = OrderedDict()107 108 # Cache statistics109 self._stats = {110 "total_requests": 0,111 "cache_hits": 0,112 "cache_misses": 0,113 "evictions": 0,114 "expirations": 0,115 "current_size_bytes": 0,116 "entry_count": 0,117 # Cache hit optimization metrics118 "cache_hit_times": [], # Track cache hit response times119 "cache_miss_times": [], # Track cache miss response times120 "sequential_hits": 0, # Track consecutive cache hits121 "sequential_misses": 0, # Track consecutive cache misses122 "hot_entries": {}, # Track frequently accessed entries123 "last_operation": None, # Track last operation type ('hit' or 'miss')124 }125 126 # Cleanup task127 self._cleanup_task: Optional[asyncio.Task] = None128 self._running = False129 130 # Lock for thread safety131 self._lock = asyncio.Lock()132 133 logger.info(f"CacheManager initialized with max_size={max_size_mb}MB, ttl={default_ttl}s")134 135 async def start(self) -> None:136 """Start the cache manager and background cleanup task."""137 if self._running:138 return139 140 self._running = True141 self._cleanup_task = asyncio.create_task(self._cleanup_loop())142 logger.info("CacheManager started")143 144 async def stop(self) -> None:145 """Stop the cache manager and cleanup task."""146 if not self._running:147 return148 149 self._running = False150 151 if self._cleanup_task:152 self._cleanup_task.cancel()153 try:154 await self._cleanup_task155 except asyncio.CancelledError:156 pass157 158 logger.info("CacheManager stopped")159 160 async def get_cached_transcription(161 self, 162 audio_fingerprint: str163 ) -> Optional[TranscriptionResult]:164 """165 Retrieve cached transcription result with optimized lookup.166 167 This method implements cache hit optimization by:168 1. Fast lookup without API calls for cached results169 2. Tracking cache hit performance metrics170 3. Optimizing frequently accessed entries171 4. Monitoring cache hit patterns172 173 Args:174 audio_fingerprint: Unique fingerprint of audio data175 176 Returns:177 Cached TranscriptionResult or None if not found178 """179 start_time = time.time()180 181 async with self._lock:182 if self.enable_stats:183 self._stats["total_requests"] += 1184 185 entry = self._cache.get(audio_fingerprint)186 187 if entry is None:188 # Cache miss - track timing and patterns189 miss_time = time.time() - start_time190 if self.enable_stats:191 self._stats["cache_misses"] += 1192 self._stats["cache_miss_times"].append(miss_time)193 # Keep only last 100 measurements for efficiency194 if len(self._stats["cache_miss_times"]) > 100:195 self._stats["cache_miss_times"] = self._stats["cache_miss_times"][-100:]196 197 # Track sequential misses198 if self._stats["last_operation"] == "miss":199 self._stats["sequential_misses"] += 1200 else:201 self._stats["sequential_misses"] = 1202 self._stats["last_operation"] = "miss"203 204 logger.debug(f"Cache miss for fingerprint: {audio_fingerprint[:16]}... "205 f"(lookup time: {miss_time*1000:.2f}ms)")206 return None207 208 # Check if entry has expired209 if entry.is_expired:210 miss_time = time.time() - start_time211 logger.debug(f"Cache entry expired for fingerprint: {audio_fingerprint[:16]}... "212 f"(lookup time: {miss_time*1000:.2f}ms)")213 await self._remove_entry(audio_fingerprint)214 if self.enable_stats:215 self._stats["cache_misses"] += 1216 self._stats["expirations"] += 1217 self._stats["cache_miss_times"].append(miss_time)218 if len(self._stats["cache_miss_times"]) > 100:219 self._stats["cache_miss_times"] = self._stats["cache_miss_times"][-100:]220 221 # Track sequential misses222 if self._stats["last_operation"] == "miss":223 self._stats["sequential_misses"] += 1224 else:225 self._stats["sequential_misses"] = 1226 self._stats["last_operation"] = "miss"227 return None228 229 # Cache hit - optimize and track performance230 hit_time = time.time() - start_time231 232 # Update access information and move to end (most recently used)233 entry.touch()234 self._cache.move_to_end(audio_fingerprint)235 236 if self.enable_stats:237 self._stats["cache_hits"] += 1238 self._stats["cache_hit_times"].append(hit_time)239 # Keep only last 100 measurements for efficiency240 if len(self._stats["cache_hit_times"]) > 100:241 self._stats["cache_hit_times"] = self._stats["cache_hit_times"][-100:]242 243 # Track hot entries (frequently accessed)244 if audio_fingerprint not in self._stats["hot_entries"]:245 self._stats["hot_entries"][audio_fingerprint] = 0246 self._stats["hot_entries"][audio_fingerprint] += 1247 248 # Keep only top 50 hot entries to prevent memory growth249 if len(self._stats["hot_entries"]) > 50:250 # Keep only the most frequently accessed entries251 sorted_entries = sorted(252 self._stats["hot_entries"].items(), 253 key=lambda x: x[1], 254 reverse=True255 )256 self._stats["hot_entries"] = dict(sorted_entries[:50])257 258 # Track sequential hits259 if self._stats["last_operation"] == "hit":260 self._stats["sequential_hits"] += 1261 else:262 self._stats["sequential_hits"] = 1263 self._stats["last_operation"] = "hit"264 265 logger.debug(f"Cache hit for fingerprint: {audio_fingerprint[:16]}... "266 f"(access count: {entry.access_count}, lookup time: {hit_time*1000:.2f}ms)")267 return entry.result268 269 async def get_cached_transcription_batch(270 self, 271 audio_fingerprints: List[str]272 ) -> Dict[str, Optional[TranscriptionResult]]:273 """274 Retrieve multiple cached transcription results in a single operation.275 276 Args:277 audio_fingerprints: List of unique fingerprints of audio data278 279 Returns:280 Dictionary mapping fingerprints to cached results (None if not found)281 """282 results = {}283 284 async with self._lock:285 for fingerprint in audio_fingerprints:286 if self.enable_stats:287 self._stats["total_requests"] += 1288 289 entry = self._cache.get(fingerprint)290 291 if entry is None:292 if self.enable_stats:293 self._stats["cache_misses"] += 1294 results[fingerprint] = None295 continue296 297 # Check if entry has expired298 if entry.is_expired:299 await self._remove_entry(fingerprint)300 if self.enable_stats:301 self._stats["cache_misses"] += 1302 self._stats["expirations"] += 1303 results[fingerprint] = None304 continue305 306 # Update access information and move to end (most recently used)307 entry.touch()308 self._cache.move_to_end(fingerprint)309 310 if self.enable_stats:311 self._stats["cache_hits"] += 1312 313 results[fingerprint] = entry.result314 315 hit_count = sum(1 for result in results.values() if result is not None)316 logger.debug(f"Batch cache lookup: {hit_count}/{len(audio_fingerprints)} hits")317 318 return results319 320 async def cache_transcription(321 self, 322 audio_fingerprint: str, 323 result: TranscriptionResult,324 ttl: Optional[int] = None325 ) -> bool:326 """327 Cache a transcription result.328 329 Args:330 audio_fingerprint: Unique fingerprint of audio data331 result: TranscriptionResult to cache332 ttl: Time to live in seconds (uses default if None)333 334 Returns:335 True if caching was successful, False otherwise336 """337 if ttl is None:338 ttl = self.default_ttl339 340 try:341 async with self._lock:342 # Create cache entry343 entry = CacheEntry(344 result=result,345 created_at=time.time(),346 last_accessed=time.time(),347 ttl=ttl348 )349 350 # Check if we need to make space351 await self._ensure_space(entry.size_bytes)352 353 # Add to cache354 self._cache[audio_fingerprint] = entry355 self._stats["current_size_bytes"] += entry.size_bytes356 self._stats["entry_count"] += 1357 358 logger.debug(f"Cached transcription for fingerprint: {audio_fingerprint[:16]}... "359 f"(size: {entry.size_bytes} bytes, ttl: {ttl}s)")360 361 return True362 363 except Exception as e:364 logger.error(f"Failed to cache transcription: {e}")365 return False366 367 async def generate_audio_fingerprint(self, audio_data: bytes) -> str:368 """369 Generate a unique fingerprint for audio data.370 371 Uses SHA-256 hashing of the audio content to create a consistent372 fingerprint that will be identical for the same audio data.373 374 Args:375 audio_data: Raw audio bytes376 377 Returns:378 Unique fingerprint string379 """380 try:381 # Use SHA-256 for consistent, collision-resistant hashing382 hasher = hashlib.sha256()383 384 # Hash the audio data in chunks to handle large files efficiently385 chunk_size = 8192386 for i in range(0, len(audio_data), chunk_size):387 chunk = audio_data[i:i + chunk_size]388 hasher.update(chunk)389 390 # Generate hexadecimal fingerprint391 fingerprint = hasher.hexdigest()392 393 logger.debug(f"Generated audio fingerprint: {fingerprint[:16]}... "394 f"(audio size: {len(audio_data)} bytes)")395 396 return fingerprint397 398 except Exception as e:399 logger.error(f"Failed to generate audio fingerprint: {e}")400 raise CacheError("fingerprint_generation", f"Failed to generate fingerprint: {e}")401 402 async def get_cache_stats(self) -> CacheStats:403 """404 Get cache performance statistics with hit optimization metrics.405 406 Returns:407 CacheStats with hit rates and usage information408 """409 async with self._lock:410 return CacheStats(411 total_requests=self._stats["total_requests"],412 cache_hits=self._stats["cache_hits"],413 cache_misses=self._stats["cache_misses"],414 hit_rate=0.0, # Will be calculated in __post_init__415 total_size_bytes=self._stats["current_size_bytes"],416 entry_count=self._stats["entry_count"],417 evictions=self._stats["evictions"]418 )419 420 async def get_detailed_cache_stats(self) -> Dict[str, Any]:421 """422 Get detailed cache performance statistics for optimization analysis.423 424 Returns:425 Dictionary with comprehensive cache metrics including hit optimization data426 """427 async with self._lock:428 stats = await self.get_cache_stats()429 430 # Calculate additional optimization metrics431 avg_entry_size = (432 stats.total_size_bytes / stats.entry_count 433 if stats.entry_count > 0 else 0434 )435 436 cache_efficiency = (437 stats.hit_rate * (1 - (stats.evictions / max(stats.total_requests, 1)))438 if stats.total_requests > 0 else 0.0439 )440 441 # Calculate cache hit optimization metrics442 hit_optimization_stats = self._calculate_hit_optimization_metrics()443 444 # Analyze access patterns445 access_pattern_stats = self._analyze_access_patterns()446 447 return {448 "basic_stats": {449 "total_requests": stats.total_requests,450 "cache_hits": stats.cache_hits,451 "cache_misses": stats.cache_misses,452 "hit_rate": stats.hit_rate,453 "miss_rate": stats.miss_rate,454 "evictions": stats.evictions455 },456 "size_stats": {457 "total_size_bytes": stats.total_size_bytes,458 "total_size_mb": stats.total_size_bytes / (1024 * 1024),459 "entry_count": stats.entry_count,460 "avg_entry_size_bytes": avg_entry_size,461 "max_size_bytes": self.max_size_bytes,462 "utilization_rate": stats.total_size_bytes / self.max_size_bytes463 },464 "performance_stats": {465 "cache_efficiency": cache_efficiency,466 "eviction_rate": stats.evictions / max(stats.total_requests, 1),467 "access_patterns": access_pattern_stats468 },469 "hit_optimization_stats": hit_optimization_stats,470 "optimization_recommendations": self._get_optimization_recommendations(stats)471 }472 473 def _calculate_hit_optimization_metrics(self) -> Dict[str, Any]:474 """475 Calculate cache hit optimization metrics.476 477 Returns:478 Dictionary with hit optimization statistics479 """480 hit_times = self._stats.get("cache_hit_times", [])481 miss_times = self._stats.get("cache_miss_times", [])482 483 # Calculate timing statistics484 avg_hit_time = sum(hit_times) / len(hit_times) if hit_times else 0.0485 avg_miss_time = sum(miss_times) / len(miss_times) if miss_times else 0.0486 487 # Calculate hit time improvement (how much faster hits are than misses)488 hit_time_improvement = (489 ((avg_miss_time - avg_hit_time) / avg_miss_time * 100) 490 if avg_miss_time > 0 else 0.0491 )492 493 # Calculate sequential operation patterns494 sequential_hits = self._stats.get("sequential_hits", 0)495 sequential_misses = self._stats.get("sequential_misses", 0)496 497 # Calculate hot entry statistics498 hot_entries = self._stats.get("hot_entries", {})499 hot_entry_count = len(hot_entries)500 total_hot_accesses = sum(hot_entries.values()) if hot_entries else 0501 avg_hot_entry_accesses = (502 total_hot_accesses / hot_entry_count if hot_entry_count > 0 else 0.0503 )504 505 # Calculate cache hit efficiency score (0-100)506 hit_rate = (507 self._stats["cache_hits"] / max(self._stats["total_requests"], 1) 508 if self._stats["total_requests"] > 0 else 0.0509 )510 511 timing_efficiency = min(hit_time_improvement / 50.0, 1.0) if hit_time_improvement > 0 else 0.0512 pattern_efficiency = min(sequential_hits / max(sequential_hits + sequential_misses, 1), 1.0)513 514 overall_efficiency = (hit_rate * 0.5 + timing_efficiency * 0.3 + pattern_efficiency * 0.2) * 100515 516 return {517 "timing_metrics": {518 "avg_hit_time_ms": avg_hit_time * 1000,519 "avg_miss_time_ms": avg_miss_time * 1000,520 "hit_time_improvement_percent": hit_time_improvement,521 "hit_samples": len(hit_times),522 "miss_samples": len(miss_times)523 },524 "pattern_metrics": {525 "sequential_hits": sequential_hits,526 "sequential_misses": sequential_misses,527 "hit_streak_ratio": sequential_hits / max(sequential_hits + sequential_misses, 1)528 },529 "hot_entry_metrics": {530 "hot_entry_count": hot_entry_count,531 "total_hot_accesses": total_hot_accesses,532 "avg_accesses_per_hot_entry": avg_hot_entry_accesses,533 "hot_entry_hit_contribution": (534 total_hot_accesses / max(self._stats["cache_hits"], 1) * 100535 if self._stats["cache_hits"] > 0 else 0.0536 )537 },538 "efficiency_score": {539 "overall_efficiency_percent": overall_efficiency,540 "hit_rate_component": hit_rate * 50, # 50% weight541 "timing_component": timing_efficiency * 30, # 30% weight542 "pattern_component": pattern_efficiency * 20 # 20% weight543 }544 }545 546 def _analyze_access_patterns(self) -> Dict[str, Any]:547 """548 Analyze cache access patterns for optimization insights.549 550 Returns:551 Dictionary with access pattern analysis552 """553 if not self._cache:554 return {555 "most_accessed_entries": 0,556 "least_accessed_entries": 0,557 "avg_access_count": 0.0,558 "access_distribution": "empty"559 }560 561 access_counts = [entry.access_count for entry in self._cache.values()]562 563 return {564 "most_accessed_entries": max(access_counts) if access_counts else 0,565 "least_accessed_entries": min(access_counts) if access_counts else 0,566 "avg_access_count": sum(access_counts) / len(access_counts) if access_counts else 0.0,567 "access_distribution": "varied" if len(set(access_counts)) > 1 else "uniform"568 }569 570 def _get_optimization_recommendations(self, stats: CacheStats) -> List[str]:571 """572 Generate optimization recommendations based on cache statistics.573 574 Args:575 stats: Current cache statistics576 577 Returns:578 List of optimization recommendations579 """580 recommendations = []581 582 # Hit rate optimization583 if stats.hit_rate < 0.5:584 recommendations.append("Consider increasing cache size - low hit rate detected")585 elif stats.hit_rate < 0.7:586 recommendations.append("Cache hit rate could be improved - consider TTL optimization")587 588 # Eviction optimization589 if stats.evictions > stats.cache_hits * 0.1:590 recommendations.append("High eviction rate - consider increasing cache size")591 592 # Size optimization593 utilization = stats.total_size_bytes / self.max_size_bytes594 if utilization > 0.9:595 recommendations.append("Cache near capacity - consider size increase or TTL reduction")596 elif utilization < 0.3:597 recommendations.append("Cache underutilized - consider size reduction for memory efficiency")598 599 # Performance optimization600 if stats.entry_count > 1000 and stats.hit_rate > 0.8:601 recommendations.append("High performance cache - consider implementing cache warming")602 603 # Cache hit optimization specific recommendations604 hit_times = self._stats.get("cache_hit_times", [])605 miss_times = self._stats.get("cache_miss_times", [])606 607 if hit_times and miss_times:608 avg_hit_time = sum(hit_times) / len(hit_times)609 avg_miss_time = sum(miss_times) / len(miss_times)610 611 if avg_hit_time > 0.001: # 1ms threshold612 recommendations.append("Cache hit times are high - consider cache structure optimization")613 614 if avg_hit_time > avg_miss_time * 0.5:615 recommendations.append("Cache hits not significantly faster than misses - investigate cache overhead")616 617 # Sequential pattern optimization618 sequential_hits = self._stats.get("sequential_hits", 0)619 sequential_misses = self._stats.get("sequential_misses", 0)620 621 if sequential_misses > sequential_hits * 2:622 recommendations.append("High sequential miss pattern - consider cache warming or preloading")623 624 # Hot entry optimization625 hot_entries = self._stats.get("hot_entries", {})626 if len(hot_entries) > 0:627 total_hot_accesses = sum(hot_entries.values())628 if total_hot_accesses > stats.cache_hits * 0.8:629 recommendations.append("High hot entry concentration - consider dedicated hot cache tier")630 631 if not recommendations:632 recommendations.append("Cache performance is optimal")633 634 return recommendations635 636 async def warm_cache(637 self, 638 audio_fingerprints: List[str],639 transcription_callback: Optional[Callable] = None640 ) -> Dict[str, bool]:641 """642 Warm the cache by preloading frequently accessed transcriptions.643 644 Args:645 audio_fingerprints: List of audio fingerprints to warm646 transcription_callback: Optional callback to generate transcriptions647 648 Returns:649 Dictionary mapping fingerprints to success status650 """651 results = {}652 653 for fingerprint in audio_fingerprints:654 try:655 # Check if already cached656 cached_result = await self.get_cached_transcription(fingerprint)657 if cached_result is not None:658 results[fingerprint] = True659 continue660 661 # If callback provided, generate and cache transcription662 if transcription_callback:663 try:664 result = await transcription_callback(fingerprint)665 if result:666 success = await self.cache_transcription(fingerprint, result)667 results[fingerprint] = success668 else:669 results[fingerprint] = False670 except Exception as e:671 logger.warning(f"Failed to generate transcription for warming: {e}")672 results[fingerprint] = False673 else:674 results[fingerprint] = False675 676 except Exception as e:677 logger.error(f"Error warming cache for fingerprint {fingerprint[:16]}...: {e}")678 results[fingerprint] = False679 680 warmed_count = sum(1 for success in results.values() if success)681 logger.info(f"Cache warming completed: {warmed_count}/{len(audio_fingerprints)} entries warmed")682 683 return results684 685 async def get_cache_hit_optimization_stats(self) -> Dict[str, Any]:686 """687 Get specialized cache hit optimization statistics.688 689 This method provides detailed metrics specifically for cache hit optimization,690 including timing analysis, pattern recognition, and performance recommendations.691 692 Returns:693 Dictionary with cache hit optimization metrics694 """695 async with self._lock:696 hit_optimization_stats = self._calculate_hit_optimization_metrics()697 basic_stats = await self.get_cache_stats()698 699 # Calculate additional hit-specific metrics700 cache_effectiveness = {701 "api_calls_avoided": basic_stats.cache_hits,702 "api_call_reduction_percent": (703 basic_stats.cache_hits / max(basic_stats.total_requests, 1) * 100704 if basic_stats.total_requests > 0 else 0.0705 ),706 "estimated_time_saved_ms": (707 len(self._stats.get("cache_hit_times", [])) * 708 (sum(self._stats.get("cache_miss_times", [0])) / max(len(self._stats.get("cache_miss_times", [1])), 1) -709 sum(self._stats.get("cache_hit_times", [0])) / max(len(self._stats.get("cache_hit_times", [1])), 1)) * 1000710 if self._stats.get("cache_hit_times") and self._stats.get("cache_miss_times") else 0.0711 )712 }713 714 # Identify optimization opportunities715 optimization_opportunities = []716 717 if basic_stats.hit_rate < 0.8:718 optimization_opportunities.append({719 "type": "hit_rate_improvement",720 "description": f"Hit rate is {basic_stats.hit_rate:.1%}, target is 80%+",721 "potential_improvement": f"{(0.8 - basic_stats.hit_rate) * 100:.1f}% hit rate increase possible"722 })723 724 hit_times = self._stats.get("cache_hit_times", [])725 if hit_times and sum(hit_times) / len(hit_times) > 0.001: # 1ms threshold726 optimization_opportunities.append({727 "type": "lookup_speed_improvement",728 "description": f"Average hit time is {sum(hit_times) / len(hit_times) * 1000:.2f}ms",729 "potential_improvement": "Consider cache structure optimization for faster lookups"730 })731 732 hot_entries = self._stats.get("hot_entries", {})733 if len(hot_entries) > 10:734 optimization_opportunities.append({735 "type": "hot_entry_optimization",736 "description": f"{len(hot_entries)} frequently accessed entries identified",737 "potential_improvement": "Consider implementing a dedicated hot cache tier"738 })739 740 return {741 "cache_effectiveness": cache_effectiveness,742 "hit_optimization_metrics": hit_optimization_stats,743 "optimization_opportunities": optimization_opportunities,744 "performance_summary": {745 "total_requests": basic_stats.total_requests,746 "cache_hits": basic_stats.cache_hits,747 "hit_rate": basic_stats.hit_rate,748 "api_calls_avoided": basic_stats.cache_hits,749 "efficiency_score": hit_optimization_stats.get("efficiency_score", {}).get("overall_efficiency_percent", 0.0)750 }751 }752 753 async def optimize_cache_performance(self) -> Dict[str, Any]:754 """755 Perform cache performance optimization operations with hit optimization focus.756 757 Returns:758 Dictionary with optimization results759 """760 async with self._lock:761 initial_stats = await self.get_cache_stats()762 initial_hit_stats = self._calculate_hit_optimization_metrics()763 764 # Remove expired entries765 expired_count = await self.evict_expired()766 767 # Analyze and optimize entry access patterns768 optimization_results = {769 "initial_stats": {770 "entry_count": initial_stats.entry_count,771 "hit_rate": initial_stats.hit_rate,772 "size_bytes": initial_stats.total_size_bytes,773 "efficiency_score": initial_hit_stats.get("efficiency_score", {}).get("overall_efficiency_percent", 0.0)774 },775 "expired_evicted": expired_count,776 "optimizations_applied": []777 }778 779 # Promote frequently accessed entries (move to end for LRU)780 frequent_entries = []781 hot_entries = self._stats.get("hot_entries", {})782 783 # Promote hot entries to the end of the cache (most recently used)784 for key, access_count in hot_entries.items():785 if key in self._cache and access_count > 5: # Threshold for frequent access786 frequent_entries.append(key)787 self._cache.move_to_end(key)788 789 if frequent_entries:790 optimization_results["optimizations_applied"].append(791 f"Promoted {len(frequent_entries)} frequently accessed entries to improve hit rates"792 )793 794 # Optimize cache structure for better hit performance795 if len(self._cache) > 100:796 # Reorganize cache to put most accessed items at the end (LRU optimization)797 sorted_entries = []798 for key, entry in self._cache.items():799 access_weight = hot_entries.get(key, 0) + entry.access_count800 sorted_entries.append((key, entry, access_weight))801 802 # Sort by access weight and reorganize803 sorted_entries.sort(key=lambda x: x[2])804 805 # Clear and rebuild cache in optimized order806 temp_cache = self._cache.copy()807 self._cache.clear()808 809 for key, entry, _ in sorted_entries:810 self._cache[key] = entry811 812 optimization_results["optimizations_applied"].append(813 "Reorganized cache structure for optimal hit performance"814 )815 816 # Clean up old timing data to prevent memory growth817 if len(self._stats.get("cache_hit_times", [])) > 100:818 self._stats["cache_hit_times"] = self._stats["cache_hit_times"][-100:]819 optimization_results["optimizations_applied"].append("Cleaned up old hit timing data")820 821 if len(self._stats.get("cache_miss_times", [])) > 100:822 self._stats["cache_miss_times"] = self._stats["cache_miss_times"][-100:]823 optimization_results["optimizations_applied"].append("Cleaned up old miss timing data")824 825 # Compact cache if fragmented (remove and re-add entries to optimize memory)826 if initial_stats.entry_count > 100 and expired_count > initial_stats.entry_count * 0.1:827 # Cache is fragmented, perform compaction828 entries_to_compact = list(self._cache.items())829 self._cache.clear()830 self._stats["current_size_bytes"] = 0831 self._stats["entry_count"] = 0832 833 # Re-add entries in access order (most recent first)834 for key, entry in reversed(entries_to_compact):835 self._cache[key] = entry836 self._stats["current_size_bytes"] += entry.size_bytes837 self._stats["entry_count"] += 1838 839 optimization_results["optimizations_applied"].append("Performed cache compaction for better memory efficiency")840 841 final_stats = await self.get_cache_stats()842 final_hit_stats = self._calculate_hit_optimization_metrics()843 844 optimization_results["final_stats"] = {845 "entry_count": final_stats.entry_count,846 "hit_rate": final_stats.hit_rate,847 "size_bytes": final_stats.total_size_bytes,848 "efficiency_score": final_hit_stats.get("efficiency_score", {}).get("overall_efficiency_percent", 0.0)849 }850 851 # Calculate optimization impact852 efficiency_improvement = (853 optimization_results["final_stats"]["efficiency_score"] - 854 optimization_results["initial_stats"]["efficiency_score"]855 )856 857 optimization_results["optimization_impact"] = {858 "efficiency_improvement_percent": efficiency_improvement,859 "hit_rate_change": final_stats.hit_rate - initial_stats.hit_rate,860 "size_reduction_bytes": initial_stats.total_size_bytes - final_stats.total_size_bytes861 }862 863 logger.info(f"Cache optimization completed: {optimization_results}")864 return optimization_results865 866 async def clear_cache(self) -> None:867 """Clear all cached entries."""868 async with self._lock:869 self._cache.clear()870 self._stats["current_size_bytes"] = 0871 self._stats["entry_count"] = 0872 logger.info("Cache cleared")873 874 async def evict_expired(self) -> int:875 """876 Remove expired cache entries.877 878 Returns:879 Number of entries evicted880 """881 async with self._lock:882 expired_keys = []883 current_time = time.time()884 885 for key, entry in self._cache.items():886 if current_time - entry.created_at > entry.ttl:887 expired_keys.append(key)888 889 evicted_count = 0890 for key in expired_keys:891 await self._remove_entry(key)892 evicted_count += 1893 if self.enable_stats:894 self._stats["expirations"] += 1895 896 if evicted_count > 0:897 logger.debug(f"Evicted {evicted_count} expired cache entries")898 899 return evicted_count900 901 async def _ensure_space(self, required_bytes: int) -> None:902 """903 Ensure there's enough space in the cache for a new entry.904 905 Args:906 required_bytes: Number of bytes needed907 """908 # Check if we have enough space909 if self._stats["current_size_bytes"] + required_bytes <= self.max_size_bytes:910 return911 912 # Need to evict entries using LRU policy913 bytes_to_free = (self._stats["current_size_bytes"] + required_bytes) - self.max_size_bytes914 bytes_freed = 0915 evicted_count = 0916 917 # Evict least recently used entries until we have enough space918 while bytes_freed < bytes_to_free and self._cache:919 # Get least recently used entry (first in OrderedDict)920 key = next(iter(self._cache))921 entry = self._cache[key]922 923 bytes_freed += entry.size_bytes924 await self._remove_entry(key)925 evicted_count += 1926 927 if self.enable_stats:928 self._stats["evictions"] += 1929 930 logger.debug(f"Evicted {evicted_count} entries to free {bytes_freed} bytes")931 932 async def _remove_entry(self, key: str) -> None:933 """934 Remove a cache entry and update statistics.935 936 Args:937 key: Cache key to remove938 """939 entry = self._cache.pop(key, None)940 if entry:941 self._stats["current_size_bytes"] -= entry.size_bytes942 self._stats["entry_count"] -= 1943 944 async def _cleanup_loop(self) -> None:945 """Background task for periodic cache cleanup."""946 while self._running:947 try:948 await asyncio.sleep(self.cleanup_interval)949 950 if not self._running:951 break952 953 # Remove expired entries954 expired_count = await self.evict_expired()955 956 # Log cache statistics periodically957 stats = await self.get_cache_stats()958 logger.debug(f"Cache stats: {stats.entry_count} entries, "959 f"{stats.total_size_bytes / 1024 / 1024:.1f}MB, "960 f"hit rate: {stats.hit_rate:.2%}")961 962 except asyncio.CancelledError:963 break964 except Exception as e:965 logger.error(f"Error in cache cleanup loop: {e}")966 967 def get_cache_info(self) -> Dict[str, Any]:968 """969 Get detailed cache information for debugging.970 971 Returns:972 Dictionary with cache details973 """974 return {975 "max_size_bytes": self.max_size_bytes,976 "current_size_bytes": self._stats["current_size_bytes"],977 "entry_count": self._stats["entry_count"],978 "default_ttl": self.default_ttl,979 "cleanup_interval": self.cleanup_interval,980 "running": self._running,981 "stats": self._stats.copy()982 }983 984 async def __aenter__(self):985 """Async context manager entry."""986 await self.start()987 return self988 989 async def __aexit__(self, exc_type, exc_val, exc_tb):990 """Async context manager exit."""991 await self.stop()992 993 994class RedisCacheManager(CacheManagerInterface):995 """996 Redis-based cache manager for distributed caching.997 998 This implementation uses Redis for persistent, distributed caching999 across multiple application instances.1000 """1001 1002 def __init__(1003 self,1004 redis_url: str,1005 key_prefix: str = "voice_control:cache:",1006 default_ttl: int = 3600,1007 max_connections: int = 101008 ):1009 """1010 Initialize Redis cache manager.1011 1012 Args:1013 redis_url: Redis connection URL1014 key_prefix: Prefix for cache keys1015 default_ttl: Default time-to-live in seconds1016 max_connections: Maximum Redis connections1017 """1018 self.redis_url = redis_url1019 self.key_prefix = key_prefix1020 self.default_ttl = default_ttl1021 self.max_connections = max_connections1022 1023 self._redis = None1024 self._stats_key = f"{key_prefix}stats"1025 1026 logger.info(f"RedisCacheManager initialized with URL: {redis_url}")1027 1028 async def start(self) -> None:1029 """Initialize Redis connection."""1030 try:1031 import redis.asyncio as redis1032 1033 self._redis = redis.from_url(1034 self.redis_url,1035 max_connections=self.max_connections,1036 decode_responses=False # We handle bytes directly1037 )1038 1039 # Test connection1040 await self._redis.ping()1041 logger.info("Redis cache manager connected")1042 1043 except ImportError:1044 raise CacheError("redis_import", "redis package not installed")1045 except Exception as e:1046 raise CacheError("redis_connection", f"Failed to connect to Redis: {e}")1047 1048 async def stop(self) -> None:1049 """Close Redis connection."""1050 if self._redis:1051 await self._redis.close()1052 logger.info("Redis cache manager disconnected")1053 1054 async def get_cached_transcription(1055 self, 1056 audio_fingerprint: str1057 ) -> Optional[TranscriptionResult]:1058 """1059 Retrieve cached transcription result from Redis.1060 1061 Args:1062 audio_fingerprint: Unique fingerprint of audio data1063 1064 Returns:1065 Cached TranscriptionResult or None if not found1066 """1067 if not self._redis:1068 raise CacheError("redis_not_connected", "Redis not connected")1069 1070 try:1071 key = f"{self.key_prefix}{audio_fingerprint}"1072 1073 # Get cached data1074 cached_data = await self._redis.get(key)1075 1076 if cached_data is None:1077 await self._increment_stat("cache_misses")1078 return None1079 1080 # Deserialize transcription result1081 import pickle1082 result = pickle.loads(cached_data)1083 1084 await self._increment_stat("cache_hits")1085 logger.debug(f"Redis cache hit for fingerprint: {audio_fingerprint[:16]}...")1086 1087 return result1088 1089 except Exception as e:1090 logger.error(f"Failed to get cached transcription from Redis: {e}")1091 await self._increment_stat("cache_misses")1092 return None1093 1094 async def cache_transcription(1095 self, 1096 audio_fingerprint: str, 1097 result: TranscriptionResult,1098 ttl: Optional[int] = None1099 ) -> bool:1100 """1101 Cache a transcription result in Redis.1102 1103 Args:1104 audio_fingerprint: Unique fingerprint of audio data1105 result: TranscriptionResult to cache1106 ttl: Time to live in seconds (uses default if None)1107 1108 Returns:1109 True if caching was successful, False otherwise1110 """1111 if not self._redis:1112 raise CacheError("redis_not_connected", "Redis not connected")1113 1114 if ttl is None:1115 ttl = self.default_ttl1116 1117 try:1118 key = f"{self.key_prefix}{audio_fingerprint}"1119 1120 # Serialize transcription result1121 import pickle1122 cached_data = pickle.dumps(result)1123 1124 # Store in Redis with TTL1125 await self._redis.setex(key, ttl, cached_data)1126 1127 logger.debug(f"Cached transcription in Redis for fingerprint: {audio_fingerprint[:16]}... "1128 f"(ttl: {ttl}s)")1129 1130 return True1131 1132 except Exception as e:1133 logger.error(f"Failed to cache transcription in Redis: {e}")1134 return False1135 1136 async def generate_audio_fingerprint(self, audio_data: bytes) -> str:1137 """1138 Generate a unique fingerprint for audio data.1139 1140 Args:1141 audio_data: Raw audio bytes1142 1143 Returns:1144 Unique fingerprint string1145 """1146 # Use the same implementation as in-memory cache1147 cache_manager = CacheManager()1148 return await cache_manager.generate_audio_fingerprint(audio_data)1149 1150 async def get_cache_stats(self) -> CacheStats:1151 """1152 Get cache performance statistics from Redis.1153 1154 Returns:1155 CacheStats with hit rates and usage information1156 """1157 if not self._redis:1158 raise CacheError("redis_not_connected", "Redis not connected")1159 1160 try:1161 # Get stats from Redis hash1162 stats_data = await self._redis.hgetall(self._stats_key)1163 1164 # Convert bytes to integers1165 stats = {}1166 for key, value in stats_data.items():1167 if isinstance(key, bytes):1168 key = key.decode('utf-8')1169 if isinstance(value, bytes):1170 value = int(value.decode('utf-8'))1171 stats[key] = value1172 1173 # Get cache size info1174 cache_keys = await self._redis.keys(f"{self.key_prefix}*")1175 entry_count = len([k for k in cache_keys if not k.decode('utf-8').endswith(':stats')])1176 1177 return CacheStats(1178 total_requests=stats.get("total_requests", 0),1179 cache_hits=stats.get("cache_hits", 0),1180 cache_misses=stats.get("cache_misses", 0),1181 hit_rate=0.0, # Will be calculated in __post_init__1182 total_size_bytes=0, # Not easily calculable in Redis1183 entry_count=entry_count,1184 evictions=stats.get("evictions", 0)1185 )1186 1187 except Exception as e:1188 logger.error(f"Failed to get cache stats from Redis: {e}")1189 return CacheStats(0, 0, 0, 0.0, 0, 0, 0)1190 1191 async def clear_cache(self) -> None:1192 """Clear all cached entries from Redis."""1193 if not self._redis:1194 raise CacheError("redis_not_connected", "Redis not connected")1195 1196 try:1197 # Get all cache keys1198 cache_keys = await self._redis.keys(f"{self.key_prefix}*")1199 1200 if cache_keys: