<commit_before># ===============================================================================
# Copyright 2015 <NAME>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ===============================================================================

# ============= enthought library imports =======================
import os
import random
import time

from numpy import array, argmin
from traits.api import Instance, Int, Property, List, \
    Any, Enum, Str, DelegatesTo, Bool, TraitError, cached_property

from pychron.core.helpers.filetools import list_directory2
from pychron.core.progress import open_progress
from pychron.core.ramper import StepRamper
from pychron.globals import globalv
from pychron.paths import paths
from pychron.pychron_constants import QTEGRA_INTEGRATION_TIMES, \
    DEFAULT_INTEGRATION_TIME, NULL_STR
from pychron.spectrometer import get_spectrometer_config_path, \
    get_spectrometer_config_name, set_spectrometer_config_name
from pychron.spectrometer.base_detector import BaseDetector
from pychron.spectrometer.thermo.detector.base import ThermoDetector
from pychron.spectrometer.thermo.magnet.base import ThermoMagnet
from pychron.spectrometer.thermo.source.base import ThermoSource
from pychron.spectrometer.thermo.spectrometer_device import SpectrometerDevice


def normalize_integration_time(it):
    """
        find the integration time closest to "it"
    """
    x = array(QTEGRA_INTEGRATION_TIMES)
    return x[argmin(abs(x - it))]


def calculate_radius(m_e, hv, mfield):
    """
        m_e= mass/charge
        hv= accelerating voltage (V)
        mfield= magnet field (H)
    """
    r = ((2 * m_e * hv) / mfield ** 2) ** 0.5

    return r


class ThermoSpectrometer(SpectrometerDevice):
    magnet = Instance(ThermoMagnet)
    source = Instance(ThermoSource)
    magnet_klass = Any
    source_klass = Any
    detector_klass = Any

    detectors = List(ThermoDetector)

    microcontroller = Any
    integration_time = Enum(QTEGRA_INTEGRATION_TIMES)

    reference_detector = Str('H1')

    magnet_dac = DelegatesTo('magnet', prefix='dac')

    magnet_dacmin = DelegatesTo('magnet', prefix='dacmin')
    magnet_dacmax = DelegatesTo('magnet', prefix='dacmin')

    current_hv = DelegatesTo('source')

    molecular_weight = Str('Ar40')
    molecular_weights = None
    isotopes = Property

    spectrometer_configuration = Str
    spectrometer_configurations = List

    sub_cup_configurations = List
    sub_cup_configuration = Property(depends_on='_sub_cup_configuration')
    _sub_cup_configuration = Str

    dc_start = Int(0)
    dc_stop = Int(500)
    dc_step = Int(50)
    dc_stepmin = Int(1)
    dc_stepmax = Int(1000)
    dc_threshold = Int(3)
    dc_npeak_centers = Int(3)

    send_config_on_startup = Bool
    use_deflection_correction = Bool(True)
    max_deflection = Int(500)

    _connection_status = False
    _config = None
    _debug_values = None
    _saved_integration = None

    def reload_mftable(self):
        self.magnet.reload_mftable()

    def get_detector_active(self, dname):
        """
        return True if dname in the list of intensity keys
        e.g.

        keys, signals = get_intensities
        return dname in keys

        :param dname:
        :return:
        """
        keys, prev = self.get_intensities()
        return dname in keys

    def test_intensity(self):
        """
        test if intensity is changing. make 2 measurements if exactlly the same for all
        detectors make third measurement if same as 1,2 make fourth measurement if same
        all four measurements same then test fails
        :return:
        """
        ret, err = True, ''
        keys, one = self.get_intensities()
        it = 0.1 if self.simulation else self.integration_time

        time.sleep(it)
        keys, two = self.get_intensities()

        if all(one == two):
            time.sleep(it)
            keys, three = self.get_intensities()
            if all(two == three):
                time.sleep(it)
                keys, four = self.get_intensities()
                if all(three == four):
                    ret = False
        return ret, err

    def test_connection(self, force=True):
        """
            if not in simulation mode send a GetIntegrationTime to the spectrometer
            if in simulation mode and the globalv.communication_simulation is disabled
            then return False

        :return: bool
        """
        self.info('testing connnection')
        ret, err = False, ''
        if not self.simulation:
            if force:
                ret = self.ask('GetIntegrationTime', verbose=True) is not None
            else:
                ret = self._connection_status
        elif globalv.communication_simulation:
            ret = True

        self._connection_status = ret
        self.microcontroller.set_simulation(not ret)
        return ret, err

    def set_gains(self, history=None):
        """

        :param history:
        :return: list
        """
        if history:
            self.debug(
                'setting gains to {}, user={}'.format(history.create_date,
                                                      history.username))
        for di in self.detectors:
            di.set_gain()

        return {di.name: di.gain for di in self.detectors}

    def load_current_detector_gains(self):
        """
        load the detector gains from the spectrometer
        """
        for di in self.detectors:
            di.get_gain()

    def get_integration_time(self, current=True):
        """
        return current or cached integration time, i.e time between intensity measurements

        :param current: bool, if True retrieve value from qtegra
        :return: float
        """
        if current:
            resp = self.ask('GetIntegrationTime')
            if resp:
                try:
                    self.integration_time = float(resp)
                    self.info(
                        'Integration Time {}'.format(self.integration_time))

                except (TypeError, ValueError, TraitError):
                    self.warning(
                        'Invalid integration time. resp={}'.format(resp))
                    self.integration_time = DEFAULT_INTEGRATION_TIME

        return self.integration_time

    def save_integration(self):
        self._saved_integration = self.integration_time

    def restore_integration(self):
        if self._saved_integration:
            self.set_integration_time(self._saved_integration)
            self._saved_integration = None

    def set_integration_time(self, it, force=False):
        """

        :param it: float, integration time
        :param force: set integration even if "it" is not different than self.integration_time
        :return: float, integration time
        """
        it = normalize_integration_time(it)
        if self.integration_time != it or force:
            self.debug('setting integration time = {}'.format(it))
            name = 'SetIntegrationTime'
            self.set_parameter(name, it)
            self.trait_setq(integration_time=it)

        return it

    def send_configuration(self, **kw):
        """
            send the configuration values to the device
        """
        self._send_configuration(**kw)

    def set_parameter(self, name, v):
        cmd = '{} {}'.format(name, v)
        self.ask(cmd)

    def get_parameter(self, cmd):
        return self.ask(cmd)

    def set_microcontroller(self, m):
        self.debug('set microcontroller {}'.format(m))
        self.magnet.microcontroller = m
        self.source.microcontroller = m
        self.microcontroller = m
        for d in self.detectors:
            d.microcontroller = m
            d.load()

    def set_deflection(self, name, value):
        det = self.get_detector(name)
        if det:
            det.deflection = value
        else:
            self.warning('Could not find detector "{}". Deflection Not Possible'.format(name))

    def get_deflection(self, name, current=False):
        """
        get deflection by detector name

        :param name: str, detector name
        :param current: bool, if True query qtegra
        :return: float
        """
        deflection = 0
        det = self.get_detector(name)
        if det:
            if current:
                det.read_deflection()
            deflection = det.deflection
        else:
            self.warning('Failed getting deflection for detector ="{}"'.format(name))

        return deflection

    def get_detector(self, name):
        """
        get Detector object by name

        :param name: str
        :return: Detector
        """

        if isinstance(name, BaseDetector):
            return name
        else:
            if name.endswith('_'):
                name = '{})'.format(name[:-1])
                name = name.replace('_', '(')

            return next((det for det in self.detectors if det.name == name), None)

    def map_isotope(self, mass):
        """
        map a mass to an isotope
        @param mass:
        @return:
        """
        molweights = self.molecular_weights
        # for k, v in molweights.iteritems():
        #     print '\t',k,v,mass, v-mass
        #     if abs(v-mass) < 0.05:
        #         print 'found'
        #         break
        print molweights, mass

        return next((k for k, v in molweights.iteritems() if abs(v - mass) < 0.15), 'Iso{:0.4f}'.format(mass))

    def map_mass(self, isotope):
        """
        map an isotope to a mass
        @param isotope:
        @return:
        """
        try:
            return self.molecular_weights[isotope]
        except KeyError:
            self.warning('Invalid isotope. Cannot map to a mass. {} not in molecular_weights'.format(isotope))

    def update_isotopes(self, isotope, detector):
        """
        update the isotope name for each detector

        called by AutomatedRun._define_detectors

        :param isotope: str
        :param detector: str or Detector
        :return:
        """
        if isotope != NULL_STR:
            det = self.get_detector(detector)
            if not det:
                self.debug('cannot update detector "{}"'.format(detector))
            else:
                det.isotope = isotope

                index = det.index
                try:
                    # nmass = int(isotope[2:])
                    nmass = self.map_mass(isotope)
                    for di in self.detectors:

                        mass = nmass + (di.index - index)

                        isotope = self.map_isotope(mass)
                        self.debug('setting detector {} to {} ({})'.format(di.name, isotope, mass))
                        di.isotope = isotope

                except BaseException, e:
                    self.warning('Cannot update isotopes. isotope={}, detector={}. error:{}'.format(isotope, detector, e))

    def get_deflection_word(self, keys):
        if self.simulation:
            x = [random.random() for i in keys]
        else:
            x = self.ask('GetDeflections {}'.format(','.join(keys)),<commit_msg>fixed update_isotopes. Detector indices need to be from High mass to Low mass, e.g H2=0, H1=1, Ax=2...<commit_after># ===============================================================================
# Copyright 2015 <NAME>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ===============================================================================

# ============= enthought library imports =======================
import os
import random
import time

from numpy import array, argmin
from traits.api import Instance, Int, Property, List, \
    Any, Enum, Str, DelegatesTo, Bool, TraitError, cached_property

from pychron.core.helpers.filetools import list_directory2
from pychron.core.progress import open_progress
from pychron.core.ramper import StepRamper
from pychron.globals import globalv
from pychron.paths import paths
from pychron.pychron_constants import QTEGRA_INTEGRATION_TIMES, \
    DEFAULT_INTEGRATION_TIME, NULL_STR
from pychron.spectrometer import get_spectrometer_config_path, \
    get_spectrometer_config_name, set_spectrometer_config_name
from pychron.spectrometer.base_detector import BaseDetector
from pychron.spectrometer.thermo.detector.base import ThermoDetector
from pychron.spectrometer.thermo.magnet.base import ThermoMagnet
from pychron.spectrometer.thermo.source.base import ThermoSource
from pychron.spectrometer.thermo.spectrometer_device import SpectrometerDevice


def normalize_integration_time(it):
    """
        find the integration time closest to "it"
    """
    x = array(QTEGRA_INTEGRATION_TIMES)
    return x[argmin(abs(x - it))]


def calculate_radius(m_e, hv, mfield):
    """
        m_e= mass/charge
        hv= accelerating voltage (V)
        mfield= magnet field (H)
    """
    r = ((2 * m_e * hv) / mfield ** 2) ** 0.5

    return r


class ThermoSpectrometer(SpectrometerDevice):
    magnet = Instance(ThermoMagnet)
    source = Instance(ThermoSource)
    magnet_klass = Any
    source_klass = Any
    detector_klass = Any

    detectors = List(ThermoDetector)

    microcontroller = Any
    integration_time = Enum(QTEGRA_INTEGRATION_TIMES)

    reference_detector = Str('H1')

    magnet_dac = DelegatesTo('magnet', prefix='dac')

    magnet_dacmin = DelegatesTo('magnet', prefix='dacmin')
    magnet_dacmax = DelegatesTo('magnet', prefix='dacmin')

    current_hv = DelegatesTo('source')

    molecular_weight = Str('Ar40')
    molecular_weights = None
    isotopes = Property

    spectrometer_configuration = Str
    spectrometer_configurations = List

    sub_cup_configurations = List
    sub_cup_configuration = Property(depends_on='_sub_cup_configuration')
    _sub_cup_configuration = Str

    dc_start = Int(0)
    dc_stop = Int(500)
    dc_step = Int(50)
    dc_stepmin = Int(1)
    dc_stepmax = Int(1000)
    dc_threshold = Int(3)
    dc_npeak_centers = Int(3)

    send_config_on_startup = Bool
    use_deflection_correction = Bool(True)
    max_deflection = Int(500)

    _connection_status = False
    _config = None
    _debug_values = None
    _saved_integration = None

    def reload_mftable(self):
        self.magnet.reload_mftable()

    def get_detector_active(self, dname):
        """
        return True if dname in the list of intensity keys
        e.g.

        keys, signals = get_intensities
        return dname in keys

        :param dname:
        :return:
        """
        keys, prev = self.get_intensities()
        return dname in keys

    def test_intensity(self):
        """
        test if intensity is changing. make 2 measurements if exactlly the same for all
        detectors make third measurement if same as 1,2 make fourth measurement if same
        all four measurements same then test fails
        :return:
        """
        ret, err = True, ''
        keys, one = self.get_intensities()
        it = 0.1 if self.simulation else self.integration_time

        time.sleep(it)
        keys, two = self.get_intensities()

        if all(one == two):
            time.sleep(it)
            keys, three = self.get_intensities()
            if all(two == three):
                time.sleep(it)
                keys, four = self.get_intensities()
                if all(three == four):
                    ret = False
        return ret, err

    def test_connection(self, force=True):
        """
            if not in simulation mode send a GetIntegrationTime to the spectrometer
            if in simulation mode and the globalv.communication_simulation is disabled
            then return False

        :return: bool
        """
        self.info('testing connnection')
        ret, err = False, ''
        if not self.simulation:
            if force:
                ret = self.ask('GetIntegrationTime', verbose=True) is not None
            else:
                ret = self._connection_status
        elif globalv.communication_simulation:
            ret = True

        self._connection_status = ret
        self.microcontroller.set_simulation(not ret)
        return ret, err

    def set_gains(self, history=None):
        """

        :param history:
        :return: list
        """
        if history:
            self.debug(
                'setting gains to {}, user={}'.format(history.create_date,
                                                      history.username))
        for di in self.detectors:
            di.set_gain()

        return {di.name: di.gain for di in self.detectors}

    def load_current_detector_gains(self):
        """
        load the detector gains from the spectrometer
        """
        for di in self.detectors:
            di.get_gain()

    def get_integration_time(self, current=True):
        """
        return current or cached integration time, i.e time between intensity measurements

        :param current: bool, if True retrieve value from qtegra
        :return: float
        """
        if current:
            resp = self.ask('GetIntegrationTime')
            if resp:
                try:
                    self.integration_time = float(resp)
                    self.info(
                        'Integration Time {}'.format(self.integration_time))

                except (TypeError, ValueError, TraitError):
                    self.warning(
                        'Invalid integration time. resp={}'.format(resp))
                    self.integration_time = DEFAULT_INTEGRATION_TIME

        return self.integration_time

    def save_integration(self):
        self._saved_integration = self.integration_time

    def restore_integration(self):
        if self._saved_integration:
            self.set_integration_time(self._saved_integration)
            self._saved_integration = None

    def set_integration_time(self, it, force=False):
        """

        :param it: float, integration time
        :param force: set integration even if "it" is not different than self.integration_time
        :return: float, integration time
        """
        it = normalize_integration_time(it)
        if self.integration_time != it or force:
            self.debug('setting integration time = {}'.format(it))
            name = 'SetIntegrationTime'
            self.set_parameter(name, it)
            self.trait_setq(integration_time=it)

        return it

    def send_configuration(self, **kw):
        """
            send the configuration values to the device
        """
        self._send_configuration(**kw)

    def set_parameter(self, name, v):
        cmd = '{} {}'.format(name, v)
        self.ask(cmd)

    def get_parameter(self, cmd):
        return self.ask(cmd)

    def set_microcontroller(self, m):
        self.debug('set microcontroller {}'.format(m))
        self.magnet.microcontroller = m
        self.source.microcontroller = m
        self.microcontroller = m
        for d in self.detectors:
            d.microcontroller = m
            d.load()

    def set_deflection(self, name, value):
        det = self.get_detector(name)
        if det:
            det.deflection = value
        else:
            self.warning('Could not find detector "{}". Deflection Not Possible'.format(name))

    def get_deflection(self, name, current=False):
        """
        get deflection by detector name

        :param name: str, detector name
        :param current: bool, if True query qtegra
        :return: float
        """
        deflection = 0
        det = self.get_detector(name)
        if det:
            if current:
                det.read_deflection()
            deflection = det.deflection
        else:
            self.warning('Failed getting deflection for detector ="{}"'.format(name))

        return deflection

    def get_detector(self, name):
        """
        get Detector object by name

        :param name: str
        :return: Detector
        """

        if isinstance(name, BaseDetector):
            return name
        else:
            if name.endswith('_'):
                name = '{})'.format(name[:-1])
                name = name.replace('_', '(')

            return next((det for det in self.detectors if det.name == name), None)

    def map_isotope(self, mass):
        """
        map a mass to an isotope
        @param mass:
        @return:
        """
        molweights = self.molecular_weights
        # for k, v in molweights.iteritems():
        #     print '\t',k,v,mass, v-mass
        #     if abs(v-mass) < 0.05:
        #         print 'found'
        #         break
        print molweights, mass

        return next((k for k, v in molweights.iteritems() if abs(v - mass) < 0.15), 'Iso{:0.4f}'.format(mass))

    def map_mass(self, isotope):
        """
        map an isotope to a mass
        @param isotope:
        @return:
        """
        try:
            return self.molecular_weights[isotope]
        except KeyError:
            self.warning('Invalid isotope. Cannot map to a mass. {} not in molecular_weights'.format(isotope))

    def update_isotopes(self, isotope, detector):
        """
        update the isotope name for each detector

        called by AutomatedRun._define_detectors

        :param isotope: str
        :param detector: str or Detector
        :return:
        """
        if isotope != NULL_STR:
            det = self.get_detector(detector)
            if not det:
                self.debug('cannot update detector "{}"'.format(detector))
            else:
                det.isotope = isotope

                index = det.index
                try:
                    # nmass = int(isotope[2:])
                    nmass = self.map_mass(isotope)
                    for di in self.detectors:

                        mass = nmass - (di.index - index)

                        isotope = self.map_isotope(mass)
                        self.debug('setting detector {} to {} ({})'.format(di.name, isotope, mass))
                        di.isotope = isotope

                except BaseException, e:
                    self.warning('Cannot update isotopes. isotope={}, detector={}. error:{}'.format(isotope, detector, e))

    def get_deflection_word(self, keys):
        if self.simulation:
            x = [random.random() for i in keys]
        else:
            x = self.ask('GetDeflections {}'.format(','.join(keys)),
{'fraction_non_alphanumeric': 0.10480592288269783, 'fraction_numerical': 0.004322334916762956, 'mean_word_length': 2.1472053872053873, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4548038', 'n_tokens_mistral': 6747, 'n_tokens_neox': 6311, 'n_words': 1787}