この記事は，ご覧のモジュールの利用でお送りします．
import numpy as np
import nidaqmx as ni
from nidaqmx import constants
***
DAQから波形を出力したい．
それも，ちょっとした波形ではなく，1時間や2時間にもわたる，高サンプリングレートの波形だ．
fs = 800000 # サンプルレート 800 kHz
dur = 3600 # 波形の時間長さ 3600 sec
n_samp = fs * dur # 波形のサンプル数
def get_wav():
t = np.linspace(0, dur, n_samp) # 高サンプリングレート の
wav = np.sin(2 * np.pi * t) # 1 Hz sin波
return wav
上記の波形をどう出力するか．
パッと思いつくのは，数時間分の波形をそのままtask.write()に渡すことである．
"""
ダメな例
"""
wf = get_wav() # 高サンプリングレート 1Hz sin波 1時間分
with ni.Task() as task:
task.ao_channels.add_ao_voltage_chan("Dev1/ao0") # タスクにoutputチャンネル追加
task.timing.cfg_samp_clk_timing(
rate = fs, # サンプリングレート
samps_per_chan = n_samp # サンプル数
)
task.write(get_wav(), auto_start=True) # 波形出力
task.wait_until_done() # 待機
task.stop() # 終了
しかし，なぜ気づかなかったのだろう．
高サンプリングレートの波形を数時間分もメモリに詰め込むと，メモリがパンクしてしまうのだ．
おかげでPCがフリーズしてしまった．
次に思いつくのは，波形をチャンクごとに生成しながら出力することである．
"""
get_wav()を，波形の一部（チャンク）を生成する関数 get_wav_chunk() に変更
"""
fs = 800000 # サンプルレート 800 kHz
dur = 3600 # 波形の時間長さ 3600 sec
n_samp = fs * dur # 波形のサンプル数
+ chunk_dur = 1 # 波形を何秒で区切るか 1 sec
+ chunk_size = fs * chunk_dur # チャンクサイズ
+ n_chunk = int(n_samp / chunk_size) # チャンク数
- def get_wav():
- t = np.linspace(0, dur, n_samp) # 高サンプリングレート の
+ def get_wav_chunk(i: int):
+ start = i * chunk_dur
+ end = start + chunk_dur
+ t = np.arange(start, end, 1/fs) # 1チャンク分の
wav = np.sin(2 * np.pi * t) # 1 Hz sin波
return wav
task.write()をfor文で回せば，チャンクごとに波形が出力されるに違いない…！
"""
微妙な例
"""
- wf = get_wav() # 高サンプリングレート 1Hz sin波
-
with ni.Task() as task:
task.ao_channels.add_ao_voltage_chan("Dev1/ao0") # タスクにoutputチャンネル追加
task.timing.cfg_samp_clk_timing(
rate = fs, # サンプリングレート
- samps_per_chan = n_samp # サンプル数
+ samps_per_chan = chunk_size # 1チャンクのサンプル数
)
- task.write(get_wav(), auto_start=True) # 波形出力
- task.wait_unitl_done() # 待機
- task.stop() # 終了
+ for i_chunk in range(n_chunk):
+ chunk = get_wav_chunk(i_chunk)
+ chunk *= 1 + i_chunk % 2 # 偶数チャンクは振幅を2Vに
+ task.write(chunk, auto_start=True) # 波形出力
+ task.wait_until_done() # 待機
+ task.stop() # 終了
これはきた…と思いきや，波形は途切れ途切れになってしまった．
どうやら，タスクを繰り返しstartしたりstopしたりしているのが原因らしい．
ならば，バッファをうまく使えば…
with ni.Task() as task:
task.ao_channels.add_ao_voltage_chan("Dev1/ao0") # タスクにoutputチャンネル追加
task.timing.cfg_samp_clk_timing(
rate = fs, # サンプリングレート
- samps_per_chan = chunk_size # 1チャンクのサンプル数
+ samps_per_chan = n_samp # 全体のサンプル数
)
+ task.out_stream.output_buf_size = chunk_size * 2 # バッファサイズをチャンク数の2倍に
+ task.out_stream.regen_mode = constants.RegenerationMode.DONT_ALLOW_REGENERATION # 波形の再生成を禁止
for i_chunk in range(n_chunk):
chunk = get_wav_chunk(i_chunk)
chunk *= 1 + i_chunk % 2 # 偶数チャンクは振幅を2Vに
task.write(chunk, auto_start=True) # 波形出力
- task.wait_until_done() # 待機
- task.stop() # 終了
+ task.wait_until_done() # 待機
+ task.stop() # 終了
こうすれば，波形を生成している間に，バッファの空きスペースに次のチャンクの波形を書き込める．
ご覧の通り，滑らかな波形を生成できた．
美しい！
***
この記事は，ご覧のコードの提供でお送りしました．
import numpy as np
import nidaqmx as ni
from nidaqmx import constants
fs = 800000 # サンプルレート 800 kHz
dur = 3600 # 波形の時間長さ 3600 sec
n_samp = fs * dur # 波形のサンプル数
chunk_dur = 1 # 波形を何秒で区切るか 2 sec
chunk_size = fs * chunk_dur # チャンクサイズ
n_chunk = int(n_samp / chunk_size) # チャンク数
def get_wav_chunk(i: int):
start = i * chunk_dur
end = start + chunk_dur
t = np.arange(start, end, 1/fs) # 1チャンク分の
wav = np.sin(2 * np.pi * t) # 1 Hz sin波
return wav
with ni.Task() as task:
task.ao_channels.add_ao_voltage_chan("Dev1/ao0") # タスクにoutputチャンネル追加
task.timing.cfg_samp_clk_timing(
rate = fs, # サンプリングレート
samps_per_chan = n_samp # 全体のサンプル数
)
task.out_stream.output_buf_size = chunk_size * 2 # バッファサイズをチャンク数の2倍に
task.out_stream.regen_mode = constants.RegenerationMode.DONT_ALLOW_REGENERATION # 波形の再生成を禁止
for i_chunk in range(n_chunk):
chunk = get_wav_chunk(i_chunk)
chunk *= 1 + i_chunk % 2 # 偶数チャンクは振幅を2Vに
task.write(chunk, auto_start=True) # 波形出力
task.wait_until_done() # 待機
task.stop() # 終了
終
制作・著作
━━━━━
SHijiMi