PythonのSimpleHTTPRequestHandlerを使用したESP32のソフトウェアアップデート

1. はじめに

私が行った研究の実験環境で用いたSimpleHTTPRequestHandlerを使用したESP32の更新の仕方を簡単にまとめたものです.

SimpleHTTPRequestHandlerとは

Pythonに付属している標準ライブラリの一つで, 簡単なHTTPサーバーを立てることができます. これを使用すると, ディレクトリからファイルを提供するためのWebサーバーを簡単に設定できます. これは, 静的ファイルの提供や, 複雑な設定を必要としない開発目的での利用には便利です.

2. 構成

サーバーを開設した後, ESP32からリクエストを送信します. サーバーはクライアント(ESP32)からのリクエストを正常に受け取り, 処理し, HTTPステータスコード200を含むレスポンスを返します. これにより, ESP32はリクエストが成功したことを確認できます.

サーバー構成：

● SimpleHTTPRequestHandlerを使用
● 言語：Python ver.3.12.0
● Ubuntu 24.04 LTS

クライアント構成：

● マイクロコントローラ：ESP32
● 言語：MicroPython
● ファームウェア：ESP32_GENERIC-20240222-v1.22.2

3. サーバープログラム

設定箇所
・SERVER_IP：扱うWi-Fiのアドレスを記入
・update_files：クライアントに送るファイルを指定

from http.server import SimpleHTTPRequestHandler, HTTPServer
import threading
import time

SERVER_IP = ''
SERVER_PORT = 8000
update_files = ['update_1.py', 'update_2.py', 'update_3.py']  # 更新ファイルのリスト
device_updates = {}
lock = threading.Lock()

class OTAHandler(SimpleHTTPRequestHandler):
    def do_GET(self):
        global device_updates

        if self.path == '/connect':
            client_ip = self.client_address[0]
            with lock:
                if client_ip not in device_updates:
                    device_updates[client_ip] = 0  # 新しいデバイスのために更新インデックスを初期化
            self.send_response(200)
            self.end_headers()
            print(f'デバイスが接続しました:')
            #print(f'デバイスが接続しました: {client_ip}')

        elif self.path.startswith('/update'):
            client_ip = self.client_address[0]
            with lock:
                update_index = device_updates.get(client_ip, 0)
                if update_index < len(update_files):
                    file_path = update_files[update_index]
                    self.send_response(200)
                    self.send_header('Content-type', 'application/octet-stream')
                    self.end_headers()
                    with open(file_path, 'rb') as file:
                        self.wfile.write(file.read())
                    print(f"更新ファイルを送信しました: {file_path} to {client_ip}")
                else:
                    self.send_response(404)
                    self.end_headers()

        elif self.path == '/complete':
            client_ip = self.client_address[0]
            self.send_response(200)
            self.end_headers()
            with lock:
                if client_ip in device_updates:
                    device_updates[client_ip] += 1  # 更新インデックスを増加
                    print(f'デバイスが更新を完了しました: 更新インデックス: {device_updates[client_ip]}')
                    if device_updates[client_ip] >= len(update_files):
                        print(f'{client_ip} のすべての更新が完了しました。')

def run_server():
    server_address = (SERVER_IP, SERVER_PORT)
    httpd = HTTPServer(server_address, OTAHandler)
    print('OTAサーバーを起動します...')
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    httpd.server_close()
    print('OTAサーバーを停止します。')

if __name__ == '__main__':
    run_server()

4. クライアント(ESP32)プログラム

設定箇所
・SSID_NAME：Wi-Fiの名前
・SSID_PASS：Wi-Fiのパスワード
・SERVER_IP：Wi-FiのIPアドレス(VM環境であればホストネーム)

・update_files：更新対象のプログラム
(ここはサーバーとファイル数を合わせなければならないので注意)

～補足～
下記の部分はESP32内臓LEDをwi-fi接続時と更新が終わった際に点灯・点滅を行い処理を可視化するために使用している
#ピン設定
led_blue = machine.Pin(2, Pin.OUT)

import network
import urequests
import machine
import time
import utime
from machine import Pin

SSID_NAME = ""
SSID_PASS = ""

# サーバーのIPアドレスとポート番号
SERVER_IP = ''
SERVER_PORT = 8000

# 更新ファイルのリスト
update_files = ['test_1.py','test_2.py','test_3.py']
update_index = 0

# ピン設定
led_blue = machine.Pin(2, Pin.OUT)

# WiFiに接続する関数
def connect_wifi(ssid, passkey, timeout=10):
    wifi = network.WLAN(network.STA_IF)
    if wifi.isconnected():
        print('WiFiに接続済み。接続をスキップします。')
        return wifi
    else:
        wifi.active(True)
        wifi.connect(ssid, passkey)
        while not wifi.isconnected() and timeout > 0:
            print('.')
            utime.sleep(1)
            timeout -= 1

    if wifi.isconnected():
        print('WiFiに接続しました')
        led_blue.value(1)
        return wifi
    else:
        print('WiFiへの接続に失敗しました')
        return None

# サーバーに接続を通知する関数
def notify_server(path):
    url = f'http://{SERVER_IP}:{path}'
    response = urequests.get(url)
    if response.status_code == 200:
        print(f'サーバーに通知しました: {path}')
    response.close()

# 更新プログラムをダウンロードする関数
def download_update(file_index):
    time.sleep(3)
    url = f'http://{SERVER_IP}:{SERVER_PORT}/update'
    response = urequests.get(url)
    if response.status_code == 200:
        file_path = update_files[file_index]
        print(f"{file_path}のプログラムを受け取りました。更新を開始します")
        #while True:
            #if not button_pin_request.value():
        with open(file_path, 'w') as file:
            file.write(response.text)
        print(f'更新プログラムをダウンロードしました: {file_path}')
        response.close()
    else:
        print(f'更新プログラムのダウンロードに失敗しました: ステータスコード {response.status_code}')
        response.close()

# LEDを点滅させる関数
def blink_led(duration, interval=0.5):
    end_time = time.time() + duration
    while time.time() < end_time:
        led_blue.value(1)
        time.sleep(interval)
        led_blue.value(0)
        time.sleep(interval)

# メインプログラムの実行部分
print('WiFiに接続します...')
connect_wifi(SSID_NAME, SSID_PASS)


while True:
    notify_server(f'{SERVER_PORT}/connect')
    print("サーバーに接続しました")
    while True:
        download_update(update_index)
        update_index += 1
        notify_server(f'{SERVER_PORT}/complete')
        print("-----------------------------")
        if update_index >= len(update_files):
            print("全ての更新終了")
            blink_led(10)
            break# 10秒間LEDを点滅させる
    break

5.実行結果

一部の載せられない部分を除き結果を表示します.
順序としては, サーバーの処理を実行し「OTAサーバーを起動します...」が表示されるのを確認します.
次にESP32の処理を実行し, Wi-Fiに接続した後, サーバーと接続し更新プロセスを開始します

クライアント

WiFiに接続します...
.
.
.
.
WiFiに接続しました
サーバーに通知しました: 8000/connect
サーバーに接続しました
test_1.pyのプログラムを受け取りました。更新を開始します
更新プログラムをダウンロードしました: test_1.py
サーバーに通知しました: 8000/complete
-----------------------------
test_2.pyのプログラムを受け取りました。更新を開始します
更新プログラムをダウンロードしました: test_2.py
サーバーに通知しました: 8000/complete
-----------------------------
test_3.pyのプログラムを受け取りました。更新を開始します
更新プログラムをダウンロードしました: test_3.py
サーバーに通知しました: 8000/complete
-----------------------------
全ての更新終了

サーバー

[SERVER_IP-ESP32] はESP32に与えられているIPアドレスが表示されます
更新インデックスは現在更新しているupdate_files内のファイルの番号を示しています.

OTAサーバーを起動します...
[SERVER_IP-ESP32] - - [11/Jul/2024 14:46:00] "GET /connect HTTP/1.0" 200 -
デバイスが接続しました:
[SERVER_IP-ESP32]  - - [11/Jul/2024 14:46:03] "GET /update HTTP/1.0" 200 -
更新ファイルを送信しました: update_1.py to [SERVER_IP-ESP32]
[SERVER_IP-ESP32]  - - [11/Jul/2024 14:46:04] "GET /complete HTTP/1.0" 200 -
デバイスが更新を完了しました: 更新インデックス: 1
[SERVER_IP-ESP32]  - - [11/Jul/2024 14:46:07] "GET /update HTTP/1.0" 200 -
更新ファイルを送信しました: update_2.py to [SERVER_IP-ESP32]
[SERVER_IP-ESP32]  - - [11/Jul/2024 14:46:07] "GET /complete HTTP/1.0" 200 -
デバイスが更新を完了しました: 更新インデックス: 2
[SERVER_IP-ESP32]  - - [11/Jul/2024 14:46:10] "GET /update HTTP/1.0" 200 -
更新ファイルを送信しました: update_3.py to [SERVER_IP-ESP32]
[SERVER_IP-ESP32]  - - [11/Jul/2024 14:46:11] "GET /complete HTTP/1.0" 200 -
デバイスが更新を完了しました: 更新インデックス: 3
[SERVER_IP-ESP32]  のすべての更新が完了しました。

以上がSimpleHTTPRequestHandlerを使用したESP32のソフトウェアのアップデートになります.

6.おわりに

今回 SimpleHTTPRequestHandlerを使用してソフトウェアのアップデートを行いましたが, 私自身インターネット関係の実装を手探りで行ったため, 所々におかしな所や間違った点があると思いますがご了承ください.
不出来なプログラムですが, 実験や実装の助けになればと思います.