diff --git a/slic/devices/general/detectors/camera_bs.py b/slic/devices/general/detectors/camera_bs.py index 15080749..f006ddfe 100644 --- a/slic/devices/general/detectors/camera_bs.py +++ b/slic/devices/general/detectors/camera_bs.py @@ -1,35 +1,21 @@ -class CameraBS: +from .camerabase import CameraBase - def __init__(self, host=None, port=None, elog=None): - self._stream_host = host - self._stream_port = port - def checkServer(self): - # Check if your instance is running on the server. - if self._instance_id not in client.get_server_info()["active_instances"]: - raise ValueError("Requested pipeline is not running.") +class CameraBS(CameraBase): - def get_images(self, N_images): - data = [] - with source(host=self._stream_host, port=self._stream_port, mode=SUB) as input_stream: - input_stream.connect() + def __init__(self, host=None, port=None): + self.host = host + self.port = port - for n in range(N_images): - data.append(input_stream.receive().data.data["image"].value) - return data + def get_image(self): + return next(self._iterate_receive(1)) - def record_images(self, fina, N_images, dsetname="images"): - ds = None - with h5py.File(fina, "w") as f: - with source(host=self._stream_host, port=self._stream_port, mode=SUB) as input_stream: - - input_stream.connect() - - for n in range(N_images): - image = input_stream.receive().data.data["image"].value - if not ds: - ds = f.create_dataset(dsetname, dtype=image.dtype, shape=(N_images,) + image.shape) - ds[n, :, :] = image + def _iterate_receive(self, n): + with source(host=self.host, port=self.port, mode=SUB) as stream: + for i in range(n): + message = stream.receive() + img = message.data.data["image"].value + yield img diff --git a/slic/devices/general/detectors/camera_ca.py b/slic/devices/general/detectors/camera_ca.py index d6ba52d7..d2e12504 100644 --- a/slic/devices/general/detectors/camera_ca.py +++ b/slic/devices/general/detectors/camera_ca.py @@ -1,50 +1,57 @@ -_cameraArrayTypes = ["monochrome", "rgb"] +from slic.utils.hastyepics import get_pv as PV + +from .camerabase import CameraBase -class CameraCA: +class CameraCA(CameraBase): - def __init__(self, pvname, cameraArrayType="monochrome", elog=None): - self.ID = pvname - self.isBS = False - self.px_height = None - self.px_width = None - self.elog = elog + def __init__(self, ID, wait_time=0.2): + self.ID = ID + self.wait_time = wait_time - def get_px_height(self): - if not self.px_height: - self.px_height = caget(self.ID + ":HEIGHT") - return int(self.px_height) - - def get_px_width(self): - if not self.px_width: - self.px_width = caget(self.ID + ":WIDTH") - return int(self.px_width) - - def get_data(self): - w = self.get_px_width() - h = self.get_px_height() - numpix = int(caget(self.ID + ":FPICTURE.NORD")) - i = caget(self.ID + ":FPICTURE", count=numpix) - return i.reshape(h, w) - - def record_images(self, fina, N_images, sleeptime=0.2): - with h5py.File(fina, "w") as f: - d = [] - for n in range(N_images): - d.append(self.get_data()) - sleep(sleeptime) - f["images"] = np.asarray(d) - - def gui(self, guiType="xdm"): - """ Adjustable convention""" - cmd = ["caqtdm", "-macro"] - - cmd.append('"NAME=%s,CAMNAME=%s"' % (self.ID, self.ID)) - cmd.append("/sf/controls/config/qt/Camera/CameraMiniView.ui") - return subprocess.Popen(" ".join(cmd), shell=True) + self.pv_image = PV(ID + ":FPICTURE") + self.pv_size = PV(ID + ":FPICTURE.NORD") + self.pv_height = PV(ID + ":HEIGHT") + self.pv_width = PV(ID + ":WIDTH") -# /sf/controls/config/qt/Camera/CameraMiniView.ui" with macro "NAME=SAROP21-PPRM138,CAMNAME=SAROP21-PPRM138 + @property + def size(self): + return int(self.pv_size.get()) + + @property + def shape(self): + return (self.height, self.width) + + @property + def height(self): + return int(self.pv_height.get()) + + @property + def width(self): + return int(self.pv_width.get()) + + + def get_image(self): + img = self.pv_image.get(count=self.size) + return img.reshape(self.shape) + + def _iterate_receive(self, n): + for i in range(n): + img = self.get_image() + yield img + sleep(self.wait_time) + + + def gui(self): + ID = self.ID + cmd = [ + "caqtdm", + "-macro", + f'"NAME={ID},CAMNAME={ID}"', + "/sf/controls/config/qt/Camera/CameraMiniView.ui" + ] + return subprocess.Popen(cmd, shell=True) diff --git a/slic/devices/general/detectors/camerabase.py b/slic/devices/general/detectors/camerabase.py new file mode 100644 index 00000000..ca4cfa2c --- /dev/null +++ b/slic/devices/general/detectors/camerabase.py @@ -0,0 +1,28 @@ +from functools import partial + + +class CameraBase: + """ + Base class translating get_images and store_images to _iterate_receive via _fill_images + """ + + def get_images(self, n): + return _fill_images(n, np.empty) + + def store_images(self, n, fname, dataset_name="images"): + with h5py.File(fname, "w") as f: + create_empty = partial(f.create_dataset, dataset_name) + _fill_images(n, create_empty) + + def _fill_images(n, create_empty): + res = None + for image in self._iterate_receive(n): + if res is None: + shape = (n,) + image.shape + dtype = image.dtype + res = create_empty(shape=shape, dtype=dtype) + res[i] = image + return res + + +