Spresenseによるガス検知ウサギ型ロボット

1900年代初頭、炭鉱の労働者たちの命を守ったのはウサギやカナリヤといった小動物でした。
炭鉱は一酸化炭素やメタンなどのガスが発生しやすく、これらのガスは人体に大きな影響を及ぼします。
ウサギやカナリヤは有害ガスの影響を受けやすく、人間に空気の異常を伝える役目を担っていました。
時は流れ1980年ごろ、電子ガス検知器の技術が進歩しました。
この時期を境目に、ガス検知精度や動物愛護の観点から、小動物を用いるガス検知は無くなりました。
さらに時は流れ2025年。
高精度なガスセンサと可愛いウサギのビジュアルを持つガス検知器があればお得じゃん、という閃きが舞い降りてきました。
本作品は、見た目で癒される可愛いウサギ型の高精度ガス検知ロボット「ANZEN-RABOT」です。
このデバイスによって、ガスによる危険作業時でも安全と癒しを提供できます。

有害ガスの危険性がある職場にANZEN-RABOTを放ち、空気の監視を行う。
ガスを検知した場合は周辺の人に警告を行う(ロボットのスタンピングおよびPC画面への警告)。
ロボットのスタンピングは、サーボモータによって後ろ足を浮かせることで実現する。
PC画面への警告は、ガス情報をBLEでPCに送信することで実現する。
ガスを検知していない場合は周囲の人に癒しを届ける。

※本来はガス検知時にウサギが転倒する仕様としていたが、タミヤのメカ・ラビットの安定性が高すぎて転倒させられなかった。
そのため、ガス検知時はウサギが威嚇時に行うスタンピング(後ろ足で地面を打ち付ける行為)する仕様とする。

以下の通りです。
構成として、メインボードにガスセンサおよびBLE通信モジュールを接続します。
また、拡張ボードにサーボモータを接続します。
サーボモータはしっぽの角度を調節でき、しっぽの角度を変えることでスタンピングを実現します。

名称	詳細	備考
マイコン	Spresense メインボード	モニター提供品
拡張IO	Spresense 拡張ボード	モニター提供品
ガスセンサモジュール	SPRESENSE用ガスセンサ BME680基板	モニター提供品
BLEモジュール	BLE1507(BLE serialization firmware)	モニター提供品
サーボモータ	SG-90	作業机の上に放置されていた
うさぎロボ	タミヤメカ・ラビット(両足キック歩行タイプ)	ヨドバシで買った

処理はシンプルに、ガス濃度の計測、BLE通信による送信、ガス濃度がしきい値を超えた場合はサーボモータの角度を変更するという内容になります。


デバイス側の処理
#include <Wire.h>
#include <Adafruit_Sensor.h>
#include <Adafruit_BME680.h>
#include "BLE1507.h"
#include <Servo.h>

// BME680のI2Cアドレス
#define BME680_I2C_ADDRESS 0x77

// BLEのUUID
#define UUID_SERVICE  0x3802
#define UUID_CHAR     0x4a02

// BLE関連
static BT_ADDR addr = {{0x19, 0x84, 0x06, 0x14, 0xAB, 0xCD}};
static char ble_name[BT_NAME_LEN] = "SPR-PERIPHERAL";
BLE1507 *ble1507;

// BME680センサーインスタンス
Adafruit_BME680 bme;

Servo myServo; // サーボモータのインスタンス
const int servoPin = 5; // サーボモータ制御用のピン (Spresense拡張ボードのピン)

void setup() {
  // シリアルモニタの初期化
  Serial.begin(115200);
  while (!Serial) delay(10);

  // BME680センサーの初期化
  Serial.println("Initializing BME680...");
  if (!bme.begin(BME680_I2C_ADDRESS)) {
    Serial.println("BME680 not found! Check wiring.");
    while (1);
  }
  Serial.println("BME680 initialized.");
  
  bme.setTemperatureOversampling(BME680_OS_8X);
  bme.setHumidityOversampling(BME680_OS_2X);
  bme.setPressureOversampling(BME680_OS_4X);
  bme.setIIRFilterSize(BME680_FILTER_SIZE_3);
  bme.setGasHeater(320, 150);

  // BLE1507の初期化
  Serial.println("Initializing BLE...");
  ble1507 = BLE1507::getInstance();
  if (ble1507 == nullptr) {
    Serial.println("Failed to get BLE instance.");
    while (1);
  }

  if (!ble1507->beginPeripheral(ble_name, addr, UUID_SERVICE, UUID_CHAR)) {
    Serial.println("Failed to initialize BLE peripheral.");
    while (1);
  }

  if (!ble1507->startAdvertise()) {
    Serial.println("Failed to start BLE advertising.");
    while (1);
  }

  Serial.println("BLE initialized and advertising started.");

  myServo.attach(servoPin, 500, 2400);
  // 初期角度設定
  myServo.write(5);
  Serial.println("Servo motor control started");
}

void loop() {
  // BME680センサーデータを読み取る
  if (!bme.performReading()) {
    Serial.println("Failed to read sensor data.");
    return;
  }

  // センサーデータをBLEで送信
  char bleData[20];
  snprintf(bleData, sizeof(bleData),
           "Gas:%.2fkOhm",
           bme.gas_resistance / 1000.0);

  ble1507->writeNotify((uint8_t*)bleData, strlen(bleData));

  if( (bme.gas_resistance / 1000.0) <= 90.0 ){
    myServo.write(175);
  }

  // デバッグ出力
  Serial.println(bleData);

  // 1秒間待機
  delay(1000);
}

こちらもシンプルに、BLE通信によるガス濃度の受信、受信データの表示、しきい値を超えていたら警告を表示となります。


PC側の処理
import asyncio
import threading
import tkinter as tk
from bleak import BleakClient, BleakScanner
from bleak.exc import BleakError

DEVICE_NAME = "SPR-PERIPHERAL"  # SPRESENSEのBLEデバイス名
CHAR_UUID = "00004a02-0000-1000-8000-00805f9b34fb"  # キャラクタリスティックUUID

class GasSensorGUI:
    def __init__(self, master):
        self.master = master
        self.master.title("Gas Sensor GUI")

        # ガスセンサの値を表示するラベル
        self.label_value = tk.Label(self.master, text="ガスセンサ値: ---", font=("", 16))
        self.label_value.pack(pady=10)

        # 90以下のときに表示するラベル
        self.label_warning = tk.Label(self.master, text="", font=("", 16), fg="red")
        self.label_warning.pack(pady=10)

        # 接続開始ボタン
        self.button_connect = tk.Button(self.master, text="BLE接続開始", command=self.start_ble_thread, font=("", 14))
        self.button_connect.pack(pady=10)

        # 終了ボタン（不要であれば省略可）
        self.button_quit = tk.Button(self.master, text="終了", command=self.master.destroy, font=("", 14))
        self.button_quit.pack(pady=10)

        # BLEスレッド用のフラグ
        self.ble_thread_running = False
        self.ble_thread = None

    def start_ble_thread(self):
        """BLE処理を別スレッドで開始する"""
        if not self.ble_thread_running:
            self.ble_thread_running = True
            self.ble_thread = threading.Thread(target=self.run_ble_loop, daemon=True)
            self.ble_thread.start()
            self.button_connect.config(state=tk.DISABLED)  # 二重起動防止のためボタンを無効化

    def run_ble_loop(self):
        """asyncioのイベントループを実行する"""
        asyncio.run(self.ble_main())

    async def ble_main(self):
        """BLEデバイスをスキャン＆接続し、通知を受信するメイン処理"""
        print("Scanning for BLE devices...")
        devices = await BleakScanner.discover()
        spresence_device = None
        for device in devices:
            print(f"Found device: {device.name} ({device.address})")
            if device.name == DEVICE_NAME:
                spresence_device = device
                break

        if spresence_device is None:
            print(f"Device with name {DEVICE_NAME} not found. Ensure SPRESENCE is advertising.")
            return

        print(f"Attempting to connect to {spresence_device.name} ({spresence_device.address})...")
        try:
            async with BleakClient(spresence_device) as client:
                if not client.is_connected:
                    print("Failed to connect to device")
                    return
                print("Connected to device")

                # キャラクタリスティック通知を有効化
                await client.start_notify(CHAR_UUID, self.notification_handler)
                print("Started receiving notifications...")

                # 通知受信を待機（適宜変更）
                while self.ble_thread_running:
                    await asyncio.sleep(0.1)

                # 通知を無効化
                await client.stop_notify(CHAR_UUID)
                print("Stopped receiving notifications")
        except BleakError as e:
            print(f"BLE Error: {e}")

    def notification_handler(self, sender, data):
        """BLE通知を受け取った際のコールバック"""
        text = data.decode("utf-8")  # 例: "Gas:12.34kOhm"
        # 値を取り出す(単純な文字列処理例)
        # "Gas:" と "kOhm" を取り除き数値に変換
        value_str = text.replace("Gas:", "").replace("kOhm", "")
        try:
            gas_value = float(value_str)
        except ValueError:
            return  # パースエラー時は無視

        # メインスレッド上のGUIを更新するため、afterを使う
        self.master.after(0, self.update_sensor_value, gas_value)

    def update_sensor_value(self, gas_value):
        """ガスセンサの値を画面に反映"""
        self.label_value.config(text=f"ガスセンサ値: {gas_value:.2f}")

        if gas_value <= 90:
            self.label_warning.config(text="ガスに注意")
        else:
            self.label_warning.config(text="")

    def on_closing(self):
        """ウィンドウが閉じられるときの処理"""
        self.ble_thread_running = False
        self.master.destroy()


def main():
    root = tk.Tk()
    app = GasSensorGUI(root)
    root.protocol("WM_DELETE_WINDOW", app.on_closing)
    root.mainloop()

if __name__ == "__main__":
    main()