MicropythonでESP32のファームウェアアップデートをWi-Fiで行う方法

はじめに

MicropythonでESP32のファームウェアアップデートをWi-Fiで行う方法を書きます．今回はサーバーにあるファームウェアをプル方式でダウンロードします．アップデートの方式はA/B パーティション方式です．ファームウェアはMicropythonのサイトのESP32/WROOMにあるOTAサポート版のものを使用します．ESP32にインストールしてあるファームウェアとアップデートに使用するファームウェアが，OTAサポート版でないとアップデートが行えないため注意してください．また，アップデートに使用するファームウェアは.app-binである必要があります．事前にESP32にインストールしておくファームウェアは.binのものにしてください．

環境

◯ESP32

• ファームウェア(アップデート前)：v1.26.0 (2025-08-09)
• ファームウェア(アップデート後)：v1.26.1 (2025-09-11)

ファイル構成

root/ 
    ├─ boot.py
    └─ ota.py

◯サーバー

• OS：ubuntu-24.04/LTS
• 実行環境：Python 3.12.3

ファイル構成

user/ 
    └─ OTA/ 
        └─ ESP32_GENERIC-OTA-20250911-v1.26.1.app-bin ⇒ ファームウェア

サンプルコード

ESP32

boot.py

boot.py

import time
from esp32 import Partition
import uos as os
import network
import ota      

def _log_info():
    run = Partition(Partition.RUNNING)
    info = run.info()
    label = info[4] if len(info) >= 5 else None
    label = None
    u = os.uname()

    fw_ver = getattr(u, 'version', None) or getattr(u, 'release', None)
    sta = network.WLAN(network.STA_IF)
    if not sta.active():
        sta.active(True)
    mb = sta.config('mac')
    mac = ':'.join('%02x' % b for b in mb)
    
    print('[BOOT] running=', label, 'fw=', fw_ver, 'mac=', mac)


def _run():
    _log_info()
    Partition.mark_app_valid_cancel_rollback()

    for i in range(3, 0, -1):
        print('[BOOT] starting OTA in', i)
        time.sleep(1)
    try:
        ota.run()
    except Exception as ex:
        print('[BOOT][ERR]', ex)

_run()

ota.py

コードを使用する際は，WIFI_SSID，WIFI_PASS，FW_URLを自身のものに変更してください．

ota.py

import time, network, urequests as rq, uhashlib as uh, esp32, machine, esp

# ===== CONFIG =====
WIFI_SSID = "Wi-FiのSSID"
WIFI_PASS = "Wi-Fiのパスワード"
FW_URL    = "http://サーバーのホスト名:任意のポート番号/OTA/ESP32_GENERIC-OTA-20250911-v1.26.1.app-bin"
CHUNK     = 1024


# ===== OTA writer (direct write to inactive OTA partition) =====
BLK = 4096

class OTADirectWriter:
    def __init__(self, total_len=0, sha16=b"\x00"*16):
        self.run  = esp32.Partition(esp32.Partition.RUNNING)
        self.part = self.run.get_next_update()
        if self.part is None:
            raise OSError("No OTA update partition")
        _t, _st, addr, psize, _label, _enc = self.part.info()
        self.addr  = addr
        self.psize = psize
        self.total = total_len or psize
        self.sha16 = sha16 or b"\x00"*16
        self.h     = uh.sha256()
        self.buf   = bytearray(BLK)
        self.boff  = 0
        self.written = 0
        # erase range based on expected total (or full slot if unknown)
        nblk = (self.total + BLK - 1) // BLK
        if hasattr(self.part, "erase_blocks"):
            self.part.erase_blocks(0, nblk)
        else:
            start_sector = addr // BLK
            for i in range(nblk):
                esp.flash_erase(start_sector + i)

    def write(self, data):
        mv = memoryview(data)
        self.h.update(mv)
        off = 0
        while off < len(mv):
            n = min(len(mv) - off, BLK - self.boff)
            self.buf[self.boff:self.boff+n] = mv[off:off+n]
            self.boff += n
            off += n
            if self.boff == BLK:
                bi = self.written // BLK
                self.part.writeblocks(bi, self.buf)
                self.written += BLK
                self.boff = 0

    def finalize(self):
        if self.boff:
            for i in range(self.boff, BLK):
                self.buf[i] = 0xFF
            bi = self.written // BLK
            self.part.writeblocks(bi, self.buf)
            self.written += BLK
            self.boff = 0
        return True

    def switch_and_reboot(self):
        try:
            try:
                esp32.Partition.set_boot(self.part)
            except Exception:
                runp = esp32.Partition(esp32.Partition.RUNNING)
                nxtp = runp.get_next_update()
                if nxtp:
                    esp32.Partition.set_boot(nxtp)
        except Exception:
            pass
        machine.reset()


# ===== HTTP stream =====
def _http_iter(url, sz=CHUNK):
    r = rq.get(url)
    if getattr(r, 'status_code', 200) != 200:
        try:
            r.close()
        except Exception:
            pass
        raise OSError("HTTP {}".format(getattr(r, 'status_code', -1)))
    try:
        while True:
            b = r.raw.read(sz)
            if not b:
                break
            yield b
    finally:
        try:
            r.close()
        except Exception:
            pass


# ===== WiFi =====
def _wifi():
    s = network.WLAN(network.STA_IF)
    if not s.active():
        s.active(True)
    if not s.isconnected():
        s.connect(WIFI_SSID, WIFI_PASS)
        t0 = time.ticks_ms()
        while not s.isconnected() and time.ticks_diff(time.ticks_ms(), t0) < 15000:
            time.sleep_ms(200)
    if not s.isconnected():
        raise OSError("WiFi connect failed")
    return s


# ===== Main =====
def run(fw_url=None):
    print("[OTA] start")
    _wifi()
    url = fw_url or FW_URL
    print("[OTA] GET", url)
    writer = OTADirectWriter(0, b"\x00"*16)
    total = 0
    for part in _http_iter(url, CHUNK):
        writer.write(part)
        total += len(part)
        if (total // CHUNK) % 64 == 0:
            print("[OTA] dl", total)
    ok = writer.finalize()
    print("[OTA] finalize:", ok, "bytes=", total)
    if ok:
        writer.switch_and_reboot()
    return ok

サーバー

コードを用意する必要はありません．HTTPサーバーが動作していて，任意のポートが開いていれば大丈夫です．(今回は例として8000番ポートを使用しました)
今回HTTPサーバーを動作させるに当たり使用したコマンドは以下の通りになります．

python3 -m http.server 8000

実行結果

ESP32

以下の画像はESP32の実行時の出力になります．reset machineの出力を堺に上部がアップデート適用前，下部がアップデート適用後になります．[BOOT]の出力にあるrunning=が現在使用しているパーテーションのラベル名，fw=が現在動作しているファームウェアのバージョンになります．適用前では，ota_0のパーテーションを使用しているのに対し，適用後ではota_1を使用していることが分かります．また，適用前のファームウェアはv1.26.0 on 2025-08-09が動作しているのに対し，適用後ではv1.26.1 on 2025-09-11が動作しており，適用したファームウェアが正常に動作していることが確認できます．

サーバー

以下の画像はサーバーでコマンドを実行してESP32を動作させた際の出力結果になります．ESP32がHTTPを使用してOTAディレクトリにあるESP32_GENERIC-OTA-20250911-v1.26.1.app-binを取得している事が分かります．