【micropython】ESP32でwebサーバーを立てる
ESP32をアクセスポイントにして、スマホ等でESP32に接続できるようにします。
インターネット環境やローカルネット環境がない場所でも、スマホ画面でESP32とデータのやり取りができるので便利です。
CでESPAsyncWebServerを使った方法が散見されますが、micropythonを使ってやっている人は少なそうなので、記事にしました。
日本では未だにC/C++が主力なんですかね。
micropythonのほうが簡単に書けるので、C/C++を使うメリットは少ないように思えるのですが。
調べた限り、日本でmicropythonを使ってwebサーバーを立てている記事は少なかったので、今後日本にもmicropythongが主流となることを願って記事にします。
開発環境
エディタはThonnyとVSCodeの両方を用意することを推奨します。
Thonnyはインストールするだけですぐに使えるのでお手軽ですが、コード補完ができなかったり、HTMLエディタがないので、動作チェック程度には使えても本格的に開発する段になって苦労します。
VSCodeの環境設定（PymakrやMicroPicoの用意、stubのインストール）については割愛します。
デメリットとしてパッケージ管理ができないので、パッケージ管理はThonnyを利用します。
今回は簡単な動作チェックだけなので、Thonnyだけでも十分です。
micopythonはv1.22を使っていますが、多少昔のバージョンでも大丈夫だと思います
ESP32をアクセスポイントにする
まずはESP32をアクセスポイントにして、スマホから接続できるか確認しましょう。
といっても、network.WLAN(network.AP_IF)と設定するだけなので特に難しいことはありません。
MY_SSIDとMY_PASSWORDは好きに設定してください
main.py
import network
MY_SSID='MY_ESP32'
MY_PASSWORD='aaaaaaaa'
# アクセスポイントインターフェースの取得
ap_if = network.WLAN(network.AP_IF)
ap_if.active(True)
ap_if.config(essid=MY_SSID, password=MY_PASSWORD)
# アクセスポイントの状態確認
if ap_if.active():
# IPアドレスの取得
ip_address = ap_if.ifconfig()[0]
print('ESP32のIPアドレス:', ip_address)
else:
print('ESP32はアクセスポイントモードで動作していません。')
スマホでWifi検索すると'MY_ESP32'がありますので、接続してみてください。
簡単な画面表示
<h1>Hello</h1>をスマホ画面に表示させるようにしましょう。
ページを返すにはpicowebというライブラリを使います。
（chatGPTなどに聞くと、いろいろ方法を答えてくれますが、大半は誤った情報です。
冒頭で述べた通り、日本ではmicropythonを使う人がまだ少ないので、chatGPTが誤った回答をするのは仕方ないかもしれません）
picowebのインストールはThonnyを使うことを勧めます。
main.py
import uasyncio as asyncio
import picoweb
# ルートパスのハンドラー
async def index(req, resp):
await picoweb.start_response(resp)
await resp.awrite("<h1>Hello</h1>")
# ルーティングの設定
routes = [("/", index)]
# Webアプリケーションの作成
app = picoweb.WebApp(__name__, routes)
# Webサーバーの起動
loop = asyncio.get_event_loop()
loop.create_task(app.run(debug=True, host="0.0.0.0", port=80))
loop.run_forever()
このコードを実行しようとすると、uloggingがありませんというエラーが出ます。
理由はpicoweb.__init__.pyのdef runの部分でuloggingを要求しているからなんですが、だからといってuloggingをインストールしても、後々メモリ不足で行き詰まります。
そもそもログを取る必要はありませんから、picowebのlog関係のところは全て消してやりましょう。
以下のようにself.logの箇所を全てpassで置き換えます（全部で4箇所ありました）
if self.debug >= 0:
pass
# self.log.error("%s: EOF on request start" % reader)
そして、def runのlog定義の場所をコメントアウトします
def run(self, host="127.0.0.1", port=8081, debug=False, lazy_init=False, log=None):
# if log is None and debug >= 0:
# import ulogging
# log = ulogging.getLogger("picoweb")
# if debug > 0:
# log.setLevel(ulogging.DEBUG)
#
# self.log = log
これで問題なく動くはずです。
HTMLを別ファイルに移す
先ほどはHTMLは文字列としてベタ書きしましたが、実際の開発ではHTMLの部分は切り分けたいはずです。
<h1>Hello</h1>
# ルートパスのハンドラー
async def index(req, resp):
await picoweb.start_response(resp)
with open('index.html', 'r') as f:
html_content = f.read()
await resp.awrite(html_content)
# あとは同じ
ESP32側で取得したデータを表示
画面を取得した後、ESP32で測定した気温や湿度などのデータを表示させるようにします。
ESP32からスマホ側にデータを送るには、BLEやMQTT通信、websocketなどいろいろな方法がありますが、今回はwebページを返すためにHTTP通信を使っているので、その延長でHTTP通信を使います。
main.py
import uasyncio as asyncio
import picoweb
# ルートパスのハンドラー
async def index(req, resp):
await picoweb.start_response(resp)
with open('index.html', 'r') as f:
html_content = f.read()
await resp.awrite(html_content)
# 追加。現在の秒を返すハンドラー
async def get_seconds(req, resp):
import utime
current_seconds = utime.localtime()[5]
print(current_seconds)
await picoweb.jsonify(resp, {"seconds": current_seconds})
# ルーティングの設定
routes = [
("/", index),
("/seconds",get_seconds) # 追加
]
# Webアプリケーションの作成
app = picoweb.WebApp(__name__, routes)
# Webサーバーの起動
loop = asyncio.get_event_loop()
loop.create_task(app.run(debug=True, host="0.0.0.0", port=80))
loop.run_forever()
index.html
<!DOCTYPE html>
<html>
<head>
<title>ESP32 HTTP Client</title>
<script>
function getSeconds() {
fetch('http://192.168.4.1/seconds')
.then(response => response.json())
.then(data => {
document.getElementById('seconds').innerText = data.seconds;
});
}
</script>
</head>
<body>
<h1>Hello</h1>
<div>
<pre id="seconds"></pre>
<button onclick="getSeconds()">現在秒を取得</button>
</div>
</body>
</html>
index.htmlをThonnyで書くのが苦しくなってきたかもしれません。
あとでリファクタリングを含めてVSCodeに移植します。
スマホでESP32を操作
POSTリクエストを送ります。
まずはHTML側
index.html
<!DOCTYPE html>
<html>
<head>
<title>ESP32 HTTP Client</title>
<script>
function getSeconds() {
fetch('http://192.168.4.1/seconds')
.then(response => response.json())
.then(data => {
document.getElementById('seconds').innerText = data.seconds;
});
}
function postForm() {
const form = document.querySelector('form');
const formData = new FormData(form);
fetch('http://192.168.4.1/post', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({name: formData.get('name'), password: formData.get('password')})
})
.then(response => response.json())
.then(data => {
console.log(data);
})
.catch(error => {
console.error('Error:', error);
});
}
</script>
<style>
div {
border: 1px solid #000;
margin: 10px;
}
</style>
</head>
<body>
<h1>Hello</h1>
<div>
<pre id="seconds"></pre>
<button onclick="getSeconds()">現在秒を取得</button>
</div>
<div>
<form method="post">
<input type="text" name="name" placeholder="名前">
<input type="text" name="password" placeholder="パスワード">
<button type="button" onclick="postForm()">送信</button>
</form>
</div>
</body>
</html>
見ての通り、JSON形式でデータを送っています。
function postForm() {
const form = document.querySelector('form');
const formData = new FormData(form);
fetch('http://192.168.4.1/post', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({name: formData.get('name'), password: formData.get('password')})
})
.then(response => response.json())
.then(data => {
console.log(data);
})
.catch(error => {
console.error('Error:', error);
});
}
続いて、ESP32側
main.py
import uasyncio as asyncio
import picoweb
import ujson
# ルートパスのハンドラー
async def index(req, resp):
await picoweb.start_response(resp)
with open('index.html', 'r') as f:
html_content = f.read()
await resp.awrite(html_content)
# 現在の秒を返すハンドラー
async def get_seconds(req, resp):
import utime
current_seconds = utime.localtime()[5]
print(current_seconds)
await picoweb.jsonify(resp, {"seconds": current_seconds})
# POSTリクエストを受け取るハンドラー
async def post_form(req, resp):
print("Headers Received:", req.headers)
content_length = int(req.headers[b'Content-Length']) # Content-Lengthヘッダーを取得
body = await req.reader.readexactly(content_length) # 指定した長さだけデータを読み込む
data = ujson.loads(body) # JSONデータを解析
print(data)
# レスポンスを送信
await picoweb.start_response(resp)
await resp.awrite("OK")
# ルーティングの設定
routes = [
("/", index),
("/seconds",get_seconds),
("/post",post_form)
]
# Webアプリケーションの作成
app = picoweb.WebApp(__name__, routes)
# Webサーバーの起動
loop = asyncio.get_event_loop()
loop.create_task(app.run(debug=True, host="0.0.0.0", port=80))
loop.run_forever()
ここの部分がPOSTされたデータを受け取る関数です。
# POSTリクエストを受け取るハンドラー
async def post_form(req, resp):
print("Headers Received:", req.headers)
content_length = int(req.headers[b'Content-Length']) # Content-Lengthヘッダーを取得
body = await req.reader.readexactly(content_length) # 指定した長さだけデータを読み込む
data = ujson.loads(body) # JSONデータを解析
print(data)
データの読み取りはawait req.reader.readexactlyを使っています。
普通はawait req.reader.read()を使いますが、この方法だといつまでもawaitが終わらない（データを完全に受信するのを待っている状態）になってしまいます。
そこでcontent_length = int(req.headers[b'Content-Length'])でデータ長を取得し、指定した長さだけデータを読み込むようにしてあげています。
Discussion