irrad_control/devices/stage/xystage.py
def _scan_row(self, row, scan_params, speed=None, scan=-1, stage_pub=None):
        """
        Method which is called by self._scan_device or self.scan_row. See docstrings there.

        Parameters
        ----------
        row : int
            Row to scan
        scan_params : dict
            dict containing all the info for doing a scan of a rectangular area.
        speed : float, None
            Scan speed in mm/s or None. If None, current speed of x-axis is used for scanning
        scan : int
            Integer indicating the scan number during self.scan_device. *scan* for single rows is -1
        stage_pub : zmq.PUB, None
            Publisher socket on which to publish data. If None, open new one
        """

        # Check socket, if no socket is given, open one
        socket_close = stage_pub is None
        if stage_pub is None:
            stage_pub = self.context.socket(zmq.PUB)
            stage_pub.set_hwm(10)
            stage_pub.bind(scan_params['tcp_address'])

        # Check whether this method is called from within self.scan_device or single row is scanned.
        # If single row is scanned, we're coming from
        from_origin = (self.x_axis.get_position(), self.y_axis.get_position()) == scan_params['origin']

        if speed is not None:
            self.set_speed(speed, self.x_axis, unit='mm/s')

        # Make x start and end variables
        x_start, x_end = scan_params['start_pos'][0], scan_params['end_pos'][0]

        # Check whether we are scanning from origin
        if from_origin:
            x_reply = self.move_absolute(x_start, self.x_axis)

            # Check reply; if something went wrong raise error
            if not self._check_reply(x_reply):
                msg = "X-axis did not move to start point. Abort"
                raise UnexpectedReplyError(msg)

        # Move to the current row
        y_reply = self.move_absolute(scan_params['rows'][row], self.y_axis)

        # Check reply; if something went wrong raise error
        if not self._check_reply(y_reply):
            msg = "Y-axis did not move to row {}. Abort.".format(row)
            raise UnexpectedReplyError(msg)

        # Send start data
        _meta = {'timestamp': time.time(), 'name': scan_params['server'], 'type': 'stage'}
        _data = {'status': 'start', 'scan': scan, 'row': row,
                 'speed': self.get_speed(self.x_axis, unit='mm/s'),
                 'x_start': self.steps_to_distance(self.position[0], unit='mm'),
                 'y_start': self.steps_to_distance(self.position[1], unit='mm')}

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

        # Scan the current row
        x_reply = self.move_absolute(x_end if self.x_axis.get_position() == x_start else x_start, self.x_axis)

        # Check reply; if something went wrong raise error
        if not self._check_reply(x_reply):
            msg = "X-axis did not scan row {}. Abort.".format(row)
            raise UnexpectedReplyError(msg)

        # Send stop data
        _meta = {'timestamp': time.time(), 'name': scan_params['server'], 'type': 'stage'}
        _data = {'status': 'stop',
                 'x_stop': self.steps_to_distance(self.position[0], unit='mm'),
                 'y_stop': self.steps_to_distance(self.position[1], unit='mm')}

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

        if socket_close:
            stage_pub.close()

        if from_origin:
            # 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)

        Method which is called by self._scan_device or self.scan_row. See docstrings there.

        Parameters
        ----------
        row : int
            Row to scan
        scan_params : dict
            dict containing all the info for doing a scan of a rectangular area.
        speed : float, None
            Scan speed in mm/s or None. If None, current speed of x-axis is used for scanning
        scan : int
            Integer indicating the scan number during self.scan_device. *scan* for single rows is -1
        stage_pub : zmq.PUB, None
            Publisher socket on which to publish data. If None, open new one
        
Method which is called by self._scan_device or self.scan_row. See docstrings there.
Parameters

row : int
Row to scan
scan_params : dict
dict containing all the info for doing a scan of a rectangular area.
speed : float, None
Scan speed in mm/s or None. If None, current speed of x-axis is used for scanning
scan : int
Integer indicating the scan number during self.scan_device. *scan* for single rows is -1
stage_pub : zmq.PUB, None
Publisher socket on which to publish data. If None, open new one
Method which is called by self . _scan_device or self . scan_row . See docstrings there . Parameters row : int Row to scan scan_params : dict dict containing all the info for doing a scan of a rectangular area . speed : float None Scan speed in mm / s or None . If None current speed of x - axis is used for scanning scan : int Integer indicating the scan number during self . scan_device . * scan * for single rows is - 1 stage_pub : zmq . PUB None Publisher socket on which to publish data . If None open new one
def _scan_row(self, row, scan_params, speed=None, scan=-1, stage_pub=None):
        socket_close = stage_pub is None
        if stage_pub is None:
            stage_pub = self.context.socket(zmq.PUB)
            stage_pub.set_hwm(10)
            stage_pub.bind(scan_params['tcp_address'])
        from_origin = (self.x_axis.get_position(), self.y_axis.get_position()) == scan_params['origin']
        if speed is not None:
            self.set_speed(speed, self.x_axis, unit='mm/s')
        x_start, x_end = scan_params['start_pos'][0], scan_params['end_pos'][0]
        if from_origin:
            x_reply = self.move_absolute(x_start, self.x_axis)
            if not self._check_reply(x_reply):
                msg = "X-axis did not move to start point. Abort"
                raise UnexpectedReplyError(msg)
        y_reply = self.move_absolute(scan_params['rows'][row], self.y_axis)
        if not self._check_reply(y_reply):
            msg = "Y-axis did not move to row {}. Abort.".format(row)
            raise UnexpectedReplyError(msg)
        _meta = {'timestamp': time.time(), 'name': scan_params['server'], 'type': 'stage'}
        _data = {'status': 'start', 'scan': scan, 'row': row,
                 'speed': self.get_speed(self.x_axis, unit='mm/s'),
                 'x_start': self.steps_to_distance(self.position[0], unit='mm'),
                 'y_start': self.steps_to_distance(self.position[1], unit='mm')}
        stage_pub.send_json({'meta': _meta, 'data': _data})
        x_reply = self.move_absolute(x_end if self.x_axis.get_position() == x_start else x_start, self.x_axis)
        if not self._check_reply(x_reply):
            msg = "X-axis did not scan row {}. Abort.".format(row)
            raise UnexpectedReplyError(msg)
        _meta = {'timestamp': time.time(), 'name': scan_params['server'], 'type': 'stage'}
        _data = {'status': 'stop',
                 'x_stop': self.steps_to_distance(self.position[0], unit='mm'),
                 'y_stop': self.steps_to_distance(self.position[1], unit='mm')}
        stage_pub.send_json({'meta': _meta, 'data': _data})
        if socket_close:
            stage_pub.close()
        if from_origin:
            self.move_absolute(scan_params['origin'][1], self.y_axis)
            self.move_absolute(scan_params['origin'][0], self.x_axis)
def _scan_row ( self , row , scan_params , speed = None , scan = - 1 , stage_pub = None ) : socket_close = stage_pub is None if stage_pub is None : stage_pub = self . context . socket ( zmq . PUB ) stage_pub . set_hwm ( 10 ) stage_pub . bind ( scan_params [ 'tcp_address' ] ) from_origin = ( self . x_axis . get_position ( ) , self . y_axis . get_position ( ) ) == scan_params [ 'origin' ] if speed is not None : self . set_speed ( speed , self . x_axis , unit = 'mm/s' ) x_start , x_end = scan_params [ 'start_pos' ] [ 0 ] , scan_params [ 'end_pos' ] [ 0 ] if from_origin : x_reply = self . move_absolute ( x_start , self . x_axis ) if not self . _check_reply ( x_reply ) : msg = "X-axis did not move to start point. Abort" raise UnexpectedReplyError ( msg ) y_reply = self . move_absolute ( scan_params [ 'rows' ] [ row ] , self . y_axis ) if not self . _check_reply ( y_reply ) : msg = "Y-axis did not move to row {}. Abort." . format ( row ) raise UnexpectedReplyError ( msg ) _meta = { 'timestamp' : time . time ( ) , 'name' : scan_params [ 'server' ] , 'type' : 'stage' } _data = { 'status' : 'start' , 'scan' : scan , 'row' : row , 'speed' : self . get_speed ( self . x_axis , unit = 'mm/s' ) , 'x_start' : self . steps_to_distance ( self . position [ 0 ] , unit = 'mm' ) , 'y_start' : self . steps_to_distance ( self . position [ 1 ] , unit = 'mm' ) } stage_pub . send_json ( { 'meta' : _meta , 'data' : _data } ) x_reply = self . move_absolute ( x_end if self . x_axis . get_position ( ) == x_start else x_start , self . x_axis ) if not self . _check_reply ( x_reply ) : msg = "X-axis did not scan row {}. Abort." . format ( row ) raise UnexpectedReplyError ( msg ) _meta = { 'timestamp' : time . time ( ) , 'name' : scan_params [ 'server' ] , 'type' : 'stage' } _data = { 'status' : 'stop' , 'x_stop' : self . steps_to_distance ( self . position [ 0 ] , unit = 'mm' ) , 'y_stop' : self . steps_to_distance ( self . position [ 1 ] , unit = 'mm' ) } stage_pub . send_json ( { 'meta' : _meta , 'data' : _data } ) if socket_close : stage_pub . close ( ) if from_origin : self . move_absolute ( scan_params [ 'origin' ] [ 1 ] , self . y_axis ) self . move_absolute ( scan_params [ 'origin' ] [ 0 ] , self . x_axis )
Method which is called by self._scan_device or self.scan_row.
Method which is called by self . _scan_device or self . scan_row .
"""
        Method which is called by self._scan_device or self.scan_row. See docstrings there.

        Parameters
        ----------
        row : int
            Row to scan
        scan_params : dict
            dict containing all the info for doing a scan of a rectangular area.
        speed : float, None
            Scan speed in mm/s or None. If None, current speed of x-axis is used for scanning
        scan : int
            Integer indicating the scan number during self.scan_device. *scan* for single rows is -1
        stage_pub : zmq.PUB, None
            Publisher socket on which to publish data. If None, open new one
        """ # Check socket, if no socket is given, open one # Check whether this method is called from within self.scan_device or single row is scanned. # If single row is scanned, we're coming from # Make x start and end variables # Check whether we are scanning from origin # Check reply; if something went wrong raise error # Move to the current row # Check reply; if something went wrong raise error # Send start data # Publish data # Scan the current row # Check reply; if something went wrong raise error # Send stop data # Publish data # Move back to origin; move y first in order to not scan over device
self
row
scan_params
speed
scan
stage_pub