<commit_before>#!/usr/bin/env python

# Copyright 2010 <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.

import ctypes
from ctypes.util import find_library
import numpy.ctypeslib as npct
import sys
import os

import numpy as np
import h5py

APS_ROOT = os.path.realpath(os.path.dirname(os.path.realpath( __file__ )) + '/../../')

# load the shared library
# try with and without "lib" prefix
libpath = find_library("aps")
if libpath is None:
    libpath = find_library("libaps")
# if we still can't find it, then look in python prefix (where conda stores binaries)
if libpath is None:
    try:
        libaps = npct.load_library("libaps", libpath)
    #Finally... load it from this directory
    except OSError:
        libpath = os.path.join(APS_ROOT, 'build')
        libaps = npct.load_library("libaps", libpath)
else:
    libaps = ctypes.CDLL(libpath)

# set up argtypes and restype for functions with arguments that aren't ints or strings
libaps.set_channel_scale.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_float]
libaps.get_channel_scale.restype = ctypes.c_float
libaps.set_channel_offset.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_float]
libaps.get_channel_offset.restype = ctypes.c_float
libaps.set_trigger_interval.argtypes = [ctypes.c_int, ctypes.c_double]
libaps.get_trigger_interval.restype = ctypes.c_double

# initialize the library
libaps.init()

class APS(object):
    # implements interface to libaps

    # class properties
    device_id = 0
    device_serial = ''

    Address = 0

    is_open = False

    ## constants
    # run modes
    RUN_SEQUENCE = 1
    RUN_WAVEFORM = 0

    # repeat modes
    CONTINUOUS = 0
    TRIGGERED = 1

    # trigger modes
    TRIGGER_INTERNAL = 0
    TRIGGER_EXTERNAL = 1

    VALID_FREQUENCIES = [1200,600,300,100,40]

    lastSeqFile = ''

    #DAC2 devices use a different bit file
    DAC2Serials = ('A6UQZB7Z', 'A6001nBU', 'A6001ixV', 'A6001nBT', 'A6001nBS')

    def __init__(self):
        pass

    def __del__(self):
        if self.is_open:
            self.disconnect()

    def enumerate(self):
        #List the number of devices attached and their serial numbers

        #First get the number of devices
        numDevices = libaps.get_numDevices()

        deviceSerials = []
        #Now, for each device, get the associated serial number
        charBuffer = ctypes.create_string_buffer(64)
        for ct in range(numDevices):
            libaps.get_deviceSerial(ct,charBuffer)
            deviceSerials.append(charBuffer.value)

        return numDevices, deviceSerials

    def connect(self, address):
        # Experiment framework function for connecting to an APS
        if self.is_open:
            self.disconnect()

        numDevices, deviceSerials = self.enumerate()

        if type(address) is int:
            if address + 1 > numDevices:
                print('APS Device: ', ID, 'not found')
                return 2
            self.device_id = address
            self.device_serial = deviceSerials[address]
            val = libaps.connect_by_ID(self.device_id)
        else:
            assert address in deviceSerials, 'Ooops!  I cannot find that device.'
            self.device_id = deviceSerials.index(address);
            self.device_serial = address
            val = libaps.connect_by_serial(address)

        if val == 0:
            self.is_open = True

        return val

    def disconnect(self):
        libaps.disconnect_by_ID(self.device_id)
        self.is_open = 0


    def readBitFileVersion(self):
        return self.librarycall('read_bitfile_version')

    def getDefaultBitFileName(self):
        #Check whether we have a DACII or APS device
        if self.device_serial in self.DAC2Serials:
            return os.path.abspath(APS_ROOT + 'bitfiles/mqco_dac2_latest')
        else:
            return os.path.abspath(APS_ROOT + 'bitfiles/mqco_aps_latest')<commit_msg>Clean up enumerate, connect, and disconnect functions.<commit_after>#!/usr/bin/env python

# Copyright 2010 <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.

import ctypes
from ctypes.util import find_library
import numpy.ctypeslib as npct
import sys
import os

import numpy as np
import h5py

APS_ROOT = os.path.realpath(os.path.dirname(os.path.realpath( __file__ )) + '/../../')

# load the shared library
# try with and without "lib" prefix
libpath = find_library("aps")
if libpath is None:
    libpath = find_library("libaps")
# if we still can't find it, then look in python prefix (where conda stores binaries)
if libpath is None:
    try:
        libaps = npct.load_library("libaps", libpath)
    #Finally... load it from this directory
    except OSError:
        libpath = os.path.join(APS_ROOT, 'build')
        libaps = npct.load_library("libaps", libpath)
else:
    libaps = ctypes.CDLL(libpath)

# set up argtypes and restype for functions with arguments that aren't ints or strings
libaps.set_channel_scale.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_float]
libaps.get_channel_scale.restype = ctypes.c_float
libaps.set_channel_offset.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_float]
libaps.get_channel_offset.restype = ctypes.c_float
libaps.set_trigger_interval.argtypes = [ctypes.c_int, ctypes.c_double]
libaps.get_trigger_interval.restype = ctypes.c_double

# initialize the library
libaps.init()

#DAC2 devices use a different bit file
DAC2_SERIALS = ('A6UQZB7Z', 'A6001nBU', 'A6001ixV', 'A6001nBT', 'A6001nBS')

def is_dacii(serial):
    """Check if a given serial number is for a DACII, as these use different bitfiles."""
    return serian in DAC2_SERIALS


class APS(object):
    """Implements an interface to the BBN APS unit via the libaps C library."""

    # class properties
    device_id = 0
    device_serial = ''

    Address = 0

    is_open = False

    ## constants
    # run modes
    RUN_SEQUENCE = 1
    RUN_WAVEFORM = 0

    # repeat modes
    CONTINUOUS = 0
    TRIGGERED = 1

    # trigger modes
    TRIGGER_INTERNAL = 0
    TRIGGER_EXTERNAL = 1

    VALID_FREQUENCIES = [1200,600,300,100,40]

    lastSeqFile = ''

    def __init__(self):
        pass

    def __del__(self):
        if self.is_open:
            self.disconnect()

    def enumerate(self):
        """List the number of devices attached and their serial numbers.

        Args:
            - None

        Returns:
            A tuple (int, list(str)). The first element is the number of APS
            units found, and the list contains their serial numbers.
        """

        numDevices = libaps.get_numDevices()
        deviceSerials = []
        #For each device, get the associated serial number
        charBuffer = ctypes.create_string_buffer(64)
        for ct in range(numDevices):
            libaps.get_deviceSerial(ct,charBuffer)
            deviceSerials.append(charBuffer.value.decode())
        return numDevices, deviceSerials

    def connect(self, address):
        """Connect to a specific APS unit. If an APS is already connected, it
        will be disconnected.

        Args:
            - address: Serial number of device that should be connected to.
        Returns:
            Status code: 0 if success, 2 if device ID not found.
        """
        if self.is_open:
            self.disconnect()
        numDevices, deviceSerials = self.enumerate()
        ##TODO: Decide if we want to keep the address-by-ID in here, as it's confusing.
        if type(address) is int:
            if address + 1 > numDevices:
                print('APS Device: ', ID, 'not found')
                return 2
            self.device_id = address
            self.device_serial = deviceSerials[address]
            val = libaps.connect_by_ID(self.device_id)
        else:
            assert address in deviceSerials, 'Ooops!  I cannot find that device.'
            self.device_id = deviceSerials.index(address);
            self.device_serial = address
            val = libaps.connect_by_serial(address.encode())
        if val == 0:
            self.is_open = True
        return val

    def disconnect(self):
        """Disconnect from the APS unit.

        Args:
            - None.
        Returns:
            - None.
        """

        if self.is_open:
            libaps.disconnect_by_ID(self.device_id)
            self.is_open = False


    def readBitFileVersion(self):
        return self.librarycall('read_bitfile_version')

    def getDefaultBitFileName(self):
        #Check whether we have a DACII or APS device
        if self.device_serial in self.DAC2Serials:
            return os.path.abspath(APS_ROOT + 'bitfiles/mqco_dac2_latest')
        else:
            return os.path.abspath(APS_ROOT + 'bitfiles/mqco_aps_latest')
{'fraction_non_alphanumeric': 0.08451286289905982, 'fraction_numerical': 0.012604607914040706, 'mean_word_length': 2.701720841300191, 'pattern_counts': {'":': 0, '<': 5, '<?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': '1602192', 'n_tokens_mistral': 3069, 'n_tokens_neox': 2903, 'n_words': 892}