"""
Classes to enable databroker to load area detector images.
Note that these have to registered ::
from databroker import catalog
from polartools.area_detector_handlers import LambdaHDF5Handler
cat = catalog['my_catalog']
cat.register_handler("AD_HDF5_lambda", LambdaHDF5Handler, overwrite=True)
.. autosummary::
~LambdaHDF5Handler
~EigerHandler
~SPEHandler
"""
# Copyright (c) 2020-2021, UChicago Argonne, LLC.
# See LICENSE file for details.
from area_detector_handlers.handlers import AreaDetectorHDF5SingleHandler
from area_detector_handlers import HandlerBase
from area_detector_handlers.eiger import EigerHandler as AD_EigerHandler
from dask.array import from_array
from os.path import getsize, join
from glob import glob
from pathlib import Path
from h5py import File
import hdf5plugin # noqa: F401
from imageio.v3 import imread
[docs]
class SPEHandler(HandlerBase):
specs = {"AD_SPE_APSPolar"} | HandlerBase.specs
"""
Handler for the Princeton detector SPE files.
"""
def __init__(self, fpath, template, filename, frame_per_point=1):
self._path = join(fpath, "")
self._fpp = frame_per_point
self._template = template
self._filename = filename
self._f_cache = dict()
def __call__(self, point_number):
if point_number not in self._f_cache:
fname = self._template % (self._path, self._filename, point_number)
imgs = imread(fname)
self._f_cache[point_number] = imgs
data = self._f_cache[point_number]
if data.shape[0] != self._fpp:
raise ValueError(
"Expected {} frames, found {} frames".format(
self._fpp, data.shape[0]
)
)
return data
def get_file_list(self, datum_kwarg_gen):
return [
self._template % (self._path, self._filename, d["point_number"])
for d in datum_kwarg_gen
]
[docs]
class LambdaHDF5Handler(AreaDetectorHDF5SingleHandler):
specs = {"AD_HDF5_Lambda250k_APSPolar"}
"""
Handler for the Lambda detector HDF5 files.
"""
def __call__(self, point_number):
return from_array(super().__call__(point_number))
[docs]
class EigerHandler(AD_EigerHandler):
"""
Modified Eiger handler -> APS seems to use a different file naming.
"""
specs = {"AD_EIGER_APSPolar"}
def __call__(self, image_num):
"""
This returns data contained in the file.
Parameters
----------
image_num int
Image number as read from eiger.cam.num_images_counter
Returns
-------
A dask array
"""
fpath = Path(
f"{self._file_prefix}_data_"
f"{1 + (image_num // self._images_per_file):06d}.h5"
).absolute()
try:
file = self._files[fpath]
except KeyError:
file = File(fpath, "r")
self._files[fpath] = file
da = from_array(file["entry/data/data"])[
image_num % self._images_per_file
]
return da.reshape((1,) + da.shape)
[docs]
def get_file_list(self):
"""
Get the file list.
"""
return glob(f"{self._file_prefix}_*")
[docs]
def get_file_sizes(self):
"""
Get the file size.
Returns size in bytes.
"""
sizes = []
file_name = self.get_file_list()
for file in file_name:
sizes.append(getsize(file))
return sizes