talkingAvater_bgk / core /optimization /inference_cache.py
oKen38461's picture
推論キャッシュと並列処理の機能を追加し、`process_talking_head_optimized`関数をキャッシュと並列処理に対応させました。また、Gradioインターフェースにキャッシュ管理機能を追加しました。
07b71bb
"""
Inference Cache System for DittoTalkingHead
Caches video generation results for faster repeated processing
"""
import hashlib
import json
import os
import pickle
import time
from pathlib import Path
from typing import Optional, Dict, Any, Tuple, Union
from functools import lru_cache
import shutil
from datetime import datetime, timedelta
class InferenceCache:
"""
Cache system for video generation results
Supports both memory and file-based caching
"""
def __init__(
self,
cache_dir: str = "/tmp/inference_cache",
memory_cache_size: int = 100,
file_cache_size_gb: float = 10.0,
ttl_hours: int = 24
):
"""
Initialize inference cache
Args:
cache_dir: Directory for file-based cache
memory_cache_size: Maximum number of items in memory cache
file_cache_size_gb: Maximum size of file cache in GB
ttl_hours: Time to live for cache entries in hours
"""
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.memory_cache_size = memory_cache_size
self.file_cache_size_bytes = int(file_cache_size_gb * 1024 * 1024 * 1024)
self.ttl_seconds = ttl_hours * 3600
# Metadata file for managing cache
self.metadata_file = self.cache_dir / "cache_metadata.json"
self.metadata = self._load_metadata()
# In-memory cache
self._memory_cache = {}
self._access_times = {}
# Clean up expired entries on initialization
self._cleanup_expired()
def _load_metadata(self) -> Dict[str, Any]:
"""Load cache metadata"""
if self.metadata_file.exists():
try:
with open(self.metadata_file, 'r') as f:
return json.load(f)
except:
return {}
return {}
def _save_metadata(self):
"""Save cache metadata"""
with open(self.metadata_file, 'w') as f:
json.dump(self.metadata, f, indent=2)
def generate_cache_key(
self,
audio_path: str,
image_path: str,
**kwargs
) -> str:
"""
Generate unique cache key based on input parameters
Args:
audio_path: Path to audio file
image_path: Path to image file
**kwargs: Additional parameters affecting output
Returns:
SHA-256 hash as cache key
"""
# Read file contents for hashing
with open(audio_path, 'rb') as f:
audio_hash = hashlib.sha256(f.read()).hexdigest()
with open(image_path, 'rb') as f:
image_hash = hashlib.sha256(f.read()).hexdigest()
# Include relevant parameters in key
key_data = {
'audio': audio_hash,
'image': image_hash,
'resolution': kwargs.get('resolution', '320x320'),
'steps': kwargs.get('steps', 25),
'seed': kwargs.get('seed', None)
}
# Generate final key
key_str = json.dumps(key_data, sort_keys=True)
return hashlib.sha256(key_str.encode()).hexdigest()
def get_from_memory(self, cache_key: str) -> Optional[str]:
"""
Get video path from memory cache
Args:
cache_key: Cache key
Returns:
Video file path if found, None otherwise
"""
if cache_key in self._memory_cache:
self._access_times[cache_key] = time.time()
return self._memory_cache[cache_key]
return None
def get_from_file(self, cache_key: str) -> Optional[str]:
"""
Get video path from file cache
Args:
cache_key: Cache key
Returns:
Video file path if found, None otherwise
"""
if cache_key not in self.metadata:
return None
entry = self.metadata[cache_key]
# Check expiration
if time.time() > entry['expires_at']:
self._remove_cache_entry(cache_key)
return None
# Check if file exists
video_path = self.cache_dir / entry['filename']
if not video_path.exists():
self._remove_cache_entry(cache_key)
return None
# Update access time
self.metadata[cache_key]['last_access'] = time.time()
self._save_metadata()
# Add to memory cache
self._add_to_memory_cache(cache_key, str(video_path))
return str(video_path)
def get(self, cache_key: str) -> Optional[str]:
"""
Get video from cache (memory first, then file)
Args:
cache_key: Cache key
Returns:
Video file path if found, None otherwise
"""
# Try memory cache first
result = self.get_from_memory(cache_key)
if result:
return result
# Try file cache
return self.get_from_file(cache_key)
def put(
self,
cache_key: str,
video_path: str,
**metadata
) -> bool:
"""
Store video in cache
Args:
cache_key: Cache key
video_path: Path to generated video
**metadata: Additional metadata to store
Returns:
True if stored successfully
"""
try:
# Copy video to cache directory
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
cache_filename = f"{cache_key[:8]}_{timestamp}.mp4"
cache_video_path = self.cache_dir / cache_filename
shutil.copy2(video_path, cache_video_path)
# Store metadata
self.metadata[cache_key] = {
'filename': cache_filename,
'created_at': time.time(),
'expires_at': time.time() + self.ttl_seconds,
'last_access': time.time(),
'size_bytes': os.path.getsize(cache_video_path),
'metadata': metadata
}
# Check cache size and clean if needed
self._check_cache_size()
# Save metadata
self._save_metadata()
# Add to memory cache
self._add_to_memory_cache(cache_key, str(cache_video_path))
return True
except Exception as e:
print(f"Error storing cache: {e}")
return False
def _add_to_memory_cache(self, cache_key: str, video_path: str):
"""Add item to memory cache with LRU eviction"""
# Check if we need to evict
if len(self._memory_cache) >= self.memory_cache_size:
# Find least recently used
lru_key = min(self._access_times, key=self._access_times.get)
del self._memory_cache[lru_key]
del self._access_times[lru_key]
self._memory_cache[cache_key] = video_path
self._access_times[cache_key] = time.time()
def _check_cache_size(self):
"""Check and maintain cache size limit"""
total_size = sum(
entry['size_bytes']
for entry in self.metadata.values()
)
if total_size > self.file_cache_size_bytes:
# Remove oldest entries until under limit
sorted_entries = sorted(
self.metadata.items(),
key=lambda x: x[1]['last_access']
)
while total_size > self.file_cache_size_bytes and sorted_entries:
key_to_remove, entry = sorted_entries.pop(0)
total_size -= entry['size_bytes']
self._remove_cache_entry(key_to_remove)
def _cleanup_expired(self):
"""Remove expired cache entries"""
current_time = time.time()
expired_keys = [
key for key, entry in self.metadata.items()
if current_time > entry['expires_at']
]
for key in expired_keys:
self._remove_cache_entry(key)
if expired_keys:
print(f"Cleaned up {len(expired_keys)} expired cache entries")
def _remove_cache_entry(self, cache_key: str):
"""Remove a cache entry"""
if cache_key in self.metadata:
# Remove file
video_file = self.cache_dir / self.metadata[cache_key]['filename']
if video_file.exists():
video_file.unlink()
# Remove from metadata
del self.metadata[cache_key]
# Remove from memory cache
if cache_key in self._memory_cache:
del self._memory_cache[cache_key]
del self._access_times[cache_key]
def clear_cache(self):
"""Clear all cache entries"""
# Remove all video files
for file in self.cache_dir.glob("*.mp4"):
file.unlink()
# Clear metadata
self.metadata = {}
self._save_metadata()
# Clear memory cache
self._memory_cache.clear()
self._access_times.clear()
print("Inference cache cleared")
def get_cache_stats(self) -> Dict[str, Any]:
"""Get cache statistics"""
total_size = sum(
entry['size_bytes']
for entry in self.metadata.values()
)
memory_hits = len(self._memory_cache)
file_entries = len(self.metadata)
return {
'memory_cache_entries': memory_hits,
'file_cache_entries': file_entries,
'total_cache_size_mb': total_size / (1024 * 1024),
'cache_size_limit_gb': self.file_cache_size_bytes / (1024 * 1024 * 1024),
'ttl_hours': self.ttl_seconds / 3600,
'cache_directory': str(self.cache_dir)
}
class CachedInference:
"""
Wrapper for cached inference execution
"""
def __init__(self, cache: InferenceCache):
"""
Initialize cached inference
Args:
cache: InferenceCache instance
"""
self.cache = cache
def process_with_cache(
self,
inference_func: callable,
audio_path: str,
image_path: str,
output_path: str,
**kwargs
) -> Tuple[str, bool, float]:
"""
Process with caching
Args:
inference_func: Function to generate video
audio_path: Path to audio file
image_path: Path to image file
output_path: Desired output path
**kwargs: Additional parameters
Returns:
Tuple of (output_path, cache_hit, process_time)
"""
start_time = time.time()
# Generate cache key
cache_key = self.cache.generate_cache_key(
audio_path, image_path, **kwargs
)
# Check cache
cached_video = self.cache.get(cache_key)
if cached_video:
# Cache hit - copy to output path
shutil.copy2(cached_video, output_path)
process_time = time.time() - start_time
print(f"✅ Cache hit! Retrieved in {process_time:.2f}s")
return output_path, True, process_time
# Cache miss - generate video
print("Cache miss - generating video...")
inference_func(audio_path, image_path, output_path, **kwargs)
# Store in cache
if os.path.exists(output_path):
self.cache.put(cache_key, output_path, **kwargs)
process_time = time.time() - start_time
return output_path, False, process_time