超ド文系の非エンジニアがエアコンをRaspberry Pi(ラズパイ)と生成AIで一番楽して操作してみる①

きっかけ

今年も夏がやってきました。今年こそは自室にエアコンをつけようということで買ったんですが、経費の関係もあってガチで冷やすことしかできないエアコンを買いました。

Hisense ルームエアコン Sシリーズe
https://www.hisense.co.jp/aircon/ha-sf-series-e/

最近の15万円ぐらいのエアコンにはスマホとの連携機能がついており、外からスイッチを入れられたりスマホから操作ができたりするようです。当然この機種にはついていません。

連携機能、ほしい。

思い立ったが吉日ということで、せっせとリサーチを始めました。この記事はその忘備録。

経歴

題名にもあるように、ガチの文系です。中学校から大学まで英語ばっかり勉強し、結果的に今は海外企業で働いています。職種はマネジメント系のものなので、非エンジニア職になります。しかも今回使うのはRaspberry Pi。Linuxなんて使ったことない僕が色々困ることになりそうです。

ただ....

典型的なパソコン音痴ではなく、むしろパソコンは大好きなほうでした。自作erとして自作パソコンを25台ぐらい作ったり(個人事業としてやっていた時期があった)、英語が問題なく使えるのもあって英語で書かれたDocsを読んだり動画を見るのは全く問題ありません。また、AIが出てきてからはAIを使ってDiscordのBotを作って業務効率化を図っていたりします。コードはあんまり書けないですが、仕組みを考えたりするのは好きで、今回も僕が考えた仕組みを元にAIとタッグを組んで作っていきます。

コンセプト

題名にもあるように今回は「一番楽」して仕組みを作ります。ということは多少お金がかかってでも開発コストを下げる(というか僕がたいした開発をできないので)ということです。駆け出しエンジニアですらない人間がホームオートメーションのシステム構築をするのは無謀ですが、逆にAIをうまく使えばド素人でもQOLが上がるものを作れるということを証明できたらと思います。

今回はこのような仕組みで構築を進めます

最初はDiscord botから直接各モジュールにコマンドを送ればいいやと思っていましたが、ここにAPIを挟んだほうが今後色々便利になりそうなのと、APIを作るとFlaskの勉強だけでなくネットワーク系のことも勉強できそうだったのでこの構成にしました。

websiteのところだけやたらとおおざっぱですが、本当はここにデータベースを作って、5分ごとにAPIにリクエストを送ってその結果をデータベースに保存、さらにそのデータを使ってグラフを書いてくれるサイトとか作りたいなと思っています。ただフロントエンドの開発経験が皆無なのと、Javascriptわからんすぎるのでもう少しリサーチがいるところです。簡単にグラフを作ってくれるツールとかあったらいいのにな...
結局AIがあってもこっちがちんぷんかんぷんだと役立たずなんですよね

使うもの

1. GPT-4o

実はかなりAIマニアで、最近出てるAIは一通り全部契約して使ってみたりしているんですが、今回はGeminiかChatGPTかで悩んでいました。最近話題になったClaude 3.5 Sonnetは、いろんなベンチマーク結果を見る限りデザイン性は高いがロジックとかを考えるのはChatGPTのほうがよいという結果でした。またこのプロジェクトではマイナーなライブラリも使うので、URLから情報をとってきてくれるChatGPTのほうが色々と都合がよさそうです。ウェブサイト制作のときはClaudeのお世話になるかもしれませんが、とりあえずAPIとBotはChatGPTで書いていきます。

2.Raspberry Pi 4B

これが無かったらはじまらないということで、Raspberry Pi 4B一式買いました。今まで作ってきたデスクトップパソコンとは全然違ってて、これはこれでオンボードPCも面白いなと思いました。ただ、M.2 SSD大好きお兄さんなのでMicroSDの読み書きの速度にはちょっとイライラしちゃいました。次買うならeMMCが載ってるものにしようかな。ちなみに最新版のRaspberry pi 5は絶対に必要なpigpioっていうライブラリが動かないらしいので、買うなら5以外にしてくださいね。正直こんなにいろんなことをさせなければNanoとかでも十分な性能ですが、僕はいろんなBotを動かしたいのと安定した動作をしてほしいので4Bにしました。

Indoor corgi RPiTPH Monitor Rev2

Indoor corgiさんから出ているRPiTPH Monitor Rev2という拡張モジュールです。温度、湿度、気圧、赤外線の送受信が行えます。あとから調べて知りましたが、RPZ-IR-Sensorという製品でも同じことができます。こちらは液晶モジュールが使えない代わりに、照度センサーがついているようです。どちらの製品も9500円ほどでした。

これらの製品にはBME280という温度計モジュールが同梱しています。エアコンの操作ではなく温度を測りたいだけの場合はこのモジュールを買って指すだけで使えます。後述するPythonのコードも使えます。

ラズパイ4Bはなんだかんだ合わせて1万円ちょっと、この拡張モジュールは9500円なので2万円上乗せでシステムを構築できそうです。今回購入したエアコンの本体価格が5万円ちょっとで(工事費は別途)、ハイエンドのエアコンが15万円ぐらいだったのを考えるととっても安上がりになりそうです。成功すれば...

--------------------------------------------------------------------

実はQiitaにはこの手の記事はいくつかあり、一番有名なものは

格安スマートリモコンの作り方
https://qiita.com/takjg/items/e6b8af53421be54b62c9

これでしょう。もちろん拝見させていただきました。
というかこの記事をみてRaspberry Piで制御しようと思い立ちました。ありがとうございます。

ただ、こちらは僕にとっては難問だらけでした

• 電子工作わからない
• ラズパイのGPIOの制御わからない(結局勉強した)
• 電子工作難しそう
• へんなとこにはんだ付けして壊したらどうしよう

ということですでにライブラリが開発されていて、基盤も完成しているRPiTPH Monitor Rev2を買いました。この記事の著者の方が言うように材料費だけなら2000円程度なんでしょうが、ライブラリ開発費、はんだ付けの手間賃などを込み込みで考えるともうちょっと高くてもいいんじゃないかと思いました。

ラズパイを準備する

長すぎる前置きでしたが、さっそく作っていきます。Raspberry Piは公式Documentation を読んでさくっとセットアップを終わらせます。新品のパソコンにWindowsを入れるのと一緒で、今回はUSBではなくMicroSDに、WindowsではなくRaspberry pi OSをいれます。Windows用のツール(https://www.raspberrypi.com/software/) があるので誰でも簡単にできます。このOSをMicroSDに入れる際にWifiの設定やキーボードの設定を予め変えられるので、やっておいたほうが後から楽かも。

この段階でGitやPythonなども入れておきます。

ターミナルを開いて$ sudo apt install git で入れることができます。

また、今回はPythonのウェブアプリ用フレームワーク Flaskを使ってREST APIを作ります。Pythonは重たいしレイテンシが大きい！っていうのは聞くんですが、まあ趣味なのでよしとしましょう。(ちなみにあとで地獄を見ます)

とりあえず温度計と赤外線センサーを使ってみる

こんな感じで接続できたので起動して使っていきます。

このモジュールの欠点としてCPUの上に設置されているので、ケースに入れた場合著しくエアフローが悪くなります。ヒートシンクありでアイドル時でも45度ぐらい、ブラウザを開いたりビルドしたりすると65度ぐらい(室温26度)です。もしケースに入れて運用することを考えている場合は一度温度を測ってみてください。少なくともヒートシンクは絶対につけたほうがいいです。

専用ライブラリcgsensorとcgirを使う準備をする

まずはこのサイトを参考にI2Cを有効にしましょう。これを有効にすると温度計と通信できるようになります。
https://www.indoorcorgielec.com/products/rpi-tph-monitor-rev2/

このサイトの環境だと設定画面から有効にしたらすぐに使えているっぽいですが、僕の環境だと設定でI2Cを有効にした後再起動しないと反映されませんでした。参考までに。

また、Pigpioの設定も済ませておきます。

sudo systemctl start pigpiod
sudo systemctl enable pigpiod

この2行をコピペでOK。Pigpioを開始して、システム開始時に自動的に起動してくれるようになります。

ライブラリをインストール・動作確認

温度計のライブラリは $ sudo python3 -m pip install -U cgsensor
赤外線モジュールのライブラリは $ sudo python3 -m pip install -U cgir

これだけ。簡単すぎる...

ちなみにこれらのライブラリはGithubに上がっています。中身を勉強できるので非常に役に立ちました。

インストールしたあとターミナルを開いて、cgsensor all と入力してみると

BME280
    気温[℃]: 25.5
    湿度[%]: 72.2
    気圧[hPa]: 1010.2

一瞬で結果がでました。優秀。一から自作していたらここまで来るのに何カ月もかかりそうです。本当に買ってよかったと思いました。温度計モジュールの動作チェックは完了です。

赤外線モジュールを使ってみる

そもそも赤外線モジュールとは何かというと、リモコンから照射された赤外線のパターンを記録して、それを同じものを照射するというものです。ラズパイはエアコンのリモコンではないですが、エアコンのリモコンから出る赤外線と同じものを真似することでリモコンかのように認識させることができます。ただ最初は真似するものがないので、cgir recを使って記録していきます。

$ cgir rec ac_onを実行すると、

赤外線コード "ac_on"を受信中...受信機に向けて赤外線を送信してください

という表示がでますので、リモコンを基盤に向けてボタンを押します。

受信コード
[9000, 4400, 640...中略]

Format NEC
Frame#1 0x50, 0xAF, 0x17, 0xE8....

赤外線コード "ac_on" を登録しました.

これで完了です。

このデータをテストするには、基盤をエアコンに向けて$ cgir send ac_onを実行します。
これでエアコンがついたらテスト完了です。

あとは同じ要領で他のボタンも登録していきます。

エアコンの温度設定についてですが、多くのエアコンの場合温度設定は本体ではなくリモコン側に記憶されています。

従って温度を下げるボタンを押した際、エアコンには「温度を1度下げてね」という信号が送信されるのではなく、「(今28度だった場合)27度にしてね」という信号になります。これは cgir recした際に見れる受信コードでも確認できます。28度→27度のときと、27度→26度のときはコードが違います。

これを注意したうえで赤外線を記録してください。温度を下げる赤外線を1通りしか記録していないと、前述のコードなら何回送信しても27度にしかなりません。

このモジュールを使うと「エアコンを自動操作して部屋を特定の温度にキープする」みたいなことができますが、赤外線の種類が少ないとロジックは良くても現実ではうまく作動しないでしょう。

APIをつくる

実際にAPIを書いていきます。まずは仕様書を作る。

1. Flaskで書く
2. 温度・湿度をAPI経由で教えてほしい
3. API経由でエアコンのON / OFF、温度調整がしたい

意外とそんなに求められることはなさそう。

温度計モジュールについてはすでにとってもいいサンプルコードがありました。

import cgsensor

bme280 = cgsensor.BME280(i2c_addr=0x76)  

bme280.forced()  # Forcedモードで測定を行い, 結果をtemperature, pressure, humidityに入れる
print('気温 {}°C'.format(bme280.temperature))  # 気温を取得して表示
print('湿度 {}%'.format(bme280.humidity))  # 湿度を取得して表示
print('気圧 {}hPa'.format(bme280.pressure))  # 気圧を取得して表示

https://github.com/IndoorCorgi/cgsensor/blob/master/examples/bme280_basic.py

cgsensorのおかげでとんでもなくシンプルなコードで温度と湿度を取得できそうです。
早速これを少しお借りしてFlaskのコードにします。

プロンプト:

This is a python code that interacts with the raspberry pi external module to measure temperature and humidity. I want to create a REST API using Flask that retrieves temperature and humidity using this module. 

import cgsensor 

bme280 = cgsensor.BME280(i2c_addr=0x76)

bme280.forced()  # Forcedモードで測定を行い, 結果をtemperature, pressure, humidityに入れる
print('気温 {}°C'.format(bme280.temperature))  # 気温を取得して表示
print('湿度 {}%'.format(bme280.humidity))  # 湿度を取得して表示
print('気圧 {}hPa'.format(bme280.pressure))  # 気圧を取得して表示

結果は...

from flask import Flask, jsonify
import cgsensor

app = Flask(__name__)

# Initialize the BME280 sensor
bme280 = cgsensor.BME280(i2c_addr=0x76)

@app.route('/temperature', methods=['GET'])
def get_temperature():
    bme280.forced()
    temperature = bme280.temperature
    return jsonify({'temperature': temperature})

@app.route('/humidity', methods=['GET'])
def get_humidity():
    bme280.forced()
    humidity = bme280.humidity
    return jsonify({'humidity': humidity})

@app.route('/pressure', methods=['GET'])
def get_pressure():
    bme280.forced()
    pressure = bme280.pressure
    return jsonify({'pressure': pressure})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

完璧

bme280 = cgsensor.BME280(i2c_addr=0x76)
の部分ですが、BME280のみを購入した場合アドレスが変わりますので0x76ではなくなります。自分のモジュールがどのI2Cに接続されているかは i2cdetect のコマンドで確認することができます。

次に赤外線モジュールのほうも実装していきます。
cgirのコードも拝見させていただいたんですが、うまく動いてくれなかったのでsubprocessをつかってコマンドを実行しちゃうことにしちゃいました。ちょっとズルした気分ですが動くのでヨシ！

プロンプト:

I want to add a feature that executes this commands using subprocess.
cgir send cooler_on

結果は...

from flask import Flask, jsonify
import subprocess
import cgsensor

app = Flask(__name__)

# Initialize the BME280 sensor
bme280 = cgsensor.BME280(i2c_addr=0x76)

@app.route('/temperature', methods=['GET'])
def get_temperature():
    bme280.forced()
    temperature = bme280.temperature
    return jsonify({'temperature': temperature})

@app.route('/humidity', methods=['GET'])
def get_humidity():
    bme280.forced()
    humidity = bme280.humidity
    return jsonify({'humidity': humidity})

@app.route('/pressure', methods=['GET'])
def get_pressure():
    bme280.forced()
    pressure = bme280.pressure
    return jsonify({'pressure': pressure})

@app.route('/cooleron', methods=['GET'])
def cooler_on():
    try:
        result = subprocess.run(['cgir', 'send', 'cooler_on'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Cooler turned on', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to turn on cooler', 'output': e.stderr.decode()}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

よさそう。これを元に他のcgirコマンドも書き足します。

今回は

• 冷房ON
• 暖房ON
• スイッチオフ
• 温度を上げる(27度→28度)
• 温度を下げる(27度→26度)
  この5つを登録していきます。

from flask import Flask, jsonify
import subprocess
import cgsensor

app = Flask(__name__)

# Initialize the BME280 sensor
bme280 = cgsensor.BME280(i2c_addr=0x76)

@app.route('/temperature', methods=['GET'])
def get_temperature():
    bme280.forced()
    temperature = bme280.temperature
    return jsonify({'temperature': temperature})

@app.route('/humidity', methods=['GET'])
def get_humidity():
    bme280.forced()
    humidity = bme280.humidity
    return jsonify({'humidity': humidity})

@app.route('/pressure', methods=['GET'])
def get_pressure():
    bme280.forced()
    pressure = bme280.pressure
    return jsonify({'pressure': pressure})

@app.route('/cooleron', methods=['GET'])
def cooler_on():
    try:
        result = subprocess.run(['cgir', 'send', 'cooler_on'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Cooler turned on', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to turn on cooler', 'output': e.stderr.decode()}), 500

@app.route('/heateron', methods=['GET'])
def heater_on():
    try:
        result = subprocess.run(['cgir', 'send', 'heater_on'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Heater turned on', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to turn on heater', 'output': e.stderr.decode()}), 500

@app.route('/tempup', methods=['GET'])
def temp_up():
    try:
        result = subprocess.run(['cgir', 'send', 'temp_up'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Temperature up', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to increase temperature', 'output': e.stderr.decode()}), 500

@app.route('/tempdown', methods=['GET'])
def temp_down():
    try:
        result = subprocess.run(['cgir', 'send', 'temp_down'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Temperature down', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to decrease temperature', 'output': e.stderr.decode()}), 500

@app.route('/turnoff', methods=['GET'])
def turn_off():
    try:
        result = subprocess.run(['cgir', 'send', 'turn_off'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Conditioner turned off', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to turn off conditioner', 'output': e.stderr.decode()}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

よさそう。特にエラーもなさそうです
ここまで15分。Flaskがわかりやすいのもありますが、これ15分でできちゃうのすごい

ここでちょっと思い出したんですが、以前Tebexというゲームコミュニティ向け決済プラットフォームのAPIを使ったときに「Headerにパスワードを入れた状態でリクエストを送信してね」っていう制限がついてました。これかなり簡単で分かりやすい仕組みだったのでパクらせていただこうと思います。

パスワードで認証する仕組みを作って、認証失敗するとステータスコード401が出るようにします。パスワードをハードコードすると良くないというのを口酸っぱく言われたので.envに入れておきましょう。

プロンプト:

I want to add a header that contains password to add a security feature. Whenever users make a request, they need to contain password in the header. otherwise api should not respond. Additionally, I want to store the password in .env file.

結果は...

from flask import Flask, jsonify, request
import subprocess
import cgsensor
from dotenv import load_dotenv
import os

# Load environment variables from .env file
load_dotenv()

app = Flask(__name__)

# Initialize the BME280 sensor
bme280 = cgsensor.BME280(i2c_addr=0x76)

# Get the required password from the environment variables
REQUIRED_PASSWORD = os.getenv('PASSWORD')

# Helper function to check password
def check_password():
    password = request.headers.get("Password")
    if not password or password != REQUIRED_PASSWORD:
        return False
    return True

@app.route('/temperature', methods=['GET'])
def get_temperature():
    if not check_password():
        return jsonify({"error": "Unauthorized"}), 401
    bme280.forced()
    temperature = bme280.temperature
    return jsonify({'temperature': temperature})

@app.route('/humidity', methods=['GET'])
def get_humidity():
    if not check_password():
        return jsonify({"error": "Unauthorized"}), 401
    bme280.forced()
    humidity = bme280.humidity
    return jsonify({'humidity': humidity})

@app.route('/pressure', methods=['GET'])
def get_pressure():
    if not check_password():
        return jsonify({"error": "Unauthorized"}), 401
    bme280.forced()
    pressure = bme280.pressure
    return jsonify({'pressure': pressure})

@app.route('/cooleron', methods=['GET'])
def cooler_on():
    if not check_password():
        return jsonify({"error": "Unauthorized"}), 401
    try:
        result = subprocess.run(['cgir', 'send', 'cooler_on'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Cooler turned on', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to turn on cooler', 'output': e.stderr.decode()}), 500

@app.route('/heateron', methods=['GET'])
def heater_on():
    if not check_password():
        return jsonify({"error": "Unauthorized"}), 401
    try:
        result = subprocess.run(['cgir', 'send', 'heater_on'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Heater turned on', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to turn on heater', 'output': e.stderr.decode()}), 500

@app.route('/tempup', methods=['GET'])
def temp_up():
    if not check_password():
        return jsonify({"error": "Unauthorized"}), 401
    try:
        result = subprocess.run(['cgir', 'send', 'temp_up'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Temperature up', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to increase temperature', 'output': e.stderr.decode()}), 500

@app.route('/tempdown', methods=['GET'])
def temp_down():
    if not check_password():
        return jsonify({"error": "Unauthorized"}), 401
    try:
        result = subprocess.run(['cgir', 'send', 'temp_down'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Temperature down', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to decrease temperature', 'output': e.stderr.decode()}), 500

@app.route('/turnoff', methods=['GET'])
def turn_off():
    if not check_password():
        return jsonify({"error": "Unauthorized"}), 401
    try:
        result = subprocess.run(['cgir', 'send', 'turn_off'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return jsonify({'message': 'Conditioner turned off', 'output': result.stdout.decode()}), 200
    except subprocess.CalledProcessError as e:
        return jsonify({'error': 'Failed to turn off conditioner', 'output': e.stderr.decode()}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

.envファイル

PASSWORD=notarealpassword

エラーなし ということでメインPCのPostmanでいくつかリクエストを出してみます。

ちなみにapp.run(host='0.0.0.0', port=5000)で指定したポートを開けていないとこのAPIは作動しません。お使いのルーターの説明書を読んでポート開放をしてくださいね。

APIをテストしてみる

まずはHeaderにパスワードを入れずにリクエストを送ってみます。

ちゃんとUnauthorizedになりました。
Headerにパスワードを入れてみましょう。

ちゃんと動いてる！
湿度やリモコンの操作もばっちりできます。

気になったので負荷テストをしてみることにしました。
やり方はここに記載されていたものを参考にしました。

https://speakerdeck.com/yokawasa/api-performance-testing-with-postman?slide=15

うーんさんざんな結果。同じネットワーク内といえど無線Wifiで接続しているのと、Raspberry Piの性能も限られているのでこんなものでしょうか。10とか20ms以内に帰ってくるAPIも多いですが、あれは職人技ですね...今回は見なかったことにしておきます。

今回作ったAPIはGithubからダウンロードできます。
git clone https://github.com/inoue-773/RemoteControlAC.git でもOKです

API編おわり

ということで一通りAPIが動くようになりました。おそらく本業の方からみたら突っ込みどころ満載のAPIですが、僕しか使わないのでよしとします。次回はDiscordと連携していきます。といってもAPIができてしまえばあとはPythonのrequestsライブラリを使ってリクエスト出すだけなので、そんなに難しくなさそう。改善点等あればぜひ教えてください！