python-microscopy · barentine · Oct 10, 2024 · Jan 29, 2025 · Jan 30, 2025 · Feb 24, 2025
diff --git a/PYME/Acquire/Hardware/Camera.py b/PYME/Acquire/Hardware/Camera.py
@@ -111,7 +111,14 @@ def fill_camera_map_metadata(self, mdh):
 class Camera(object):
     # Frame format - PYME previously supported frames in a custom format, but numpy_frames should always be true for current code
     numpy_frames = 1 #Frames are delivered as numpy arrays.
-
+    # what data type is returned by the camera? For full support it is recommended to use 
+    # a format supported in `PYME.IO.PZFFormat`
+    dtype = 'uint16'  # uint16 is the default for PYME, and historically the only supported dtype of PYMEAcquire
+
+    # number of channels in the camera output. This should be 1 for monochrome cameras, which should use splitter or 
+    # multiview approaches for implementing multi-color imaging.
+    n_channels = 1
+
     # Acquisition modes
     MODE_SINGLE_SHOT = 0
     MODE_CONTINUOUS = 1

diff --git a/PYME/Acquire/Hardware/pointscan_shim/init_sim_pointscan_shim.py b/PYME/Acquire/Hardware/pointscan_shim/init_sim_pointscan_shim.py
@@ -0,0 +1,96 @@
+
+from PYME.Acquire.ExecTools import joinBGInit, HWNotPresent, init_gui, init_hardware
+import time
+
+
+
+@init_hardware('Pointscan Camera')
+def pointscan_cam(scope):
+    from PYME.Acquire.Hardware.pointscan_shim import pointscan_camera
+    import threading  
+    import queue
+    import numpy as np
+    import time
+
+    class TestScanner(pointscan_camera.BaseScanner):
+        """ A (buffered) fake image generator for testing the camera shim.
+        """
+        n_channels = 3
+        def __init__(self, scan_params=None, axes_order=('x', 'y')):
+            super().__init__(scan_params, axes_order)
+
+        @pointscan_camera.BaseScanner.frame_integ_time.getter
+        def frame_integ_time(self):
+            return 1
+
+        def wait_for_finished_buffer(self, timeout):
+            t0 = time.time()
+            while time.time() - t0 < timeout:
+                if self.n_full > 0:
+                    with self.full_buffer_lock:
+                        return self.full_buffers.get()
+                time.sleep(0.05)
+            raise TimeoutError('Timed out waiting for scanner buffer')
+
+        def get_serial_number(self):
+            return 'Number0'
+
+        def stop(self):
+            pass
+
+        def _scan(self):
+            time.sleep(self.frame_integ_time)
+            for ind in range(self.n_channels):
+                buf = self.free_buffers.get_nowait()
+                buf[:] = ind * np.ones((self.height, self.width), dtype=self.dtype)
+                with self.full_buffer_lock:
+                    self.full_buffers.put(buf)
+                    self.n_full += 1
+
+        def scan(self):
+            """raster-scan an image and add each channel as a separate frame to the full frame
+            buffer
+
+            """
+            t = threading.Thread(target=self._scan)
+            t.start()
+
+        # def allocate_buffers(self, n_buffers):
+        #     """queue up a number of single-frame buffers
+        #     Note that each channel will be slotted into the queue as a separate frame (for now)
+        #     Parameters
+        #     ----------
+        #     n_buffers : int
+        #         number of single-frame (XY) buffers to allocate
+
+        #     """
+        #     self.free_buffers = queue.Queue()
+        #     self.full_buffers = queue.Queue()
+        #     self.n_full = 0
+        #     for ind in range(n_buffers):
+        #         self.free_buffers.put(np.zeros((self.width, self.height), 
+        #                                     dtype=self.dtype))
+
+        # def destroy_buffers(self):
+        #     with self.full_buffer_lock:
+        #         self.n_full = 0
+        #         while not self.full_buffers.empty():
+        #             try:
+        #                 self.full_buffers.get_nowait()
+        #             except queue.Empty:
+        #                 pass
+
+
+    cam = pointscan_camera.PointscanCameraShim(position_scanner=TestScanner(scan_params={
+        'n_x': 10,  # [px]
+        'n_y': 10,  # [px]
+    }))
+
+    scope.register_camera(cam, 'Test')
+
+#must be here!!!
+joinBGInit() #wait for anyhting which was being done in a separate thread
+
+
+time.sleep(.5)
+scope.initDone = True
diff --git a/PYME/Acquire/Hardware/pointscan_shim/init_sim_software_scanner.py b/PYME/Acquire/Hardware/pointscan_shim/init_sim_software_scanner.py
@@ -0,0 +1,38 @@
+
+from PYME.Acquire.ExecTools import joinBGInit, HWNotPresent, init_gui, init_hardware
+import time
+
+
+
+@init_hardware('Pointscan Camera')
+def pointscan_cam(scope):
+    from PYME.Acquire.Hardware.pointscan_shim import pointscan_camera
+    from PYME.Acquire.Hardware.pointscan_shim import software_scanner
+    from PYME.Acquire.Hardware.Simulator import fakePiezo
+
+    scope.fakeXPiezo = fakePiezo.FakePiezo(10000)
+    scope.register_piezo(scope.fakeXPiezo, 'x')
+
+    scope.fakeYPiezo = fakePiezo.FakePiezo(10000)
+    scope.register_piezo(scope.fakeYPiezo, 'y')
+
+    scope.fakeZPiezo = fakePiezo.FakePiezo(100)
+    scope.register_piezo(scope.fakeYPiezo, 'z')
+
+    sig_provider = software_scanner.TestSignalProvider(scope.fakeXPiezo, scope.fakeYPiezo)
+    scanner = software_scanner.SoftwareStageScanner(scope.fakeXPiezo, scope.fakeYPiezo, sig_provider, kwargs={
+        'scan_params': {
+            'n_x': 11,  # [px]
+            'n_y': 11
+            }
+        })
+    cam = pointscan_camera.PointscanCameraShim(position_scanner=scanner)
+
+    scope.register_camera(cam, 'Test')
+
+#must be here!!!
+joinBGInit() #wait for anyhting which was being done in a separate thread
+
+
+time.sleep(.5)
+scope.initDone = True
diff --git a/PYME/Acquire/Hardware/pointscan_shim/nidaq_scanner.py b/PYME/Acquire/Hardware/pointscan_shim/nidaq_scanner.py
@@ -0,0 +1,246 @@
+
+import numpy as np
+from PYME.Acquire.Hardware.pointscan_shim import pointscan_camera
+import nidaqmx as ni
+from nidaqmx.constants import TerminalConfiguration, Edge, FrequencyUnits, VoltageUnits
+import threading
+
+# drive_channels = {
+#     'x': {
+#         'channel': '/Dev1/ao0',
+#         'min_val': -1,  # [V]
+#         'max_val': 1,  # [V]
+#         'min_position': -10,  # [um] The position min_val corresponds to
+#         'max_position': 10,  # [um] The position max_val corresponds to
+#         'units': VoltageUnits.VOLTS,
+#     },
+#     'y':{
+#         'channel': '/Dev1/ao1',
+#         'min_val': -1,
+#         'max_val': 1,
+#         'min_position': -5,  # [um] The position min_val corresponds to
+#         'max_position': 5,  # [um] The position max_val corresponds to
+#         'units': VoltageUnits.VOLTS,
+#     }
+# }
+
+# signal_channels = {
+#     'SX': {
+#         'channel': '/Dev1/ai0',
+#         'min_val': -1,
+#         'max_val': 1,
+#         'units': VoltageUnits.VOLTS,
+#         'terminal_config': TerminalConfiguration.RSE
+#     },
+# }
+
+# counter_channel = 'Dev1/ctr0'
+
+class NIDAQScanner(pointscan_camera.BaseScanner):
+    dtype = 'float64'
+    def __init__(self, drive_channels, signal_channels, counter_channel='Dev1/ctr0',
+                 duty_cycle=0.9, pixel_clock_rate=1000, bidirectional=True, 
+                 return_to_start=True, kwargs=None):
+        super().__init__(**kwargs)
+        self._scanning_lock = threading.Lock()
+        self._drive_channels = drive_channels
+        self._signal_channels = signal_channels
+        self._counter_channel = counter_channel
+
+        self._duty_cycle = duty_cycle  # (0, 1]
+        self.pixel_clock_rate = pixel_clock_rate  # [Hz]
+        self._return_to_start = return_to_start
+        self._bidirectional = bidirectional
+
+        self.map_drive_voltages_to_positions()
+        self._x_c = self._drive_channels['x']['min_position'] + self._x_pos_range / 2
+        self._y_c = self._drive_channels['y']['min_position'] + self._y_pos_range / 2
+
+        # get device properties
+        dev_name = self._counter_channel.lstrip('/').split('/')[0]
+
+        self._model = ni.system.System.local().devices[dev_name].product_type
+        self._serial = str(ni.system.System.local().devices[dev_name].serial_num)
+
+    def map_drive_voltages_to_positions(self):
+        self._x_pos_range = self._drive_channels['x']['max_position'] - self._drive_channels['x']['min_position']
+        self._x_volt_range = self._drive_channels['x']['max_val'] - self._drive_channels['x']['min_val']
+        self.x_um_per_volt = self._x_pos_range / self._x_volt_range
+
+        self._y_pos_range = self._drive_channels['y']['max_position'] - self._drive_channels['y']['min_position']
+        self._y_volt_range = self._drive_channels['y']['max_val'] - self._drive_channels['y']['min_val']
+        self.y_um_per_volt = self._y_pos_range / self._y_volt_range
+
+        self.x_volt_from_um = lambda x: (x - self._drive_channels['x']['min_position']) / self.x_um_per_volt + self._drive_channels['x']['min_val']
+        self.y_volt_from_um = lambda y: (y - self._drive_channels['y']['min_position']) / self.y_um_per_volt + self._drive_channels['y']['min_val']
+
+    @property
+    def scan_center(self):
+        return self._x_c, self._y_c
+
+    @scan_center.setter
+    def scan_center(self, center):
+        self._x_c, self._y_c = center
+
+    @property
+    def return_to_start(self):
+        return self._return_to_start
+
+    @property
+    def n_channels(self):
+        return len(self._signal_channels)
+
+    def prepare(self):
+        """
+        map pixel positions to voltage values
+        """
+        x0, y0 = self.scan_center
+        self._axes_voltages = {}
+        # self.axis_voltages = {}
+        # self.n_scan_positions = 1
+
+        # generate the grid, in position space, then map to control voltages
+        # FIXME - doesn't handle even sized grids
+        half_x = self.width // 2
+        x_pixelsize = self._scan_params['voxelsize.x']
+        x_pos = np.linspace(-half_x, half_x, self.width, endpoint=True) * x_pixelsize + x0
+
+        half_y = self.height // 2
+        y_pixelsize = self._scan_params['voxelsize.y']
+        y_pos = np.linspace(-half_y, half_y, self.height, endpoint=True) * y_pixelsize + y0
+
+        # raise an error if the scan goes outside the drive range
+        if np.any(x_pos < self._drive_channels['x']['min_position']) or np.any(x_pos > self._drive_channels['x']['max_position']):
+            raise ValueError('X scan position out of range')
+        if np.any(y_pos < self._drive_channels['y']['min_position']) or np.any(y_pos > self._drive_channels['y']['max_position']):
+            raise ValueError('Y scan position out of range')
+
+        if self.axes_order[0] == 'x':
+            # make x scan faster
+            if not self._bidirectional:
+                self._axes_voltages['x'] = self.x_volt_from_um(np.tile(x_pos, self.height))
+            else: # bi-directional on fast axis
+                self._axes_voltages['x'] = self.x_volt_from_um(x_pos)  # 0th iteration
+                for h_ind in range(1, self.height):  # 1-thru
+                    self._axes_voltages['x'] = np.concatenate([self._axes_voltages['x'], self.x_volt_from_um(x_pos[::-1]) if h_ind % 2 else self.x_volt_from_um(x_pos)])
+            # slow axis
+            self._axes_voltages['y'] = self.y_volt_from_um(np.repeat(y_pos, self.width))
+        else: # make y fast axis
+            if not self._bidirectional:
+                self._axes_voltages['y'] = self.y_volt_from_um(np.tile(y_pos, self.width))  # 0th iteration
+            else:  # bi-directional on fast axis
+                self._axes_voltages['y'] = self.y_volt_from_um(y_pos)
+                for w_ind in range(1, self.width):  # 1-thru
+                    self._axes_voltages['y'] = np.concatenate([self._axes_voltages['y'], self.y_volt_from_um(y_pos[::-1]) if w_ind % 2 else self.y_volt_from_um(y_pos)])
+            # slow axis
+            self._axes_voltages['x'] = self.x_volt_from_um(np.repeat(x_pos, self.height))
+
+        self.n_steps = len(self._axes_voltages['x'])
+
+    def scan(self, wait_until_done=False):
+        t = threading.Thread(target=self._scan)
+        t.start()
+        if wait_until_done:
+            t.join()
+
+    def _scan_continuously(self):
+        while self.keep_scanning:
+            self.scan(wait_until_done=True)
+
+    def scan_continuously(self):
+        threading.Thread(target=self._scan_continuously).start()
+
+    def _scan(self):
+        with self._scanning_lock, ni.Task() as ctr_task, ni.Task() as ai_task, ni.Task() as ao_task:
+            self.prepare()
+
+            # set up counter / pixel clock task
+            ctr_task.co_channels.add_co_pulse_chan_freq(
+                counter=self._counter_channel, 
+                units=FrequencyUnits.HZ, freq=self._pixel_clock_rate, 
+                duty_cycle=self._duty_cycle
+            )
+            ctr_task.timing.cfg_implicit_timing(
+                sample_mode=ni.constants.AcquisitionType.FINITE,
+                samps_per_chan=self.n_steps  # for finite mode this sets the number of pulses
+            )
+
+            # set up Analog Input task
+            for k, v in self._signal_channels.items():
+                ai_task.ai_channels.add_ai_voltage_chan(v['channel'],
+                                                        terminal_config=v['terminal_config'],
+                                                        min_val=v['min_val'], max_val=v['max_val'],
+                                                        units=v['units'])
+            # read signal on the falling edge of each pixel clock pulse
+            ai_task.timing.cfg_samp_clk_timing(source=f"/{self._counter_channel}InternalOutput",
+                                            rate=self._pixel_clock_rate, sample_mode=ni.constants.AcquisitionType.FINITE,
+                                            samps_per_chan=self.n_steps,
+                                            active_edge=Edge.FALLING)
+
+            # set up Analog Output task, order of X,Y here regardless of scan speed order
+            for v in [self._drive_channels['x'], self._drive_channels['y']]:
+                ao_task.ao_channels.add_ao_voltage_chan(v['channel'],
+                                                            min_val=v['min_val'], max_val=v['max_val'], 
+                                                            units=v['units'])
+            # move the stage on the rising edge of each pixel clock pulse
+            ao_task.timing.cfg_samp_clk_timing(source=f"/{self._counter_channel}InternalOutput",
+                                            rate=self._pixel_clock_rate, sample_mode=ni.constants.AcquisitionType.FINITE,
+                                            samps_per_chan=self.n_steps,
+                                            active_edge=Edge.RISING)
+
+
+            # "start" AI task (it will actually start when the clock starts)
+            ai_task.start()
+            # "start" the AO task (it will actually start when the clock starts)
+            ao_task.write(np.stack((self._axes_voltages['x'], 
+                                    self._axes_voltages['y'])), auto_start=True)
+            # start counter task (which should actually kick this whole thing off)
+            ctr_task.start()
+            # wait until we're done
+            ctr_task.wait_until_done(timeout=2 * self.n_steps / self._pixel_clock_rate)
+            # read data from AI task
+            data = ai_task.read(number_of_samples_per_channel=self.n_steps)
+            # print(data)
+
+        # write the scan buffer to the full frame buffer
+        for ind in range(self.n_channels):
+            buf = self.free_buffers.get_nowait()
+            if self.n_channels > 1:
+                # thought data should be (n_channels, n_steps), but comes as list of lists
+                buf[:] = np.asarray(data[ind]).reshape(self.width, self.height)
+            else:
+                buf[:] = np.asarray(data).reshape(self.width, self.height)
+            if self._bidirectional:  # have to unwrap the bi-directional scan
+                if self.axes_order[0] == 'x':
+                    buf[1::2, :] = buf[1::2, ::-1]
+                else:
+                    buf[:, 1::2] = buf[::-1, 1::2]
+            with self.full_buffer_lock:
+                self.full_buffers.put(buf)
+                self.n_full += 1
+
+        if self.return_to_start:
+            x0, y0 = self.scan_center
+            self.move_to_position(x0, y0)
+
+    def move_to_position(self, x, y, wait_until_done=True):
+        with ni.Task() as ao_task:
+            ao_task.ao_channels.add_ao_voltage_chan(self._drive_channels['x']['channel'],
+                                                    min_val=self._drive_channels['x']['min_val'], 
+                                                    max_val=self._drive_channels['x']['max_val'], 
+                                                    units=self._drive_channels['x']['units'])
+            ao_task.ao_channels.add_ao_voltage_chan(self._drive_channels['y']['channel'],
+                                                    min_val=self._drive_channels['y']['min_val'], 
+                                                    max_val=self._drive_channels['y']['max_val'], 
+                                                    units=self._drive_channels['y']['units'])
+            ao_task.write([self.x_volt_from_um(x),
+                        self.y_volt_from_um(y)], auto_start=True)
+            if wait_until_done:
+                ao_task.wait_until_done()
+
+
+    def get_serial_number(self):
+        return self._serial
+
+    def stop(self):
+        self.keep_scanning = False