<commit_before>import datetime
import logging
import traceback

from .command import Command, CommandArgumentError, CommandError, CommandKilledError
from ..instrument.privileges import PRIV_BEAMSTOP, PRIV_MOVEMOTORS, PRIV_PINHOLE

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)


class GeneralScan(Command):
    name = '__generalscan'

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        assert self.name in ['scan', 'scanrel']
        if self.kwargs:
            raise CommandArgumentError('Command {} does not support keyword arguments.'.format(self.name))
        if self.name == 'scan':
            if len(self.args) != 6:
                raise CommandArgumentError('Command {} needs exactly six positional arguments.'.format(self.name))
        else:
            if len(self.args) != 5:
                raise CommandArgumentError('Command {} needs exactly five positional arguments.'.format(self.name))
        self.motorname = self.args[0]
        try:
            self.get_motor(self.motorname)
        except KeyError:
            raise CommandArgumentError('Unknown motor: {}'.format(self.motorname))
        if not self.services['accounting'].has_privilege(PRIV_MOVEMOTORS):
            raise CommandError('Insufficient privileges to move motors.')
        if (self.motorname in ['BeamStop_X', 'BeamStop_Y'] and
                not self.services['accounting'].has_privilege(PRIV_BEAMSTOP)):
            raise CommandError('Insufficient privileges to move the beamstop')
        if (self.motorname in ['PH1_X', 'PH1_Y', 'PH2_X', 'PH2_Y', 'PH3_X', 'PH3_Y'] and
                not self.services['accounting'].has_privilege(PRIV_PINHOLE)):
            raise CommandError('Insufficient privileges to move the pinholes')
        if self.name == 'scan':
            self.start = float(self.args[1])
            self.end = float(self.args[2])
            self.npoints = int(self.args[3])
            self.exptime = float(self.args[4])
            self.halfwidth = 0.5 * (self.end - self.start)
            self.comment = str(self.args[5])
        else:
            self.halfwidth = float(self.args[1])
            self.npoints = int(self.args[2])
            self.start = None  # to be determined just before execution
            self.end = None  # to be determined just before execution
            self.exptime = float(self.args[3])
            self.comment = str(self.args[4])
        if self.npoints < 2:
            raise CommandArgumentError('At least 2 points are required in a scan.')
        if self.exptime < 1e-6 or self.exptime > 1e6:
            raise CommandArgumentError('Exposure time must be between 1e-6 and 1e6 seconds.')
        self.idx = None
        self.scanfsn = None
        self.motorpos = None
        self.required_devices = ['pilatus', 'Motor_' + self.motorname]
        self.prefix = self.config['path']['prefixes']['scn']
        self.fsn_being_exposed = None
        self.file_being_exposed = None
        self.exposure_startdate = None
        self._ea_connection = None
        self._outstanding_scanpoints=0
        self._work_status=None
        self.killed = None

    def validate(self):
        motor = self.get_motor(self.motorname)
        if self.name == 'scanrel':
            pos = motor.where()
            self.start = pos - self.halfwidth
            self.end = pos + self.halfwidth
        else:
            assert self.name == 'scan'
        if not motor.checklimits(self.start):
            raise CommandArgumentError(
                'Start position is outside the software limits for motor {}'.format(self.motorname))
        if not motor.checklimits(self.end):
            raise CommandArgumentError(
                'Start position is outside the software limits for motor {}'.format(self.motorname))
        return True

    def on_variable_change(self, device, variablename, newvalue):
        if self.killed is not None:
            if self.killed:
                logger.debug('Calling die_on_kill() from on_variable_change()')
                self.die_on_kill()
            return False
        if device.name == 'pilatus' and variablename == '_status' and newvalue == 'idle' and self._work_status=='Exposing':
            # exposure ready. Submit it to exposureanalyzer.
            self.motorpos = self.get_motor(self.motorname).where()
            self.services['filesequence'].new_exposure(
                self.fsn_being_exposed, self.file_being_exposed, self.prefix, self.exposure_startdate,
                position=self.motorpos, scanfsn=self.scanfsn)
            # don't wait for exposureanalyzer, move to the next point
            self.idx += 1
            self.emit('progress', 'Scan running: {:d}/{:d}'.format(self.idx, self.npoints), self.idx / self.npoints)
            if self.idx < self.npoints:
                nextpos = self.start + (self.end - self.start) / (self.npoints - 1) * self.idx
                # self.emit('message', 'Moving motor {} to {:.3f}'.format(self.motorname, nextpos))
                self._work_status='Moving'
                self.get_motor(self.motorname).moveto(nextpos)
            else:
                # Otherwise this was the last point. We wait for exposureanalyzer to finish all its jobs.
                # We will test that case in self.on_scanpoint()
                self._work_status='Finalizing'
                self.emit('message', 'Finalizing scan #{:d}.'.format(self.scanfsn))
        if device.name == 'pilatus' and variablename == 'imgpath' and self._work_status == 'Initializing detector':
            self.emit('message', 'Moving motor {} to start position ({:.3f})'.format(self.motorname, self.start))
            self.get_motor(self.motorname).moveto(self.start)
            self._work_status = 'Moving'
        return False

    def on_scanpoint(self, exposureanalyzer, prefix, fsn, pos, counters):
        logger.debug('Scan command::on_scanpoint')
        self._outstanding_scanpoints -= 1
        if (self.idx >= self.npoints) and (self._outstanding_scanpoints <= 0):
            # the last scan point has been received.
            self.emit('message', 'Scan #{:d} finished.'.format(self.scanfsn))
            self.idle_return(self.scanfsn)
        return False

    def on_motor_stop(self, motor, targetreached):
        if self._work_status != 'Moving':
            # unexpected stop message, disregard it.
            return False
        assert (motor.name == self.motorname)
        if self.killed is not None:
            if self.killed:
                logger.debug('Calling die_on_kill() from on_motor_stop()')
                self.die_on_kill()
            return False
        if not targetreached:
            try:
                raise CommandError('Target position of motor {} not reached.'.format(self.motorname))
            except CommandError as ce:
                self.emit('fail', ce, traceback.format_exc())
            self.idle_return(None)
            return False
        # otherwise start the exposure
        self.fsn_being_exposed = self.services['filesequence'].get_nextfreefsn(self.prefix)
        self.file_being_exposed = self.services['filesequence'].exposurefileformat(
            self.prefix, self.fsn_being_exposed) + '.cbf'
        self.exposure_startdate = datetime.datetime.now()
        self._work_status='Exposing'
        self.get_device('pilatus').expose(self.file_being_exposed)
        self._outstanding_scanpoints += 1

    def execute(self):
        self.idx = 0
        self.killed = False
        try:
            cmdline = self.namespace['commandline']<commit_msg>Fixed bugs introduced in the scan command by previous commits<commit_after>import datetime
import logging
import traceback

from .command import Command, CommandArgumentError, CommandError, CommandKilledError
from ..instrument.privileges import PRIV_BEAMSTOP, PRIV_MOVEMOTORS, PRIV_PINHOLE

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)


class GeneralScan(Command):
    name = '__generalscan'

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        assert self.name in ['scan', 'scanrel']
        if self.kwargs:
            raise CommandArgumentError('Command {} does not support keyword arguments.'.format(self.name))
        if self.name == 'scan':
            if len(self.args) != 6:
                raise CommandArgumentError('Command {} needs exactly six positional arguments.'.format(self.name))
        else:
            if len(self.args) != 5:
                raise CommandArgumentError('Command {} needs exactly five positional arguments.'.format(self.name))
        self.motorname = self.args[0]
        try:
            self.get_motor(self.motorname)
        except KeyError:
            raise CommandArgumentError('Unknown motor: {}'.format(self.motorname))
        if not self.services['accounting'].has_privilege(PRIV_MOVEMOTORS):
            raise CommandError('Insufficient privileges to move motors.')
        if (self.motorname in ['BeamStop_X', 'BeamStop_Y'] and
                not self.services['accounting'].has_privilege(PRIV_BEAMSTOP)):
            raise CommandError('Insufficient privileges to move the beamstop')
        if (self.motorname in ['PH1_X', 'PH1_Y', 'PH2_X', 'PH2_Y', 'PH3_X', 'PH3_Y'] and
                not self.services['accounting'].has_privilege(PRIV_PINHOLE)):
            raise CommandError('Insufficient privileges to move the pinholes')
        if self.name == 'scan':
            self.start = float(self.args[1])
            self.end = float(self.args[2])
            self.npoints = int(self.args[3])
            self.exptime = float(self.args[4])
            self.halfwidth = 0.5 * (self.end - self.start)
            self.comment = str(self.args[5])
        else:
            self.halfwidth = float(self.args[1])
            self.npoints = int(self.args[2])
            self.start = None  # to be determined just before execution
            self.end = None  # to be determined just before execution
            self.exptime = float(self.args[3])
            self.comment = str(self.args[4])
        if self.npoints < 2:
            raise CommandArgumentError('At least 2 points are required in a scan.')
        if self.exptime < 1e-6 or self.exptime > 1e6:
            raise CommandArgumentError('Exposure time must be between 1e-6 and 1e6 seconds.')
        self.idx = None
        self.scanfsn = None
        self.motorpos = None
        self.required_devices = ['pilatus', 'Motor_' + self.motorname]
        self.prefix = self.config['path']['prefixes']['scn']
        self.fsn_being_exposed = None
        self.file_being_exposed = None
        self.exposure_startdate = None
        self._ea_connection = None
        self._outstanding_scanpoints=0
        self._work_status=None
        self.killed = None

    def validate(self):
        motor = self.get_motor(self.motorname)
        if self.name == 'scanrel':
            pos = motor.where()
            self.start = pos - self.halfwidth
            self.end = pos + self.halfwidth
        else:
            assert self.name == 'scan'
        if not motor.checklimits(self.start):
            raise CommandArgumentError(
                'Start position is outside the software limits for motor {}'.format(self.motorname))
        if not motor.checklimits(self.end):
            raise CommandArgumentError(
                'Start position is outside the software limits for motor {}'.format(self.motorname))
        return True

    def on_variable_change(self, device, variablename, newvalue):
        if self.killed is not None:
            if self.killed:
                logger.debug('Calling die_on_kill() from on_variable_change()')
                self.die_on_kill()
            return False
        elif device.name == 'pilatus' and variablename == '_status' and newvalue == 'idle' and self._work_status == 'Exposing':
            # exposure ready. Submit it to exposureanalyzer.
            self.motorpos = self.get_motor(self.motorname).where()
            self.services['filesequence'].new_exposure(
                self.fsn_being_exposed, self.file_being_exposed, self.prefix, self.exposure_startdate,
                position=self.motorpos, scanfsn=self.scanfsn)
            # don't wait for exposureanalyzer, move to the next point
            self.idx += 1
            self.emit('progress', 'Scan running: {:d}/{:d}'.format(self.idx, self.npoints), self.idx / self.npoints)
            if self.idx < self.npoints:
                nextpos = self.start + (self.end - self.start) / (self.npoints - 1) * self.idx
                # self.emit('message', 'Moving motor {} to {:.3f}'.format(self.motorname, nextpos))
                self._work_status='Moving'
                self.get_motor(self.motorname).moveto(nextpos)
            else:
                # Otherwise this was the last point. We wait for exposureanalyzer to finish all its jobs.
                # We will test that case in self.on_scanpoint()
                self._work_status='Finalizing'
                self.emit('message', 'Finalizing scan #{:d}.'.format(self.scanfsn))
        elif (device.name == 'pilatus') and (variablename == 'imgpath') and (
            self._work_status == 'Initializing detector'):
            self.emit('message', 'Moving motor {} to start position ({:.3f})'.format(self.motorname, self.start))
            self.get_motor(self.motorname).moveto(self.start)
            self._work_status = 'Moving'
        return False

    def on_scanpoint(self, exposureanalyzer, prefix, fsn, pos, counters):
        logger.debug('Scan command::on_scanpoint')
        self._outstanding_scanpoints -= 1
        if (self.idx >= self.npoints) and (self._outstanding_scanpoints <= 0):
            # the last scan point has been received.
            self.emit('message', 'Scan #{:d} finished.'.format(self.scanfsn))
            self.idle_return(self.scanfsn)
        return False

    def on_motor_stop(self, motor, targetreached):
        if self._work_status != 'Moving':
            # unexpected stop message, disregard it.
            return False
        assert (motor.name == self.motorname)
        if self.killed is not None:
            if self.killed:
                logger.debug('Calling die_on_kill() from on_motor_stop()')
                self.die_on_kill()
            return False
        if not targetreached:
            try:
                raise CommandError('Target position of motor {} not reached.'.format(self.motorname))
            except CommandError as ce:
                self.emit('fail', ce, traceback.format_exc())
            self.idle_return(None)
            return False
        # otherwise start the exposure
        self.fsn_being_exposed = self.services['filesequence'].get_nextfreefsn(self.prefix)
        self.file_being_exposed = self.services['filesequence'].exposurefileformat(
            self.prefix, self.fsn_being_exposed) + '.cbf'
        self.exposure_startdate = datetime.datetime.now()
        self._work_status='Exposing'
        self.get_device('pilatus').expose(self.file_being_exposed)
        self._outstanding_scanpoints += 1

    def execute(self):
        self.idx = 0
        self.killed = None
        try:
            cmdline = self.namespace['commandline']
{'fraction_non_alphanumeric': 0.10861031143679666, 'fraction_numerical': 0.005103376079560324, 'mean_word_length': 2.312026002166847, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '781248', 'n_tokens_mistral': 4462, 'n_tokens_neox': 4279, 'n_words': 1029}