"""Bounded, expiring CPU transcript cache; never retains audio or CUDA objects.""" import hashlib import time from collections import OrderedDict from threading import Lock class TranscriptCache: def __init__(self, max_entries=16, ttl_seconds=600, clock=time.monotonic): self.max_entries = max_entries self.ttl_seconds = ttl_seconds self.clock = clock self.entries = OrderedDict() self.lock = Lock() @staticmethod def key(waveform, settings): digest = hashlib.sha256() digest.update(repr((waveform.shape, waveform.dtype.str, settings)).encode()) digest.update(waveform.tobytes()) return digest.digest() def _expire(self, now): for key, (expires, _) in list(self.entries.items()): if expires <= now: del self.entries[key] def get(self, key): with self.lock: self._expire(self.clock()) entry = self.entries.get(key) if entry is None: return None self.entries.move_to_end(key) return entry[1] def put(self, key, value): with self.lock: now = self.clock() self._expire(now) self.entries[key] = (now + self.ttl_seconds, value) self.entries.move_to_end(key) while len(self.entries) > self.max_entries: self.entries.popitem(last=False)