Using Python and the hidapi library with ADU USB Data Acquisition Products (Linux & OS X)

View the ADU series of USB based Data Acquisition Products


Introduction


Communicating with USB devices via software involves a few simple steps. Unlike RS232 based devices which are connected to physical COM ports, USB devices are assigned a logical handle by operating systems when they are first plugged in. This process is known as enumeration. Once a USB device has been enumerated, it is ready for use by the host computer software. For the host application software to communicate with the USB device, it must first obtain the handle assigned to the USB device during the enumeration process. The handle can be obtained using an open function along with some specific information about the USB device. Information that can be used to obtain a handle to a USB device include, serial number, product ID, or vendor ID.

Once we obtain a USB device handle, we can read and write information to and from the USB device via our application. Once the application has finished with all communication with the USB device, the handle is closed. The handle is generally closed when the application terminates.

The sample source code outlines the basics of communicating directly with an ADU device on Linux and OS X using Python and libhidapi. Basics of opening a USB device handle, writing and reading data, as well as closing the handle of the ADU usb device is provided as an example. An alternate way of working with ADU devices in Linux is to use the adutux kernel driver to access the device as a file descriptor (outlined here: https://www.ontrak.net/Linux/APG.htm). The recommended way of working with ADU devices via Python on OS X is via the hidapi (this example).

All source code is provided so that you may review details that are not highlighted here.

Lets have a look at the code......


This example illustrates the basics of reading and writing to ADU devices using the hidapi library.

NOTE: When running the example, it must be run with root privileges in order to access the USB device.

hidapi is a library that provides simple access to HID compliant USB devices (https://github.com/libusb/hidapi). We will need a vendor ID and product ID in order to open the USB device. The VENDOR_ID define will always remain the same as this is OnTrak's USB vendor ID, however, PRODUCT_ID must be set to match the product that is connected via USB. See this link for a list of OnTrack product IDs: https://www.ontrak.net/Nodll.htm. A Python interface to the C library is available here: https://github.com/trezor/cython-hidapi.

First we'll import the hid library. If you haven't yet installed it, you may do so in the command line via pip install hidapi or by installing via requirements.txt (pip install -r requirements.txt). If you receive errors when installing the hidapi Python module on Linux, check that udev is installed (on Debian/Ubuntu: sudo apt-get install libudev1 libudev-dev)

We'll declare OnTrak's vendor ID and the product ID for the ADU device we wish to use (in our case 200 for ADU200).


import hid

VENDOR_ID = 0x0a07 # OnTrak Control Systems Inc. vendor ID
PRODUCT_ID = 200 # ADU200 Device product name - change this to match your product
 

Next, we'll open the connected USB device that matches our vendor and product ID. This device will be used for all of our interactions with the ADU via hidapi (opening, closing, reading and writing commands).


device = hid.device()
device.open(VENDOR_ID, PRODUCT_ID)
print('Connected to ADU{}\n'.format(PRODUCT_ID))

Now that we have successfully opened our device, we can write commands to the ADU device and read the result.

Two convenience functions have been written to help properly format command packets to send to the ADU device, as well as to read a result from the ADU device: write_to_adu() and read_from_adu(). We'll cover the internals of these functions at the end of this page.


# write_to_adu() returns the number of bytes written on success, and None on failure
bytes_written = write_to_adu(device, 'RK0') # set relay 0, note: device does not send a response for this
bytes_written = write_to_adu(device, 'SK0') # reset relay 0

In order to read from the ADU device, we can send a command that requests a return value (as defined in our product documentation). Such a command for the ADU200 is RPK0. This requests the value of relay 0, which we previously set with the RK0 and SK0 commands in the above code block.

We can then use read_from_adu() to read the result. read_from_adu() returns the data read in integer format on success, and None on failure. A timeout is supplied for the maximum amount of time that the host (computer) will wait for data from the read request.


    bytes_written = write_to_adu(device, 'RPA') # request the status of PORT A in binary format

    data = read_from_adu(device, 200) # read the response from above PA request
    if data != None:
		print('Received string: {}'.format(data))
		# data_int = int(data) # if you wish to work with the data in integer format
		# print( 'Received int: {}'.format(data_int))

When we are finished communicating with the device, we should close the device. This is generally donw when the application is closed.


		device.close()

Further Details


If you're interested in the internals of write_to_adu() and read_from_adu() as well as how to structure a command packet, the details are below.

All ADU commands have their first byte set to 0x01 and the following bytes contain the ASCII representation of the command. The ADU command packet format is described here: https://www.ontrak.net/Nodll.htm. As described in the link, the remaining bytes in the command buffer must be null padded (0x00).

We'll use hid_write() to write our command to the device. After sending, we check to result to make sure the transfer succeeded. The number of bytes sent should match our command buffer's size (8 bytes)


def write_to_adu(dev, msg_str):
    print('Writing command: {}'.format(msg_str))

    # message structure:
    #   message is an ASCII string containing the command
    #   8 bytes in length
    #   0th byte must always be 0x01 (decimal 1)
    #   bytes 1 to 7 are ASCII character values representing the command
    #   remainder of message is padded to 8 bytes with character code 0

    byte_str = chr(0x01) + msg_str + chr(0) * max(7 - len(msg_str), 0)

    try:
        num_bytes_written = dev.write(byte_str.encode())
    except IOError as e:
        print ('Error writing command: {}'.format(e))
        return None 

    return num_bytes_written
 

If the write, we should now have a result to read from the command we previously sent. We can use hid.device.read() to read the value. The arguments are the USB device and a timeout. read() should return the data read from the device.

If reading from the device was successful, let's extract the data we are interested in. It's important to note that the data returned is 8 bytes in length. The first byte of the data returned is 0x01 and is followed by an ASCII representation of the number. The remainder of the bytes are padded with 0x00 (NULL) values. We can construct a string from the second byte to the eighth byte and strip out the null '\x00' characters. The ASCII string is then returned.


def read_from_adu(dev, timeout):
	try:
        # read a maximum of 8 bytes from the device, with a user specified timeout
        data = dev.read(8, timeout)
    except IOError as e:
        print ('Error reading response: {}'.format(e))
        return None

    byte_str = ''.join(chr(n) for n in data[1:]) # construct a string out of the read values, starting from the 2nd byte

	result_str = byte_str.split('\x00',1)[0] # remove the trailing null '\x00' characters

    if len(result_str) == 0:
        return None

	return result_str

If you wish to view a list of all connected devices, you may use the enumerate() function of hidapi:


print('Connected ADU devices:')
for d in hid.enumerate(VENDOR_ID):
	print('    ADU{}'.format(d['product_id']))
print('')

When newly connecting to a device, you may want to read until there is nothing left to read from the device. This is to clear any pending reads that was not initiated by us. We'll review the read_adu function later on.


# clear the read buffer of any unread values
# this is important so that we don't read old values from previous requests sent to the device
while True:
	if read_adu(device, 200) == None:
		break

Download Linux & OS Python Example File in ZIP Format