in ez_wsi_dicomweb/patch_embedding_endpoints.py [0:0]
def _load_patch_bytes(self) -> List[np.ndarray]:
"""Loads patch bytes for each request."""
if (
self._slide_embedding_source is None
or not self._slide_embedding_source.patches
):
return []
first_patch = self._slide_embedding_source.patches[0].patch
if isinstance(first_patch, gcs_image.GcsPatch):
# Make a shallow copy of source patch Source GcsImage to ensure
# if image bytes are are not retained after the request is processed.
source_copy = copy.copy(first_patch.source)
image_bytes = []
icc_profile_normalization = source_copy.create_icc_profile_transformation(
self._target_icc_profile_bytes
)
for patch in self._slide_embedding_source.patches:
p = patch.patch
if (
self._require_fully_in_source_image
and not p.is_patch_fully_in_source_image()
):
raise ez_wsi_errors.PatchOutsideOfImageDimensionsError(
'A portion of the patch does not overlap the image.'
)
# create a patch with same coordinates a the temporary source.
# require in source image is not relevant to the temp patch. Setting to
# false.
temp_patch = gcs_image.GcsPatch(
source_copy, p.x, p.y, p.width, p.height, False
)
image_bytes.append(temp_patch.image_bytes(icc_profile_normalization))
return image_bytes
if isinstance(first_patch, dicom_slide.DicomPatch):
# Load in slide DICOM slide imaging using frame cache that is not shared
# to scope imageing bytes to the embedding request and also avoid.
# possible LRU cache eviction across parallel reads.
# Make a shallow copy of source patch Source DicomSlide or
# or Dicom Microscopy image
source_copy = copy.copy(first_patch.source)
# Init new frame cache on the shallow copy source.
# Source and copy will no longer share the cache.
fc = source_copy.init_slide_frame_cache()
# Construct list of patches to return embedding for.
patch_list = typing.cast(
List[dicom_slide.DicomPatch],
[patch.patch for patch in self._slide_embedding_source.patches],
)
# Preload list of patches into frame cache. Copy across any loaded
# imaging that was loaded in the original cache.
source_copy.preload_patches_in_frame_cache(
patch_list, False, first_patch.source.slide_frame_cache
)
if (
self._target_icc_profile_bytes is None
or not self._target_icc_profile_bytes
):
icc_profile_normalization = None
else:
dicom_path = str(source_copy.path)
with self._icc_profile_cache_lock:
source_icc_profile_bytes = self._icc_profile_cache.get(dicom_path)
if source_icc_profile_bytes is None:
if isinstance(source_copy, dicom_slide.DicomSlide):
source_icc_profile_bytes = source_copy.get_icc_profile_bytes()
elif isinstance(source_copy, dicom_slide.DicomMicroscopeImage):
source_icc_profile_bytes = source_copy.get_level_icc_profile_bytes(
first_patch.level
)
else:
raise ValueError('Unexpected object')
with self._icc_profile_cache_lock:
self._icc_profile_cache[dicom_path] = source_icc_profile_bytes
icc_profile_normalization = (
dicom_slide.create_icc_profile_transformation(
source_icc_profile_bytes, self._target_icc_profile_bytes
)
)
fc.block_until_frames_are_loaded()
# Generate image bytes for each patch.
image_bytes = []
for p in patch_list:
if (
self._require_fully_in_source_image
and not p.is_patch_fully_in_source_image()
):
raise ez_wsi_errors.PatchOutsideOfImageDimensionsError(
'A portion of the patch does not overlap the image.'
)
# create copy of the patch.
# set the patch to point to the copied source to make patch image
# retrieval read from the copied sources frame cache.
temp_patch = dicom_slide.DicomPatch(
p.get_pyramid_imaging_source_level(),
p.x,
p.y,
p.width,
p.height,
source_copy,
p.level,
# ensuring patch falls inside image dimensions is not relevant
# to the temp patch.
False,
)
image_bytes.append(temp_patch.image_bytes(icc_profile_normalization))
return image_bytes
raise ValueError('Unexpected object')