irrad_control/devices/stage/xystage.py
def _scan_device(self, scan_params):
        """
        Method which is supposed to be called by self.scan_device. See docstring there.

        Parameters
        ----------
        scan_params : dict
            dict containing all the info for doing a scan of a rectangular area.
        """

        # initialize zmq data publisher
        stage_pub = self.context.socket(zmq.PUB)
        stage_pub.set_hwm(10)
        stage_pub.bind(scan_params['tcp_address'])

        # Move to start point
        self.move_absolute(scan_params['start_pos'][0], self.x_axis)
        self.move_absolute(scan_params['start_pos'][1], self.y_axis)

        # Set the scan speed
        self.set_speed(scan_params['speed'], self.x_axis, unit='mm/s')

        # Initialize scan
        _meta = {'timestamp': time.time(), 'name': scan_params['server'], 'type': 'stage'}
        _data = {'status': 'init', 'y_step': scan_params['step_size'], 'n_rows': scan_params['n_rows']}

        # Send init data
        stage_pub.send_json({'meta': _meta, 'data': _data})

        try:

            # Loop until fluence is reached and self.stop_scan event is set
            # Each scan is counted as one coverage of the entire area
            scan = 0
            while not (self.stop_scan.wait(1e-1) or self.finish_scan.wait(1e-1)):

                # Determine whether we're going from top to bottom or opposite
                _tmp_rows = list(range(scan_params['n_rows']) if scan % 2 == 0 else reversed(range(scan_params['n_rows'])))

                # Loop over rows
                for row in _tmp_rows:

                    # Check for emergency stop; if so, raise error
                    if self.stop_scan.wait(1e-1):
                        msg = "Scan was stopped manually"
                        raise UnexpectedReplyError(msg)

                    # Wait for beam current to be sufficient / beam to be on for scan
                    while self.no_beam.wait(1e-1):
                        msg = "Low beam current or no beam in row {} of scan {}. " \
                              "Waiting for beam current to rise.".format(row, scan)
                        logging.warning(msg)
                        time.sleep(1)

                        # If beam does not recover and we need to stop manually
                        if self.stop_scan.wait(1e-1):
                            msg = "Scan was stopped manually"
                            raise UnexpectedReplyError(msg)

                    # Scan row
                    self._scan_row(row=row, scan_params=scan_params, scan=scan, stage_pub=stage_pub)

                # Increment
                scan += 1

        # Some axis command didn't succeed or emergency exit was issued
        except UnexpectedReplyError:
            logging.exception("Scan aborted!")
            pass

        finally:

            # Send finished data
            _meta = {'timestamp': time.time(), 'name': scan_params['server'], 'type': 'stage'}
            _data = {'status': 'finished'}

            # Publish data
            stage_pub.send_json({'meta': _meta, 'data': _data})

            # Reset speeds
            self.set_speed(10, self.x_axis, unit='mm/s')
            self.set_speed(10, self.y_axis, unit='mm/s')

            # Move back to origin; move y first in order to not scan over device
            self.move_absolute(scan_params['origin'][1], self.y_axis)
            self.move_absolute(scan_params['origin'][0], self.x_axis)

            # Reset signal so one can scan again
            if self.stop_scan.is_set():
                self.stop_scan.clear()

            if self.finish_scan.is_set():
                self.finish_scan.clear()

            if self.no_beam.is_set():
                self.no_beam.clear()

            # Close publish socket
            stage_pub.close()

        Method which is supposed to be called by self.scan_device. See docstring there.

        Parameters
        ----------
        scan_params : dict
            dict containing all the info for doing a scan of a rectangular area.
        
Method which is supposed to be called by self.scan_device. See docstring there.
Parameters

scan_params : dict
dict containing all the info for doing a scan of a rectangular area.
Method which is supposed to be called by self . scan_device . See docstring there . Parameters scan_params : dict dict containing all the info for doing a scan of a rectangular area .
def _scan_device(self, scan_params):
        stage_pub = self.context.socket(zmq.PUB)
        stage_pub.set_hwm(10)
        stage_pub.bind(scan_params['tcp_address'])
        self.move_absolute(scan_params['start_pos'][0], self.x_axis)
        self.move_absolute(scan_params['start_pos'][1], self.y_axis)
        self.set_speed(scan_params['speed'], self.x_axis, unit='mm/s')
        _meta = {'timestamp': time.time(), 'name': scan_params['server'], 'type': 'stage'}
        _data = {'status': 'init', 'y_step': scan_params['step_size'], 'n_rows': scan_params['n_rows']}
        stage_pub.send_json({'meta': _meta, 'data': _data})
        try:
            scan = 0
            while not (self.stop_scan.wait(1e-1) or self.finish_scan.wait(1e-1)):
                _tmp_rows = list(range(scan_params['n_rows']) if scan % 2 == 0 else reversed(range(scan_params['n_rows'])))
                for row in _tmp_rows:
                    if self.stop_scan.wait(1e-1):
                        msg = "Scan was stopped manually"
                        raise UnexpectedReplyError(msg)
                    while self.no_beam.wait(1e-1):
                        msg = "Low beam current or no beam in row {} of scan {}. " \
                              "Waiting for beam current to rise.".format(row, scan)
                        logging.warning(msg)
                        time.sleep(1)
                        if self.stop_scan.wait(1e-1):
                            msg = "Scan was stopped manually"
                            raise UnexpectedReplyError(msg)
                    self._scan_row(row=row, scan_params=scan_params, scan=scan, stage_pub=stage_pub)
                scan += 1
        except UnexpectedReplyError:
            logging.exception("Scan aborted!")
            pass
        finally:
            _meta = {'timestamp': time.time(), 'name': scan_params['server'], 'type': 'stage'}
            _data = {'status': 'finished'}
            stage_pub.send_json({'meta': _meta, 'data': _data})
            self.set_speed(10, self.x_axis, unit='mm/s')
            self.set_speed(10, self.y_axis, unit='mm/s')
            self.move_absolute(scan_params['origin'][1], self.y_axis)
            self.move_absolute(scan_params['origin'][0], self.x_axis)
            if self.stop_scan.is_set():
                self.stop_scan.clear()
            if self.finish_scan.is_set():
                self.finish_scan.clear()
            if self.no_beam.is_set():
                self.no_beam.clear()
            stage_pub.close()
def _scan_device ( self , scan_params ) : stage_pub = self . context . socket ( zmq . PUB ) stage_pub . set_hwm ( 10 ) stage_pub . bind ( scan_params [ 'tcp_address' ] ) self . move_absolute ( scan_params [ 'start_pos' ] [ 0 ] , self . x_axis ) self . move_absolute ( scan_params [ 'start_pos' ] [ 1 ] , self . y_axis ) self . set_speed ( scan_params [ 'speed' ] , self . x_axis , unit = 'mm/s' ) _meta = { 'timestamp' : time . time ( ) , 'name' : scan_params [ 'server' ] , 'type' : 'stage' } _data = { 'status' : 'init' , 'y_step' : scan_params [ 'step_size' ] , 'n_rows' : scan_params [ 'n_rows' ] } stage_pub . send_json ( { 'meta' : _meta , 'data' : _data } ) try : scan = 0 while not ( self . stop_scan . wait ( 1e-1 ) or self . finish_scan . wait ( 1e-1 ) ) : _tmp_rows = list ( range ( scan_params [ 'n_rows' ] ) if scan % 2 == 0 else reversed ( range ( scan_params [ 'n_rows' ] ) ) ) for row in _tmp_rows : if self . stop_scan . wait ( 1e-1 ) : msg = "Scan was stopped manually" raise UnexpectedReplyError ( msg ) while self . no_beam . wait ( 1e-1 ) : msg = "Low beam current or no beam in row {} of scan {}. " "Waiting for beam current to rise." . format ( row , scan ) logging . warning ( msg ) time . sleep ( 1 ) if self . stop_scan . wait ( 1e-1 ) : msg = "Scan was stopped manually" raise UnexpectedReplyError ( msg ) self . _scan_row ( row = row , scan_params = scan_params , scan = scan , stage_pub = stage_pub ) scan += 1 except UnexpectedReplyError : logging . exception ( "Scan aborted!" ) pass finally : _meta = { 'timestamp' : time . time ( ) , 'name' : scan_params [ 'server' ] , 'type' : 'stage' } _data = { 'status' : 'finished' } stage_pub . send_json ( { 'meta' : _meta , 'data' : _data } ) self . set_speed ( 10 , self . x_axis , unit = 'mm/s' ) self . set_speed ( 10 , self . y_axis , unit = 'mm/s' ) self . move_absolute ( scan_params [ 'origin' ] [ 1 ] , self . y_axis ) self . move_absolute ( scan_params [ 'origin' ] [ 0 ] , self . x_axis ) if self . stop_scan . is_set ( ) : self . stop_scan . clear ( ) if self . finish_scan . is_set ( ) : self . finish_scan . clear ( ) if self . no_beam . is_set ( ) : self . no_beam . clear ( ) stage_pub . close ( )
Method which is supposed to be called by self.scan_device.
Method which is supposed to be called by self . scan_device .
"""
        Method which is supposed to be called by self.scan_device. See docstring there.

        Parameters
        ----------
        scan_params : dict
            dict containing all the info for doing a scan of a rectangular area.
        """ # initialize zmq data publisher # Move to start point # Set the scan speed # Initialize scan # Send init data # Loop until fluence is reached and self.stop_scan event is set # Each scan is counted as one coverage of the entire area # Determine whether we're going from top to bottom or opposite # Loop over rows # Check for emergency stop; if so, raise error # Wait for beam current to be sufficient / beam to be on for scan # If beam does not recover and we need to stop manually # Scan row # Increment # Some axis command didn't succeed or emergency exit was issued # Send finished data # Publish data # Reset speeds # Move back to origin; move y first in order to not scan over device # Reset signal so one can scan again # Close publish socket
self
scan_params