In this demo we connect to a single MonoDAQ and setup:
The digital output channel is toggled at a rate ~10 Hz. We may wire it to stimulate other analog and digital channels as a feedback signal. Then we show:
from isotel.idm import monodaq, signal, gateway
mdu = monodaq.MonoDAQ_U()
# or connect to MonoDAQ with username and password
#mdu = monodaq.MonoDAQ_U(gateway.Group(username='user', password='test'))
First we call reset() function to begin from start, then we configure the channels, and we print out the configuration. Configuration could also be stored in a 1-wire memory attached to the connector, which is automatically reloaded when connector with such memory is detected.
When directly accessing the parameters, you may want to use a Parameters view of the IDM, which helps you with browsing thru the device data structure.
mdu.reset()
mdu['ch.function.function2'] = 'Voltage Input'
mdu['ch.function.function3'] = 'Digital Input'
mdu['ch.function.function4'] = 'Voltage Input'
mdu['ch.function.function5'] = 'Digital Output'
mdu['ch.function.function10'] = 'Current Input'
mdu['ch.set.set5'] = '~1' # We set to High but as MDU will return actual voltage level 5 we need to add ~
mdu.print_setup()
The following thread starts a parallel independent instance of mdu to toggle the DO1 pin at rate of about 10 Hz.
A channel number is obtained from a pin name.
It uses mdu.time() to find the last received update from the MonoDAQ.
import threading
def toggle_digital__():
mdu = monodaq.MonoDAQ_U()
val = 0
while True:
mdu.set_channel_value('DO1', '~' + str(val), after=mdu.time() + 0.1)
val ^= 1
threading.Thread(target=toggle_digital__).start()
The fetch() is a python generator and fetches N samples and returns tuples organized as:
For example:
s[0][1][0] means from stream0, channels, and select 0-th channelGenerator provides meta information on channels in the first line (row).
list( mdu.fetch(N=2) )
Scope offers real-time representation of data while being acquired.
Data must be re-arranged to a dict of lists via collect2dict() which has an optional parameter split=<N>. By specifying it, the collect2dict() cuts the input stream into slices of N. No data is lost.
The scope() updates the screen each time it receives suc a slice.
If the split parameter is not given then entire stream is collected and then passed further to be drawn.
signal.scope( signal.collect2dict(mdu.fetch(100000), split=20000), title="Demo");
Python generators offer easy cascading and propagation of information. In the following example we filter out the digital stream, and leave only the first stream 0.
def filter_din(source):
yield (next(source)[0],) # copy header
for s in source:
yield (s[0],)
signal.scope(signal.collect2dict(filter_din(mdu.fetch(200))), title="Analog Only");
Example below shows how to wait for a rising edge on voltage channel V1 and by specifying condition and optional pre-condition. At the time condition triggers P pre-buffered samples are yielded, and continues for N samples.
if signal.scope(signal.collect2dict(signal.trigger(mdu.fetch(8000), precond='V1 < 1', cond='V1 > 1',
P=50, N=50)), title="Rising Edge Trigger") == None:
print('Signal not detected')
Shows how to capture data in a list, and then re-generate the stream through triggering function.
sig1 = list( mdu.fetch(8000) );
Now lets post-process the signal in sig1 by passing it thru the signal's trigger() method. Note that in python regeneration is done as: (s for s in sig1)
sig2 = list(signal.trigger((s for s in sig1), cond='V1 > 2', precond='V1 < 1', P=15, N=15))
Let's plot the result in sig2:
signal.scope( signal.collect2dict( (s for s in sig2) ) );
Let's reuse the last captured data, and pass it thru the addmath() function to calculate power of the V1+ [V] and CUR+ [mA] signals. One may specify just sufficient number of starting characters that uniquely define the channel name. The 2nd parameter defines the name of channel, and 3rd parameter defines an expression to evaluate.
signal.scope( signal.collect2dict( signal.addmath((s for s in sig2), 'P [mW]', expr='V1 * CUR')) );
Importing data from collected dictionaries to panda is demonstrated below.
We re-play sig2 into a math function, collect it, and take the fist sice out of the generated list to feed result to the panda DataFrame.
import pandas as pd
res = list( signal.collect2dict( signal.addmath((s for s in sig2), 'P [mW]', expr='V1 * CUR')) )[0]
d = pd.DataFrame.from_dict(res['P [mW]'])
d.head()
d.describe()
Let's write a simple sink that re-formats the streams' tuples into a TSV output:
def to_tsv(stream):
def print_tsv(tup):
line = ''
for s in tup:
line += str(s[0]) + '\t'
for y in s[1]:
line += str(y) + '\t'
return line
for s in stream:
yield print_tsv(s)
for ln in to_tsv( (s for s in sig2) ):
print(ln)
Let's fetch directly or stream to a file:
for ln in to_tsv( mdu.fetch(5) ):
print(ln)
with open('measurement.tsv', 'w') as file:
for ln in to_tsv( (s for s in sig2) ):
file.write(ln + '\n')
!cat measurement.tsv