前回のとおり、Raspberry Pi Pico W から XIAO-ESP32-C6 に基板を変えてリトライ!Socket周りでメモリエラーやら諸々トラブル続きでしたが、なんとか突破してWiFiとUDP(たぶんTCPも)接続は安定させられたのではないかと思います(執筆時点、prremote未リリースのHEADからビルドしたバージョンを使ってます、リリースは後ほど)。
以下が、ESP32-C6に書き込んだコードです。モータードライバには秋月電子通商のTB6612使用 Dual DCモータードライブキットを利用しています。せっかく電源を分離しやすい仕様になっているのに、配線をサボって共用しているため、電圧降下対策に470μFコンデンサを挟んでいます。1μFのノイズ避けコンデンサも一応挟んでますが、どのくらい効いてるのかは不明です。
# XIAO ESP32C6 版 RCカー制御スクリプト
# pico-w/main5.rb (Raspberry Pi Pico W 版) を Seeed Studio XIAO ESP32C6 へ移植したもの。
#
# - WiFi : ESP32C6 内蔵 WiFi。
# - LED : XIAO ESP32C6 のオンボード LED は GPIO15 / アクティブ Low(0=点灯 / 1=消灯)。
# - ADC : ESP32C6 の ADC1 は GPIO0-6。ADC#read は Pico と同じく 16bit 換算
# (0-65520、内部で 12bit を <<4)なのでホスト側プロトコルは変更不要。
# - GPIO/PWM: ピン番号を XIAO のピン配置(D0-D10)に合わせて再割り当て(下記)。
# PWM(LEDC)は最大 6ch。USB(GPIO12/13)・ストラップ(GPIO8/9/15)は使用不可。
#
# WIFI_SSID = "your-ssid"
# WIFI_PASS = "your-password"
# HOST = "raspberrypi.local"
# PORT = 2000
# UDP 受信タイムアウト関連
# UDPSocket#recvfrom はタイムアウトを持たず永久ブロックするため、
# プリミティブの recvfrom_nonblock を POLL_MS 間隔でポーリングして打ち切る。
RECV_TIMEOUT_MS = 200 # この時間サーバ応答が無ければ「今サイクルは応答なし」とみなす
POLL_MS = 10 # recvfrom_nonblock のポーリング間隔
MAX_MISSES = 5 # 連続でこの回数応答が無ければ通信断とみなしセッションを切る
# XIAO ESP32C6 オンボード LED は GPIO15・アクティブ Low(0=点灯 / 1=消灯)
LED_ON = 0
LED_OFF = 1
led = GPIO.new(15, GPIO::OUT) # オンボード黄色 LED (D13)
led.write(LED_OFF)
# ADC1 は GPIO0-6。XIAO のアナログピン A0=GPIO0 / A1=GPIO1 を使用
adc1 = ADC.new(0) # A0 (GPIO0 / D0)
adc2 = ADC.new(1) # A1 (GPIO1 / D1)
# TB6612: AIN1/AIN2/BIN1/BIN2 = direction, PWMA/PWMB = speed
# STBY ピンは 3.3V に接続しておくこと
# NOTE: 以下は USB(GPIO12/13)・LED/ストラップ(GPIO8/9/15)・ADC(GPIO0/1)を避け、
# XIAO のピンに実在する GPIO のみを使った一例。実際の配線に合わせて変更可
# (括弧内は XIAO ESP32C6 のシルク表記)。
ma1 = GPIO.new(2, GPIO::OUT) # AIN1 (D2)
ma2 = GPIO.new(16, GPIO::OUT) # AIN2 (D6)
mb1 = GPIO.new(17, GPIO::OUT) # BIN1 (D7)
mb2 = GPIO.new(21, GPIO::OUT) # BIN2 (D3)
pwma = PWM.new(18, frequency: 1000, duty_u16: 0) # PWMA (D10)
pwmb = PWM.new(19, frequency: 1000, duty_u16: 0) # PWMB (D8)
ma1.write(0); ma2.write(0); mb1.write(0); mb2.write(0)
# スピードは固定
pwma.duty_u16 = 20000
pwmb.duty_u16 = 20000
WiFi.init
def stop_motors(ma1, ma2, mb1, mb2, pwma, pwmb)
pwma.duty_u16 = 0; pwmb.duty_u16 = 0
ma1.write(0); ma2.write(0); mb1.write(0); mb2.write(0)
end
# motor 値は 1=正転, -1=逆転, 0=停止
def set_motor(pin1, pin2, dir)
if dir > 0
pin1.write(1); pin2.write(0)
elsif dir < 0
pin1.write(0); pin2.write(1)
else
pin1.write(0); pin2.write(0)
end
end
# 切断中の点滅待機 (on_ms=150, off_ms=850 → 1秒サイクル)
# LED はアクティブ Low なので LED_ON=0(点灯)/ LED_OFF=1(消灯)
def blink_wait(led, seconds, on_ms: 150, off_ms: 850)
cycle = (on_ms + off_ms) / 1000.0
(seconds / cycle).to_i.times do
led.write(LED_ON)
sleep(on_ms / 1000.0)
led.write(LED_OFF)
sleep(off_ms / 1000.0)
end
end
# 応答データグラムを最大 timeout_ms 待つ。来なければ nil を返す(=タイムアウト)。
# UDPSocket#recvfrom は永久ブロックなので、nil を返す recvfrom_nonblock を自前ポーリングする。
def recv_with_timeout(sock, maxlen, timeout_ms)
(timeout_ms / POLL_MS).times do
result = sock.recvfrom_nonblock(maxlen) # データ無しなら nil
return result[0] if result # result = [data, addr] なので本体だけ返す
sleep_ms POLL_MS
end
nil
end
def run_session(sock, adc1, adc2, ma1, ma2, mb1, mb2)
misses = 0
loop do
sensor1 = adc1.read
sensor2 = adc2.read
# UDP は 1 send = 1 データグラム。行区切り(\n)は不要
sock.send("#{sensor1},#{sensor2}", 0)
# サーバ応答を待つ(タイムアウトあり)。サーバーは "150,180" を 1 データグラムで返す
response = recv_with_timeout(sock, 64, RECV_TIMEOUT_MS)
if response.nil? || response.empty?
# 応答が来ないサイクルは安全のためモーターを止める(惰性停止)。
# 古い指令のまま走り続けるのを防ぐフェイルセーフ。
set_motor(ma1, ma2, 0)
set_motor(mb1, mb2, 0)
misses += 1
puts "No response (#{misses}/#{MAX_MISSES})"
# 連続で応答が無ければ通信断とみなしてセッションを抜ける(→ 再接続)
break if misses >= MAX_MISSES
next
end
misses = 0 # 応答が来たらカウンタをリセット
motor1, motor2 = response.split(",").map { |x| x.to_i }
set_motor(ma1, ma2, motor1)
set_motor(mb1, mb2, motor2)
puts "Sensors: #{sensor1}, #{sensor2} -> Motors: #{motor1}, #{motor2}"
sleep 0.1
end
end
socket = nil
loop do
begin
if WiFi.link_status != WiFi::LINK_UP
puts "Connecting to #{WIFI_SSID}..."
WiFi.connect(WIFI_SSID, WIFI_PASS, WiFi::Auth::WPA2_MIXED_PSK, 30)
puts "WiFi connected: #{WiFi.ipv4_address}"
end
# UDP はコネクションレスだが、connect で送信先を固定すると
# send/recv に毎回アドレスを書かずに済む(TCP の接続確立とは別物)
socket = UDPSocket.new
socket.connect(HOST, PORT)
led.write(LED_ON)
puts "Connected to #{HOST}:#{PORT}"
run_session(socket, adc1, adc2, ma1, ma2, mb1, mb2)
puts "Session ended"
rescue WiFi::ConnectError => e
puts "WiFi auth error: #{e.message}"
blink_wait(led, 10)
rescue WiFi::ConnectTimeout => e
puts "WiFi timeout: #{e.message}"
rescue SocketError => e
puts "Socket error: #{e.message}"
rescue IOError, EOFError => e
puts "IO error: #{e.message}"
rescue StandardError => e
puts "Error: #{e.class} #{e.message}"
ensure
socket.close if socket && !socket.closed?
led.write(LED_OFF)
stop_motors(ma1, ma2, mb1, mb2, pwma, pwmb)
end
puts "Retry in 3s..."
blink_wait(led, 3)
end
以下は実際に動かしてみた時の動画です。RaspberryPi側にUDPのサーバが立っていて、こちらのキー入力に応じた信号をレスポンスとして返すようになっています。
esp32c6向けのUDPSocket通信も上々!mruby/cが入ってます。ライブデモするとボロが出そうだからこれは動画だけお披露目で😅 pic.twitter.com/fk6KtXqEDz
— ITO Yosei (@YoseiIto) July 3, 2026