0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@mobility_career(Mobility Career)

Python(PyTorch)×ArduinoでEnd-to-End自動運転ミニカーを作る【コード全公開】

0
Last updated at Posted at 2026-10-03

はじめに

カメラ画像から操作を直接推論する End-to-End（E2E）の自動運転 を、手のひらサイズのミニカーで再現しました。

  • 人間がキーボードで操縦して「画像と操作のペア」を集める（Behavioral Cloning）
  • PyTorchでCNNを学習させる
  • 推論結果をBluetoothでArduinoに送り、モーターを動かす
  • Arduino側に、AIとは独立したフェールセーフを2つ持たせる

この記事では、そのまま動かせるコード一式 と、組んでみてわかった技術的な注意点をまとめます。

実車の開発現場との比較や、作ってみて感じたことは、ブログに書いています。

動作の様子

走行テストの動画

システム構成

重いAI処理を行う上位のPCと、モーターを制御する下位のマイコンを分けています。実車の「SoC（AI処理）」と「MCU（車両制御）」の分離をまねた構成です。

スマホカメラ ──Wi-Fi(IP Webcam)──▶ PC（Python / PyTorch）
                                     │  F / L / R / S の1文字
                                     ▼ Bluetooth（HC-05）
                                  Arduino UNO ──▶ L298N ──▶ DCモーター×2
                                     ▲
                       超音波センサー（HC-SR04）
役割 実車では ミニカーでは
認知 車載カメラ スマホのカメラ
判断 AI処理用のSoC（ADAS ECU） PC上のPython（PyTorch）
命令の伝送 CAN、車載Ethernet Bluetooth（HC-05）
操作 車両制御用のECU（パワトレ、ブレーキのECU） Arduino UNO
安全の見張り役 冗長センサー、安全監視 超音波センサー（HC-SR04）

部品

役割 部品
MCU Arduino UNO R3
上位の判断 PC（GPUがあれば学習が速い）
カメラ スマートフォン（IP Webcamアプリで映像を配信）＋スマホマウント
フェールセーフ用センサー 超音波センサー HC-SR04
通信 Bluetoothモジュール HC-05
モーター モータードライバー L298N ＋ DCモーター×2
モーター用電源 単3電池4本の電池ボックス

配線

ミニカーを真上から見た配線。①HC-05 ②Arduino UNO ③L298N ④HC-SR04

①HC-05　②Arduino UNO　③L298N　④HC-SR04（車体の前方）

配線の注意点

  • モーターの電源はArduinoと分ける。 起動時の突入電流で電圧が下がり（ブラウンアウト）、Arduinoが再起動を繰り返す原因になります。
  • GNDは共通にする。 電源を分けても、Arduino・L298N・電池のGNDはつなぎます。
  • HC-05のRXは3.3V。 ArduinoのTX（5V）からは、抵抗で分圧してつなぎます。
  • 書き込み時はHC-05を外す。 0・1番ピンにつないだままだと、スケッチを書き込めません。
  • モーターのノイズ対策。 DCモーターの端子に0.1μFのコンデンサを入れると、誤動作が減ります。

コード

PC側は4つのファイル、Arduino側は1つのスケッチです。

.
├── e2e_common.py       # モデル・前処理・接続設定（共通）
├── data_collection.py  # Phase 2：データ収集
├── train_model.py      # Phase 3：学習
├── inference.py        # Phase 4：推論と制御
└── minicar_mcu.ino     # Arduino：命令の受信とフェールセーフ

必要なライブラリは次のとおりです。

pip install torch torchvision opencv-python pyserial pandas pillow

Arduino：命令の受信と、独立したフェールセーフ

minicar_mcu.ino
// --- Arduino側：命令の受信と、独立したフェールセーフ ---
const int TRIG = 9, ECHO = 10;                 // 超音波センサー HC-SR04
const int IN1 = 4, IN2 = 5, IN3 = 6, IN4 = 7;  // モータードライバー L298N
const int STOP_DISTANCE_CM = 20;               // これより近いと強制停止
const unsigned long CMD_TIMEOUT_MS = 300;      // 命令がこの時間届かなければ停止

char command = 'S';
unsigned long lastCmdMs = 0;

void setup() {
  Serial.begin(9600);  // HC-05は0・1番ピン（ハードウェアシリアル）に接続
  pinMode(TRIG, OUTPUT);
  pinMode(ECHO, INPUT);
  pinMode(IN1, OUTPUT); pinMode(IN2, OUTPUT);
  pinMode(IN3, OUTPUT); pinMode(IN4, OUTPUT);
  drive(LOW, LOW, LOW, LOW);
}

long readDistanceCm() {
  digitalWrite(TRIG, LOW);  delayMicroseconds(2);
  digitalWrite(TRIG, HIGH); delayMicroseconds(10);
  digitalWrite(TRIG, LOW);
  long us = pulseIn(ECHO, HIGH, 25000);  // 25msで打ち切る（初期値の1秒だと制御が止まる）
  if (us == 0) return -1;                // 反射が返ってこない
  return us / 58;                        // 往復の時間 → 距離[cm]
}

void drive(int a, int b, int c, int d) {
  digitalWrite(IN1, a); digitalWrite(IN2, b);
  digitalWrite(IN3, c); digitalWrite(IN4, d);
}

void loop() {
  // 1. 上位（Python）からの命令を受け取る
  while (Serial.available() > 0) {
    command = Serial.read();
    lastCmdMs = millis();
  }

  // 2. フェールセーフ①：障害物が近ければ、上位の命令を無視して停止
  long d = readDistanceCm();
  if (d > 0 && d < STOP_DISTANCE_CM) {
    drive(LOW, LOW, LOW, LOW);
    return;
  }

  // 3. フェールセーフ②：通信が途絶えたら停止（ウォッチドッグ）
  if (millis() - lastCmdMs > CMD_TIMEOUT_MS) {
    drive(LOW, LOW, LOW, LOW);
    return;
  }

  // 4. 上位の判断どおりに動かす（左右は配線によって逆になることがある）
  switch (command) {
    case 'F': drive(HIGH, LOW, HIGH, LOW); break;  // 前進
    case 'L': drive(LOW, LOW, HIGH, LOW);  break;  // 左旋回（右のモーターだけ回す）
    case 'R': drive(HIGH, LOW, LOW, LOW);  break;  // 右旋回（左のモーターだけ回す）
    default:  drive(LOW, LOW, LOW, LOW);   break;  // 停止
  }
}

ポイントは次の3つです。

  • フェールセーフ①（障害物）： 20cm以内に物があれば、AIが「前進」と言っていても止まります。
  • フェールセーフ②（通信の途絶）： 命令が0.3秒届かなければ止まるウォッチドッグです。これがないと、Bluetoothが切れた瞬間の命令が「前進」なら走り続けます。
  • pulseIn() のタイムアウト： 初期値（1秒）のままだと、反射が返らないたびに制御ループが止まります。25msで打ち切っています。

共通部品

学習時と推論時で前処理がずれると精度が落ちるので、1つのファイルにまとめています。CNNはNVIDIAのPilotNetを小さくした構成です。

e2e_common.py
# --- 共通部品 (e2e_common.py)：モデル・前処理・接続設定 ---
import torch.nn as nn
from torchvision import transforms

STREAM_URL = "http://192.168.x.x:8080/video"  # IP Webcamの映像配信URL
BT_PORT = "/dev/cu.HC-05"                     # macOSの例（Windowsは "COM5" など）
LABELS = ["F", "L", "R"]                      # 前進・左・右


class E2E_Model(nn.Module):
    """NVIDIAのPilotNetを小さくした構成のCNN。画像から3つの操作のどれかを出す"""

    def __init__(self):
        super().__init__()
        self.conv_layers = nn.Sequential(
            nn.Conv2d(3, 24, kernel_size=5, stride=2), nn.ReLU(),
            nn.Conv2d(24, 36, kernel_size=5, stride=2), nn.ReLU(),
            nn.Conv2d(36, 48, kernel_size=5, stride=2), nn.ReLU(),
            nn.Flatten(),
        )
        self.fc_layers = nn.Sequential(
            nn.Linear(48 * 27 * 37, 100), nn.ReLU(),  # 240×320の入力で 27×37 になる
            nn.Linear(100, len(LABELS)),
        )

    def forward(self, x):
        return self.fc_layers(self.conv_layers(x))


# 学習時と推論時で、必ず同じ前処理を使う
preprocess = transforms.Compose([
    transforms.Resize((240, 320)),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])

Phase 2：データ収集

キーボード（W／A／D／S）で操縦しながら、映像と操作を保存します。keyboard ライブラリはmacOSで管理者権限が必要になることが多いので、OpenCVのキー入力を使っています。

data_collection.py
# --- Phase 2：データ収集 (data_collection.py) ---
# W：前進 / A：左 / D：右 / S：停止 / Q：終了（映像のウィンドウを選択した状態で押す）
import os
import time

import cv2
import serial

from e2e_common import BT_PORT, STREAM_URL

os.makedirs("dataset/images", exist_ok=True)
log_file = open("dataset/driving_log.csv", "a")  # 追記：何回かに分けて集められる

cap = cv2.VideoCapture(STREAM_URL)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)  # 古いフレームをためこまない
bt = serial.Serial(BT_PORT, 9600, timeout=0)
time.sleep(2)

KEYS = {ord("w"): "F", ord("a"): "L", ord("d"): "R", ord("s"): "S"}
action = "S"
frame_count = len(os.listdir("dataset/images"))  # 続きの番号から保存する

try:
    while True:
        ok, frame = cap.read()
        if not ok:
            continue

        cv2.imshow("Data Collection", frame)
        key = cv2.waitKey(1) & 0xFF
        if key == ord("q"):
            break
        action = KEYS.get(key, action)  # 押されたキーで操作を切り替える

        # Arduinoへ毎フレーム送る（ラジコンとして動かしつつ、生存確認を兼ねる）
        bt.write(action.encode())

        # 停止中以外の「画像と操作」のペアを保存する
        if action != "S":
            img_name = f"frame_{frame_count:06d}.jpg"
            cv2.imwrite(f"dataset/images/{img_name}", frame)
            log_file.write(f"{img_name},{action}\n")
            frame_count += 1
finally:
    bt.write(b"S")
    log_file.close()
    cap.release()
    bt.close()
    cv2.destroyAllWindows()

データ集めのコツ

  • コースから外れかけた位置から「立て直す運転」も記録する。うまく走った映像だけだと、AIはずれたときの戻り方を知りません。
  • 時間帯や照明を変えて集める。

Phase 3：学習

train_model.py
# --- Phase 3：学習 (train_model.py) ---
import pandas as pd
import torch
import torch.nn as nn
from PIL import Image
from torch.utils.data import DataLoader, Dataset, Subset
from torchvision import transforms

from e2e_common import LABELS, E2E_Model, preprocess

# 照明の違いに強くするため、学習時だけ明るさ・色をランダムに揺らす
augment = transforms.ColorJitter(brightness=0.4, contrast=0.4, saturation=0.3)


class DrivingDataset(Dataset):
    def __init__(self, csv_file, img_dir, train):
        self.data = pd.read_csv(csv_file, names=["image", "label"])
        self.img_dir = img_dir
        self.train = train

    def __len__(self):
        return len(self.data)

    def __getitem__(self, idx):
        name, label = self.data.iloc[idx]
        image = Image.open(f"{self.img_dir}/{name}").convert("RGB")
        if self.train:
            image = augment(image)
        return preprocess(image), LABELS.index(label)


train_data = DrivingDataset("dataset/driving_log.csv", "dataset/images", train=True)
val_data = DrivingDataset("dataset/driving_log.csv", "dataset/images", train=False)
print("ラベルの内訳:", train_data.data["label"].value_counts().to_dict())  # 偏りを確認する

n_val = max(1, len(train_data) // 5)  # 2割を検証用に取り分ける
order = torch.randperm(len(train_data)).tolist()
train_set, val_set = Subset(train_data, order[n_val:]), Subset(val_data, order[:n_val])
train_loader = DataLoader(train_set, batch_size=32, shuffle=True)
val_loader = DataLoader(val_set, batch_size=32)

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = E2E_Model().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

for epoch in range(10):
    model.train()
    for images, labels in train_loader:
        images, labels = images.to(device), labels.to(device)
        optimizer.zero_grad()
        loss = criterion(model(images), labels)
        loss.backward()
        optimizer.step()

    # 学習に使っていないデータで、正解率を確かめる
    model.eval()
    correct = 0
    with torch.no_grad():
        for images, labels in val_loader:
            preds = model(images.to(device)).argmax(1)
            correct += (preds == labels.to(device)).sum().item()
    print(f"Epoch {epoch + 1}: loss {loss.item():.4f}, 検証の正解率 {correct / n_val:.1%}")

torch.save(model.state_dict(), "e2e_model.pth")
print("e2e_model.pth に保存しました")
  • 学習時だけ ColorJitter で明るさや色を揺らし、照明の変化に強くしています。
  • 2割を検証用に分けて、学習に使っていないデータで正解率を確認します。

Phase 4：推論と制御

inference.py
# --- Phase 4：推論と制御 (inference.py) ---
import time

import cv2
import serial
import torch
from PIL import Image

from e2e_common import BT_PORT, LABELS, STREAM_URL, E2E_Model, preprocess

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = E2E_Model().to(device)
model.load_state_dict(torch.load("e2e_model.pth", map_location=device))
model.eval()

cap = cv2.VideoCapture(STREAM_URL)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
bt = serial.Serial(BT_PORT, 9600, timeout=0)
time.sleep(2)

try:
    while True:
        ok, frame = cap.read()
        if not ok:
            break
        t0 = time.perf_counter()

        # OpenCVの画像(BGR)を、学習時と同じ形のテンソルに変換する
        rgb = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
        x = preprocess(rgb).unsqueeze(0).to(device)

        # AIによる判断
        with torch.no_grad():
            action = LABELS[model(x).argmax(1).item()]

        bt.write(action.encode())  # 毎フレーム送る＝Arduinoへの生存確認にもなる

        ms = (time.perf_counter() - t0) * 1000
        cv2.putText(frame, f"{action}  {ms:.0f} ms", (10, 30),
                    cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)  # 判断と処理時間を表示
        cv2.imshow("Inference", frame)
        if cv2.waitKey(1) & 0xFF == ord("q"):
            break
finally:
    bt.write(b"S")  # どんな終わり方でも、最後に停止命令を送る
    cap.release()
    bt.close()
    cv2.destroyAllWindows()
  • 毎フレーム命令を送ることが、Arduinoのウォッチドッグへの生存確認にもなります。
  • 1フレームの処理時間を画面に表示するので、自分の環境の遅延を測れます。

組んでみてわかった注意点

1. 遅延は継ぎ目ごとに積み重なる

カメラ → Wi-Fi → 推論 → Bluetooth → Arduino の各段階で遅れが生じます。特に次の2つは見落としやすいです。

  • OpenCVのバッファ： 映像をためこむので、処理が追いつかないと古いフレームで判断します。CAP_PROP_BUFFERSIZE を1にしています（バックエンドによっては効かないことがあります）。
  • pulseIn() のブロッキング： 前述のとおり、タイムアウトを短くしないと制御ループ全体が止まります。

2. 照明が変わるだけで精度が落ちる（ドメインシフト）

学習時と走行時で部屋の明るさが違うだけで、判断が大きくぶれます。データ集めの段階で照明を変えることと、学習時の ColorJitter が有効です。

3. 正解率の高さにだまされる（データの偏り）

コースを走ると、ラベルの大半が「F（前進）」になります。たとえば7割が前進なら、常に「前進」と答えるだけで正解率は約7割 になり、カーブで曲がれないモデルでも数字は良く見えます。学習スクリプトでラベルの内訳を表示しているのはこのためです。カーブのデータを増やす、左右反転した画像を足す（LとRを入れ替える）などで偏りをならします。

おわりに

数千円の部品でも、「AIの判断を、遅延のある通信越しに、物理的なモーターへ渡す」という実車と同じ構造の難しさが現れます。フェールセーフをAIとは独立した下位のマイコンに持たせる設計は、ミニカーでも実車でも変わらない原則です。

実車の開発現場との比較は、ブログに書いています。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?