from collections import OrderedDict from threading import Lock from typing import Dict, Optional import openslide class SlideCache: def __init__(self, max_items: int = 8) -> None: self._max_items = max_items self._slides: Dict[str, openslide.OpenSlide] = OrderedDict() self._lock = Lock() def get(self, slide_id: str) -> Optional[openslide.OpenSlide]: with self._lock: slide = self._slides.get(slide_id) if slide is None: return None self._slides.move_to_end(slide_id) return slide def put(self, slide_id: str, path: str) -> openslide.OpenSlide: with self._lock: if slide_id in self._slides: self._slides.move_to_end(slide_id) return self._slides[slide_id] slide = openslide.OpenSlide(path) self._slides[slide_id] = slide self._slides.move_to_end(slide_id) while len(self._slides) > self._max_items: _, oldest = self._slides.popitem(last=False) try: oldest.close() except Exception: pass return slide cache = SlideCache() def load_slide(slide_id: str, path: str) -> openslide.OpenSlide: return cache.put(slide_id, path) def get_slide(slide_id: str) -> Optional[openslide.OpenSlide]: return cache.get(slide_id)