HC-SR04で距離測定｜ESP32とPythonでリアルタイム表示・CSV保存【超音波センサー#1】

Contents

概要
プログラムコード
1. マイコン（ESP32）
2. Python

概要

今回は、超音波距離センサー「HC-SR04」を使って、対象物までの距離を測定します。

ESP32で距離データを取得し、Pythonに送信してリアルタイム表示。さらに、測定結果をCSVファイルとして保存し、あとから分析できるようにします。

超音波センサーシリーズ第1回では、今後のフィルタ処理、速度計算、状態判定、機械学習、2D位置推定につなげるための基本となる測定環境を作ります。

ちょっとしたデジタル化のトライアルをお考えの方に、少しでも参考になればうれしいです。

詳しくは、以下のYouTube動画をご覧ください。

【第1回】超音波センサーで距離測定！リアルタイム表示とCSV保存をやってみた

プログラムコード

マイコン（ESP32）

超音波を発射
対象物で反射
戻ってくるまでの時間を測定
音速から距離を計算

#define TRIG_PIN 13
#define ECHO_PIN 14

// 最大測定距離[cm]
#define MAX_DISTANCE 700

// タイムアウト時間[μs]
float timeOut = MAX_DISTANCE * 60;

// 音速[m/s]
const int SOUND_VELOCITY = 340;

void setup()
{
    pinMode(TRIG_PIN, OUTPUT);
    pinMode(ECHO_PIN, INPUT);

    Serial.begin(115200);
}

void loop()
{
    float distance = getDistance();

    Serial.print("Distance: ");
    Serial.print(distance);
    Serial.println(" cm");

    delay(100);
}


/*************************************************
 * 距離測定関数
 *************************************************/
float getDistance()
{
    unsigned long pingTime;

    // 超音波パルス送信
    digitalWrite(TRIG_PIN, HIGH);
    delayMicroseconds(10);
    digitalWrite(TRIG_PIN, LOW);

    // 反射波の時間測定
    pingTime = pulseIn(
        ECHO_PIN,
        HIGH,
        timeOut
    );

    // 距離計算
    float distance =
        (float)pingTime *
        SOUND_VELOCITY /
        2 /
        10000;

    return distance;
}

Python

ESP32と接続
CSVファイル準備
リアルタイムでデータ受信
グラフ表示
CSV保存

import serial
import time
import re
import csv
from collections import deque

import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation


# =====================================
# 設定
# =====================================

PORT = "COM5"          # ESP32のCOMポート
BAUDRATE = 115200      # 通信速度
MAX_POINTS = 100       # グラフに表示する最大点数

CSV_FILE = "ultrasonic_distance_log.csv"


# =====================================
# シリアル接続
# =====================================

ser = serial.Serial(
    PORT,
    BAUDRATE,
    timeout=1
)

# ESP32起動待ち
time.sleep(2)


# =====================================
# CSVファイル準備
# =====================================

csv_file = open(
    CSV_FILE,
    mode="w",
    newline="",
    encoding="utf-8-sig"
)

csv_writer = csv.writer(csv_file)

# ヘッダー
csv_writer.writerow([
    "elapsed_time_s",
    "distance_cm",
    "raw_text"
])


# =====================================
# データ保存用
# =====================================

times = deque(maxlen=MAX_POINTS)
distances = deque(maxlen=MAX_POINTS)

start_time = time.time()


# =====================================
# グラフ準備
# =====================================

fig, ax = plt.subplots(figsize=(10, 5))

line, = ax.plot(
    [],
    [],
    marker="o"
)

ax.set_title(
    "Real-time Ultrasonic Distance"
)

ax.set_xlabel("Time [s]")
ax.set_ylabel("Distance [cm]")
ax.grid(True)


# =====================================
# グラフ更新処理
# =====================================

def update(frame):

    while ser.in_waiting > 0:

        line_text = (
            ser.readline()
            .decode("utf-8", errors="ignore")
            .strip()
        )

        # 例:
        # Distance: 52.34cm

        match = re.search(
            r"Distance:\s*(\d+\.?\d*)cm",
            line_text
        )

        if match:

            distance = float(
                match.group(1)
            )

            elapsed_time = (
                time.time()
                - start_time
            )

            # データ保存
            times.append(elapsed_time)
            distances.append(distance)

            # CSV保存
            csv_writer.writerow([
                f"{elapsed_time:.3f}",
                f"{distance:.2f}",
                line_text
            ])

            csv_file.flush()

            # コンソール表示
            print(
                f"{elapsed_time:.2f}s, "
                f"{distance:.2f}cm"
            )

    # グラフ更新
    if len(times) > 0:

        line.set_data(
            times,
            distances
        )

        ax.set_xlim(
            max(0, times[0]),
            times[-1] + 1
        )

        y_min = max(
            0,
            min(distances) - 5
        )

        y_max = (
            max(distances) + 5
        )

        ax.set_ylim(
            y_min,
            y_max
        )

    return line,


# =====================================
# 終了処理
# =====================================

def on_close(event):

    print("終了処理を実行します。")

    csv_file.close()
    ser.close()

    print(
        f"CSVを保存しました: "
        f"{CSV_FILE}"
    )


fig.canvas.mpl_connect(
    "close_event",
    on_close
)


# =====================================
# アニメーション開始
# =====================================

ani = FuncAnimation(
    fig,
    update,
    interval=100,
    cache_frame_data=False
)

plt.show()