diff --git a/DCAMCamera/blacs_tabs.py b/DCAMCamera/blacs_tabs.py
index b6acf27e7c61683a70dd924f33f27ad71663444d..b55abdf431c570e2e71c8c4bd72095988612b870 100644
--- a/DCAMCamera/blacs_tabs.py
+++ b/DCAMCamera/blacs_tabs.py
@@ -3,11 +3,14 @@
# /user_devices/DCAMCamera/blacs_tabs.py #
# #
# Jan 2023, Marvin Holten #
+# 2025, Johannes Schabbauer #
# #
# #
#####################################################################
from labscript_devices.IMAQdxCamera.blacs_tabs import IMAQdxCameraTab
+import labscript_utils.h5_lock
+import h5py
class DCAMCameraTab(IMAQdxCameraTab):
"""Thin sub-class of obj:`IMAQdxCameraTab`.
@@ -16,4 +19,30 @@ class DCAMCameraTab(IMAQdxCameraTab):
:obj:`DCAMCameraWorker`."""
# override worker class
- worker_class = 'user_devices.DCAMCamera.blacs_workers.DCAMCameraWorker'
\ No newline at end of file
+ worker_class = 'user_devices.DCAMCamera.blacs_workers.DCAMCameraWorker'
+
+ def initialise_workers(self):
+ table = self.settings['connection_table']
+ connection_table_properties = table.find_by_name(self.device_name).properties
+ # The device properties can vary on a shot-by-shot basis, but at startup we will
+ # initially set the values that are configured in the connection table, so they
+ # can be used for manual mode acquisition:
+ with h5py.File(table.filepath, 'r') as f:
+ device_properties = labscript_utils.properties.get(
+ f, self.device_name, "device_properties"
+ )
+ worker_initialisation_kwargs = {
+ 'serial_number': connection_table_properties['serial_number'],
+ 'orientation': connection_table_properties['orientation'],
+ 'camera_attributes': device_properties['camera_attributes'],
+ 'manual_mode_camera_attributes': connection_table_properties[
+ 'manual_mode_camera_attributes'
+ ],
+ 'mock': connection_table_properties['mock'],
+ 'image_receiver_port': self.image_receiver.port,
+ 'occupation_receiver_port' : connection_table_properties['occupation_receiver_port']
+ }
+ self.create_worker(
+ 'main_worker', self.worker_class, worker_initialisation_kwargs
+ )
+ self.primary_worker = "main_worker"
\ No newline at end of file
diff --git a/DCAMCamera/blacs_workers.py b/DCAMCamera/blacs_workers.py
index 03bacb39cf5631a3dcd876a617cbccdc599cf8ea..a98ac1371ff95b4956bf54e43281b4fe1ed8d92e 100644
--- a/DCAMCamera/blacs_workers.py
+++ b/DCAMCamera/blacs_workers.py
@@ -3,11 +3,20 @@
# /user_devices/DCAMCamera/blacs_workers.py #
# #
# Jan 2023, Marvin Holten #
+# 2025, Johannes Schabbauer #
# #
# #
#####################################################################
from labscript_devices.IMAQdxCamera.blacs_workers import IMAQdxCameraWorker
+import threading
+import numpy as np
+import labscript_utils.h5_lock
+import h5py
+import labscript_utils.properties
+import zmq
+from labscript_utils.ls_zprocess import Context
+from labscript_utils.shared_drive import path_to_local
# Don't import API yet so as not to throw an error, allow worker to run as a dummy
# device, or for subclasses to import this module to inherit classes without requiring API
@@ -204,7 +213,7 @@ class DCAM_Camera(object):
return image
- def grab_multiple(self, n_images, images):
+ def grab_multiple(self, n_images, images, tweezer_socket=None, tweezer_img_no=None, get_occupation=None):
"""Grab n_images into images array during buffered acquistion.
Grab method involves a continuous loop with fast timeout in order to
@@ -222,6 +231,16 @@ class DCAM_Camera(object):
self._abort_acquisition = False
return
images.append(self.grab(bufferNo=i))
+ # Send image to occupation receiver
+ if tweezer_socket and tweezer_img_no==i:
+ occupation = get_occupation(images[-1])
+ metadata = dict(dtype=str(occupation.dtype), shape=occupation.shape)
+ tweezer_socket.send_json(metadata, zmq.SNDMORE)
+ tweezer_socket.send(occupation, copy=False)
+ print(f"Trying to send image {len(images)} to occupation receiver...", end="\r")
+ response = tweezer_socket.recv()
+ assert response == b'ok', response
+ print(f"Sent image {len(images)} to occupation receiver.")
print(f"Got image {i+1} of {n_images}.")
print(f"Got {len(images)} of {n_images} images.")
@@ -248,6 +267,16 @@ class DCAMCameraWorker(IMAQdxCameraWorker):
:obj:`get_attributes_as_dict` to use DCAMCameraWorker.get_attributes() method."""
interface_class = DCAM_Camera
+ def init(self):
+ super().init()
+
+ # Connect to occupation matrix receiver port for conditional Tweezer programming
+ if self.occupation_receiver_port is not None:
+ self.tweezer_socket = Context().socket(zmq.REQ)
+ self.tweezer_socket.connect(
+ f'tcp://{self.parent_host}:{self.occupation_receiver_port}'
+ )
+
def get_attributes_as_dict(self, visibility_level):
"""Return a dict of the attributes of the camera for the given visibility
level
@@ -260,5 +289,91 @@ class DCAMCameraWorker(IMAQdxCameraWorker):
return IMAQdxCameraWorker.get_attributes_as_dict(self,visibility_level)
else:
return self.camera.get_attributes(visibility_level)
+
+ def transition_to_buffered(self, device_name, h5_filepath, initial_values, fresh):
+ if getattr(self, 'is_remote', False):
+ h5_filepath = path_to_local(h5_filepath)
+ if self.continuous_thread is not None:
+ # Pause continuous acquistion during transition_to_buffered:
+ self.stop_continuous(pause=True)
+ with h5py.File(h5_filepath, 'r') as f:
+ group = f['devices'][self.device_name]
+ if not 'EXPOSURES' in group:
+ return {}
+ self.h5_filepath = h5_filepath
+ self.exposures = group['EXPOSURES'][:]
+ self.n_images = len(self.exposures)
+
+ # Get the camera_attributes from the device_properties
+ properties = labscript_utils.properties.get(
+ f, self.device_name, 'device_properties'
+ )
+ camera_attributes = properties['camera_attributes']
+ self.stop_acquisition_timeout = properties['stop_acquisition_timeout']
+ self.exception_on_failed_shot = properties['exception_on_failed_shot']
+ saved_attr_level = properties['saved_attribute_visibility_level']
+ self.camera.exception_on_failed_shot = self.exception_on_failed_shot
+
+ ### ADDED CODE TO PASS OCCUPATION RECEIVER ARGUMENTS TO grab_multiple() ###
+ self.images = []
+ if properties["occupation_receiver_image_index"] is not None:
+ tweezers_centers = f["globals"].attrs["tweezers_centers"]
+ thresholds = f["globals"].attrs["thresholds"]
+ ROI_size = f["globals"].attrs["ROI_size"]
+ get_occupation = lambda image: self.get_occupation(image,tweezers_centers,thresholds, ROI_size)
+ args = (self.n_images, self.images,self.tweezer_socket, properties["occupation_receiver_image_index"], get_occupation)
+ else:
+ # Standard args from IMAQdxCameraWorker
+ args = (self.n_images, self.images)
+
+
+ # Only reprogram attributes that differ from those last programmed in, or all of
+ # them if a fresh reprogramming was requested:
+ if fresh:
+ self.smart_cache = {}
+ self.set_attributes_smart(camera_attributes)
+ # Get the camera attributes, so that we can save them to the H5 file:
+ if saved_attr_level is not None:
+ self.attributes_to_save = self.get_attributes_as_dict(saved_attr_level)
+ else:
+ self.attributes_to_save = None
+ print(f"Configuring camera for {self.n_images} images.")
+ self.camera.configure_acquisition(continuous=False, bufferCount=self.n_images)
+
+
+ self.acquisition_thread = threading.Thread(
+ target=self.camera.grab_multiple,
+ args=args,
+ daemon=True,
+ )
+ self.acquisition_thread.start()
+ return {}
+
+ def get_occupation(self, image, tweezer_centers, thresholds, size_px=3):
+ """
+ Calculate the Tweezer occupancy of Tweezers, either 0 or 1.
+
+ Parameters
+ ----------
+ image : ndarray
+ Monochrome picture.
+ tweezers_centers : ndarray
+ Indices where the Tweezers aer in the image
+ thresholds : float or ndarray
+ Threshold in px counts, that distinghishes no atom or 1 atom.
+ Same threashold for all Tweezers, if a single number. If ndarray,
+ the size has to be equal to the size of tweezers_centers.
+ size_px : int
+ ROI around each tweezer center, where the pixels get summed up.
+ """
+ px_sum = np.zeros(tweezer_centers.shape[0])
+ lower = size_px//2
+ upper = size_px-lower
+ for i in range(tweezer_centers.shape[0]):
+ x,y = tweezer_centers[i,:]
+ px_sum[i] = image[x-lower:x+upper, y-lower:y+upper].sum()
+
+ return px_sum > thresholds
+
diff --git a/DCAMCamera/labscript_devices.py b/DCAMCamera/labscript_devices.py
index 79a9f2689a2964f2cbb447dd2c032d50eb1f07de..bd52783d4d09a8ddeba9995f12f141b0ae4ed119 100644
--- a/DCAMCamera/labscript_devices.py
+++ b/DCAMCamera/labscript_devices.py
@@ -3,10 +3,14 @@
# /user_devices/DCAMCamera/labscript_devices.py #
# #
# Jan 2023, Marvin Holten #
+# 2025, Johannes Schabbauer #
+
# #
# #
#####################################################################
+import h5py
+from labscript import set_passed_properties, LabscriptError
from labscript_devices.IMAQdxCamera.labscript_devices import IMAQdxCamera
class DCAMCamera(IMAQdxCamera):
@@ -14,3 +18,29 @@ class DCAMCamera(IMAQdxCamera):
description = 'DCAM Camera'
+ @set_passed_properties(
+ property_names={
+ "connection_table_properties": [
+ "occupation_receiver_port",
+ ],
+ }
+ )
+ def __init__(self, name, parent_device, connection, serial_number, orientation=None, pixel_size=..., magnification=1, trigger_edge_type='rising', trigger_duration=None, minimum_recovery_time=0, camera_attributes=None, manual_mode_camera_attributes=None, stop_acquisition_timeout=5, exception_on_failed_shot=True, saved_attribute_visibility_level='intermediate', occupation_receiver_port=None, mock=False, **kwargs):
+ self.occupation_receiver_image_index = None
+ self.exposure_times = []
+ super().__init__(name, parent_device, connection, serial_number, orientation, pixel_size, magnification, trigger_edge_type, trigger_duration, minimum_recovery_time, camera_attributes, manual_mode_camera_attributes, stop_acquisition_timeout, exception_on_failed_shot, saved_attribute_visibility_level, mock, **kwargs)
+
+ def expose(self, t, name, frametype='frame', trigger_duration=None, send_image_to_occupation_receiver=False):
+ self.exposure_times.append(t)
+ if send_image_to_occupation_receiver:
+ if getattr(self,"occupation_receiver_time",None) is not None:
+ raise LabscriptError(f"{self.name}: In the current implementation only a single image can be sent to the 'occupation receiver'")
+ self.occupation_receiver_time = t
+
+ return super().expose(t, name, frametype, trigger_duration)
+
+ def generate_code(self, hdf5_file):
+ super().generate_code(hdf5_file)
+ if getattr(self,"occupation_receiver_time",None) is not None:
+ self.occupation_receiver_image_index = self.exposure_times.index(self.occupation_receiver_time)
+ hdf5_file[f"devices/{self.name}"].attrs["occupation_receiver_image_index"] = self.occupation_receiver_image_index