Skip to content

Commit

Permalink
Minimal style cleanup.
Browse files Browse the repository at this point in the history
  • Loading branch information
Robert Wlodarczyk committed Jan 5, 2016
1 parent fa6f992 commit 5860bca
Showing 1 changed file with 52 additions and 47 deletions.
99 changes: 52 additions & 47 deletions usbtmc/usbtmc.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@

USBTMC_HEADER_SIZE = 12


def parse_visa_resource_string(resource_string):
# valid resource strings:
# USB::1234::5678::INSTR
Expand All @@ -82,19 +83,20 @@ def parse_visa_resource_string(resource_string):

if m is not None:
return dict(
type = m.group('type').upper(),
prefix = m.group('prefix'),
arg1 = m.group('arg1'),
arg2 = m.group('arg2'),
arg3 = m.group('arg3'),
suffix = m.group('suffix')
type=m.group('type').upper(),
prefix=m.group('prefix'),
arg1=m.group('arg1'),
arg2=m.group('arg2'),
arg3=m.group('arg3'),
suffix=m.group('suffix')
)


# Exceptions
class UsbtmcException(Exception):
em = {0: "No error"}

def __init__(self, err = None, note = None):
def __init__(self, err=None, note=None):
self.err = err
self.note = note
self.msg = ''
Expand All @@ -115,19 +117,21 @@ def __init__(self, err = None, note = None):
def __str__(self):
return self.msg


def list_devices():
"List all connected USBTMC devices"

def is_usbtmc_device(dev):
for cfg in dev:
d = usb.util.find_descriptor(cfg, bInterfaceClass = USBTMC_bInterfaceClass,
bInterfaceSubClass = USBTMC_bInterfaceSubClass)
d = usb.util.find_descriptor(cfg, bInterfaceClass=USBTMC_bInterfaceClass,
bInterfaceSubClass=USBTMC_bInterfaceSubClass)
is_advantest = dev.idVendor == 0x1334
return d is not None or is_advantest

return list(usb.core.find(find_all = True, custom_match = is_usbtmc_device))
return list(usb.core.find(find_all=True, custom_match=is_usbtmc_device))


def find_device(idVendor = None, idProduct = None, iSerial = None):
def find_device(idVendor=None, idProduct=None, iSerial=None):
"Find USBTMC instrument"

devs = list_devices()
Expand Down Expand Up @@ -155,6 +159,7 @@ def find_device(idVendor = None, idProduct = None, iSerial = None):

return None


class Instrument(object):
"USBTMC instrument interface client"
def __init__(self, *args, **kwargs):
Expand Down Expand Up @@ -359,13 +364,13 @@ def get_capabilities(self):
if not self.connected:
self.open()

b=self.device.ctrl_transfer(
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USBTMC_REQUEST_GET_CAPABILITIES,
0x0000,
self.iface.index,
0x0018,
timeout=int(self.timeout*1000))
b = self.device.ctrl_transfer(
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USBTMC_REQUEST_GET_CAPABILITIES,
0x0000,
self.iface.index,
0x0018,
timeout=int(self.timeout*1000))
if (b[0] == USBTMC_STATUS_SUCCESS):
self.bcdUSBTMC = (b[3] << 8) + b[2]
self.support_pulse = b[4] & 4 != 0
Expand Down Expand Up @@ -395,12 +400,12 @@ def pulse(self):

if self.support_pulse:
b = self.device.ctrl_transfer(
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USBTMC_REQUEST_INDICATOR_PULSE,
0x0000,
self.iface.index,
0x0001,
timeout=int(self.timeout*1000))
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USBTMC_REQUEST_INDICATOR_PULSE,
0x0000,
self.iface.index,
0x0001,
timeout=int(self.timeout*1000))
if (b[0] != USBTMC_STATUS_SUCCESS):
raise UsbtmcException("Pulse failed", 'pulse')

Expand Down Expand Up @@ -477,7 +482,7 @@ def read_raw(self, num=-1):
self.open()

read_len = self.max_transfer_size
if num > 0 and num < read_len:
if 0 < num < read_len:
read_len = num

eom = False
Expand Down Expand Up @@ -532,7 +537,7 @@ def ask_raw(self, data, num=-1):
if self.advantest_quirk and not was_locked:
self.unlock()

def write(self, message, encoding = 'utf-8'):
def write(self, message, encoding='utf-8'):
"Write string to instrument"
if type(message) is tuple or type(message) is list:
# recursive call for a list of commands
Expand All @@ -542,11 +547,11 @@ def write(self, message, encoding = 'utf-8'):

self.write_raw(str(message).encode(encoding))

def read(self, num=-1, encoding = 'utf-8'):
def read(self, num=-1, encoding='utf-8'):
"Read string from instrument"
return self.read_raw(num).decode(encoding).rstrip('\r\n')

def ask(self, message, num=-1, encoding = 'utf-8'):
def ask(self, message, num=-1, encoding='utf-8'):
"Write then read string"
if type(message) is tuple or type(message) is list:
# recursive call for a list of commands
Expand Down Expand Up @@ -578,13 +583,13 @@ def read_stb(self):
rstb_btag = 2
self.last_rstb_btag = rstb_btag

b=self.device.ctrl_transfer(
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USB488_READ_STATUS_BYTE,
rstb_btag,
self.iface.index,
0x0003,
timeout=int(self.timeout*1000))
b = self.device.ctrl_transfer(
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USB488_READ_STATUS_BYTE,
rstb_btag,
self.iface.index,
0x0003,
timeout=int(self.timeout*1000))
if (b[0] == USBTMC_STATUS_SUCCESS):
# check btag
if rstb_btag != b[1]:
Expand Down Expand Up @@ -625,23 +630,23 @@ def clear(self):

# Send INITIATE_CLEAR
b = self.device.ctrl_transfer(
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USBTMC_REQUEST_INITIATE_CLEAR,
0x0000,
self.iface.index,
0x0001,
timeout=int(self.timeout*1000))
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USBTMC_REQUEST_INITIATE_CLEAR,
0x0000,
self.iface.index,
0x0001,
timeout=int(self.timeout*1000))
if (b[0] == USBTMC_STATUS_SUCCESS):
# Initiate clear succeeded, wait for completion
while True:
# Check status
b = self.device.ctrl_transfer(
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USBTMC_REQUEST_CHECK_CLEAR_STATUS,
0x0000,
self.iface.index,
0x0002,
timeout=int(self.timeout*1000))
usb.util.build_request_type(usb.util.CTRL_IN, usb.util.CTRL_TYPE_CLASS, usb.util.CTRL_RECIPIENT_INTERFACE),
USBTMC_REQUEST_CHECK_CLEAR_STATUS,
0x0000,
self.iface.index,
0x0002,
timeout=int(self.timeout*1000))
if (b[0] == USBTMC_STATUS_PENDING):
time.sleep(0.1)
else:
Expand Down

0 comments on commit 5860bca

Please sign in to comment.