<commit_before># Copyright 2016 <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



from .instrument import Instrument, SCPIInstrument, VisaInterface, MetaInstrument
from auspex.log import logger

from types import MethodType
from unittest.mock import MagicMock
import auspex.globals

# Dirty trick to avoid loading libraries when scraping
# This code using quince.
if auspex.globals.auspex_dummy_mode:
    fake_aps2 = True
else:
    try:
        import aps2
        fake_aps2 = False
    except:
        logger.warning("Could not find APS2 python driver.")
        fake_aps2 = True

class DigitalAttenuator(SCPIInstrument):
    """BBN 3 Channel Instrument"""

    NUM_CHANNELS = 3

    def __init__(self, resource_name=None, name='Unlabeled Digital Attenuator'):
        super(DigitalAttenuator, self).__init__(resource_name=resource_name,
            name=name)

    def connect(self, resource_name=None, interface_type=None):
        if resource_name is not None:
            self.resource_name = resource_name
        super(DigitalAttenuator, self).connect(resource_name=self.resource_name,
            interface_type=interface_type)
        self.interface._resource.baud_rate = 115200
        self.interface._resource.read_termination = u"\r\n"
        self.interface._resource.write_termination = u"\n"
        #Override query to look for ``end``
        def query(self, query_string):
            val = self._resource.query(query_string)
            assert self.read() == "END"
            return val
        self.interface.query = MethodType(query, self.interface)

    @classmethod
    def channel_check(cls, chan):
        """ Assert the channel requested is feasbile """
        assert chan > 0 and chan <= cls.NUM_CHANNELS, "Invalid channel requested: channel ({:d}) must be between 1 and {:d}".format(chan, cls.NUM_CHANNELS)

    def get_attenuation(self, chan):
        DigitalAttenuator.channel_check(chan)
        return float(self.interface.query("GET {:d}".format(chan)))

    def set_attenuation(self, chan, val):
        DigitalAttenuator.channel_check(chan)
        self.interface.write("SET {:d} {:.1f}".format(chan, val))
        assert self.interface.read() == "Setting channel {:d} to {:.2f}".format(chan, val)
        assert self.interface.read() == "END"

    @property
    def ch1_attenuation(self):
        return self.get_attenuation(1)
    @ch1_attenuation.setter
    def ch1_attenuation(self, value):
        self.set_attenuation(1, value)

    @property
    def ch2_attenuation(self):
        return self.get_attenuation(2)
    @ch2_attenuation.setter
    def ch2_attenuation(self, value):
        self.set_attenuation(2, value)

    @property
    def ch3_attenuation(self):
        return self.get_attenuation(3)
    @ch3_attenuation.setter
    def ch3_attenuation(self, value):
        self.set_attenuation(3, value)

class SpectrumAnaylzer(SCPIInstrument):
    """BBN USB Spectrum Analyzer"""

    IF_FREQ = 0.0107 # 10.7 MHz IF notch filter

    def __init__(self, resource_name=None, *args, **kwargs):
        super(SpectrumAnaylzer, self).__init__(resource_name, *args, **kwargs)

    def connect(self, resource_name=None, interface_type=None):
        super(SpectrumAnaylzer, self).connect(resource_name, interface_type)
        self.interface._resource.timeout = 0.1
        self.interface._resource.read_termination = u"\r\n"
        self.interface._resource.write_termination = u"\n"

    def get_voltage(self):
        volt = None
        for ct in range(10):
            try:
                volt = float(self.interface.query("READ "))
            except ValueError:
                pass
        if volt is None:
            logger.warning("Failed to get data from BBN Spectrum Analyzer "+
                " at {}.".format(self.resource_name))
        return volt

    @property
    def voltage(self):
        return self.get_voltage()

    def peak_amplitude(self):
        volt = self.get_voltage()
        if volt is None:
            return None
        else:
            interp = -100. + (volt - 75.) * (8/45)
            return interp



class MakeSettersGetters(MetaInstrument):
    def __init__(self, name, bases, dct):
        super(MakeSettersGetters, self).__init__(name, bases, dct)

        for k,v in dct.items():
            if isinstance(v, property):
                logger.debug("Adding '%s' command to APS", k)
                setattr(self, 'set_'+k, v.fset)
                setattr(self, 'get_'+k, v.fget)

class APS2(Instrument, metaclass=MakeSettersGetters):
    """BBN APS2"""
    instrument_type = "AWG"

    def __init__(self, resource_name, name="Unlabeled APS2"):
        self.name = name
        self.resource_name = resource_name

        if fake_aps2:
            self.wrapper = MagicMock()
        else:
            self.wrapper = aps2.APS2()

        self.set_amplitude = self.wrapper.set_channel_scale
        self.set_offset    = self.wrapper.set_channel_offset
        self.set_enabled   = self.wrapper.set_channel_enabled

        self.get_amplitude = self.wrapper.get_channel_scale
        self.get_offset    = self.wrapper.get_channel_offset
        self.get_enabled   = self.wrapper.get_channel_enabled

        self.run           = self.wrapper.run
        self.stop          = self.wrapper.stop
        self.connected     = False

    def connect(self, resource_name=None):
        if resource_name is None and self.resource_name is None:
            raise Exception("Must supply a resource name to 'connect' if the instrument was initialized without one.")
        elif resource_name is not None:
            self.resource_name = resource_name

        self.wrapper.connect(self.resource_name)
        self.connected = True

    def disconnect(self):
        if self.resource_name and self.connected:
            self.wrapper.disconnect()
            self.connected = False

    def set_all(self, settings_dict):
        # Pop the channel settings
        settings = settings_dict.copy()
        channel_settings = settings.pop('channels')

        # Call the non-channel commands
        super(APS2, self).set_all(settings)

        for chan, ch_settings in enumerate(channel_settings):
            for name, value in ch_settings.items():
                if hasattr(self, 'set_' + name):
                    getattr(self, 'set_' + name)(chan, value)

    @property
    def seq_file(self):
        return None
    @seq_file.setter
    def seq_file(self, filename):
        self.wrapper.load_sequence_file(filename)

    @property
    def trigger_source(self):<commit_msg>APS2 should stop playing before quitting<commit_after># Copyright 2016 <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



from .instrument import Instrument, SCPIInstrument, VisaInterface, MetaInstrument
from auspex.log import logger

from types import MethodType
from unittest.mock import MagicMock
import auspex.globals

# Dirty trick to avoid loading libraries when scraping
# This code using quince.
if auspex.globals.auspex_dummy_mode:
    fake_aps2 = True
else:
    try:
        import aps2
        fake_aps2 = False
    except:
        logger.warning("Could not find APS2 python driver.")
        fake_aps2 = True

class DigitalAttenuator(SCPIInstrument):
    """BBN 3 Channel Instrument"""

    NUM_CHANNELS = 3

    def __init__(self, resource_name=None, name='Unlabeled Digital Attenuator'):
        super(DigitalAttenuator, self).__init__(resource_name=resource_name,
            name=name)

    def connect(self, resource_name=None, interface_type=None):
        if resource_name is not None:
            self.resource_name = resource_name
        super(DigitalAttenuator, self).connect(resource_name=self.resource_name,
            interface_type=interface_type)
        self.interface._resource.baud_rate = 115200
        self.interface._resource.read_termination = u"\r\n"
        self.interface._resource.write_termination = u"\n"
        #Override query to look for ``end``
        def query(self, query_string):
            val = self._resource.query(query_string)
            assert self.read() == "END"
            return val
        self.interface.query = MethodType(query, self.interface)

    @classmethod
    def channel_check(cls, chan):
        """ Assert the channel requested is feasbile """
        assert chan > 0 and chan <= cls.NUM_CHANNELS, "Invalid channel requested: channel ({:d}) must be between 1 and {:d}".format(chan, cls.NUM_CHANNELS)

    def get_attenuation(self, chan):
        DigitalAttenuator.channel_check(chan)
        return float(self.interface.query("GET {:d}".format(chan)))

    def set_attenuation(self, chan, val):
        DigitalAttenuator.channel_check(chan)
        self.interface.write("SET {:d} {:.1f}".format(chan, val))
        assert self.interface.read() == "Setting channel {:d} to {:.2f}".format(chan, val)
        assert self.interface.read() == "END"

    @property
    def ch1_attenuation(self):
        return self.get_attenuation(1)
    @ch1_attenuation.setter
    def ch1_attenuation(self, value):
        self.set_attenuation(1, value)

    @property
    def ch2_attenuation(self):
        return self.get_attenuation(2)
    @ch2_attenuation.setter
    def ch2_attenuation(self, value):
        self.set_attenuation(2, value)

    @property
    def ch3_attenuation(self):
        return self.get_attenuation(3)
    @ch3_attenuation.setter
    def ch3_attenuation(self, value):
        self.set_attenuation(3, value)

class SpectrumAnaylzer(SCPIInstrument):
    """BBN USB Spectrum Analyzer"""

    IF_FREQ = 0.0107 # 10.7 MHz IF notch filter

    def __init__(self, resource_name=None, *args, **kwargs):
        super(SpectrumAnaylzer, self).__init__(resource_name, *args, **kwargs)

    def connect(self, resource_name=None, interface_type=None):
        super(SpectrumAnaylzer, self).connect(resource_name, interface_type)
        self.interface._resource.timeout = 0.1
        self.interface._resource.read_termination = u"\r\n"
        self.interface._resource.write_termination = u"\n"

    def get_voltage(self):
        volt = None
        for ct in range(10):
            try:
                volt = float(self.interface.query("READ "))
            except ValueError:
                pass
        if volt is None:
            logger.warning("Failed to get data from BBN Spectrum Analyzer "+
                " at {}.".format(self.resource_name))
        return volt

    @property
    def voltage(self):
        return self.get_voltage()

    def peak_amplitude(self):
        volt = self.get_voltage()
        if volt is None:
            return None
        else:
            interp = -100. + (volt - 75.) * (8/45)
            return interp



class MakeSettersGetters(MetaInstrument):
    def __init__(self, name, bases, dct):
        super(MakeSettersGetters, self).__init__(name, bases, dct)

        for k,v in dct.items():
            if isinstance(v, property):
                logger.debug("Adding '%s' command to APS", k)
                setattr(self, 'set_'+k, v.fset)
                setattr(self, 'get_'+k, v.fget)

class APS2(Instrument, metaclass=MakeSettersGetters):
    """BBN APS2"""
    instrument_type = "AWG"

    def __init__(self, resource_name, name="Unlabeled APS2"):
        self.name = name
        self.resource_name = resource_name

        if fake_aps2:
            self.wrapper = MagicMock()
        else:
            self.wrapper = aps2.APS2()

        self.set_amplitude = self.wrapper.set_channel_scale
        self.set_offset    = self.wrapper.set_channel_offset
        self.set_enabled   = self.wrapper.set_channel_enabled

        self.get_amplitude = self.wrapper.get_channel_scale
        self.get_offset    = self.wrapper.get_channel_offset
        self.get_enabled   = self.wrapper.get_channel_enabled

        self.run           = self.wrapper.run
        self.stop          = self.wrapper.stop
        self.connected     = False

    def connect(self, resource_name=None):
        if resource_name is None and self.resource_name is None:
            raise Exception("Must supply a resource name to 'connect' if the instrument was initialized without one.")
        elif resource_name is not None:
            self.resource_name = resource_name

        self.wrapper.connect(self.resource_name)
        self.connected = True

    def disconnect(self):
        if self.resource_name and self.connected:
            self.stop()
            self.wrapper.disconnect()
            self.connected = False

    def set_all(self, settings_dict):
        # Pop the channel settings
        settings = settings_dict.copy()
        channel_settings = settings.pop('channels')

        # Call the non-channel commands
        super(APS2, self).set_all(settings)

        for chan, ch_settings in enumerate(channel_settings):
            for name, value in ch_settings.items():
                if hasattr(self, 'set_' + name):
                    getattr(self, 'set_' + name)(chan, value)

    @property
    def seq_file(self):
        return None
    @seq_file.setter
    def seq_file(self, filename):
        self.wrapper.load_sequence_file(filename)

    @property
    def trigger_source(self):
{'fraction_non_alphanumeric': 0.09758439831572727, 'fraction_numerical': 0.00997266750387826, 'mean_word_length': 2.6082089552238807, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '218329', 'n_tokens_mistral': 4169, 'n_tokens_neox': 3913, 'n_words': 971}