<commit_before>import traceback

from gi.repository import GLib

from .command import Command, CommandError


class Moveto(Command):
    """Move motor

    Invocation: moveto(<motorname>, <position>)

    Arguments:
        <motorname>: name of the motor
        <position>: target position (physical units)

    Remarks:
        None
    """
    name = 'moveto'

    def execute(self, interpreter, arglist, instrument, namespace):
        motorname = arglist[0]
        position = arglist[1]

        motor = instrument.motors[motorname]
        self._connections = [motor.connect('stop', self.on_stop, motorname),
                             motor.connect(
                                 'position-change', self.on_position_change, motorname),
                             motor.connect('error', self.on_motorerror, motorname)]

        if instrument.motors[motorname].where()==position:
            GLib.idle_add(lambda m=instrument.motors[motorname],tr=True, mn=motorname:self.on_stop(m,tr,mn))
        instrument.motors[motorname].moveto(position)

    def on_position_change(self, motor, newpos, motorname):
        self.emit('pulse', 'Moving motor %s: %-8.2f' % (motorname, newpos))

    def on_stop(self, motor, targetreached, motorname):
        try:
            for c in self._connections:
                motor.disconnect(c)
            del self._connections
        except AttributeError:
            pass
        self.emit('return', targetreached)
        return False

    def on_motorerror(self, motor, propname, exc, tb, motorname):
        self.emit('fail', exc, tb)

class Moverel(Command):
    """Move motor relatively

    Invocation: moverel(<motorname>, <position>)

    Arguments:
        <motorname>: name of the motor
        <position>: target position (physical units), relative to the present

    Remarks:
        None
    """
    name = 'moverel'

    def execute(self, interpreter, arglist, instrument, namespace):
        motorname = arglist[0]
        position = arglist[1]

        motor = instrument.motors[motorname]
        self._connections = [motor.connect('stop', self.on_stop, motorname),
                             motor.connect(
                                 'position-change', self.on_position_change, motorname),
                             motor.connect('error', self.on_motorerror, motorname)]
        if position==0:
            GLib.idle_add(lambda m=instrument.motors[motorname],tr=True, mn=motorname:self.on_stop(m,tr,mn))
        instrument.motors[motorname].moverel(position)

    def on_position_change(self, motor, newpos, motorname):
        self.emit('pulse', 'Moving motor %s: %-8.2f' % (motorname, newpos))

    def on_stop(self, motor, targetreached, motorname):
        try:
            for c in self._connections:
                motor.disconnect(c)
            del self._connections
        except AttributeError:
            pass
        self.emit('return', targetreached)
        return False

    def on_motorerror(self, motor, propname, exc, tb, motorname):
        self.emit('fail', exc, tb)

class Where(Command):
    """Get current motor position(s)

    Invocation: where([<motorname>])

    Arguments:
        <motorname>: if given, return with the position of just this motor

    Remarks:
        None
    """

    name = 'where'

    def execute(self, interpreter, arglist, instrument, namespace):
        if arglist:
            ret = instrument.motors[arglist[0]].where()
            txt = arglist[0] + ': %8.3f' % ret
        else:
            longestmotorname = max(
                max([len(m) for m in instrument.motors]), len('Motor name'))

            heading = '| ' + \
                ('%%-%ds' % longestmotorname) % 'Motor name' + \
                ' | Position |'
            separator = '+' + '-' * (len(heading) - 2) + '+'
            ret = dict([(m, instrument.motors[m].where())
                        for m in instrument.motors])
            txt = '\n'.join(
                [separator, heading, separator] + [('| %%-%ds | %%8.3f |' % longestmotorname) % (m, ret[m]) for m in sorted(ret)] + [separator])
        GLib.idle_add(lambda m=txt, r=ret: self._idlefunc(m, r))

    def _idlefunc(self, message, ret):
        self.emit('message', message)
        self.emit('return', ret)
        return False


class Beamstop(Command):
    """Query or adjust the beamstop

    Invocation:
        beamstop()           -- returns the current state ('in', 'out', 'none')
        beamstop(<value>)    -- moves the beamstop in or out

    Arguments:
        <value>: if 'in', True, 1: moves the beamstop in. If 'out', False, 0:
            moves the beamstop out

    Remarks:
        None
    """

    name = 'beamstop'

    def execute(self, interpreter, arglist, instrument, namespace):
        self._instrument = instrument
        if not arglist:
            GLib.idle_add(self.end_command)
            return
        if (arglist[0] == 'in') or (arglist[0]):
            self._xpos, self._ypos = self._instrument.config['beamstop']['in']
        elif (arglist[0] == 'out') or (not arglist[0]):
            self._xpos, self._ypos = self._instrument.config['beamstop']['out']
        else:
            raise ConnectionError('Invalid argument: ' + str(arglist[0]))
        self._motorconnections = [self._instrument.motors['BeamStop_X'].connect('stop', self.on_stop, 'BeamStop_X'),
                                  self._instrument.motors['BeamStop_X'].connect('variable-change', self.on_varchange, 'BeamStop_X')]
        self._startpos=self._instrument.motors['BeamStop_X'].where()
        self._instrument.motors['BeamStop_X'].moveto(self._xpos)

    def on_stop(self, motor, targetpositionreached, motorname):
        for c in self._motorconnections:
            motor.disconnect(c)
        del self._motorconnections
        if not targetpositionreached:
            try:
                raise CommandError(
                    'Error on moving beamstop: target position could not be reached with motor ' + motorname)
            except Exception as ce:
                self.emit('fail', (ce, traceback.format_exc()))
        if motorname == 'BeamStop_X':
            self._motorconnections = [self._instrument.motors['BeamStop_Y'].connect('stop', self.on_stop, 'BeamStop_Y'),
                                      self._instrument.motors['BeamStop_Y'].connect('variable-change', self.on_varchange, 'BeamStop_Y')]
            self._startpos=self._instrument.motors['BeamStop_Y'].where()
            try:
                self._instrument.motors['BeamStop_Y'].moveto(self._ypos)
            except Exception as exc:
                self._instrument.motors['BeamStop_Y'].disconnect(self._stopconnection)
                del self._stopconnection
                self.emit('fail',exc,traceback.format_exc())
                self.emit('return',None)
        else:
            self.end_command()

    def on_varchange(self, device, variablename, newvalue, motorname):
        if variablename=='actualposition':
            target=device.get_variable('targetposition')
            self.emit('progress', 'Moving motor %s to %.3f. Now at: %.3f'%(motorname, target, newvalue),
                      1-(newvalue-target)/(self._startpos-target))


    def end_command(self):
        xpos = self._instrument.motors['BeamStop_X'].where()
        ypos = self._instrument.motors['BeamStop_Y'].where()
        if ((abs(xpos - self._instrument.config['beamstop']['in'][0]) < 0.001) and
                (abs(ypos - self._instrument.config['beamstop']['in'][1]) < 0.001)):
            self.emit('return', 'in')
        elif ((abs(xpos - self._instrument.config['beamstop']['out'][0]) < 0.001) and
                  (abs(ypos - self._instrument.config['beamstop']['out'][1]) < 0.001)):<commit_msg>Corrected argument recognition in beamstop command<commit_after>import traceback

from gi.repository import GLib

from .command import Command, CommandError
import logging
logger=logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

class Moveto(Command):
    """Move motor

    Invocation: moveto(<motorname>, <position>)

    Arguments:
        <motorname>: name of the motor
        <position>: target position (physical units)

    Remarks:
        None
    """
    name = 'moveto'

    def execute(self, interpreter, arglist, instrument, namespace):
        motorname = arglist[0]
        position = arglist[1]

        motor = instrument.motors[motorname]
        self._connections = [motor.connect('stop', self.on_stop, motorname),
                             motor.connect(
                                 'position-change', self.on_position_change, motorname),
                             motor.connect('error', self.on_motorerror, motorname)]

        if instrument.motors[motorname].where()==position:
            GLib.idle_add(lambda m=instrument.motors[motorname],tr=True, mn=motorname:self.on_stop(m,tr,mn))
        instrument.motors[motorname].moveto(position)

    def on_position_change(self, motor, newpos, motorname):
        self.emit('pulse', 'Moving motor %s: %-8.2f' % (motorname, newpos))

    def on_stop(self, motor, targetreached, motorname):
        try:
            for c in self._connections:
                motor.disconnect(c)
            del self._connections
        except AttributeError:
            pass
        self.emit('return', targetreached)
        return False

    def on_motorerror(self, motor, propname, exc, tb, motorname):
        self.emit('fail', exc, tb)

class Moverel(Command):
    """Move motor relatively

    Invocation: moverel(<motorname>, <position>)

    Arguments:
        <motorname>: name of the motor
        <position>: target position (physical units), relative to the present

    Remarks:
        None
    """
    name = 'moverel'

    def execute(self, interpreter, arglist, instrument, namespace):
        motorname = arglist[0]
        position = arglist[1]

        motor = instrument.motors[motorname]
        self._connections = [motor.connect('stop', self.on_stop, motorname),
                             motor.connect(
                                 'position-change', self.on_position_change, motorname),
                             motor.connect('error', self.on_motorerror, motorname)]
        if position==0:
            GLib.idle_add(lambda m=instrument.motors[motorname],tr=True, mn=motorname:self.on_stop(m,tr,mn))
        instrument.motors[motorname].moverel(position)

    def on_position_change(self, motor, newpos, motorname):
        self.emit('pulse', 'Moving motor %s: %-8.2f' % (motorname, newpos))

    def on_stop(self, motor, targetreached, motorname):
        try:
            for c in self._connections:
                motor.disconnect(c)
            del self._connections
        except AttributeError:
            pass
        self.emit('return', targetreached)
        return False

    def on_motorerror(self, motor, propname, exc, tb, motorname):
        self.emit('fail', exc, tb)

class Where(Command):
    """Get current motor position(s)

    Invocation: where([<motorname>])

    Arguments:
        <motorname>: if given, return with the position of just this motor

    Remarks:
        None
    """

    name = 'where'

    def execute(self, interpreter, arglist, instrument, namespace):
        if arglist:
            ret = instrument.motors[arglist[0]].where()
            txt = arglist[0] + ': %8.3f' % ret
        else:
            longestmotorname = max(
                max([len(m) for m in instrument.motors]), len('Motor name'))

            heading = '| ' + \
                ('%%-%ds' % longestmotorname) % 'Motor name' + \
                ' | Position |'
            separator = '+' + '-' * (len(heading) - 2) + '+'
            ret = dict([(m, instrument.motors[m].where())
                        for m in instrument.motors])
            txt = '\n'.join(
                [separator, heading, separator] + [('| %%-%ds | %%8.3f |' % longestmotorname) % (m, ret[m]) for m in sorted(ret)] + [separator])
        GLib.idle_add(lambda m=txt, r=ret: self._idlefunc(m, r))

    def _idlefunc(self, message, ret):
        self.emit('message', message)
        self.emit('return', ret)
        return False


class Beamstop(Command):
    """Query or adjust the beamstop

    Invocation:
        beamstop()           -- returns the current state ('in', 'out', 'none')
        beamstop(<value>)    -- moves the beamstop in or out

    Arguments:
        <value>: if 'in', True, 1: moves the beamstop in. If 'out', False, 0:
            moves the beamstop out

    Remarks:
        None
    """

    name = 'beamstop'

    def execute(self, interpreter, arglist, instrument, namespace):
        self._instrument = instrument
        if not arglist:
            GLib.idle_add(self.end_command)
            return
        if (arglist[0] == 'in') or ((isinstance(arglist[0],int) or isinstance(arglist[0], bool) or isinstance(arglist[0],float) and bool(arglist[0]))):
            self._xpos, self._ypos = self._instrument.config['beamstop']['in']
            self._direction='in'
        elif (arglist[0] == 'out') or ((isinstance(arglist[0],int) or isinstance(arglist[0], bool) or isinstance(arglist[0],float) and not bool(arglist[0]))):
            self._xpos, self._ypos = self._instrument.config['beamstop']['out']
            self._direction='out'
        else:
            raise ConnectionError('Invalid argument: ' + str(arglist[0]))
        self._motorconnections = [self._instrument.motors['BeamStop_X'].connect('stop', self.on_stop, 'BeamStop_X'),
                                  self._instrument.motors['BeamStop_X'].connect('variable-change', self.on_varchange, 'BeamStop_X')]
        self._startpos=self._instrument.motors['BeamStop_X'].where()
        self._instrument.motors['BeamStop_X'].moveto(self._xpos)

    def on_stop(self, motor, targetpositionreached, motorname):
        for c in self._motorconnections:
            motor.disconnect(c)
        del self._motorconnections
        if not targetpositionreached:
            try:
                raise CommandError(
                    'Error on moving beamstop: target position could not be reached with motor ' + motorname)
            except Exception as ce:
                self.emit('fail', (ce, traceback.format_exc()))
        if motorname == 'BeamStop_X':
            self._motorconnections = [self._instrument.motors['BeamStop_Y'].connect('stop', self.on_stop, 'BeamStop_Y'),
                                      self._instrument.motors['BeamStop_Y'].connect('variable-change', self.on_varchange, 'BeamStop_Y')]
            self._startpos=self._instrument.motors['BeamStop_Y'].where()
            try:
                self._instrument.motors['BeamStop_Y'].moveto(self._ypos)
            except Exception as exc:
                self._instrument.motors['BeamStop_Y'].disconnect(self._stopconnection)
                del self._stopconnection
                self.emit('fail',exc,traceback.format_exc())
                self.emit('return',None)
        else:
            self.end_command()

    def on_varchange(self, device, variablename, newvalue, motorname):
        if variablename=='actualposition':
            target=device.get_variable('targetposition')
            self.emit('progress', 'Moving beamstop %s. Motor %s to %.3f, Now at: %.3f'%(self._direction, motorname, target, newvalue),
                      1-(newvalue-target)/(self._startpos-target))


    def end_command(self):
        xpos = self._instrument.motors['BeamStop_X'].where()
        ypos = self._instrument.motors['BeamStop_Y'].where()
        if ((abs(xpos - self._instrument.config['beamstop']['in'][0]) < 0.001) and
                (abs(ypos - self._instrument.config['beamstop']['in'][1]) < 0.001)):
            self.emit('return', 'in')
        elif ((abs(xpos - self._instrument.config['beamstop']['out'][0]) < 0.001) and
                  (abs(ypos - self._instrument.config['beamstop']['out'][1]) < 0.001)):
{'fraction_non_alphanumeric': 0.13015157209069272, 'fraction_numerical': 0.006138043342101967, 'mean_word_length': 2.3320116861435727, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1848493', 'n_tokens_mistral': 4861, 'n_tokens_neox': 4673, 'n_words': 1011}