"""Custom Exception meant for dealing specifically with U12 Exceptions.

"""

def __init__(self, rawByte = None, labelPrefix = "bit", labelList = None, zeroLabel = "0", oneLabel = "1"):

"""

# Do labels first, so that self.something = something works.

self.__dict__['labels'] = []

self.labelPrefix = labelPrefix

if labelList is None:

self.labelList = list(range(8))

else:

self.labelList = list(reversed(labelList))

self.zeroLabel = zeroLabel

self.oneLabel = oneLabel

self.rawValue = 0

self.rawBits = [ 0 ] * 8

self.data = [ self.zeroLabel ] * 8

items = min(8, len(self.labelList))

for i in reversed(range(items)):

self.labels.append("%s%s" % (self.labelPrefix, self.labelList[i]))

if rawByte is not None:

self.fromByte(rawByte)

def fromByte(self, raw):

"""

self.rawValue = raw

self.rawBits = []

self.data = []

items = min(8, len(self.labelList))

for i in reversed(range(items)):

self.rawBits.append( ((raw >> (i)) & 1) )

self.data.append(self.oneLabel if bool(((raw >> (i)) & 1)) else self.zeroLabel)

def asByte(self):

"""

byteVal = 0

for i, v in enumerate(reversed(self.rawBits)):

byteVal += ( 1 << i ) * v

return byteVal

def asBin(self):

result = "0b"

for i in self.rawBits:

result += "%s" % i

return result

def __len__(self):

return len(self.data)

def __repr__(self):

result = "["

for i in range(len(self.data)):

result += " %s = %s (%s)," % (self.labels[i], self.data[i], self.rawBits[i])

result = result.rstrip(',')

result += " ]"

return "<BitField object: %s >" % result

def __str__(self):

result = "["

for i in range(len(self.data)):

result += " %s = %s," % (self.labels[i], self.data[i])

result = result.rstrip(',')

result += " ]"

return result

def __getattr__(self, label):

try:

i = self.labels.index(label)

return self.rawBits[i]

except ValueError:

raise AttributeError(label)

def __setattr__(self, label, value):

try:

i = self.labels.index(label)

self.rawBits[i] = int(bool(value))

self.data[i] = self.oneLabel if bool(value) else self.zeroLabel

except ValueError:

self.__dict__[label] = value

def __getitem__(self, key):

return self.rawBits[key]

def __setitem__(self, key, value):

self.rawBits[key] = int(bool(value))

self.data[key] = self.oneLabel if bool(value) else self.zeroLabel

def __iter__(self):

return iter(self.rawBits)

def items(self):

"""

return list(zip(self.labels, self.data))

def __int__(self):

return self.asByte()

def __hex__(self):

return hex(self.asByte())

def __add__(self, other):

"""

if ret == -1:

try:

ec = ctypes.get_errno()

raise U12Exception("Exodriver returned error number %s" % ec)

except AttributeError:

raise U12Exception("Exodriver returned an error, but LabJackPython is unable to read the error code. Upgrade to Python 2.6 for this functionality.")

else:

l = ctypes.CDLL("liblabjackusb.so", use_errno=True)

l.LJUSB_Stream.errcheck = errcheck

l.LJUSB_Read.errcheck = errcheck

try:

l = ctypes.CDLL("liblabjackusb.dylib", use_errno=True)

except:

#Try to load with full path.

l = ctypes.CDLL("/usr/local/lib/liblabjackusb.dylib", use_errno=True)

l.LJUSB_Stream.errcheck = errcheck

l.LJUSB_Read.errcheck = errcheck

"""_loadLibrary()

global _os_name

_os_name = "nt"

try:

if sys.platform.startswith("win32"):

#Windows detected

return ctypes.WinDLL("ljackuw")

if sys.platform.startswith("cygwin"):

#Cygwin detected. WinDLL not available, but CDLL seems to work.

return ctypes.CDLL("ljackuw")

except Exception:

e = sys.exc_info()[1]

raise U12Exception("Could not load LabJack UW driver.\n\n The error was: %s" % e)

_os_name = "posix"

addStr = "Exodriver"

try:

if sys.platform.startswith("linux"):

#Linux detected

addStr = "Linux SO"

return _loadLinuxSo()

if sys.platform.startswith("darwin"):

#Mac detected

addStr = "Mac Dylib"

return _loadMacDylib()

#Other OS? Just try to load the Exodriver like a Linux SO

addStr = "Other SO"

return _loadLinuxSo()

except OSError:

e = sys.exc_info()[1]

raise U12Exception("Could not load the Exodriver driver.\n\nCheck that the Exodriver is installed, and the permissions are set correctly.\nThe error message was: %s" % e)

except Exception:

e = sys.exc_info()[1]

"""

def __init__(self, id = -1, serialNumber = None, debug = False):

self.id = id

self.serialNumber = serialNumber

self.deviceName = "U12"

self.streaming = False

self.handle = None

self.debug = debug

self._autoCloseSetup = False

if _os_name != "nt":

# Save some variables to save state.

self.pwmAVoltage = 0

self.pwmBVoltage = 0

self.open(id, serialNumber)

def open(self, id = -1, serialNumber = None):

"""

if _os_name == "nt":

pass

else:

if self.debug:

print("open called")

devType = ctypes.c_ulong(1)

openDev = staticLib.LJUSB_OpenDevice

openDev.restype = ctypes.c_void_p

if serialNumber is not None:

numDevices = staticLib.LJUSB_GetDevCount(devType)

for i in range(numDevices):

handle = openDev(i+1, 0, devType)

if handle != 0 and handle is not None:

self.handle = ctypes.c_void_p(handle)

try:

serial = self.rawReadSerial()

except Exception:

serial = self.rawReadSerial()

if serial == int(serialNumber):

break

else:

self.close()

if self.handle is None:

raise U12Exception("Couldn't find a U12 with a serial number matching %s" % serialNumber)

elif id != -1:

numDevices = staticLib.LJUSB_GetDevCount(devType)

for i in range(numDevices):

handle = openDev(i+1, 0, devType)

if handle != 0 and handle is not None:

self.handle = ctypes.c_void_p(handle)

try:

unitId = self.rawReadLocalId()

except Exception:

unitId = self.rawReadLocalId()

if unitId == int(id):

break

else:

self.close()

if self.handle is None:

raise U12Exception("Couldn't find a U12 with a local ID matching %s" % id)

elif id == -1:

handle = openDev(1, 0, devType)

if handle == 0 or handle is None:

raise Exception("Couldn't open a U12. Check that one is connected and try again.")

else:

self.handle = ctypes.c_void_p(handle)

# U12 ignores first command, so let's write a command.

command = [ 0 ] * 8

command[5] = 0x57 # 0b01010111

try:

self.write(command)

self.read()

except:

pass

self.id = self.rawReadLocalId()

else:

raise Exception("Invalid combination of parameters.")

if not self._autoCloseSetup:

# Only need to register auto-close once per device.

atexit.register(self.close)

self._autoCloseSetup = True

def close(self):

if _os_name == "nt":

pass

else:

staticLib.LJUSB_CloseDevice(self.handle)

self.handle = None

def write(self, writeBuffer):

if _os_name == "nt":

pass

else:

if self.handle is None:

raise U12Exception("The U12's handle is None. Please open a U12 with open()")

if self.debug:

print("Writing: " + hexWithoutQuotes(writeBuffer))

newA = (ctypes.c_byte*len(writeBuffer))(0)

for i in range(len(writeBuffer)):

newA[i] = ctypes.c_byte(writeBuffer[i])

writeBytes = staticLib.LJUSB_Write(self.handle, ctypes.byref(newA), len(writeBuffer))

if writeBytes != len(writeBuffer):

raise U12Exception( "Could only write %s of %s bytes." % (writeBytes, len(writeBuffer) ) )

return writeBuffer

def read(self, numBytes=8, timeout=1000):

if _os_name == "nt":

pass

else:

if self.handle is None:

raise U12Exception("The U12's handle is None. Please open a U12 with open()")

newA = (ctypes.c_byte*numBytes)()

readBytes = staticLib.LJUSB_ReadTO(self.handle, ctypes.byref(newA), numBytes, timeout)

# Return a list of integers in command-response mode

result = [(newA[i] & 0xff) for i in range(readBytes)]

if self.debug:

print("Received: " + hexWithoutQuotes(result))

return result

# Low-level helpers

def rawReadSerial(self):

"""

results = self.rawReadRAM()

return unpack(">I", pack("BBBB", results['DataByte3'], results['DataByte2'], results['DataByte1'], results['DataByte0']))[0]

def rawReadLocalId(self):

"""

results = self.rawReadRAM(0x08)

return results['DataByte0']

# Begin Section 5 Functions

def rawAISample(self, channel0PGAMUX = 8, channel1PGAMUX = 9, channel2PGAMUX = 10, channel3PGAMUX = 11, UpdateIO = False, LEDState = True, IO3toIO0States = 0, EchoValue = 0):

"""

command = [ 0 ] * 8

# Bits 6-4: PGA for 1st Channel

# Bits 3-0: MUX command for 1st Channel

command[0] = int(channel0PGAMUX)

tempNum = command[0] & 7 # 7 = 0b111

channel0Number = tempNum if (command[0] & 0xf) > 7 else tempNum+8

channel0Gain = (command[0] >> 4) & 7 # 7 = 0b111

command[1] = int(channel1PGAMUX)

tempNum = command[1] & 7 # 7 = 0b111

channel1Number = tempNum if (command[1] & 0xf) > 7 else tempNum+8

channel1Gain = (command[1] >> 4) & 7 # 7 = 0b111

command[2] = int(channel2PGAMUX)

tempNum = command[2] & 7 # 7 = 0b111

channel2Number = tempNum if (command[2] & 0xf) > 7 else tempNum+8

channel2Gain = (command[2] >> 4) & 7 # 7 = 0b111

command[3] = int(channel3PGAMUX)

tempNum = command[3] & 7 # 7 = 0b111

channel3Number = tempNum if (command[3] & 0xf) > 7 else tempNum+8

channel3Gain = (command[3] >> 4) & 7 # 7 = 0b111

# Bit 1: Update IO

# Bit 0: LED State

bf = BitField()

bf.bit1 = int(UpdateIO)

bf.bit0 = int(LEDState)

command[4] = int(bf)

# Bit 7-4: 1100 (Command/Response)

# Bit 3-0: Bits for IO3 through IO0 States

bf.fromByte(0)

bf.bit7 = 1

bf.bit6 = 1

bf.fromByte( int(bf) | int(IO3toIO0States) )

command[5] = int(bf)

command[7] = EchoValue

self.write(command)

results = self.read()

bf = BitField()

bf.fromByte(results[0])

if bf.bit7 != 1 or bf.bit6 != 0:

raise U12Exception("Expected a AIStream response, got %s instead." % results[0])

returnDict = {}

returnDict['EchoValue'] = results[1]

returnDict['PGAOvervoltage'] = bool(bf.bit4)

returnDict['IO3toIO0States'] = BitField(results[0], "IO", list(range(3, -1, -1)), "Low", "High")

channel0 = (results[2] >> 4) & 0xf

channel1 = (results[2] & 0xf)

channel2 = (results[5] >> 4) & 0xf

channel3 = (results[5] & 0xf)

channel0 = (channel0 << 8) + results[3]

returnDict['Channel0'] = self.bitsToVolts(channel0Number, channel0Gain, channel0)

channel1 = (channel1 << 8) + results[4]

returnDict['Channel1'] = self.bitsToVolts(channel1Number, channel1Gain, channel1)

channel2 = (channel2 << 8) + results[6]

returnDict['Channel2'] = self.bitsToVolts(channel2Number, channel2Gain, channel2)

channel3 = (channel3 << 8) + results[7]

returnDict['Channel3'] = self.bitsToVolts(channel3Number, channel3Gain, channel3)

return returnDict

def rawDIO(self, D15toD8Directions = 0, D7toD0Directions = 0, D15toD8States = 0, D7toD0States = 0, IO3toIO0DirectionsAndStates = 0, UpdateDigital = False):

"""

command = [ 0 ] * 8

# Bits for D15 through D8 Direction

command[0] = int(D15toD8Directions)

# Bits for D7 through D0 Direction ( 0 = Output, 1 = Input)

command[1] = int(D7toD0Directions)

# Bits for D15 through D8 State ( 0 = Low, 1 = High)

command[2] = int(D15toD8States)

# Bits for D7 through D0 State ( 0 = Low, 1 = High)

command[3] = int(D7toD0States)

# Bits 7-4: Bits for IO3 through IO0 Direction

# Bits 3-0: Bits for IO3 through IO0 State

command[4] = int(IO3toIO0DirectionsAndStates)

# 01X10111 (DIO)

command[5] = 0x57 # 0b01010111

# Bit 0: Update Digital

command[6] = int(bool(UpdateDigital))

#XXXXXXXX

# command[7] = XXXXXXXX

self.write(command)

results = self.read()

returnDict = {}

if results[0] != 87:

raise U12Exception("Expected a DIO response, got %s instead." % results[0])

returnDict['D15toD8States'] = BitField(results[1], "D", list(range(15, 7, -1)), "Low", "High")

returnDict['D7toD0States'] = BitField(results[2], "D", list(range(7, -1, -1)), "Low", "High")

returnDict['D15toD8Directions'] = BitField(results[4], "D", list(range(15, 7, -1)), "Output", "Input")

returnDict['D7toD0Directions'] = BitField(results[5], "D", list(range(7, -1, -1)), "Output", "Input")

returnDict['D15toD8OutputLatchStates'] = BitField(results[6], "D", list(range(15, 7, -1)))

returnDict['D7toD0OutputLatchStates'] = BitField(results[7], "D", list(range(7, -1, -1)))

returnDict['IO3toIO0States'] = BitField((results[3] >> 4), "IO", list(range(3, -1, -1)), "Low", "High")

return returnDict

def rawCounter(self, StrobeEnabled = False, ResetCounter = False):

"""

command = [ 0 ] * 8

bf = BitField()

bf.bit1 = int(StrobeEnabled)

bf.bit0 = int(ResetCounter)

command[0] = int(bf)

bf.fromByte(0)

bf.bit6 = 1

bf.bit4 = 1

bf.bit1 = 1

command[5] = int(bf)

self.write(command)

results = self.read()

returnDict = {}

if results[0] != command[5]:

raise U12Exception("Expected a Counter response, got %s instead." % results[0])

returnDict['D15toD8States'] = BitField(results[1], "D", list(range(15, 7, -1)), "Low", "High")

returnDict['D7toD0States'] = BitField(results[2], "D", list(range(7, -1, -1)), "Low", "High")

returnDict['IO3toIO0States'] = BitField((results[3] >> 4), "IO", list(range(3, -1, -1)), "Low", "High")

counter = results[7]

counter += results[6] << 8

counter += results[5] << 16

counter += results[4] << 24

returnDict['Counter'] = counter

return returnDict

def rawCounterPWMDIO(self, D15toD8Directions = 0, D7toD0Directions = 0, D15toD8States = 0, D7toD0States = 0, IO3toIO0DirectionsAndStates = 0, ResetCounter = False, UpdateDigital = 0, PWMA = 0, PWMB = 0):

"""

command = [ 0 ] * 8

# Bits for D15 through D8 Direction

command[0] = int(D15toD8Directions)

# Bits for D7 through D0 Direction ( 0 = Output, 1 = Input)

command[1] = int(D7toD0Directions)

# Bits for D15 through D8 State ( 0 = Low, 1 = High)

command[2] = int(D15toD8States)

# Bits for D7 through D0 State ( 0 = Low, 1 = High)

command[3] = int(D7toD0States)

# Bits 7-4: Bits for IO3 through IO0 Direction

# Bits 3-0: Bits for IO3 through IO0 State

command[4] = int(IO3toIO0DirectionsAndStates)

bf = BitField()

bf.bit5 = int(ResetCounter)

bf.bit4 = int(UpdateDigital)

binPWMA = int((1023 * (float(PWMA)/5.0)))

binPWMB = int((1023 * (float(PWMB)/5.0)))