広告メディア事業部【第3回】Pythonによるオシロスコープ波形データ解析の秘訣【アナログデータ編】
広告メディア事業部IT技術
Python でオシロスコープ波形データを解析しよう！
前回は、オシロスコープで波形データを取得する意義について解説しました。
今回は、オシロスコープで得た「csv 形式の波形データを、Python で解析する方法」を紹介します。
Python で波形データを解析するメリット
波形データ解析に Python を用いると、以下のようなメリットがあります。
- スクリプト型言語なので、トライアンドエラーでデバッグが迅速
- 数学に強いライブラリが充実しているため、複雑な演算も可能
- グラフ描画ライブラリで、解析結果をすぐに可視化できる
- PyVISA との連携で、データ取得から解析まで一括でできる
実際にどうやって解析するのか？
今回ご紹介するのは、PC メモリを圧迫しない「時系列逐次処理」を使った解析方法です。
実際のアナログ波形データを想定した正弦波と RC 回路の過渡応答波形を用いて、最大・最小と立ち上がり時間の測定を行います。
最後に、csv 読み込みからデータ取得まで行えるサンプルコードを記載していますので、活用してみてください。
前回の記事はこちら
環境準備
実行環境は、「Python（3.6）」「Spyder（3.2.6）」となっています。
また、コード実行には、以下のモジュールが波形データ生成に必要です。
- Pandas モジュール
- math モジュール
- random モジュール
解析する波形データ
使用する波形データは、Python で簡単に生成できますので、下記のコードを実行してみてください。
1import pandas as pd
2import random
3import math
4
5X_SAMPLE = 3000
6NOISE_SEED_OFF = 123
7#sin_wave
8SIN_INT = 1000
9NOISE_AMP_1=0.1
10AMP_1= 5.0
11
12#rise_wave
13RISE_TIME=300
14AMP_2 =5
15NOISE_AMP_2=0.1
16RC=300
17
18xx = [ i for i in range(3000)]
19analog_wave_1=[]
20noise_1=[]
21for i in range(len(xx)):
22 random.seed(NOISE_SEED_OFF+i)
23 noise_1.append(NOISE_AMP_1*random.randint(-100,100)/100)
24 analog_wave_1.append(AMP_1* math.sin(2/SIN_INT*xx[i]*math.pi)+noise_1[i])
25
26analog_wave_2=[]
27noise_2=[]
28for i in range(len(xx)):
29 random.seed(NOISE_SEED_OFF+i)
30 noise_2.append(NOISE_AMP_2*random.randint(-100,100)/100)
31
32 if i<RISE_TIME:
33 analog_wave_2.append(noise_2[i])
34 else:
35 rise_dt = AMP_2*(1 - math.exp(-1/RC*xx[i-RISE_TIME]))
36 analog_wave_2.append(rise_dt+noise_2[i])
37
38wave_data= pd.DataFrame({"wave_1":analog_wave_1,"wave_2":analog_wave_2})
39wave_data.to_csv("analog_wave_test.csv")実際のアナログデータを想定しているので、上記にはランダムノイズも含まれます。
今回生成した波形データのサンプル数は3,000個で、仕様は以下のとおりです。
- 振幅 5V の正弦波（チャンネル1）
- RC 回路の過渡応答を想定した 5V の立ち上がり波形（チャンネル2）
チャンネル1
理解を深めるため、グラフで見ていきましょう。
振幅 5V の正弦波（チャンネル1）は、このようになっています。
チャンネル2
一方、RC 回路の過渡応答を想定した5Vの立ち上がり波形（チャンネル2）は、こんな感じです。
求める特性値
これで求める特性値は、以下の2つです。
- 振幅 5V の正弦波における最大・最小値
- RC 回路における立ち上がり時間
時系列逐次処理でデータ解析！
それでは、いよいよ実際にデータを解析してみましょう！
ただし、csv ファイルをそのまま読み込むとメモリを大幅に圧迫してしまいます。
そこで、使うのが「時系列逐次処理」です。
時系列逐次処理とは？
時系列逐次処理は、一行ごとに csv データを読み捨てるため、メモリをほとんど使用せずに解析ができます。
オシロスコープの波形が、横軸を時間とする時系列データだからこそ、採用できる方法です。
あるトリガタイミングから過去に遡ってデータ解析を行うことは難しいですが、大抵の特性値なら簡単に取得することができます。
1wave_path = "./analog_wave_test.csv"
2wave_file = open(wave_path,'r')
3skip_row_count = 1
4
5#ヘッダ行等のスキップ
6while skip_row_count !=0:
7 row_data = wave_file.readline()
8 skip_row_count-=1
9
10#データの読み込み開始
11##最初の１行目はループ前に読み込む
12row_data = wave_file.readline()
13split_data = row_data.split(",")
14
15dt_cnt=0
16while row_data:
17 index_dt = float(split_data[0])
18 ch1_dt = float(split_data[1])
19 ch2_dt = float(split_data[2])
20 ###########この間に解析処理を入れる##########
21
22 #######################################
23 ##次の行を読み込む。ここで最終行なら空白が返ってくるためループ終了
24 dt_cnt+=1
25 row_data = wave_file.readline()
26 split_data = row_data.split(",")
27wave_file.close()読み込んだ csv データの１列目がインデックスで、２列目がチャンネル１、３列目がチャンネル2のデータです。
csv ファイル中のデータ開始行が先頭からでないケースに備えて、開始行までスキップする機能も含まれています。
今回のサンプル波形データは１行目がヘッダになっているので、１行分スキップしていますね。
以降の処理は、この繰り返し文の中に記述していく形になります。
①正弦波データの最大・最小
まず、正弦波データの最大・最小値から確認していきましょう。
全データを一括で読み込んだ場合なら「Max」「Min」のメソッドを使えば完了ですが、逐次処理なので少々ややこしい手順となります。
以下のコードで、最大値と最小値を大きいものから順に5個測定できます。
※別途、各種リストの初期化処理が必要なので、最後のサンプルコードを参照ください。
1 #最大値とそのインデックスを大きい順に５つ格納する
2 for i in range(len(ch1_max)):
3 if ch1_dt > ch1_max[i][1]:
4 ch1_max.insert(i,[index_dt,ch1_dt])
5 if len(ch1_max) >MAX_REC:
6 del ch1_max[-1]
7 break
8
9 #最小値とそのインデックスを大きい順に５つ格納する
10 for i in range(len(ch1_min)):
11 if ch1_dt < ch1_min[i][1]:
12 ch1_min.insert(i,[index_dt,ch1_dt])
13 if len(ch1_min) >MAX_REC:
14 del ch1_min[-1]
15 break最大値
行を読み込んでいく過程で最大値を更新した場合に、検出した値とインデックス番号を、ch1_max というリストへ保存していきます。
リスト用の「insert」メソッドを活用すれば、更新データを自由に挿入できるため、自動的に順位を更新可能です。
溢れた6個目のデータは、「del」メソッドで削除されていきます。
最小値
最小値も、基本的な動作は最大値と同じです。
不等号が逆な点と、使用するリスト名が異なるくらいしか違いはありません。
②RC 回路の立ち上がり時間
次に、RC 回路立ち上がりデータにおける、電圧20%から80%までの時間を測定してみましょう。
コードは以下になります。
1 #CH2：RC回路立ち上がり
2 #バッファを使って閾値の超過タイミングを検出する
3 ##サンプル間隔を開けてノイズの影響を軽減する
4 if index_dt % DET_SMP_INT == 0 and rise_st_f:
5 #リングバッファへの保存
6 for i in range(len(ch2_rise_st_buf)-1):
7 ch2_rise_st_buf[len(ch2_rise_st_buf)-1-i] = ch2_rise_st_buf[len(ch2_rise_st_buf)-1-i-1]
8 ch2_rise_st_buf[0]=ch2_dt
9 #5回連続で値が閾値を超えた場合に閾値超過と判定する
10 for i in range(len(ch2_rise_st_buf)):
11 if ch2_rise_st_buf[i] > CH2_20:
12 if i >4 :
13 ch2_rise_st.append([index_dt,ch2_dt])
14 ch1_at_ch2_rise.append([index_dt,ch1_dt])
15 rise_st_f=False
16 break
17 else:
18 breakタイミング検出のポイント
ノイズ重畳のせいで、指定された電圧値の到達タイミング検出が難しくなっています。
そのため、要素数10個のリングバッファを使い、5回連続で指定電圧を超えたタイミングを記録します。
この時、以下のポイントを意識してください。
- 検出処理は1サンプルごとではなく、任意に行う（ノイズに強い検出ができる）
- rise_st_f フラグを使い、検出が終了したら以後は行わない（フラグ管理しないと延々と検出・更新され続ける）
電圧80%の検出タイミングも、変数や閾値が違うだけで基本は同じです。
③両データの連携：立ち上がりの指定タイミングでの正弦波の値
最後に、あるチャンネルの指定タイミングにおけるもう片方のチャンネルの測定値を取得する方法をご紹介します。
チャンネル2で電圧20%を検出したときに、チャンネル1の電圧値を取得してみましょう。
②のコード内にあるch1_at_ch2_rise.append([index_dt,ch1_dt]) がその処理にあたります。
処理中は全チャンネル時間的な同期がとれているので、片方のチャンネル処理中に割り込む形で処理を追加すればいいのです。
サンプルコード：波形データの最大・最小と立ち上がり時間の解析
本記事で紹介したテクニックをまとめて、1つのコードにしてみました。
コード自体は今回使用するサンプルの波形データに合わせたものですが、ファイル名やスキップする行を変えれば、どんな波形データでも対応できますよ。
1wave_path = "./analog_wave_test.csv"
2wave_file = open(wave_path,'r')
3skip_row_count = 1
4
5#設定値関連
6MAX_REC = 5 #最大・最小を記録する際に何個まで記録するか
7CH2_20 = 5*0.2 #チャンネル２で検出する電圧の値（既定電圧の20%）
8CH2_80 = 5*0.8 #チャンネル２で検出する電圧の値（既定電圧の80%）
9DET_SMP_INT = 2 #電圧閾値を検出する際にデータ比較をするサンプル間隔
10
11#ヘッダ行等のスキップ
12while skip_row_count !=0:
13 row_data = wave_file.readline()
14 skip_row_count-=1
15
16#変数の定義
17ch1_max=[]
18ch2_max=[]
19ch1_min=[]
20ch2_min=[]
21ch2_rise_st=[]
22ch2_rise_ed=[]
23ch2_rise_st_buf =[ 0 for i in range(10)]
24ch2_rise_ed_buf =[ 0 for i in range(10)]
25ch1_at_ch2_rise=[]
26#立ち上がり検出を1回だけ実施するためのフラグ
27rise_st_f = True
28rise_ed_f = True
29#データの読み込み開始
30##最初の１行目はループ前に読み込む
31row_data = wave_file.readline()
32split_data = row_data.split(",")
33
34#各測定値の初期値を設定
35ch1_max.insert(0,[float(split_data[0]),float(split_data[1])])
36ch2_max.insert(0,[float(split_data[0]),float(split_data[2])])
37ch1_min.insert(0,[float(split_data[0]),float(split_data[1])])
38ch2_min.insert(0,[float(split_data[0]),float(split_data[2])])
39
40#2行目以降の読み出し処理を開始
41dt_cnt=0
42while row_data: #読み出した値がデータ終了後の空白なら終了
43 index_dt = float(split_data[0])
44 ch1_dt = float(split_data[1])
45 ch2_dt = float(split_data[2])
46 ###########この間に解析処理を入れる##########
47
48 #CH1：正弦波の処理
49 #最大値とそのインデックスを大きい順に５つ格納する
50 for i in range(len(ch1_max)):
51 if ch1_dt > ch1_max[i][1]:
52 ch1_max.insert(i,[index_dt,ch1_dt])
53 if len(ch1_max) >MAX_REC:
54 del ch1_max[-1]
55 break
56
57 #最小値とそのインデックスを大きい順に５つ格納する
58 for i in range(len(ch1_min)):
59 if ch1_dt < ch1_min[i][1]:
60 ch1_min.insert(i,[index_dt,ch1_dt])
61 if len(ch1_min) >MAX_REC:
62 del ch1_min[-1]
63 break
64
65 #CH2：RC回路立ち上がり
66 #バッファを使って閾値の超過タイミングを検出する
67 ##サンプル間隔を開けてノイズの影響を軽減する
68 if index_dt % DET_SMP_INT == 0 and rise_st_f:
69 #リングバッファへの保存
70 for i in range(len(ch2_rise_st_buf)-1):
71 ch2_rise_st_buf[len(ch2_rise_st_buf)-1-i] = ch2_rise_st_buf[len(ch2_rise_st_buf)-1-i-1]
72 ch2_rise_st_buf[0]=ch2_dt
73 #5回連続で値が閾値を超えた場合に閾値超過と判定する
74 for i in range(len(ch2_rise_st_buf)):
75 if ch2_rise_st_buf[i] > CH2_20:
76 if i >4 :
77 ch2_rise_st.append([index_dt,ch2_dt])
78 ch1_at_ch2_rise.append([index_dt,ch1_dt])
79 rise_st_f=False
80 break
81 else:
82 break
83
84 #バッファを使って閾値の超過タイミングを検出する
85 ##サンプル間隔を開けてノイズの影響を軽減する
86 if index_dt % DET_SMP_INT == 0 and rise_ed_f:
87 #リングバッファへの保存
88 for i in range(len(ch2_rise_ed_buf)-1):
89 ch2_rise_ed_buf[len(ch2_rise_ed_buf)-1-i] = ch2_rise_ed_buf[len(ch2_rise_ed_buf)-1-i-1]
90 ch2_rise_ed_buf[0]=ch2_dt
91 #5回連続で値が閾値を超えた場合に閾値超過と判定する
92 for i in range(len(ch2_rise_ed_buf)):
93 if ch2_rise_ed_buf[i] > CH2_80:
94 if i >4 :
95 ch2_rise_ed.append([index_dt,ch2_dt])
96 rise_ed_f=False
97 break
98 else:
99 break
100
101 #######################################
102
103 ##次の行を読み込む。ここで最終行なら空白が返ってくるためループ終了
104 dt_cnt+=1
105 row_data = wave_file.readline()
106 split_data = row_data.split(",")
107wave_file.close()
108
109#測定結果を表示
110print("正弦波の最大値："+str(ch1_max[0][1]))
111print("正弦波の最小値："+str(ch1_min[0][1]))
112print("RC回路の立ち上がり時間："+str(ch2_rise_ed[0][0]-ch2_rise_st[0][0]))
113print("RC回路の20%時の正弦波の電圧値："+str(ch1_at_ch2_rise[0][1]))さいごに
今回は、電圧の最大・最小や立ち上がり時間といった「アナログ特性の取得方法」を紹介しました。
1つの波形データだけなら、オシロスコープのカーソル機能や測定機能だけでも事足りるでしょう。
しかし、Python を使えば、大量のデータ処理ができる上、解析結果の集約も非常に簡単です。
回路動作を自動化し、オシロスコープと連携してしまえば、無人測定も可能ですよ！
そして、次回は「シリアル通信の解析」を行っていきたいと思います。
お楽しみに！
第4回はこちら！
第1回はこちら！
こちらの記事もオススメ！
2020.07.17ライトコード的「やってみた！」シリーズ「やってみた！」を集めました！（株）ライトコードが今まで作ってきた「やってみた！」記事を集めてみました！※作成日が新し...
2020.07.30Python 特集実装編※最新記事順Responder + Firestore でモダンかつサーバーレスなブログシステムを作ってみた！P...
ライトコードでは、エンジニアを積極採用中！
ライトコードでは、エンジニアを積極採用しています！社長と一杯しながらお話しする機会もご用意しております。そのほかカジュアル面談等もございますので、くわしくは採用情報をご確認ください。
採用情報へ
広告メディア事業部
「好きを仕事にするエンジニア集団」の（株）ライトコードです！ ライトコードは、福岡、東京、大阪の3拠点で事業展開するIT企業です。 現在は、国内を代表する大手IT企業を取引先にもち、ITシステムの受託事業が中心。 いずれも直取引で、月間PV数1億を超えるWebサービスのシステム開発・運営、インフラの構築・運用に携わっています。 システム開発依頼・お見積もり大歓迎！ また、現在「WEBエンジニア」「モバイルエンジニア」「営業」「WEBデザイナー」「WEBディレクター」を積極採用中です！ インターンや新卒採用も行っております。 以下よりご応募をお待ちしております！ https://rightcode.co.jp/recruit
おすすめ記事
移転したライトコード大阪オフィスを調査せよ！
広告メディア事業部
2024.04.03
日常
【GCP】BIG QUERYを触り程度に理解してみる
かねまさ（エンジニア）
2024.04.02
IT技術
【Android】Github ActionsでFirebase Test Labの実行を分散する
笹川（エンジニア）
2024.04.02
IT技術
【Next.js】App Router で使用できるキャッシュまとめ
モーリー（エンジニア）
2024.03.29
IT技術
GitHubActionsのランナーに触れてみた
こやまん（エンジニア）
2024.03.28
IT技術
