File size: 21,239 Bytes
f559cc0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
"""
inference_cache.py β€” Hash-based caching for repeated image predictions.

Caches prediction results keyed by image hash to avoid redundant computation
for identical or near-identical images. Useful when:
- Users retake the same image without meaningful changes
- Batch processing includes duplicate images
- A/B testing sends the same image through multiple model versions

Cache Strategy
--------------
- Primary key: SHA-256 hash of resized, normalized image bytes
- Secondary: Perceptual hash (pHash) for near-duplicate detection
- Tertiary: Average hash (aHash) for fast pre-filtering
- TTL: Configurable, default 24 hours
- Max size: LRU eviction when cache exceeds size limit
- Persistent: Optional disk-backed cache for cross-session persistence

Feature Extraction Cache
------------------------
Additionally caches intermediate feature extraction results to speed up
repeated feature extraction with the same image but different model configs.
"""
from __future__ import annotations

import hashlib
import json
import logging
import time
from collections import OrderedDict
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Any

import numpy as np
from PIL import Image

log = logging.getLogger("anemialens.cache")


@dataclass
class CacheEntry:
    """A single cached prediction result."""
    image_hash: str
    phash: int
    prediction: dict[str, Any]
    timestamp: float
    hit_count: int = 0
    model_version: str = ""
    quality_metrics: dict[str, float] = field(default_factory=dict)


@dataclass
class FeatureCacheEntry:
    """Cached feature extraction result."""
    image_hash: str
    features: dict[str, float]
    timestamp: float
    hit_count: int = 0
    config_hash: str = ""  # Hash of feature extraction config


@dataclass(frozen=True)
class CacheStats:
    """Comprehensive cache statistics."""
    size: int
    max_size: int
    hits: int
    misses: int
    hit_rate: float
    ttl_seconds: int
    oldest_entry_age_seconds: float
    newest_entry_age_seconds: float
    total_predictions_cached: int
    total_features_cached: int
    evictions: int
    disk_persisted: bool
    disk_path: str | None
    disk_entries: int


class InferenceCache:
    """
    LRU cache for image prediction results.

    Usage
    -----
    cache = InferenceCache(max_size=500, ttl_seconds=86400)
    key = cache.compute_hash(image)
    result = cache.get(key, phash)
    if result is not None:
        return result  # Cache hit!
    # ... run prediction ...
    cache.put(key, phash, prediction, model_version="v8")
    """

    def __init__(
        self,
        max_size: int = 500,
        ttl_seconds: int = 86400,
        phash_threshold: int = 8,
        persist_path: str | Path | None = None,
    ) -> None:
        self.max_size = max_size
        self.ttl_seconds = ttl_seconds
        self.phash_threshold = phash_threshold  # Max Hamming distance for near-duplicate
        self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self._phash_index: dict[int, str] = {}  # phash β†’ hash_key
        self._ahash_index: dict[int, str] = {}  # ahash β†’ hash_key (fast pre-filter)
        self.hits = 0
        self.misses = 0
        self.evictions = 0
        self._total_predictions_cached = 0
        self._total_features_cached = 0

        # Feature extraction cache
        self._feature_cache: dict[str, FeatureCacheEntry] = {}

        # Persistent storage
        self._persist_path = Path(persist_path) if persist_path else None
        self._disk_entries = 0
        if self._persist_path and self._persist_path.exists():
            self._load_from_disk()

    def compute_hash(self, image: Image.Image) -> str:
        """
        Compute SHA-256 hash of image content.

        Uses resized, normalized representation for consistent hashing.
        """
        # Resize to fixed size for consistent hashing
        resized = image.resize((64, 64)).convert("RGB")
        data = np.asarray(resized, dtype=np.uint8).tobytes()
        return hashlib.sha256(data).hexdigest()

    @staticmethod
    def compute_phash(image: Image.Image) -> int:
        """
        Compute perceptual hash (simplified DCT-based).

        Returns a 64-bit integer hash.
        """
        gray = image.resize((32, 32)).convert("L")
        pixels = np.asarray(gray, dtype=np.float64)

        # Simple DCT approximation using mean thresholding
        mean_val = pixels.mean()
        bits = (pixels > mean_val).flatten()

        # Convert bit array to integer
        phash = 0
        for i, bit in enumerate(bits[:64]):
            if bit:
                phash |= (1 << i)

        return phash

    @staticmethod
    def compute_ahash(image: Image.Image) -> int:
        """
        Compute average hash (aHash) for fast pre-filtering.

        Simpler and faster than phash, good for quick rejection.
        Returns a 64-bit integer hash.
        """
        gray = image.resize((8, 8)).convert("L")
        pixels = np.asarray(gray, dtype=np.float64)
        mean_val = pixels.mean()
        bits = (pixels > mean_val).flatten()

        ahash = 0
        for i, bit in enumerate(bits):
            if bit:
                ahash |= (1 << i)

        return ahash

    def get(
        self,
        image_hash: str,
        phash: int | None = None,
        ahash: int | None = None,
    ) -> dict[str, Any] | None:
        """
        Look up a cached prediction.

        First tries exact hash match, then falls back to perceptual hash
        near-duplicate detection, then average hash for fast pre-filtering.

        Parameters
        ----------
        image_hash : SHA-256 hash of the image
        phash : Perceptual hash for near-duplicate detection
        ahash : Average hash for fast pre-filtering

        Returns
        -------
        Cached prediction dict or None
        """
        now = time.time()

        # Try exact match
        if image_hash in self._cache:
            entry = self._cache[image_hash]
            # Check TTL
            if now - entry.timestamp > self.ttl_seconds:
                self._remove(image_hash)
                self.misses += 1
                return None
            entry.hit_count += 1
            self.hits += 1
            # Move to end (most recently used)
            self._cache.move_to_end(image_hash)
            log.debug(
                "Cache HIT for image %s (hit #%d)",
                image_hash[:8], entry.hit_count,
            )
            return entry.prediction

        # Try near-duplicate via phash
        if phash is not None:
            for stored_phash, stored_hash in self._phash_index.items():
                if self._hamming_distance(phash, stored_phash) <= self.phash_threshold:
                    if stored_hash in self._cache:
                        entry = self._cache[stored_hash]
                        if now - entry.timestamp > self.ttl_seconds:
                            self._remove(stored_hash)
                            continue
                        entry.hit_count += 1
                        self.hits += 1
                        self._cache.move_to_end(stored_hash)
                        log.debug(
                            "Cache HIT (near-duplicate phash) for phash %x", phash
                        )
                        return entry.prediction

        # Fast pre-filter via ahash (wider threshold for speed)
        if ahash is not None:
            for stored_ahash, stored_hash in self._ahash_index.items():
                if self._hamming_distance(ahash, stored_ahash) <= 4:  # Tighter threshold for ahash
                    if stored_hash in self._cache:
                        entry = self._cache[stored_hash]
                        if now - entry.timestamp > self.ttl_seconds:
                            self._remove(stored_hash)
                            continue
                        entry.hit_count += 1
                        self.hits += 1
                        self._cache.move_to_end(stored_hash)
                        log.debug(
                            "Cache HIT (near-duplicate ahash) for ahash %x", ahash
                        )
                        return entry.prediction

        self.misses += 1
        return None

    def put(
        self,
        image_hash: str,
        phash: int | dict[str, Any],
        prediction: dict[str, Any] | None = None,
        model_version: str = "",
        quality_metrics: dict[str, float] | None = None,
        ahash: int | None = None,
    ) -> None:
        """
        Store a prediction in the cache.

        Parameters
        ----------
        image_hash : SHA-256 hash of the image
        phash : Perceptual hash
        prediction : Prediction result dict
        model_version : Version string of the model used
        quality_metrics : Quality metrics at prediction time
        ahash : Average hash for fast pre-filtering
        """
        if prediction is None and isinstance(phash, dict):
            prediction = phash
            phash = 0

        if prediction is None:
            raise ValueError("prediction payload is required when phash is provided explicitly")

        now = time.time()

        # Evict if at capacity
        if image_hash not in self._cache and len(self._cache) >= self.max_size:
            self._evict_lru()

        entry = CacheEntry(
            image_hash=image_hash,
            phash=int(phash),
            prediction=prediction,
            timestamp=now,
            model_version=model_version,
            quality_metrics=quality_metrics or {},
        )

        self._cache[image_hash] = entry
        self._cache.move_to_end(image_hash)
        self._phash_index[phash] = image_hash
        if ahash is not None:
            self._ahash_index[ahash] = image_hash

        self._total_predictions_cached += 1

        # Persist to disk if configured
        if self._persist_path:
            self._persist_to_disk()

        log.debug("Cache PUT for image %s", image_hash[:8])

    def clear(self) -> None:
        """Clear all cached entries."""
        self._cache.clear()
        self._phash_index.clear()
        self.hits = 0
        self.misses = 0

    def stats(self) -> CacheStats:
        """Return comprehensive cache statistics."""
        total = self.hits + self.misses
        hit_rate = self.hits / max(total, 1)
        now = time.time()

        oldest_age = 0.0
        newest_age = 0.0
        if self._cache:
            ages = [now - entry.timestamp for entry in self._cache.values()]
            oldest_age = max(ages)
            newest_age = min(ages)

        return CacheStats(
            size=len(self._cache),
            max_size=self.max_size,
            hits=self.hits,
            misses=self.misses,
            hit_rate=round(hit_rate, 3),
            ttl_seconds=self.ttl_seconds,
            oldest_entry_age_seconds=round(oldest_age, 1),
            newest_entry_age_seconds=round(newest_age, 1),
            total_predictions_cached=self._total_predictions_cached,
            total_features_cached=self._total_features_cached,
            evictions=self.evictions,
            disk_persisted=self._persist_path is not None,
            disk_path=str(self._persist_path) if self._persist_path else None,
            disk_entries=self._disk_entries,
        )

    def cleanup_expired(self) -> int:
        """Remove expired entries. Returns count of removed entries."""
        now = time.time()
        expired = [
            key for key, entry in self._cache.items()
            if now - entry.timestamp > self.ttl_seconds
        ]
        for key in expired:
            self._remove(key)
        return len(expired)

    # ──────────────────────────────────────────────────────────────────────
    # Private helpers
    # ──────────────────────────────────────────────────────────────────────

    def _remove(self, key: str) -> None:
        """Remove an entry from both cache and hash indices."""
        if key in self._cache:
            entry = self._cache.pop(key)
            self._phash_index.pop(entry.phash, None)
            # Remove from ahash index too
            ahash_to_remove = None
            for ahash_val, stored_hash in list(self._ahash_index.items()):
                if stored_hash == key:
                    ahash_to_remove = ahash_val
                    break
            if ahash_to_remove is not None:
                self._ahash_index.pop(ahash_to_remove, None)

    def _evict_lru(self) -> None:
        """Evict the least recently used entry."""
        if self._cache:
            key, entry = self._cache.popitem(last=False)
            self._phash_index.pop(entry.phash, None)
            # Remove from ahash index
            ahash_to_remove = None
            for ahash_val, stored_hash in list(self._ahash_index.items()):
                if stored_hash == key:
                    ahash_to_remove = ahash_val
                    break
            if ahash_to_remove is not None:
                self._ahash_index.pop(ahash_to_remove, None)
            self.evictions += 1
            log.debug("Cache LRU eviction: %s", key[:8])

    @staticmethod
    def _hamming_distance(a: int, b: int) -> int:
        """Count differing bits between two integers."""
        xor = a ^ b
        return bin(xor).count("1")

    # ──────────────────────────────────────────────────────────────────────
    # Disk persistence
    # ──────────────────────────────────────────────────────────────────────

    def _persist_to_disk(self) -> None:
        """Serialize cache to disk for cross-session persistence."""
        if not self._persist_path:
            return

        try:
            self._persist_path.parent.mkdir(parents=True, exist_ok=True)
            data = {
                "entries": [
                    {
                        "image_hash": e.image_hash, "phash": e.phash,
                        "prediction": e.prediction, "timestamp": e.timestamp,
                        "hit_count": e.hit_count, "model_version": e.model_version,
                        "quality_metrics": e.quality_metrics,
                    }
                    for e in self._cache.values()
                ],
                "metadata": {
                    "hits": self.hits, "misses": self.misses,
                    "evictions": self.evictions,
                    "total_predictions_cached": self._total_predictions_cached,
                    "saved_at": time.time(),
                },
            }
            temp_path = self._persist_path.with_suffix(".tmp")
            with open(temp_path, "w") as f:
                json.dump(data, f, default=str)
            temp_path.replace(self._persist_path)
            self._disk_entries = len(data["entries"])
        except Exception as e:
            log.warning("Failed to persist cache to disk: %s", e)

    def _load_from_disk(self) -> None:
        """Load cache from disk if available."""
        if not self._persist_path or not self._persist_path.exists():
            return
        try:
            with open(self._persist_path) as f:
                data = json.load(f)
            now = time.time()
            loaded = 0
            for entry_data in data.get("entries", []):
                if now - entry_data["timestamp"] > self.ttl_seconds:
                    continue
                entry = CacheEntry(
                    image_hash=entry_data["image_hash"], phash=entry_data["phash"],
                    prediction=entry_data["prediction"], timestamp=entry_data["timestamp"],
                    hit_count=entry_data.get("hit_count", 0),
                    model_version=entry_data.get("model_version", ""),
                    quality_metrics=entry_data.get("quality_metrics", {}),
                )
                self._cache[entry.image_hash] = entry
                self._phash_index[entry.phash] = entry.image_hash
                loaded += 1
            metadata = data.get("metadata", {})
            self.hits = metadata.get("hits", 0)
            self.misses = metadata.get("misses", 0)
            self.evictions = metadata.get("evictions", 0)
            self._total_predictions_cached = metadata.get("total_predictions_cached", loaded)
            self._disk_entries = loaded
        except Exception as e:
            log.warning("Failed to load cache from disk: %s", e)

    # ──────────────────────────────────────────────────────────────────────
    # Feature extraction cache
    # ──────────────────────────────────────────────────────────────────────

    def get_cached_features(self, image_hash: str, config_hash: str = "") -> dict[str, float] | None:
        """Look up cached feature extraction result."""
        key = f"{image_hash}:{config_hash}"
        if key in self._feature_cache:
            entry = self._feature_cache[key]
            if time.time() - entry.timestamp > self.ttl_seconds:
                del self._feature_cache[key]
                return None
            entry.hit_count += 1
            return entry.features
        return None

    def cache_features(self, image_hash: str, features: dict[str, float], config_hash: str = "") -> None:
        """Store feature extraction result in cache."""
        key = f"{image_hash}:{config_hash}"
        if key not in self._feature_cache and len(self._feature_cache) >= 200:
            oldest_key = min(self._feature_cache, key=lambda k: self._feature_cache[k].timestamp)
            del self._feature_cache[oldest_key]
        self._feature_cache[key] = FeatureCacheEntry(
            image_hash=image_hash, features=features, timestamp=time.time(), config_hash=config_hash,
        )
        self._total_features_cached += 1

    def clear_features(self) -> None:
        """Clear all cached feature extraction results."""
        self._feature_cache.clear()


# ─────────────────────────────────────────────────────────────────────────────
# Module-level singleton
# ─────────────────────────────────────────────────────────────────────────────

_default_cache: InferenceCache | None = None


def get_inference_cache(
    max_size: int = 500,
    ttl_seconds: int = 86400,
) -> InferenceCache:
    """Get or create the singleton inference cache."""
    global _default_cache
    if _default_cache is None:
        _default_cache = InferenceCache(max_size=max_size, ttl_seconds=ttl_seconds)
    return _default_cache


def cache_prediction(
    image: Image.Image,
    prediction: dict[str, Any],
    model_version: str = "",
    quality_metrics: dict[str, float] | None = None,
) -> None:
    """Convenience function to cache a prediction."""
    cache = get_inference_cache()
    image_hash = cache.compute_hash(image)
    phash = cache.compute_phash(image)
    ahash = cache.compute_ahash(image)
    cache.put(image_hash, phash, prediction, model_version, quality_metrics, ahash)


def get_cached_prediction(
    image: Image.Image,
) -> dict[str, Any] | None:
    """Convenience function to look up a cached prediction."""
    cache = get_inference_cache()
    image_hash = cache.compute_hash(image)
    phash = cache.compute_phash(image)
    ahash = cache.compute_ahash(image)
    return cache.get(image_hash, phash, ahash)


def cache_feature_extraction(
    image: Image.Image,
    features: dict[str, float],
    config_hash: str = "",
) -> None:
    """Convenience function to cache feature extraction results."""
    cache = get_inference_cache()
    image_hash = cache.compute_hash(image)
    cache.cache_features(image_hash, features, config_hash)


def get_cached_features(
    image: Image.Image,
    config_hash: str = "",
) -> dict[str, float] | None:
    """Convenience function to look up cached feature extraction."""
    cache = get_inference_cache()
    image_hash = cache.compute_hash(image)
    return cache.get_cached_features(image_hash, config_hash)


def _compute_image_hash(image: Image.Image) -> str:
    """Legacy helper preserved for existing tests and scripts."""
    return get_inference_cache().compute_hash(image)