1
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?

現実世界を動かす技術をシェアしよう！～身近なものづくりから最先端テクノロジーまで大募集！～
@nak435

ELM327 OBDⅡアダプタ（2）ESP32-C3 と接続する

1
Posted at

使用した OBDⅡアダプタ については、前回の記事を参照してください。

このアダプタは、BLE接続ができるので、ESP32-C3と接続してみました。

1. ESP32-C3 と接続

  • 使用した言語
    MicroPython
    v1.28.0 on 2026-04-06; ESP32C3 module with ESP32C3

  • 使用したライブラリ
    aioble
    This library provides an object-oriented, asyncio-based wrapper for MicroPython's bluetooth API.

  • コード
    次のコードで接続できます 
    import asyncio
import bluetooth
import aioble
import utime

# --- 設定項目 ---
TARGET_BLE_NAME = "OBD" # 「OBD」が含まれれば接続

# BLE UUID設定
UART_SERVICE_UUID = bluetooth.UUID(0xFFF0)
TX_CHAR_UUID      = bluetooth.UUID(0xFFF1) # Notify（データ受信）
RX_CHAR_UUID      = bluetooth.UUID(0xFFF2) # Write（コマンド送信）

# --- タイムスタンプ生成関数 ---
def get_ts():
    """起動からの経過時間を [秒.ミリ秒] 形式の文字列で返す"""
    ms = utime.ticks_ms()
    return f"[{ms // 1000:4d}.{ms % 1000:03d}]"

async def main():
    print(f"{get_ts()} 【OBD2 BLEアダプター】'{TARGET_BLE_NAME}'を検索中...")
    device = None
    
    # 10秒間のスキャンを最大10回試行
    for attempt in range(1, 11):
        print(f"{get_ts()} スキャン試行 {attempt}/10 回目 (10秒間)...")
        async with aioble.scan(duration_ms=10000, interval_us=30000, window_us=30000, active=True) as scanner:
            async for result in scanner:
                
                # デバイス名を取得
                raw_name = result.name()
                if raw_name is None:
                    continue
                
                # ヌル文字(\x00)や改行、前後のスペースをクリーンアップ
                clean_name = raw_name.replace("\x00", "").strip()
                
                # MACアドレス（ログ用）
                addr_str = ":".join(f"{b:02x}" for b in result.device.addr)
                
                # デバッグ用：発見したすべてのBLEデバイスをコンソールに薄く出す
                #print(f"  [スキャン中] 発見: '{clean_name}' [{addr_str}]")
                
                # "obd" が含まれているか判定
                if TARGET_BLE_NAME.lower() in clean_name.lower():
                    device = result.device
                    print(f"{get_ts()} 発見しました！ Target: {clean_name} [{addr_str}]")
                    break
            
            if device:
                break
            
    if not device: 
        return print(f"{get_ts()} エラー: 「{TARGET_BLE_NAME}」を含むBLEデバイスが見つかりません。")
    
    print(f"{get_ts()} 接続中...")
    async with await device.connect(timeout_ms=5000, min_conn_interval_us=7500, max_conn_interval_us=15000) as connection:
        print(f"{get_ts()} 接続成功。通信の安定化を待っています...")
        await asyncio.sleep_ms(500)  # 電波を安定させるためのウェイト
        
        print(f"{get_ts()} [最適化] サービス（UUID: FFF0）のみをピンポイント検索中...")
        
        # 明示的に上位スコープのUUIDを使用して検索
        srv_uuid = bluetooth.UUID(0xFFF0)
        uart_service = await connection.service(srv_uuid)
        
        if uart_service is None:
            return print(f"{get_ts()} エラー: サービス FFF0 が見つかりませんでした。")
                
        print(f"{get_ts()} キャラクタリスティック（FFF1/FFF2）を特定中...")
        tx_char = await uart_service.characteristic(bluetooth.UUID(0xFFF1))
        rx_char = await uart_service.characteristic(bluetooth.UUID(0xFFF2))

        # 最終チェック
        if tx_char is None or rx_char is None:
            return print(f"{get_ts()} エラー: 通信キャラクタリスティックの特定に失敗しました。")

        print(f"{get_ts()} 受信(Notify)のsubscribeを開始します...")
        await tx_char.subscribe(notify=True)
        print(f"{get_ts()} subscribe成功。通信リンク確立")

# プログラムの実行
asyncio.run(main())

  • 実行ログ 
    MPY: soft reboot
[   5.121] 【OBD2 BLEアダプター】'OBD'を検索中...
[   5.122] スキャン試行 1/10 回目 (10秒間)...
[   5.636] 発見しました！ Target: OBDII [77:70:07:31:9d:dc]
[   5.640] 接続中...
[   6.223] 接続成功。通信の安定化を待っています...
[   6.732] [最適化] サービス（UUID: FFF0）のみをピンポイント検索中...
[   7.072] キャラクタリスティック（FFF1/FFF2）を特定中...
[   7.193] 受信(Notify)のsubscribeを開始します...
[   7.253] subscribe成功。通信リンク確立
>>>

    続けて、ATZなどの初期化コマンドを送信すればよいです...

2. 自由にATコマンドを打つ

OBDⅡアダプタに接続できたので、自由にATコマンドを打ってみて、どんな応答が返るのか見てみたいと思い、次のコードを作成しました。
ESP32-C3はシングルコアですが、asyncio を使ってマルチスレッドで処理することで、ATコマンドを手打ちできるようにしています。
途中、応答の一部が受信できない aioble のバグに遭遇しましたが、AI に助けてもらい完成したコードです。

  • ATコマンドを自由に打てるコード 
    import asyncio
import bluetooth
import aioble
import utime
import sys
import select
from collections import deque

# ============================================================
# 設定項目
# ============================================================
TARGET_BLE_NAME = "OBD"

UART_SERVICE_UUID = bluetooth.UUID(0xFFF0)
TX_CHAR_UUID      = bluetooth.UUID(0xFFF1)  # Notify（データ受信）
RX_CHAR_UUID      = bluetooth.UUID(0xFFF2)  # Write（コマンド送信）

# ============================================================
# グローバル状態
# ============================================================
START_TICK = utime.ticks_ms()
_rx_buffer = "" # 受信データの蓄積バッファ（取りこぼし防止のため常時保持）
_cmd_queue = asyncio.ThreadSafeFlag() if False else None  # 未使用（将来拡張用に残した）


def get_ts():
    diff = utime.ticks_diff(utime.ticks_ms(), START_TICK)
    sec = diff // 1000
    ms = diff % 1000
    return f"[{sec:4d}.{ms:03d}]"


# ============================================================
# 受信常駐タスク：BLE Notifyを監視し続け、'>' が来るたびに1応答として出力
# ============================================================
async def notify_listener(tx_char):
    global _rx_buffer
    print(f"{get_ts()} [System] 受信常駐タスク起動（キュー直接ドレインモード）")
    pkt_count = 0

    while True:
        try:
            # 最初の1件だけ notified() で待つ（キューが空の時のみ安全に動作する）
            first = await tx_char.notified()
            pkt_count += 1
            #print(f"{get_ts()} [PKT#{pkt_count}] len={len(first)} raw={repr(first)}")
            _rx_buffer += first.decode('utf-8', 'ignore')

            # 後続パケットがキューに溜まるのを少し待つ
            await asyncio.sleep_ms(80)

            # aiobleバグ回避(issue#934)：notified()を再度呼ばず、キューを直接空になるまで吸い出す
            q = tx_char._notify_queue
            while q:
                data = q.popleft()
                pkt_count += 1
                #print(f"{get_ts()} [PKT#{pkt_count}] len={len(data)} raw={repr(data)}")
                _rx_buffer += data.decode('utf-8', 'ignore')

            # 次回のnotified()呼び出しのため、イベントフラグを明示的にクリア
            tx_char._notify_event.clear()

            # バッファ内に完成した応答（'>'まで）があれば切り出して表示
            while ">" in _rx_buffer:
                idx = _rx_buffer.find(">")
                raw = _rx_buffer[:idx + 1]
                _rx_buffer = _rx_buffer[idx + 1:]
                pretty = raw.replace("\r\r", "\n").replace("\r", "\n").strip()
                print(f"{get_ts()} [応答RAW]  {repr(raw)}")
                if pretty:
                    #print(f"{get_ts()} [応答] {pretty}")
                    pass

        except Exception as e:
            print(f"{get_ts()} [System] 受信タスクでエラー: {e}")
            await asyncio.sleep_ms(100)

# ============================================================
# 送信常駐タスク：_next_cmd にセットされたコマンドを検知して送信
# ============================================================
_next_cmd = None

async def command_sender(rx_char):
    global _next_cmd
    print(f"{get_ts()} [System] 送信常駐タスク起動")

    while True:
        if _next_cmd is not None:
            cmd = _next_cmd
            _next_cmd = None
            print(f"{get_ts()} >> [TX] {cmd}")
            try:
                await rx_char.write((cmd + "\r").encode())
            except Exception as e:
                print(f"{get_ts()} [System] 送信エラー: {e}")

        await asyncio.sleep_ms(50)


# ============================================================
# REPL入力監視タスク：標準入力を非ブロッキングで監視し、_next_cmdにセット
# ============================================================
async def repl_command_listener():
    global _next_cmd
    print(f"{get_ts()} [System] REPL入力監視タスク起動。ATコマンドを入力してEnterを押してください。")

    poller = select.poll()
    poller.register(sys.stdin, select.POLLIN)

    while True:
        # 0ms待機（ノンブロッキング）で入力があるかだけチェック
        events = poller.poll(0)
        if events:
            line = sys.stdin.readline().strip()
            if line:
                _next_cmd = line
        await asyncio.sleep_ms(50)


# ============================================================
# 初期化コマンド送信（起動時の自動シーケンス用）
# 受信は notify_listener が並行して処理しているので、ここでは送信して少し待つだけ
# ============================================================
async def send_init_cmd(rx_char, cmd_str, wait_ms=700):
    print(f"{get_ts()} [自動初期化] 送信: {cmd_str}")
    await rx_char.write((cmd_str + "\r").encode())
    await asyncio.sleep_ms(wait_ms)


# ============================================================
# BLEスキャン＆接続
# ============================================================
async def scan_and_connect():
    print(f"{get_ts()} 【OBD2 BLEアダプター】'{TARGET_BLE_NAME}' を検索中...")
    device = None

    for attempt in range(1, 11):
        print(f"{get_ts()} スキャン試行 {attempt}/10 回目...")
        async with aioble.scan(duration_ms=10000, interval_us=30000, window_us=30000, active=True) as scanner:
            async for result in scanner:
                raw_name = result.name()
                if raw_name is None:
                    continue
                clean_name = raw_name.replace("\x00", "").strip()
                addr_str = ":".join(f"{b:02x}" for b in result.device.addr)
                #print(f"  [スキャン中] 発見: '{clean_name}' [{addr_str}]")
                if TARGET_BLE_NAME.lower() in clean_name.lower():
                    device = result.device
                    print(f"{get_ts()} ★発見しました！ Target: {clean_name} [{addr_str}]")
                    break
            if device:
                break

    if not device:
        print(f"{get_ts()} エラー: 「{TARGET_BLE_NAME}」を含むBLEデバイスが見つかりません。")
        return None

    print(f"{get_ts()} 接続中...")
    connection = await device.connect(timeout_ms=5000,
        min_conn_interval_us=7500,   # 7.5ms（BLE規格上の最短値）
        max_conn_interval_us=15000,  # 15ms
    )
    print(f"{get_ts()} 接続成功。安定化を待っています...")
    await asyncio.sleep_ms(500)

    # 接続直後にMTU拡張を試みる（対応していれば大きめのMTUを要求）
    try:
        mtu = await connection.exchange_mtu(247)
        print(f"{get_ts()} MTU交渉成功: {mtu}")
    except Exception as e:
        print(f"{get_ts()} MTU交渉失敗またはメソッド無し: {e}")

    return connection


# ============================================================
# メイン処理
# ============================================================
async def main():
    connection = await scan_and_connect()
    if connection is None:
        return

    uart_service = await connection.service(UART_SERVICE_UUID)
    if uart_service is None:
        print(f"{get_ts()} エラー: サービス FFF0 が見つかりませんでした。")
        return

    tx_char = await uart_service.characteristic(TX_CHAR_UUID)
    rx_char = await uart_service.characteristic(RX_CHAR_UUID)
    if tx_char is None or rx_char is None:
        print(f"{get_ts()} エラー: 通信キャラクタリスティックの特定に失敗しました。")
        return

    print(f"{get_ts()} 受信(Notify)の購読を開始します...")
    await tx_char.subscribe(notify=True)
    tx_char._notify_queue = deque((), 20)
    print(f"{get_ts()} subscribe成功。通知キューを深さ20に拡張しました。")
    await asyncio.sleep_ms(200)
        
    # --- 受信・送信・REPL監視の3タスクを先に起動 ---
    # ここから先は、届いたデータはnotify_listenerが拾ってくれる
    listener_task = asyncio.create_task(notify_listener(tx_char))
    sender_task = asyncio.create_task(command_sender(rx_char))
    repl_task = asyncio.create_task(repl_command_listener())

    # --- 自動初期化シーケンス（必要に応じてコメントアウト可） ---
    print(f"\n{get_ts()} --- ELM327自動初期化シーケンス開始 ---")
    await send_init_cmd(rx_char, "ATZ", wait_ms=1500)
    await send_init_cmd(rx_char, "ATE1", wait_ms=700)
    await send_init_cmd(rx_char, "ATL0", wait_ms=700)
    await send_init_cmd(rx_char, "ATS1", wait_ms=700)
    await send_init_cmd(rx_char, "ATSP6", wait_ms=700)

    print(f"\n{get_ts()} [準備完了] REPLからATコマンドを自由に入力してください。")
    print(f"{get_ts()} (プログラムを止めずに動き続けます。Ctrl+Cで終了)")

    # --- 3つの常駐タスクを永続的に並行実行 ---
    await asyncio.gather(listener_task, sender_task, repl_task)


# ============================================================
# 実行
# ============================================================
try:
    asyncio.run(main())
except KeyboardInterrupt:
    print(f"\n{get_ts()} ユーザーにより中断されました。")
finally:
    asyncio.new_event_loop()  # 次回実行時のためにイベントループをリセット

  • 実行ログ 
    [   0.001] 【OBD2 BLEアダプター】'OBD' を検索中...
[   0.002] スキャン試行 1/10 回目...
[   0.331] ★発見しました！ Target: OBDII [77:70:07:31:9d:dc]
[   0.333] 接続中...
[   1.419] 接続成功。安定化を待っています...
[   1.959] MTU交渉成功: 247
[   2.439] 受信(Notify)の購読を開始します...
[   2.499] subscribe成功。通知キューを深さ20に拡張しました。

[   2.708] --- ELM327自動初期化シーケンス開始 ---
[   2.710] [自動初期化] 送信: ATZ
[   2.713] [System] 受信常駐タスク起動（キュー直接ドレインモード）
[   2.719] [System] 送信常駐タスク起動
[   2.721] [System] REPL入力監視タスク起動。ATコマンドを入力してEnterを押してください。
[   3.629] [応答RAW]  'ATZ\rELM327 v1.5\r\r>'
[   4.268] [自動初期化] 送信: ATE1
[   4.409] [応答RAW]  'ATE1\rOK\r\r>'
[   5.028] [自動初期化] 送信: ATL0
[   5.189] [応答RAW]  'ATL0\rOK\r\r>'
[   5.808] [自動初期化] 送信: ATS1
[   5.969] [応答RAW]  'ATS1\rOK\r\r>'
[   6.588] [自動初期化] 送信: ATSP6
[   6.749] [応答RAW]  'ATSP6\rOK\r\r>'

[   7.378] [準備完了] REPLからATコマンドを自由に入力してください。
[   7.388] (プログラムを止めずに動き続けます。Ctrl+Cで終了)

ATRV
[  56.418] >> [TX] ATRV
[  56.699] [応答RAW]  'ATRV\r12.1V\r\r>'

010C
[  64.168] >> [TX] 010C
[  65.039] [応答RAW]  '010C\rNO DATA\r\r>'

010D
[  82.288] >> [TX] 010D
[  83.149] [応答RAW]  '010D\rNO DATA\r\r>'
    前回同様、実車に接続していないため、車速や回転数は取得できていません。
    次回は、実車に接続して確認してみます。

次回は、実車に接続して確認してみます。

実車に接続して実行しましたが、車速や回転数は取得できませんでした。
接続したクルマは この OBDⅡアダプタ と相性が悪いようです。

このOBDⅡアダプタメーカーのサイトで紹介されていたアプリを使ってみましたが、やはり、車速や回転数は取得できませんでした。

と いうことで、この先に予定していた 後続記事は中止します。

技適もありませんので、これ以降使うこともないでしょう。
簡単に基板が取り出せたので、写真を載せておきます。

IMG_8874.jpeg
  • 中央 YMB02 のすぐ下の16pのICはYM1130：ELM327のコマンドエミュレーションとBluetooth通信機能
  • その右横のクリスタル：40MHz
  • 上部 YMIOT 横の8pのICは表面に印字無し：SN65HVD230等の互換品のCANトランシーバーICと推測

以上

1
0
3

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
1
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?