#!/usr/bin/env python
# -*- coding: UTF-8 -*-
"""
UL call demonstrated: CtrDevice.c_config_scan()
Purpose: Performs a continuous scan of the
specified encoder channel
Demonstration: Displays the event counter data for
the specified encoders
"""
from __future__ import print_function
from time import sleep
from os import system
from sys import stdout
import ctypes
from uldaq import (get_daq_device_inventory, InterfaceType, ScanStatus,
ScanOption, CInScanFlag, CounterMeasurementType,
CounterMeasurementMode, CounterEdgeDetection,
CounterTickSize, CounterDebounceMode, CounterDebounceTime,
CConfigScanFlag, create_int_buffer, DaqDevice)
def main():
"""Counter input scan with encoder example."""
daq_device = None
ctr_device = None
status = ScanStatus.IDLE
interface_type = InterfaceType.USB
FirstChannel = 0
SecondChannel = 1
ChannelCount = 2
sample_rate = 100.0 # Hz
samples_per_channel = 100
scan_options = ScanOption.CONTINUOUS
scan_flags = CInScanFlag.DEFAULT + CInScanFlag.CTR32_BIT
DistancePerTick = 0.001 # inches per encoder count (arbitrary value for demonstration)
try:
# Get descriptors for all of the available DAQ devices.
devices = get_daq_device_inventory(interface_type)
number_of_devices = len(devices)
if number_of_devices == 0:
raise RuntimeError('Error: No DAQ devices found')
descriptor_index = 999
print('Found', number_of_devices, 'DAQ device(s):')
for i in range(number_of_devices):
if devices[i].product_name == "USB-QUAD08":
descriptor_index = i
descriptor_index = int(descriptor_index)
if descriptor_index not in range(number_of_devices):
raise RuntimeError('Error: Invalid descriptor index')
# Create the DAQ device from the descriptor at the specified index.
daq_device = DaqDevice(devices[descriptor_index])
# Establish a connection to the DAQ device.
descriptor = daq_device.get_descriptor()
print('\nConnecting to', descriptor.dev_string, '- please wait...')
daq_device.connect(connection_code=0)
# Get the CtrDevice object and verify that it is valid.
ctr_device = daq_device.get_ctr_device()
if ctr_device is None:
raise RuntimeError('Error: The DAQ device does not support '
'counters')
# configure counter 0 for encoder
ctr_device.c_config_scan(FirstChannel,
CounterMeasurementType.ENCODER,
CounterMeasurementMode.ENCODER_X1,
CounterEdgeDetection.RISING_EDGE,
CounterTickSize.TICK_20PT83ns,
CounterDebounceMode.NONE,
CounterDebounceTime.DEBOUNCE_0ns,
CConfigScanFlag.DEFAULT)
# configure counter 1 for period mode
ctr_device.c_config_scan(SecondChannel,
CounterMeasurementType.PERIOD,
CounterMeasurementMode.DEFAULT,
CounterEdgeDetection.RISING_EDGE,
CounterTickSize.TICK_20PT83ns,
CounterDebounceMode.TRIGGER_AFTER_STABLE,
CounterDebounceTime.DEBOUNCE_500ns,
CConfigScanFlag.DEFAULT)
# Allocate a buffer to receive the data.
data = create_int_buffer(ChannelCount, samples_per_channel)
print('\n', descriptor.dev_string, ' ready', sep='')
print(' Function demonstrated: ctr_device.c_config_scan()')
print(' Counter(s):', FirstChannel, '-', SecondChannel)
print(' Sample rate:', sample_rate, 'Hz')
# Start the scan
ctr_device.c_in_scan(FirstChannel,
SecondChannel,
samples_per_channel,
sample_rate,
scan_options,
scan_flags,
data)
sleep(0.5)
system('clear')
try:
while True:
try:
status, transfer_status = ctr_device.get_scan_status()
reset_cursor()
print('Please enter CTRL + C to terminate the process\n')
print('Active DAQ device: ', descriptor.dev_string, ' (',
descriptor.unique_id, ')\n', sep='')
print('actual scan rate = ', '{:.6f}'.format(sample_rate),
'Hz\n')
index = transfer_status.current_index
print('currentScanCount = ',
transfer_status.current_scan_count)
print('currentTotalCount = ',
transfer_status.current_total_count)
print('currentIndex = ', index, '\n')
# convert encoder count to signed number
# sign indicates direction
position = ctypes.c_int32(data[index]).value * DistancePerTick
# convert period count to time
time = data[index+1] * 0.00000002083
frequency = 1 / time
rpm = frequency * 60
print('chan =', (0), ': ', '{:.3f}'.format(position),' inches')
print('chan =', (1), ': ', '{:.1f}'.format(rpm),' RPM')
sleep(0.1)
if status != ScanStatus.RUNNING:
break
except (ValueError, NameError, SyntaxError):
break
except KeyboardInterrupt:
pass
except RuntimeError as error:
print('\n', error)
finally:
if daq_device:
if status == ScanStatus.RUNNING:
ctr_device.scan_stop()
if daq_device.is_connected():
daq_device.disconnect()
daq_device.release()
def reset_cursor():
"""Reset the cursor in the terminal window."""
stdout.write('\033[1;1H')
if __name__ == '__main__':
main()