Skip to content

image_cache

anomaly_match.prediction.image_cache

Least Recently Used (LRU) image cache for on-demand loading of prediction source images.

Images are loaded from disk using the same fitsbolt pipeline used during training, ensuring visual consistency between training and prediction views.

ImageCache(cfg, search_dir, cache_size=200)

LRU cache that loads source images on demand.

Keeps up to cache_size images in memory. When the cache is full the least-recently-used entry is evicted. Images are loaded from search_dir using the fitsbolt pipeline configured in cfg.

Parameters:

Name Type Description Default
cfg DotMap

Configuration object with normalisation settings and fitsbolt_cfg for the image processing pipeline.

 required
search_dir str

Root directory to search for image files.

 required
cache_size int

Maximum number of images to keep in memory.

 200
Source code in anomaly_match/prediction/image_cache.py 
40
41
42
43
44
45
46
47
48
49
def __init__(
    self,
    cfg: DotMap,
    search_dir: str,
    cache_size: int = 200,
) -> None:
    self._cfg = cfg
    self._search_dir = search_dir
    self._cache_size = cache_size
    self._cache: OrderedDict[str, np.ndarray] = OrderedDict()

size property

Return number of currently cached images.

Returns:

Type Description
int

Current cache occupancy.

max_size property

Return maximum cache capacity.

Returns:

Type Description
int

Maximum number of images the cache can hold.

get_image(filename)

Return the image for filename, loading from disk if needed.

On cache hit the entry is promoted to most-recently-used. On cache miss the image is loaded via fitsbolt's load_and_process_single_wrapper and inserted into the cache, evicting the LRU entry if the cache is full.

Parameters:

Name Type Description Default
filename str

Source filename (basename or relative path).

 required

Returns:

Type Description
ndarray | None

HWC uint8 numpy array, or None if the file cannot be
ndarray | None

loaded.
Source code in anomaly_match/prediction/image_cache.py 
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
def get_image(self, filename: str) -> np.ndarray | None:
    """Return the image for *filename*, loading from disk if needed.

    On cache hit the entry is promoted to most-recently-used.  On
    cache miss the image is loaded via fitsbolt's
    ``load_and_process_single_wrapper`` and inserted into the cache,
    evicting the LRU entry if the cache is full.

    Args:
        filename: Source filename (basename or relative path).

    Returns:
        HWC uint8 numpy array, or ``None`` if the file cannot be
        loaded.
    """
    # Cache hit — promote to most-recently-used
    if filename in self._cache:
        self._cache.move_to_end(filename)
        return self._cache[filename]

    # Cache miss — load from disk
    image = self._load_from_disk(filename)
    if image is None:
        return None

    # Ensure uint8 HWC
    image = _ensure_uint8_hwc(image)

    # Insert and evict if necessary
    self._cache[filename] = image
    if len(self._cache) > self._cache_size:
        evicted_key, _ = self._cache.popitem(last=False)
        logger.trace("ImageCache evicted {}", evicted_key)

    return image

prefetch(filenames)

Pre-load a batch of images into the cache.

Already-cached filenames are skipped.

Parameters:

Name Type Description Default
filenames list[str]

List of filenames to pre-load.

 required
Source code in anomaly_match/prediction/image_cache.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def prefetch(self, filenames: list[str]) -> None:
    """Pre-load a batch of images into the cache.

    Already-cached filenames are skipped.

    Args:
        filenames: List of filenames to pre-load.
    """
    for fn in filenames:
        if fn not in self._cache:
            self.get_image(fn)

put(filename, image)

Manually insert an image into the cache.

Useful for caching images loaded via alternative paths (e.g. Cutana cutouts) that bypass get_image.

Parameters:

Name Type Description Default
filename str

Cache key.

 required
image ndarray

HWC uint8 numpy array.

 required
Source code in anomaly_match/prediction/image_cache.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def put(self, filename: str, image: np.ndarray) -> None:
    """Manually insert an image into the cache.

    Useful for caching images loaded via alternative paths (e.g.
    Cutana cutouts) that bypass ``get_image``.

    Args:
        filename: Cache key.
        image: HWC uint8 numpy array.
    """
    image = _ensure_uint8_hwc(image)
    self._cache[filename] = image
    if len(self._cache) > self._cache_size:
        self._cache.popitem(last=False)

clear()

Drop all cached images.

Source code in anomaly_match/prediction/image_cache.py 
118
119
120
def clear(self) -> None:
    """Drop all cached images."""
    self._cache.clear()