ข้ามไปยังเนื้อหา
TESA Open Knowledge

ลงมือทำ: ส่งเหตุการณ์ที่ fuse แล้วขึ้น MQTT

โมดูล 6 — แอป Edge AI · สไลด์: slides.md · ภาพรวมโมดูล · หน้าหลักสูตร

เติมห้าจุดใน s17_fusion_iot.py ให้เลือกโมเดล อ่าน verdict อ่าน gyro ดิบ รวมสองสัญญาณด้วย AND แล้ว publish เหตุการณ์ที่ผ่านการยืนยันเป็น JSON ขึ้น MQTT broker ผ่าน WiFi ครั้งเดียวต่อเหตุการณ์ พร้อมเขียนโค้ดชุดเดียวที่ถอยเป็นโหมด [SIM] เองเมื่อไม่มีเน็ต

เป้าหมาย

หัวข้อที่มีชื่อว่า “เป้าหมาย”

เมื่อจบบทเรียนนี้ คุณจะ:

  1. เติมห้าจุดใน practice/s17_fusion_iot.py จนเหตุการณ์ fused ถูก publish ขึ้น topic ครั้งเดียวต่อการเขย่าหนึ่งครั้ง และตรวจเห็นข้อความได้จากฝั่งผู้รับ (เช่น mosquitto_sub หรือเว็บ MQTT client)
  2. หาท่าที่โมเดลตอบคลาสเป้าหมายแต่ประตูดิบไม่ผ่าน (หรือกลับกัน) แล้วยืนยันว่าไม่มีเหตุการณ์ถูกส่ง
  3. อธิบายการ degrade อย่างสง่างามด้วย try/except ImportError และ is_connected() และบอกได้ว่าทำไมควรส่งเฉพาะเหตุการณ์ที่สรุปแล้วพร้อมหลักฐานดิบใน payload

ก่อนเริ่ม

หัวข้อที่มีชื่อว่า “ก่อนเริ่ม”

ผ่านบทเรียน 6.5 มาแล้ว เข้าใจเงื่อนไข fused และ edge-trigger เตรียมโปรแกรมรับข้อความ MQTT เช่น mosquitto_sub -h test.mosquitto.org -t "tesaiot/edge-ai/s17/#" -v หรือเว็บ MQTT client ที่ต่อ broker เดียวกันได้

  • อุปกรณ์: บอร์ด TESAIoT Dev Kit ที่ลงเฟิร์มแวร์ MicroPython ของ BENTO แล้ว หรือ BENTO Emulator ใน BENTO IDE — Emulator จำลอง WiFi ให้ต่อติด และส่ง MQTT ถึง broker สาธารณะจริงผ่าน WebSocket (ต่อไม่ได้จะใช้ broker จำลอง) แต่ gyro บน Emulator ค้างใกล้ศูนย์ ซ้อมบน Emulator จึงต้องเปลี่ยนประตูเป็น accel ชั่วคราว บนบอร์ดต้องใส่ชื่อและรหัส WiFi ของคุณเอง
  • เรียนมาก่อน: บทเรียน 6.5 — sensor fusion: verdict ของโมเดลกับเซนเซอร์ดิบ

แนวคิด

หัวข้อที่มีชื่อว่า “แนวคิด”

ทั้งไฟล์อ่านเป็นประโยคเดียว: ต่อเน็ต → หาโมเดลและสั่งรัน → วนอ่าน verdict กับ gyro ดิบ → พอสองสัญญาณเห็นตรงกันก็ publish → หยุดตอนออก ส่วนต่อเน็ตให้ไว้แล้ว: import wifi, mqtt อยู่ใน try/except ImportError (ไม่มีโมดูลก็ตั้ง HAVE_NET = False), wifi.connect(WIFI_SSID, WIFI_PASS) คืน True/False และบล็อกจนต่อได้หรือหมดเวลา, wifi.ip() บอก IP, mqtt.connect(BROKER, PORT, client_id=CLIENT_ID) แล้วเช็ก mqtt.is_connected() ถ้าไม่ติดจะเข้าโหมด [SIM] ที่พิมพ์ payload ลง console แทน ห้าจุดที่เติมคือ (1) edge_ai.select(model['index']) (2) r = edge_ai.result() (3) ax, ay, az, gx, gy, gz = sensors.bmi270.motion() (4) fused = model_hit and raw_ok และ (5) mqtt.publish(TOPIC, payload) ใน publish_event() ที่ห่อ try/except OSError ไว้ เพราะ publish โยน OSError เมื่อการเชื่อมต่อหลุด

MQTT เป็น publish/subscribe ผ่าน broker ผู้ส่งไม่ต้องรู้ว่าใครฟัง payload ของเราคือ JSON สั้น ๆ {"event": ..., "conf": ..., "gyro": ..., "ts": ...} ใส่ทั้ง verdict และหลักฐานดิบให้ปลายทางตรวจการตัดสินใจย้อนหลังได้ และส่งเฉพาะเหตุการณ์ที่สรุปแล้ว ไม่ส่งสตรีมดิบ ประหยัด bandwidth และรักษาความเป็นส่วนตัว ธง fired ทำให้หนึ่งเหตุการณ์คือหนึ่งข้อความ ไม่ spam broker สาธารณะ broker สาธารณะมีคนใช้ร่วมกันมาก ให้เปลี่ยน CLIENT_ID และ TOPIC ให้มีชื่อของคุณต่อท้าย ถ้า client_id ซ้ำกัน broker จะเตะการเชื่อมต่อเก่าออก และอย่าส่งข้อมูลส่วนตัวขึ้น broker สาธารณะ บน Emulator ให้เปลี่ยนประตูชั่วคราวเป็น gmag = abs(ax) + abs(ay) + abs(az - 9.81) กับ MOTION_FLOOR = 5.0 เพราะแผง HW ขยับเฉพาะ accel แล้วกดปุ่ม Shake บนบอร์ดจริงใช้ประตู gyro ตามเดิม

ตัวอย่างสมบูรณ์

หัวข้อที่มีชื่อว่า “ตัวอย่างสมบูรณ์”

s17_fusion_iot_full.py ให้เลือกประตูได้สองแบบด้วย GATE_MODE: "imu" (gyro ดิบ แบบ corroboration) หรือ "radar" (มีคนอยู่ไหม แบบ multi-modal คนละเซนเซอร์กับโมเดล บน Emulator ปุ่ม Shake ทำให้ presence เป็นจริง) นับจำนวนเหตุการณ์ ต่อ broker ใหม่อัตโนมัติเมื่อหลุด และแสดง RSSI จาก wifi.status()["rssi"]

ไฟล์ ไฟล์นี้สอน
examples/s17_fusion_iot_full.py fusion + IoT ฉบับขัดเรียบร้อย

ฝึกเติม

หัวข้อที่มีชื่อว่า “ฝึกเติม”

คอมเมนต์ # เติม: อยู่ที่บรรทัด 127 (select), 142 (result), 151 (อ่าน motion()), 162 (fused) และ 115 (mqtt.publish ใน publish_event) ถ้า gyro ค้าง 0 ตลอดบนบอร์ด จุดที่ 151 ยังว่าง ถ้าเห็นบรรทัด MQTT TX แต่ฝั่งผู้รับไม่เห็นอะไร ตรวจจุดที่ 115 และชื่อ topic ทั้งสองฝั่ง

ไฟล์ฝึก เรื่อง
practice/s17_fusion_iot.py รวม verdict ของโมเดลกับเซนเซอร์ดิบ แล้วสตรีมขึ้นคลาวด์ (ฉบับฝึกเติมโค้ด)

เฉลย

หัวข้อที่มีชื่อว่า “เฉลย”

เปิดเฉลยหลังจากลองเองแล้วอย่างน้อยหนึ่งรอบ และอ่าน วิธีใช้เฉลย ก่อน

เฉลย คู่กับ
solution/s17_fusion_iot.py practice/s17_fusion_iot.py

เช็กความเข้าใจ

หัวข้อที่มีชื่อว่า “เช็กความเข้าใจ”

คำถามชุดเดียวกันอยู่ใน quiz.yaml สำหรับระบบที่ตรวจอัตโนมัติ

  1. เขย่าแรงหนึ่งครั้งแต่ฝั่งผู้รับเห็นข้อความสิบกว่าข้อความ ส่วนใดของโค้ดหายไป (เลือกหนึ่งข้อ · เป้าหมายข้อ 1)

    • ก) try/except OSError
    • ข) ธง fired ที่ยิงเฉพาะขอบขาขึ้นของ fused
    • ค) wifi.ip()
    • ง) ประตู gyro
    เฉลย

    ข — fused เป็นจริงหลายเฟรมติดกันระหว่างเขย่า ถ้าไม่มี fired ทุกเฟรมจะถูก publish

  2. บนบอร์ด gyro บนจอค้าง 0 ตลอดและไม่มีเหตุการณ์ถูกส่งเลย จุดใดยังว่าง (เลือกหนึ่งข้อ · เป้าหมายข้อ 1)

    • ก) จุดที่ 3 อ่าน sensors.bmi270.motion()
    • ข) จุดที่ 5 mqtt.publish
    • ค) จุดที่ 1 select
    • ง) ไม่มีจุดใดผิด
    เฉลย

    ก — ค่าเริ่มต้น gx = gy = gz = 0.0 ทำให้ gmag เป็น 0 ประตูดิบจึงไม่เคยเปิด fused ก็ไม่เคยจริง

  3. ยกบอร์ดเร็ว ๆ โมเดลตอบ shaking 0.7 แต่ gmag = 25 ที่ MOTION_FLOOR = 40 เกิดอะไร (เลือกหนึ่งข้อ · เป้าหมายข้อ 2)

    • ก) ส่งเหตุการณ์ เพราะโมเดลมั่นใจ
    • ข) ไม่ส่ง เพราะประตูดิบไม่ผ่าน fusion กรอง false positive นี้ไว้
    • ค) แอปพัง
    • ง) ส่งแบบ [SIM]
    เฉลย

    ข — AND ต้องผ่านทั้งสองด่าน นี่คือตัวอย่างของ false positive ที่ fusion มีไว้กรอง

  4. ทำไมจึงห่อ import wifi, mqtt ด้วย try/except ImportError (เลือกหนึ่งข้อ · เป้าหมายข้อ 3)

    • ก) ให้ import เร็วขึ้น
    • ข) ให้โค้ดชุดเดียวรันได้แม้เฟิร์มแวร์ไม่มีโมดูลเน็ต โดยถอยเป็นโหมด offline แทนการพังตั้งแต่บรรทัด import
    • ค) เพื่อซ่อนรหัส WiFi
    • ง) ไม่จำเป็น
    เฉลย

    ข — degrade อย่างสง่างามคือเช็กก่อนใช้ส่วนที่ต้องมีเน็ต ส่วนที่เหลือของแอปยังทำงานและแสดงกลไก fusion ได้

  5. payload แบบใดเหมาะที่สุดกับหลัก edge-to-cloud ของบทเรียนนี้ (เลือกหนึ่งข้อ · เป้าหมายข้อ 3)

    • ก) สตรีม IMU ดิบ 50 ครั้งต่อวินาที
    • ข) {“event”:“shaking”,“conf”:0.92,“gyro”:180,“ts”:123456} ครั้งเดียวต่อเหตุการณ์
    • ค) ไฟล์เสียงดิบทุกวินาที
    • ง) ภาพหน้าจอทุกเฟรม
    เฉลย

    ข — คิดที่ขอบแล้วส่งเฉพาะข้อสรุปพร้อมหลักฐานดิบที่จำเป็น ประหยัด bandwidth และรักษาความเป็นส่วนตัว

แล็บ

หัวข้อที่มีชื่อว่า “แล็บ”

MVP ของชุดบทเรียน 6.5–6.6: การตัดสินใจแบบ fused (verdict AND ประตูดิบ) ถูก publish ขึ้น MQTT ได้จริง ครั้งเดียวต่อเหตุการณ์ ส่วนการขยับเบา ๆ ไม่ถูกส่ง

  • ตั้ง CLIENT_ID กับ TOPIC ให้มีชื่อของคุณ เติมไฟล์ฝึกครบห้าจุด แล้วรันบนบอร์ดหรือ Emulator (เปลี่ยนเป็นประตู accel เมื่ออยู่บน Emulator)
  • เปิดโปรแกรมรับข้อความ subscribe topic ของคุณ แล้วเขย่าจนเห็นข้อความฝั่งผู้รับ จดตัวอย่าง payload ลงบันทึกการเรียน
  • หาท่าที่ด่านใดด่านหนึ่งไม่ผ่าน แล้วยืนยันว่าไม่มีข้อความถูกส่ง
  • ถอดเน็ต (หรือใส่ชื่อ WiFi ผิด) แล้วยืนยันว่าแอปถอยเป็นโหมด [SIM] โดยไม่พัง

ไปต่อ

หัวข้อที่มีชื่อว่า “ไปต่อ”

โมดูลถัดไป (ใต้ฝากระโปรง) เราจะมุดลงไปดูสแตก Edge AI จริง ทั้ง tri-core, ai_engine, IPC model link และ TFLite-Micro บน NPU ว่า verdict ที่ใช้มาทั้งคอร์สเกิดขึ้นอย่างไร

บทเรียนถัดไป: บทเรียน 7.1 — สแตก Edge AI: tri-core, ai_engine, IPC model link และ TFLite-Micro

สะท้อนคิด

หัวข้อที่มีชื่อว่า “สะท้อนคิด”
  • ถ้าบอร์ดร้อยตัวส่งเหตุการณ์ขึ้น topic เดียวกัน คุณจะออกแบบชื่อ topic และ payload อย่างไรให้ปลายทางแยกได้
  • ข้อมูลใดไม่ควรขึ้น broker สาธารณะเด็ดขาด และคุณจะส่งมันไปที่ไหนแทน

คำถามทบทวน

ลองตอบเองก่อน แล้วค่อยเปิดดูเฉลย

  1. เขย่าแรงหนึ่งครั้งแต่ฝั่งผู้รับเห็นข้อความสิบกว่าข้อความ ส่วนใดของโค้ดหายไป (เป้าหมายข้อ 1)

    1. try/except OSError
    2. ธง fired ที่ยิงเฉพาะขอบขาขึ้นของ fused
    3. wifi.ip()
    4. ประตู gyro
    ดูเฉลย

    คำตอบ: B. ธง fired ที่ยิงเฉพาะขอบขาขึ้นของ fused

    fused เป็นจริงหลายเฟรมติดกันระหว่างเขย่า ถ้าไม่มี fired ทุกเฟรมจะถูก publish

  2. บนบอร์ด gyro บนจอค้าง 0 ตลอดและไม่มีเหตุการณ์ถูกส่งเลย จุดใดยังว่าง (เป้าหมายข้อ 1)

    1. จุดที่ 3 อ่าน sensors.bmi270.motion()
    2. จุดที่ 5 mqtt.publish
    3. จุดที่ 1 select
    4. ไม่มีจุดใดผิด
    ดูเฉลย

    คำตอบ: A. จุดที่ 3 อ่าน sensors.bmi270.motion()

    ค่าเริ่มต้น gx = gy = gz = 0.0 ทำให้ gmag เป็น 0 ประตูดิบจึงไม่เคยเปิด fused ก็ไม่เคยจริง

  3. ยกบอร์ดเร็ว ๆ โมเดลตอบ shaking 0.7 แต่ gmag = 25 ที่ MOTION_FLOOR = 40 เกิดอะไร (เป้าหมายข้อ 2)

    1. ส่งเหตุการณ์ เพราะโมเดลมั่นใจ
    2. ไม่ส่ง เพราะประตูดิบไม่ผ่าน fusion กรอง false positive นี้ไว้
    3. แอปพัง
    4. ส่งแบบ [SIM]
    ดูเฉลย

    คำตอบ: B. ไม่ส่ง เพราะประตูดิบไม่ผ่าน fusion กรอง false positive นี้ไว้

    AND ต้องผ่านทั้งสองด่าน นี่คือตัวอย่างของ false positive ที่ fusion มีไว้กรอง

  4. ทำไมจึงห่อ import wifi, mqtt ด้วย try/except ImportError (เป้าหมายข้อ 3)

    1. ให้ import เร็วขึ้น
    2. ให้โค้ดชุดเดียวรันได้แม้เฟิร์มแวร์ไม่มีโมดูลเน็ต โดยถอยเป็นโหมด offline แทนการพังตั้งแต่บรรทัด import
    3. เพื่อซ่อนรหัส WiFi
    4. ไม่จำเป็น
    ดูเฉลย

    คำตอบ: B. ให้โค้ดชุดเดียวรันได้แม้เฟิร์มแวร์ไม่มีโมดูลเน็ต โดยถอยเป็นโหมด offline แทนการพังตั้งแต่บรรทัด import

    degrade อย่างสง่างามคือเช็กก่อนใช้ส่วนที่ต้องมีเน็ต ส่วนที่เหลือของแอปยังทำงานและแสดงกลไก fusion ได้

  5. payload แบบใดเหมาะที่สุดกับหลัก edge-to-cloud ของบทเรียนนี้ (เป้าหมายข้อ 3)

    1. สตรีม IMU ดิบ 50 ครั้งต่อวินาที
    2. {"event":"shaking","conf":0.92,"gyro":180,"ts":123456} ครั้งเดียวต่อเหตุการณ์
    3. ไฟล์เสียงดิบทุกวินาที
    4. ภาพหน้าจอทุกเฟรม
    ดูเฉลย

    คำตอบ: B. {"event":"shaking","conf":0.92,"gyro":180,"ts":123456} ครั้งเดียวต่อเหตุการณ์

    คิดที่ขอบแล้วส่งเฉพาะข้อสรุปพร้อมหลักฐานดิบที่จำเป็น ประหยัด bandwidth และรักษาความเป็นส่วนตัว

อ้างอิงบทเรียนนี้

ถ้านำบทเรียนนี้ไปสอน ทำสไลด์ หรือทำเอกสารต่อ ให้อ้างอิงด้วยข้อความนี้ ถ้าดัดแปลงเนื้อหา ให้เติม (ดัดแปลง)ต่อท้ายชื่อบทเรียน

"ลงมือทำ: ส่งเหตุการณ์ที่ fuse แล้วขึ้น MQTT" จาก TESA Open Knowledge โดยสมาคมสมองกลฝังตัวไทย (Thai Embedded Systems Association: TESA) https://github.com/tesaiot/tesa-qualification-program สัญญาอนุญาต CC BY-NC 4.0

ข้อความอ้างอิงภาษาอังกฤษ: "Hands-on: send the fused event over MQTT" from TESA Open Knowledge by the Thai Embedded Systems Association (TESA), https://github.com/tesaiot/tesa-qualification-program, licensed under CC BY-NC 4.0

ลิงก์บทเรียน: https://tesaiot.github.io/tesa-qualification-program/courses/edge-ai-developer/m06-apps/l06-fusion-iot-lab/

วิธีอ้างอิง TESA ฉบับเต็ม

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · CC BY-NC 4.0

เนื้อหาเผยแพร่ภายใต้ CC BY-NC 4.0 นำไปใช้ต่อในงานที่ไม่ใช่เพื่อการค้าได้ โปรดอ้างอิงสมาคมสมองกลฝังตัวไทย (TESA) ทุกครั้ง · วิธีอ้างอิง TESA