โมดูล wifi คือหน้าต่างต่ออินเทอร์เน็ตของบอร์ด (WiFi รันบน CM33_NS) ชุดบทเรียนนี้ใช้แค่ 3 คำสั่ง:
| คำสั่ง | ทำอะไร |
|---|---|
wifi.connect(ssid, password) |
ต่อเข้า AP (บล็อกจนต่อได้หรือ timeout) |
wifi.is_connected() |
คืน True/False — ต่อสำเร็จหรือยัง |
wifi.ip() |
คืน IP ที่ได้มาเป็นข้อความ (เช็กว่าได้ที่อยู่จริง) |
import wifi
if not wifi.is_connected():
wifi.connect("<ชื่อ WiFi ของคุณ>", "<รหัส WiFi ของคุณ>") # ใส่ค่าจริงของคุณตอนรัน (อย่า commit รหัสจริง)
if wifi.is_connected():
print("WiFi ok:", wifi.ip())
wifi.connect() ใช้เวลาสักครู่ (สแกน + จับมือ) — เรียกครั้งเดียวก่อนเข้าลูป ไม่ใช่ทุกรอบโมดูลนี้มีคำสั่งอื่นอีก (
wifi.scan(),wifi.status(),wifi.ping()) แต่ชุดบทเรียนนี้เอาแค่ต่อให้ติดพอ — ลองโชว์wifi.status()["rssi"]เป็นโจทย์ต่อยอด
MQTT เป็นโปรโตคอลส่งข้อความแบบ publish/subscribe ผ่านตัวกลางชื่อ broker เบามาก เหมาะกับอุปกรณ์ IoT
tesaiot/edge-ai/s17 — ใครสนใจก็ subscribe ช่องนี้test.mosquitto.org (ฟรี ไม่ต้องล็อกอิน)publish/subscribe แยกผู้ส่งกับผู้รับออกจากกัน (decoupled) — บอร์ดไม่ต้องรู้ว่าใครฟังอยู่ แค่โยนข้อความเข้า topic ที่ตกลงกันไว้ นี่คือหัวใจว่าทำไม IoT ชอบ MQTT
mqtt ห่อ MQTT ให้เหลือไม่กี่คำสั่ง ชุดบทเรียนนี้ใช้ 4 ตัวหลัก:
| คำสั่ง | ทำอะไร |
|---|---|
mqtt.connect(broker, port, client_id=...) |
ต่อเข้า broker (ต้องต่อ WiFi ก่อน) |
mqtt.publish(topic, payload) |
ส่งข้อความเข้า topic (QoS 0 โดยปริยาย) |
mqtt.is_connected() |
เช็กว่ายังต่อ broker อยู่ไหม |
mqtt.disconnect() |
ตัดการเชื่อมต่อให้เรียบร้อยตอนจบ |
import mqtt
mqtt.connect("test.mosquitto.org", 1883, client_id="bento-s17")
mqtt.publish("tesaiot/edge-ai/s17", '{"event":"shaking","conf":0.92}')
client_id ควรตั้งให้ไม่ซ้ำใคร (broker สาธารณะมีหลายคนใช้) — เราใส่คำนำหน้า bento-mqtt.publish() โยน OSError ได้ถ้าการเชื่อมต่อหลุด — ห่อด้วย try/except เสมอมีคำสั่งอื่นอีก (
subscribe,get_message) สำหรับ "รับ" ข้อความ แต่ชุดบทเรียนนี้เราโฟกัส "ส่ง" (publish) — การรับเก็บไว้เป็น โจทย์ต่อยอด
เราส่งเหตุการณ์เป็น JSON เพราะทุกระบบปลายทาง (dashboard, Node-RED, มือถือ) อ่านได้ทันที ประกอบ payload จากข้อมูล fusion:
payload = '{"event":"%s","conf":%.2f,"gyro":%.0f,"ts":%d}' % (
TARGET_CLASS, # เหตุการณ์อะไร (จากโมเดล)
conf_val, # โมเดลมั่นใจแค่ไหน
gmag, # เซนเซอร์ดิบยืนยันแรงเท่าไร
time.ticks_ms(), # ประทับเวลา (บอกลำดับเหตุการณ์)
)
event, conf) และ หลักฐานดิบ (gyro) ลงไป — ปลายทางเห็นว่าตัดสินจากอะไร% ก็พอสำหรับ payload สั้นๆ (MicroPython มี json ให้ใช้ด้วย ถ้าอยากเป๊ะ)การใส่ทั้ง
conf(จากโมเดล) และgyro(จากเซนเซอร์ดิบ) ในข้อความเดียว = ปลายทางตรวจสอบการตัดสินใจ fusion ของเราย้อนหลังได้ นี่คือ observability ที่ดี
บอร์ดจริงมี WiFi/MQTT ส่วน Emulator มี wifi แบบจำลองและ mqtt ที่ต่อ broker สาธารณะของจริงผ่าน WebSocket แต่ที่ไหนก็อาจเจอเน็ตไม่ติดหรือไม่มีโมดูลเน็ต — เราต้องเขียนให้โค้ด ชุดเดียว รันได้ทุกที่ ไม่พังถ้าไม่มีเน็ต
try:
import wifi, mqtt
HAVE_NET = True
except ImportError:
HAVE_NET = False # ไม่มีโมดูลเน็ต → โหมด offline
def publish_event(conf_val, gmag):
payload = build_payload(conf_val, gmag)
if HAVE_NET and mqtt.is_connected():
mqtt.publish(TOPIC, payload) # บอร์ด: ส่งขึ้นคลาวด์จริง
else:
lcd.console('<span class=info> [SIM] ' + payload + '</span>') # เน็ตไม่ติด: โชว์แทน
try/except ImportError = ถ้าไม่มีโมดูล wifi/mqtt ก็ตั้งธง offline แทนที่จะ crash ตอน import[SIM]นี่คือนิสัย multi-surface ของทั้งคอร์ส (บทเรียน 1.1–1.3): โค้ดบรรทัดเดียวกันต้องรันได้ทั้ง Emulator และบอร์ด — เราแค่ห่อส่วนที่ต้องมีฮาร์ดแวร์จริงด้วยการเช็กก่อน
เปิด s17_fusion_iot.py บน BENTO Edge AI Emulator ในเบราว์เซอร์ได้ทันที WiFi เป็นแบบจำลองที่ต่อติด และ MQTT ส่งถึง broker สาธารณะจริงผ่าน WebSocket (ต่อไม่ได้จะใช้ broker จำลอง) ข้อควรรู้: แผง HW ของ Emulator ขยับเฉพาะ accel (ลากเอียงและปุ่ม Shake) ส่วน gyro ค้างใกล้ศูนย์ ประตู gyro จึงไม่เปิด ถ้าจะซ้อมบน Emulator ให้เปลี่ยนประตูเป็น accel ชั่วคราว เช่น gmag = abs(ax) + abs(ay) + abs(az - 9.81) กับ MOTION_FLOOR = 5.0
บนจอของแอปนี้จะเห็น verdict, ค่าประตูดิบ และบรรทัด MQTT TX หรือ [SIM] ในหน้าเดียว
MOTION_FLOOR หรือยังMQTT TX (หรือ [SIM] ถ้าเน็ตไม่ติด) จะพิมพ์ payload JSON ที่ถูก publishซ้อมบน Emulator ให้เห็น fusion กับ MQTT TX ก่อน แล้วค่อยย้ายไปบอร์ดจริงที่ต่อ WiFi ของจริงและใช้ประตู gyro — โค้ดชุดเดียวกัน เปลี่ยนแค่พื้นผิวที่รัน
รวมทุกจังหวะเป็นภาพเดียว: จากเซนเซอร์บนบอร์ด จนถึงข้อความที่โผล่บน dashboard — นี่คือวงจรที่แอปชุดบทเรียนนี้เดินทุก ~250 ms
ครึ่งซ้าย (เซนเซอร์ → verdict) เกิดบน CM55 เราไม่แตะ · ครึ่งขวา (fuse → trigger → publish) คือโค้ด Python ของเราบน CM33 — ชุดบทเรียนนี้เราคุมครึ่งขวาทั้งหมด แล้วต่อสายออกไปถึงคลาวด์
ทั้งไฟล์อ่านเป็นประโยคเดียว: "ต่อเน็ต → หาโมเดล + สั่งรัน → วนอ่าน verdict + gyro ดิบ → พอสองสัญญาณเห็นตรงกัน ก็ publish เหตุการณ์ขึ้น MQTT → หยุดตอนออก"
ตัวเลขช่อง (1–5) ชี้จุดที่คุณต้องเติมในไฟล์ฝึก — ช่อง 1–2 เป็นของเดิม (ทวน) ช่อง 3–5 คือ fusion + IoT ของใหม่ของชุดบทเรียนนี้
ช่องเติมที่ 1: หาโมเดลจากชื่อ (ให้ไว้แล้ว) แล้วสั่งให้ CM55 เริ่มรัน — การต่อ WiFi/MQTT ทำให้แล้วด้านบน:
model = find_model(MODEL_KEYWORD) # ให้ไว้แล้ว: คืน dict ทั้งก้อน
labels = model['labels']
try:
# เติม: สั่งให้ CM55 รันโมเดลนี้ ด้วย edge_ai.select(model['index'])
pass
lcd.console('<span class=ok> เริ่มอนุมาน %s</span>' % model['name'])
except OSError as e:
lcd.console('<span class=error> โหลดไม่สำเร็จ: %s</span>' % e)
pass ด้วย edge_ai.select(model['index']) — เอา index จาก dict ไปสั่ง (เหมือนบทเรียน 1.6–1.7)try เพราะ select() โยน OSError ได้ (ข้ามคอร์อาจพลาด)--- เพราะไม่มีโมเดลรัน result() คืน None ตลอด
MODEL_KEYWORD+TARGET_CLASSอยู่บนหัวไฟล์ — remix ได้ทันที เปลี่ยนเป็น Cough/Siren แอปก็เปลี่ยนเหตุการณ์ที่สตรีมโดยไม่แตะ logic
ช่องเติมที่ 2 และ 3: จังหวะ 2 (verdict เดิม) + จังหวะ fuse ครึ่งแรก (อ่านเซนเซอร์ดิบมาประกบ):
# เติม: อ่านผลอนุมานล่าสุดมาเก็บใน r -> r = edge_ai.result()
r = None
pass
if r and r['seq'] != last_seq:
last_seq = r['seq']
verdict.text(r['label'] or '-')
conf.text("conf: %.0f %%" % (r['conf'] * 100))
# เติม: อ่านเซนเซอร์ดิบมายืนยัน -> ax,ay,az,gx,gy,gz = sensors.bmi270.motion()
gx = gy = gz = 0.0
pass
gmag = abs(gx) + abs(gy) + abs(gz)
raw_lab.text("gyro: %.0f" % gmag)
pass ด้วย r = edge_ai.result() (เหมือนบทเรียน 1.1–1.3 / 1.6–1.7)pass ด้วย ax, ay, az, gx, gy, gz = sensors.bmi270.motion() — อ่านค่าดิบมาทำ gateseq ก่อนเสมอ วาดจอ + อ่านเซนเซอร์เฉพาะตอนมี verdict ใหม่ช่อง 2 คือ "ของเดิม" ที่คุณทำเป็นแล้ว ช่อง 3 คือของใหม่ — เริ่มมีเซนเซอร์ สองเส้นทาง ในลูปเดียว: ผ่านโมเดล กับ อ่านดิบ
ช่องเติมที่ 4: จังหวะ fuse ครึ่งหลัง — เอา verdict มา AND กับ gate ของเซนเซอร์ดิบ:
model_hit = (r['label'] == TARGET_CLASS
and r['conf'] >= edge_ai.CONF_FLOOR) # โมเดลผ่าน
raw_ok = gmag > MOTION_FLOOR # เซนเซอร์ดิบผ่าน
# เติม: รวมสองสัญญาณด้วย AND -> fused = model_hit and raw_ok
fused = False
pass
if fused and not fired:
publish_event(r['conf'], gmag) # สองสัญญาณเห็นตรงกัน → ส่ง
fired = True
elif not fused:
fired = False
state.text("รอเหตุการณ์ ...")
state.color(DIM)
pass ด้วย fused = model_hit and raw_ok — หัวใจของ fusion อยู่บรรทัดนี้and not fired = edge-trigger ยิงครั้งเดียวต่อการเจอ (ไม่ spam broker)fused เป็น False ตลอด → ไม่มีอะไรถูกส่งเลย แม้โมเดล+เซนเซอร์จะเห็นตรงกันบรรทัดนี้คือความต่างจากบทเรียน 6.3–6.4 — action ไม่ได้ยิงจาก verdict อย่างเดียว แต่ยิงจาก verdict ที่เซนเซอร์ดิบรับรอง
ช่องเติมที่ 5: ภายใน publish_event() — ที่ที่ fused decision กลายเป็นข้อความบนคลาวด์:
def publish_event(conf_val, gmag):
payload = '{"event":"%s","conf":%.2f,"gyro":%.0f,"ts":%d}' % (
TARGET_CLASS, conf_val, gmag, time.ticks_ms())
state.text("! ส่งเหตุการณ์ !"); state.color(GREEN)
if HAVE_NET and mqtt.is_connected():
try:
# เติม: ส่ง payload ขึ้น broker -> mqtt.publish(TOPIC, payload)
pass
lcd.console('<span class=ok> MQTT TX: ' + payload + '</span>')
except OSError as e:
lcd.console('<span class=error> ส่งไม่สำเร็จ: %s</span>' % e)
else:
lcd.console('<span class=info> [SIM] ' + payload + '</span>') # offline
pass ด้วย mqtt.publish(TOPIC, payload) — ส่งเหตุการณ์ที่ผ่าน fusion ขึ้น topictry/except OSError เพราะการเชื่อมต่ออาจหลุด — แอปต้องไม่ล้มเพราะเน็ตสะดุดelse โชว์ [SIM] — กลไกครบ แค่ไม่ส่งจริงเติมครบ 5 ช่องเมื่อไร คุณได้แอป Edge AI ที่ตัดสินใจแบบ fused แล้วสตรีมขึ้นคลาวด์ — นี่คือ MVP ของบทเรียน 6.5–6.6 ของชุดบทเรียนนี้
เปิด s17_fusion_iot.py ในไฟล์มี pass วางไว้ 5 จุด ตรงที่ต้องเติมคำสั่งจริง:
| # | จุด | เติมด้วย | ถ้าลืม |
|---|---|---|---|
| 1 | หลัง find_model | edge_ai.select(model['index']) |
จอค้าง --- ไม่มี verdict |
| 2 | ในลูป | r = edge_ai.result() |
จอไม่ขึ้นคลาสเลย |
| 3 | มี verdict ใหม่ | ax, ay, az, gx, gy, gz = sensors.bmi270.motion() |
gyro ค้าง 0 ประตูไม่มีวันเปิด |
| 4 | รวมสัญญาณ | fused = model_hit and raw_ok |
ไม่มีเหตุการณ์ถูกส่งเลย |
| 5 | ใน publish_event | mqtt.publish(TOPIC, payload) |
fused แล้วแต่ไม่ถึง broker |
ขั้นตอน:
MODEL_KEYWORD / TARGET_CLASS / MOTION_FLOOR ที่อยากลอง (เริ่มจาก Motion/shaking/40)# เติม: ทีละจุด แล้วแทน pass ด้วยคำสั่งตามคำใบ้ช่อง 1–2 คือของเดิม (ทวน) · ช่อง 3–5 คือ fusion + IoT ของใหม่ — เติมครบเมื่อไร แอปของคุณ "ตัดสินใจแบบเชื่อถือได้" แล้ว "พูดคุยกับคลาวด์" ได้
อยากต่อยอดเรื่อง sensor fusion และ IoT/MQTT ลองดูของดีที่คนทำจริงเขาอธิบายไว้:
วิดีโอ (ช่องการศึกษาที่น่าเชื่อถือ)
อ่าน/ภาพอ้างอิง
วิดีโอ/ภาพภายนอกเป็นของเจ้าของต้นฉบับ ใช้เพื่อการศึกษา อ้างอิงลิงก์ต้นทาง — เราไม่ได้ฝังคลิปไว้ในสไลด์ ให้กดลิงก์ไปดูที่ต้นทางเอง
MVP ของบทเรียน 6.5–6.6 (เกณฑ์ผ่านของชุดบทเรียน): คุณทำให้ การตัดสินใจแบบ fused (verdict AND gyro ดิบ) ถูก publish ขึ้น MQTT ได้จริง — บนบอร์ดหรือ Emulator (ถ้าเน็ตไม่ติดจะเห็น [SIM] บน console แทน)
and และทำไมต้อง edge-trigger ก่อน publish"ส่งได้" ไม่ใช่แค่ "เห็น TX บนจอ" — คุณต้องบอกได้ว่า fusion กรอง false positive ยังไง และเหตุการณ์นี้เดินทางจากบอร์ดถึง broker ทางไหน
ถ้าติด ให้ไต่บันไดนี้ทีละขั้น อย่าเพิ่งกระโดดไปดูเฉลย เพราะของจะเข้าหัวตอนที่คุณพยายามเองก่อน:
# เติม: ทั้ง 5 จุดในไฟล์ฝึก + ตารางหน้าที่แล้ว บอกว่าแต่ละช่องเติมอะไรs17_fusion_iot.py มีโครงครบทั้งไฟล์แล้ว (รวม WiFi/MQTT + degrade) เหลือแค่ 5 บรรทัดให้เติมs17_fusion_iot.py เติมครบพร้อมคอมเมนต์อธิบายทุกช่อง (อ่านให้เข้าใจ ปิดไฟล์ แล้วพิมพ์เอง)s17_fusion_iot_full.py เพิ่ม gate เป็นเรดาร์ (multi-modal) + นับจำนวนเหตุการณ์ + reconnect broker + โชว์ RSSIลองเขียนเองให้สุดก่อนนะ ถ้าติดจริงๆ ค่อยเปิดเฉลยดูทีละช่อง แล้วกลับมาพิมพ์เอง — เดี๋ยวเราค่อย ๆ แกะไปด้วยกัน
การรวม verdict กับเซนเซอร์ดิบแล้วสตรีมขึ้นคลาวด์ ซ่อนแนวคิดที่ปิดกล่อง โมดูล 6 (Apps) พอดี:
ฝั่ง Edge AI / ระบบ
AND raw feature (how strong) → ตัดสินใจที่ทน false positiveconf + gyro ให้ปลายทางตรวจสอบการตัดสินใจย้อนหลังฝั่ง MicroPython / โครงโปรแกรม
wifi + mqtt — ต่อเน็ต + publish/subscribe แบบ decoupledtry/except ImportError + เช็ก is_connected() → โค้ดชุดเดียวรันทุกพื้นผิวfinally: stop() + mqtt.disconnect() คืนทุกอย่างสู่สถานะที่รู้แน่ทั้งหมดยังยืนบนคำสั่งเดิมของ
edge_ai+sensors— ชุดบทเรียนนี้แค่ต่อ "IoT layer" บนแอปที่คุณทำเป็นแล้ว
pattern "รวมสัญญาณให้เชื่อถือได้ แล้วสตรีมเหตุการณ์ขึ้นคลาวด์" คือแก่นของระบบ Edge AI + IoT จริงทุกตัว:
สังเกตว่าทุกระบบใช้ pattern เดียวกับแอปชุดบทเรียนนี้เป๊ะ — ต่างกันแค่โมเดล, ประตูยืนยัน, และ topic ปลายทาง เราแค่กำลังเริ่มจากปลายทางเพื่อย้อนไปสร้างระบบเต็มเองได้
งานทำเอง (ท้ายบทเรียน):
s17_fusion_iot.py ให้ครบทั้ง 5 ช่อง รันได้จริงบน Emulator หรือบอร์ดshaking แต่ gyro ดิบไม่ถึง MOTION_FLOOR (หรือกลับกัน) แล้วยืนยันว่า ไม่มีเหตุการณ์ถูกส่งMODEL_KEYWORD/TARGET_CLASS/MOTION_FLOOR เป็นชุดอื่น แล้วอธิบายว่า topic/payload เปลี่ยนยังไงใบ้ข้อ 2 — fusion เป็น and ดังนั้นแค่ด่านเดียวไม่ผ่านก็พอ: โมเดลมั่นใจแต่ gyro เบา = ไม่ส่ง หรือ gyro แรงแต่โมเดลตอบผิดคลาส = ไม่ส่ง นี่คือหัวใจว่า fusion กัน false positive ยังไง
วันนี้เราได้: เข้าใจ sensor fusion (verdict + raw gate) · อ่านเซนเซอร์ดิบมายืนยัน verdict · ต่อ WiFi + publish MQTT · เขียนโค้ดชุดเดียวที่ degrade ได้ทั้งบอร์ดและ Emulator · ปิดกล่อง โมดูล 6 (Apps)
ชุดบทเรียนถัดไป (บทเรียน 7.1–7.2) เราปิด โมดูล 6 (Apps) แล้วเปิด โมดูล 7 (Researcher) — ใต้ฝากระโปรง: มุดลงไปดูสแตก Edge AI จริงๆ (tri-core,
ai_engine, IPC model-link, TFLite-Micro → NPU) ว่า verdict ที่เราใช้ทั้งคอร์สเกิดขึ้นได้ยังไงใต้พื้น เจอกันครับ