ESP32とガベージコレクション
メモリ不足発生
次のPythonプログラムはESP32開発ボード(ESP32-C3M)に接続されたAHT21で5分毎に温度と湿度を測定して、IoTクラウドのThingSpeakにアップロードしてグラフを描くプログラムです。
プログラムを動かすと4回目のアップロードで次のエラーが出ました。
-10368, 'MBEDTLS_ERR_X509_ALLOC_FAILED'
これはMbed TLSライブラリにおいて、メモリの割り当てに失敗したことを示すエラーであり、必要なメモリを確保できなかった場合に返されます。
# AHT21 and ThingSpeak
# Mar. 7th 2025
import requests
from machine import Pin, I2C
import network
import time
from aht21 import AHT21
import json
import gc
DEBUG = True
WIFI_SSID = 'SSID'
WIFI_PASS = 'パスワード'
# ThingSpeakのAPIキーとURL
API_KEY = 'APIキー'
THINGSPEAK_URL = 'https://api.thingspeak.com/update.json'
headers = {
'Content-Type' : 'application/json'
}
i2c = I2C(0)
aht21 = AHT21(i2c)
wifi = network.WLAN(network.STA_IF)
if not wifi.isconnected():
print("Connecting to WiFi...")
wifi.active(True)
wifi.connect(WIFI_SSID, WIFI_PASS)
while not wifi.isconnected():
pass
print('IP:', wifi.ifconfig()[0])
gc.enable()
gc.collect()
mem_free = str(gc.mem_free())
print("mem_free:", mem_free)
try:
n = 0
while True:
temperature = '{:.1f}'.format(aht21.temperature)
humidity = '{:.1f}'.format(aht21.relative_humidity)
print('TEMP:',temperature, 'HUM:', humidity)
# アップロードするデータ
data = {
"api_key": API_KEY,
"field1": str(temperature), # フィールド1のデータ
"field2": str(humidity), # フィールド2のデータ
"field3": mem_free
}
# 辞書型からJSONに変換
data = json.dumps(data).encode('utf-8')
if DEBUG == True:
print(data)
# POSTリクエストを送信
response = requests.post(THINGSPEAK_URL, headers=headers, data=data)
print(response.status_code)
# レスポンスの確認
if response.status_code == 200:
print("データが正常にアップロードされました。")
else:
print("データのアップロードに失敗しました:", response.status_code)
time.sleep(60 * 5) # 5分に1回測定、アップロードするためにお休み
# メモリ解放
# gc.collect()
n += 1
mem_free = str(gc.mem_free())
print("n:", str(n), "mem_free:", mem_free)
except OSError as e:
print("mem_free:", str(gc.mem_free()))
print('エラーが発生しました。: {}'.format(e))
ちなみに、アップロードするたびに取得しているgc.mem_free()は次のようになりました。
実行直後 mem_free: 162144
n: 1 mem_free: 123136
n: 2 mem_free: 94384
n: 3 mem_free: 60512
4回目のアップロードでメモリ不足のエラーが発生しました。
エラー発生時 mem_free: 56496
gc.collect()の有効化
上のプログラムの79行目にあるgc.collect()のコメントを外して、アップロードするたびに実行するようにしてみると次のようになり、メモリ不足のエラーが発生しなくなりました。
実行直後 91488
n: 1 mem_free: 88256
n: 2 mem_free: 88256
n: 3 mem_free: 88256
n: 4 mem_free: 88256
:
:
n: 9 mem_free: 88256
n: 10 mem_free: 88256
n: 11 mem_free: 88256
:
:
gc.mem_free() の値をThingSpeakで表示させると次のようになり、メモリ解放が行われていることが分かります。
ちなみに、温度、湿度は次のように表示されています。
モジュールのダイエット
上のプログラムでgc.collect()を有効にして放置していたところ、32回目で
n: 32 mem_free: 72896
mem_free: 70000
[Errno 12] ENOMEM
のエラーが発生し、プログラムが停止してしまいました。
またもやメモリー不足です。
そこでimportしているモジュールから必要なクラスだけをインポートするように、次のようにダイエットし、プログラムをESP32開発ボードにmain.pyのファイル名で転送し、PCを接続せず、ボードだけで動かしてみました。
from requests import post
from machine import Pin, I2C
from network import WLAN
from time import sleep
from aht21 import AHT21
from json import dumps
import gc
9時過ぎから18時過ぎまでのグラフを以下に示します。
メモリ不足のエラーなしに、約9時間、5分に1回、計100回以上アップロードできているのが分かります。
24時間経過後のグラフ。今のところ、問題なく動いています。
ダイエット版のサンプルコード
# AHT21 and ThingSpeak
# Mar. 7th 2025
from requests import post
from machine import Pin, I2C
from network import WLAN
from time import sleep
from aht21 import AHT21
from json import dumps
import gc
DEBUG = True
WIFI_SSID = 'WiFiのSSID'
WIFI_PASS = 'パスワード'
# ThingSpeakのAPIキーとURL
API_KEY = 'APIキー'
THINGSPEAK_URL = 'https://api.thingspeak.com/update.json'
headers = {
'Content-Type' : 'application/json'
}
i2c = I2C(0)
aht21 = AHT21(i2c)
wifi = WLAN(0) # network.STA_IF = 0
if not wifi.isconnected():
print("Connecting to WiFi...")
wifi.active(True)
wifi.connect(WIFI_SSID, WIFI_PASS)
while not wifi.isconnected():
pass
print('IP:', wifi.ifconfig()[0])
gc.enable()
gc.collect()
mem_free = str(gc.mem_free())
print("mem_free:", mem_free)
try:
n = 0
while True:
temperature = '{:.1f}'.format(aht21.temperature)
humidity = '{:.1f}'.format(aht21.relative_humidity)
print('TEMP:',temperature, 'HUM:', humidity)
# アップロードするデータ
data = {
"api_key": API_KEY,
"field1": str(temperature), # フィールド1のデータ
"field2": str(humidity), # フィールド2のデータ
"field3": mem_free
}
# 辞書型からJSONに変換
data = dumps(data).encode('utf-8')
if DEBUG == True:
print(data)
# POSTリクエストを送信
response = post(THINGSPEAK_URL, headers=headers, data=data)
print(response.status_code)
# レスポンスの確認
if response.status_code == 200:
print("データが正常にアップロードされました。")
else:
print("データのアップロードに失敗しました:", response.status_code)
sleep(60 * 5) # 5分に1回測定、アップロードするためにお休み
# メモリ解放
gc.collect()
n += 1
mem_free = str(gc.mem_free())
print("n:", str(n), "mem_free:", mem_free)
except OSError as e:
print("mem_free:", str(gc.mem_free()))
print('エラーが発生しました。: {}'.format(e))
gc.threshold(gc.mem_free() // 4 + gc.mem_alloc())を追加
さらに Micropythonのドキュメント に
gc.collect()
gc.threshold(gc.mem_free() // 4 + gc.mem_alloc())
とすれば良いとの記述があったので試してみました。
これは、ヒープサイズの25％に現在使用中のメモリサイズを加算し、この値を超えるとガベージコレクションが自動的に行われるというものです。
ついでに、オンボードLEDを点灯させるようにし、アップロードに失敗した場合はLEDを消灯するコードを追加してみました。
gc.threshold(gc.mem_free() // 4 + gc.mem_alloc())を追加したコード
# AHT21 and ThingSpeak
# Mar. 7th 2025
# Last modify: Mar. 10th 2025
from requests import post
from machine import Pin, I2C
from network import WLAN
from time import sleep
from aht21 import AHT21
from json import dumps
import gc
DEBUG = True
ONBOARD_LED = 0
WIFI_SSID = 'WiFiのSSID'
WIFI_PASS = 'WiFiのパスワード'
# ThingSpeakのAPIキーとURL
API_KEY = 'APIキー'
THINGSPEAK_URL = 'https://api.thingspeak.com/update.json'
headers = {
'Content-Type' : 'application/json'
}
i2c = I2C(0)
aht21 = AHT21(i2c)
led = Pin(ONBOARD_LED, Pin.OUT)
# オンボードLEDをn回点滅させる関数
def blink(n):
for i in range(n):
led.on()
sleep(0.1)
led.off()
sleep(0.1)
wifi = WLAN(0) # network.STA_IF = 0
if not wifi.isconnected():
print("Connecting to WiFi...")
wifi.active(True)
wifi.connect(WIFI_SSID, WIFI_PASS)
while not wifi.isconnected():
blink(1)
pass
print('IP:', wifi.ifconfig()[0])
gc.enable()
gc.collect()
mem_free = str(gc.mem_free())
print("mem_free:", mem_free)
try:
n = 0
while True:
temperature = '{:.1f}'.format(aht21.temperature)
humidity = '{:.1f}'.format(aht21.relative_humidity)
print('TEMP:',temperature, 'HUM:', humidity)
# アップロードするデータ
data = {
"api_key": API_KEY,
"field1": str(temperature), # フィールド1のデータ
"field2": str(humidity), # フィールド2のデータ
"field3": mem_free
}
# 辞書型からJSONに変換
data = dumps(data).encode('utf-8')
if DEBUG == True:
print(data)
# POSTリクエストを送信
response = post(THINGSPEAK_URL, headers=headers, data=data)
print(response.status_code)
# レスポンスの確認
if response.status_code == 200:
print("データが正常にアップロードされました。")
blink(5)
led.on()
else:
print("データのアップロードに失敗しました:", response.status_code)
blink(10)
led.off()
# gc.thresholdを設定
gc.threshold(gc.mem_free() // 4 + gc.mem_alloc())
sleep(60 * 5) # 5分に1回測定、アップロードするためにお休み
# メモリ解放
gc.collect()
n += 1
mem_free = str(gc.mem_free())
print("n:", str(n), "mem_free:", mem_free)
except OSError as e:
print("mem_free:", str(gc.mem_free()))
print('エラーが発生しました。: {}'.format(e))
上のコードで19時30分過ぎから23時40分過ぎまで約50回の計測結果を下図に示します。
空きメモリが85Kbytesを大きく下回ることはなくなったようです。
gc.threshold(gc.mem_free() // 4 + gc.mem_alloc())追加後のグラフ
最終的に
やはり上記の修正を施しても、しばらく放置していると止まってしまうようです。
そこで、exceptに入ったらハードウエアリセットするように次のように書き換えました。
except OSError as e:
print("mem_free:", str(gc.mem_free()))
print('エラーが発生しました。: {}'.format(e))
# ハードウェアリセット
machine.reset()
これでしばらく様子を見てみます。
Discussion