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

Apps III: sensor fusion + IoT · เอา verdict ของโมเดล มารวมกับเซนเซอร์ดิบ แล้วสตรีมขึ้นคลาวด์

โมดูล 6 — แอป Edge AI

ต่อจากบทเรียน 6.5 — sensor fusion: verdict ของโมเดลกับเซนเซอร์ดิบ

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

IoT — ทำไมต้องสตรีมขึ้นคลาวด์

Edge AI ตัดสินใจบนบอร์ดได้เอง (นั่นคือจุดเด่น) แต่บ่อยครั้งเราอยาก รวมเหตุการณ์จากหลายบอร์ด ไว้ที่เดียว เพื่อดูภาพรวม/แจ้งเตือน/เก็บสถิติ

บอร์ด A fused event บอร์ด B fused event MQTT broker ตัวรวมข้อความ (คลาวด์) Dashboard Grafana/Node-RED แจ้งเตือนมือถือ subscribe topic
  • สำคัญ: เราสตรีมแค่ เหตุการณ์ที่สรุปแล้ว (fused event) ไม่ใช่ข้อมูลดิบทั้งสตรีม — ประหยัด bandwidth + รักษาความเป็นส่วนตัว
  • นี่คือความงามของ Edge AI + IoT: คิดที่ขอบ ส่งเฉพาะข้อสรุป ไม่ใช่ยกทุกอย่างขึ้นคลาวด์แบบ Cloud AI

ย้อนกลับไปบทเรียน 1.1–1.3: Edge AI มีอยู่เพื่อไม่ต้องอัปโหลดข้อมูลดิบ — IoT ที่ดีจึงส่งแค่ "verdict ที่ผ่านการยืนยัน" ไม่ใช่เสียง/ภาพดิบ นี่คือหลักที่เราจะยึด

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

WiFi — ต่อเน็ตด้วยโมดูล wifi

โมดูล 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())
  • ตอนรันจริงให้ใส่ SSID/รหัสของ WiFi ที่คุณใช้ (โมดูลเดียวกัน)
  • wifi.connect() ใช้เวลาสักครู่ (สแกน + จับมือ) — เรียกครั้งเดียวก่อนเข้าลูป ไม่ใช่ทุกรอบ

โมดูลนี้มีคำสั่งอื่นอีก (wifi.scan(), wifi.status(), wifi.ping()) แต่ชุดบทเรียนนี้เอาแค่ต่อให้ติดพอ — ลองโชว์ wifi.status()["rssi"] เป็นโจทย์ต่อยอด

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

MQTT — โมเดล publish / subscribe

MQTT เป็นโปรโตคอลส่งข้อความแบบ publish/subscribe ผ่านตัวกลางชื่อ broker เบามาก เหมาะกับอุปกรณ์ IoT

บอร์ด (publisher) publish(topic, payload) broker test.mosquitto.org:1883 ผู้รับ (subscriber) subscribe(topic) ส่งไป topic แจกให้คนที่ sub
  • topic = "ช่อง" ตั้งชื่อเป็นลำดับชั้น เช่น tesaiot/edge-ai/s17 — ใครสนใจก็ subscribe ช่องนี้
  • payload = เนื้อข้อความ (เราส่งเป็น JSON) — ใครก็อ่านต่อได้ ไม่ต้องรู้จักผู้ส่ง
  • broker = ตัวกลาง ชุดบทเรียนนี้ใช้ broker สาธารณะ test.mosquitto.org (ฟรี ไม่ต้องล็อกอิน)

publish/subscribe แยกผู้ส่งกับผู้รับออกจากกัน (decoupled) — บอร์ดไม่ต้องรู้ว่าใครฟังอยู่ แค่โยนข้อความเข้า topic ที่ตกลงกันไว้ นี่คือหัวใจว่าทำไม IoT ชอบ MQTT

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

โมดูล 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) — การรับเก็บไว้เป็น โจทย์ต่อยอด

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

payload JSON — โครงของเหตุการณ์

เราส่งเหตุการณ์เป็น JSON เพราะทุกระบบปลายทาง (dashboard, Node-RED, มือถือ) อ่านได้ทันที ประกอบ payload จากข้อมูล fusion:

payload = '{"event":"%s","conf":%.2f,"gyro":%.0f,"ts":%d}' % (
    TARGET_CLASS,        # เหตุการณ์อะไร (จากโมเดล)
    conf_val,            # โมเดลมั่นใจแค่ไหน
    gmag,                # เซนเซอร์ดิบยืนยันแรงเท่าไร
    time.ticks_ms(),     # ประทับเวลา (บอกลำดับเหตุการณ์)
)
  • ใส่ทั้ง verdict (event, conf) และ หลักฐานดิบ (gyro) ลงไป — ปลายทางเห็นว่าตัดสินจากอะไร
  • ประกอบ string เองด้วย % ก็พอสำหรับ payload สั้นๆ (MicroPython มี json ให้ใช้ด้วย ถ้าอยากเป๊ะ)
  • payload สั้น = ส่งเร็ว ประหยัดพลังงาน — ส่งแค่ข้อสรุป ไม่ใช่สตรีมดิบ (หลักจากสไลด์ IoT)

การใส่ทั้ง conf (จากโมเดล) และ gyro (จากเซนเซอร์ดิบ) ในข้อความเดียว = ปลายทางตรวจสอบการตัดสินใจ fusion ของเราย้อนหลังได้ นี่คือ observability ที่ดี

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

รันได้ทั้งบอร์ดและ Emulator — degrade อย่างสง่างาม

บอร์ดจริงมี 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
  • โหมด offline ยัง แสดง fusion + edge-trigger ครบ แค่พิมพ์ payload ลง console แทนส่งจริง
  • ทำให้ MVP สาธิตได้ทุกที่: กลไก fusion กับการส่งเห็นได้ทั้งบน Emulator และบอร์ด ถ้าเน็ตไม่ติดก็ยังเห็นกลไกผ่าน [SIM]

นี่คือนิสัย multi-surface ของทั้งคอร์ส (บทเรียน 1.1–1.3): โค้ดบรรทัดเดียวกันต้องรันได้ทั้ง Emulator และบอร์ด — เราแค่ห่อส่วนที่ต้องมีฮาร์ดแวร์จริงด้วยการเช็กก่อน

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

ลองเลยบนจอ 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] ในหน้าเดียว

  • ด้านบนคือ verdict ของโมเดล (label + conf) เหมือนบนบอร์ดจริง
  • ค่า gyro ดิบที่เอามา gate ก็โชว์ให้เห็นว่ามันข้ามเกณฑ์ MOTION_FLOOR หรือยัง
  • พอ fused ครบทั้งสองด่าน บรรทัด MQTT TX (หรือ [SIM] ถ้าเน็ตไม่ติด) จะพิมพ์ payload JSON ที่ถูก publish

ซ้อมบน Emulator ให้เห็น fusion กับ MQTT TX ก่อน แล้วค่อยย้ายไปบอร์ดจริงที่ต่อ WiFi ของจริงและใช้ประตู gyro — โค้ดชุดเดียวกัน เปลี่ยนแค่พื้นผิวที่รัน

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

ไล่หนึ่งรอบเต็ม — sensor → model → fuse → cloud

รวมทุกจังหวะเป็นภาพเดียว: จากเซนเซอร์บนบอร์ด จนถึงข้อความที่โผล่บน dashboard — นี่คือวงจรที่แอปชุดบทเรียนนี้เดินทุก ~250 ms

เซนเซอร์ IMU โมเดล (M55) verdict fuse + gyro ดิบ edge-trigger fired? publish mqtt / sim คลาวด์ dashboard CM55 + NPU โค้ด Python บน CM33 วนรอบใหม่ทุก ~250 ms

ครึ่งซ้าย (เซนเซอร์ → verdict) เกิดบน CM55 เราไม่แตะ · ครึ่งขวา (fuse → trigger → publish) คือโค้ด Python ของเราบน CM33 — ชุดบทเรียนนี้เราคุมครึ่งขวาทั้งหมด แล้วต่อสายออกไปถึงคลาวด์

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

โครงของไฟล์ s17_fusion_iot.py

ทั้งไฟล์อ่านเป็นประโยคเดียว: "ต่อเน็ต → หาโมเดล + สั่งรัน → วนอ่าน verdict + gyro ดิบ → พอสองสัญญาณเห็นตรงกัน ก็ publish เหตุการณ์ขึ้น MQTT → หยุดตอนออก"

ต่อ WiFi/MQTT (ให้ไว้แล้ว) select ช่อง 1 result + gyro ช่อง 2 + 3 fused? ออก→stop publish_event ช่อง 4 + 5 · ยิงครั้งเดียว no yes → publish ช่อง 4 = fused decision · ช่อง 5 = mqtt.publish ภายใน publish_event

ตัวเลขช่อง (1–5) ชี้จุดที่คุณต้องเติมในไฟล์ฝึก — ช่อง 1–2 เป็นของเดิม (ทวน) ช่อง 3–5 คือ fusion + IoT ของใหม่ของชุดบทเรียนนี้

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

ไล่โค้ด (1) — select + ต่อเน็ต

ช่องเติมที่ 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

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

ไล่โค้ด (2) — อ่าน verdict + gyro ดิบ

ช่องเติมที่ 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)
  • ช่อง 2: แทน pass ด้วย r = edge_ai.result() (เหมือนบทเรียน 1.1–1.3 / 1.6–1.7)
  • ช่อง 3: แทน pass ด้วย ax, ay, az, gx, gy, gz = sensors.bmi270.motion() — อ่านค่าดิบมาทำ gate
  • เช็ก seq ก่อนเสมอ วาดจอ + อ่านเซนเซอร์เฉพาะตอนมี verdict ใหม่

ช่อง 2 คือ "ของเดิม" ที่คุณทำเป็นแล้ว ช่อง 3 คือของใหม่ — เริ่มมีเซนเซอร์ สองเส้นทาง ในลูปเดียว: ผ่านโมเดล กับ อ่านดิบ

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

ไล่โค้ด (3) — เงื่อนไข fused

ช่องเติมที่ 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 ที่เซนเซอร์ดิบรับรอง

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

ไล่โค้ด (4) — publish ขึ้น MQTT

ช่องเติมที่ 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 ขึ้น topic
  • ห่อด้วย try/except OSError เพราะการเชื่อมต่ออาจหลุด — แอปต้องไม่ล้มเพราะเน็ตสะดุด
  • โหมด offline (เน็ตไม่ติด) ไปทาง else โชว์ [SIM] — กลไกครบ แค่ไม่ส่งจริง

เติมครบ 5 ช่องเมื่อไร คุณได้แอป Edge AI ที่ตัดสินใจแบบ fused แล้วสตรีมขึ้นคลาวด์ — นี่คือ MVP ของบทเรียน 6.5–6.6 ของชุดบทเรียนนี้

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

ลงมือทำ — เติมช่องว่างทั้ง 5 จุด

เปิด 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

ขั้นตอน:

  1. ตั้ง MODEL_KEYWORD / TARGET_CLASS / MOTION_FLOOR ที่อยากลอง (เริ่มจาก Motion/shaking/40)
  2. ไล่หา # เติม: ทีละจุด แล้วแทน pass ด้วยคำสั่งตามคำใบ้
  3. บนบอร์ด: ใส่ SSID/รหัสจริงในหัวไฟล์ · บน Emulator: WiFi จำลองต่อติดเอง และเปลี่ยนประตูเป็น accel ชั่วคราวเพราะ gyro บน Emulator ค้างใกล้ศูนย์
  4. กด Run / Program to Device แล้วเขย่าบอร์ดจริงจนเหตุการณ์ถูกส่ง

ช่อง 1–2 คือของเดิม (ทวน) · ช่อง 3–5 คือ fusion + IoT ของใหม่ — เติมครบเมื่อไร แอปของคุณ "ตัดสินใจแบบเชื่อถือได้" แล้ว "พูดคุยกับคลาวด์" ได้

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

แหล่งเรียนรู้เพิ่มเติม

อยากต่อยอดเรื่อง sensor fusion และ IoT/MQTT ลองดูของดีที่คนทำจริงเขาอธิบายไว้:

วิดีโอ (ช่องการศึกษาที่น่าเชื่อถือ)

อ่าน/ภาพอ้างอิง

วิดีโอ/ภาพภายนอกเป็นของเจ้าของต้นฉบับ ใช้เพื่อการศึกษา อ้างอิงลิงก์ต้นทาง — เราไม่ได้ฝังคลิปไว้ในสไลด์ ให้กดลิงก์ไปดูที่ต้นทางเอง

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

ชัยชนะที่เห็นได้ + MVP ของชุดบทเรียนนี้

ชัยชนะที่เห็นได้ของชุดบทเรียนนี้ · เหตุการณ์ Edge AI ที่ผ่านการยืนยันด้วยเซนเซอร์ดิบ ถูกสตรีมขึ้น MQTT broker (เห็นบน dashboard / mosquitto_sub)

MVP ของบทเรียน 6.5–6.6 (เกณฑ์ผ่านของชุดบทเรียน): คุณทำให้ การตัดสินใจแบบ fused (verdict AND gyro ดิบ) ถูก publish ขึ้น MQTT ได้จริง — บนบอร์ดหรือ Emulator (ถ้าเน็ตไม่ติดจะเห็น [SIM] บน console แทน)

  • fusion ต้องเห็นผล: เขย่าเบาๆ (โมเดลไม่มั่นใจ/gyro ไม่ถึงเกณฑ์) → ไม่ส่ง · เขย่าจริง → ส่ง
  • อธิบายได้ว่าทำไม fusion ใช้ and และทำไมต้อง edge-trigger ก่อน publish

"ส่งได้" ไม่ใช่แค่ "เห็น TX บนจอ" — คุณต้องบอกได้ว่า fusion กรอง false positive ยังไง และเหตุการณ์นี้เดินทางจากบอร์ดถึง broker ทางไหน

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

บันไดช่วยเหลือ — ใบ้ → เริ่มจากโครง → เฉลย → ฉบับเต็ม

ถ้าติด ให้ไต่บันไดนี้ทีละขั้น อย่าเพิ่งกระโดดไปดูเฉลย เพราะของจะเข้าหัวตอนที่คุณพยายามเองก่อน:

  • ใบ้ — คำใบ้อยู่ในคอมเมนต์ # เติม: ทั้ง 5 จุดในไฟล์ฝึก + ตารางหน้าที่แล้ว บอกว่าแต่ละช่องเติมอะไร
  • เริ่มจากโครง — s17_fusion_iot.py มีโครงครบทั้งไฟล์แล้ว (รวม WiFi/MQTT + degrade) เหลือแค่ 5 บรรทัดให้เติม
  • เฉลย — s17_fusion_iot.py เติมครบพร้อมคอมเมนต์อธิบายทุกช่อง (อ่านให้เข้าใจ ปิดไฟล์ แล้วพิมพ์เอง)
  • ฉบับเต็ม — s17_fusion_iot_full.py เพิ่ม gate เป็นเรดาร์ (multi-modal) + นับจำนวนเหตุการณ์ + reconnect broker + โชว์ RSSI

ลองเขียนเองให้สุดก่อนนะ ถ้าติดจริงๆ ค่อยเปิดเฉลยดูทีละช่อง แล้วกลับมาพิมพ์เอง — เดี๋ยวเราค่อย ๆ แกะไปด้วยกัน

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

เชื่อมโยงรากฐาน — วันนี้เราแตะอะไรไปบ้าง

การรวม verdict กับเซนเซอร์ดิบแล้วสตรีมขึ้นคลาวด์ ซ่อนแนวคิดที่ปิดกล่อง โมดูล 6 (Apps) พอดี:

ฝั่ง Edge AI / ระบบ

  • Sensor fusion — verdict (what) AND raw feature (how strong) → ตัดสินใจที่ทน false positive
  • Verdict → verified action — ต่อจากบทเรียน 6.3–6.4: action ยิงเมื่อ สองสัญญาณเห็นตรงกัน
  • Edge-to-cloud — ส่งเฉพาะ เหตุการณ์ที่สรุปแล้ว ไม่ใช่สตรีมดิบ (รักษา privacy + bandwidth)
  • Observability — payload พก conf + gyro ให้ปลายทางตรวจสอบการตัดสินใจย้อนหลัง

ฝั่ง MicroPython / โครงโปรแกรม

  • wifi + mqtt — ต่อเน็ต + publish/subscribe แบบ decoupled
  • degrade อย่างสง่างาม — try/except ImportError + เช็ก is_connected() → โค้ดชุดเดียวรันทุกพื้นผิว
  • เก็บกวาดตอนจบ — finally: stop() + mqtt.disconnect() คืนทุกอย่างสู่สถานะที่รู้แน่

ทั้งหมดยังยืนบนคำสั่งเดิมของ edge_ai + sensors — ชุดบทเรียนนี้แค่ต่อ "IoT layer" บนแอปที่คุณทำเป็นแล้ว

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

ใช้จริงที่ไหน — fusion + IoT ในโลกจริง

pattern "รวมสัญญาณให้เชื่อถือได้ แล้วสตรีมเหตุการณ์ขึ้นคลาวด์" คือแก่นของระบบ Edge AI + IoT จริงทุกตัว:

โรงงาน — เฝ้าเครื่องจักร โมเดลเสียง "ผิดปกติ" + IMU สั่นเกินเกณฑ์ → fused publish เหตุการณ์ → dashboard ทั้งไลน์การผลิต = verdict + raw gate + MQTT ของเราวันนี้ บ้าน — กันขโมยไม่ปลุกพลาด โมเดล motion + เรดาร์ยืนยันมีคน → ค่อยแจ้ง ส่งเฉพาะเหตุการณ์ ไม่สตรีมภาพ → privacy = corroboration gate ของเราวันนี้ เกษตร — ฝูงสัตว์/ไร่ โมเดลท่าทาง + อุณหภูมิดิบ → fused event บอร์ดหลายร้อยตัว → broker เดียว รวมภาพ = many publishers → 1 broker ของเราวันนี้ ร่วมกัน — fused event → cloud ทุกระบบ = รวมสัญญาณ + ยืนยัน + ส่งข้อสรุป ต่างกันแค่ "โมเดล + gate + topic" ปลายทาง คือ Capstone (8.1–8.2) ของเรา

สังเกตว่าทุกระบบใช้ pattern เดียวกับแอปชุดบทเรียนนี้เป๊ะ — ต่างกันแค่โมเดล, ประตูยืนยัน, และ topic ปลายทาง เราแค่กำลังเริ่มจากปลายทางเพื่อย้อนไปสร้างระบบเต็มเองได้

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0

งานทำเอง + สรุปบทเรียน

งานทำเอง (ท้ายบทเรียน):

  1. เติม s17_fusion_iot.py ให้ครบทั้ง 5 ช่อง รันได้จริงบน Emulator หรือบอร์ด
  2. หา false positive ที่ fusion กรองได้: หาท่าที่ทำให้โมเดลตอบ shaking แต่ gyro ดิบไม่ถึง MOTION_FLOOR (หรือกลับกัน) แล้วยืนยันว่า ไม่มีเหตุการณ์ถูกส่ง
  3. remix: เปลี่ยน 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 ที่เราใช้ทั้งคอร์สเกิดขึ้นได้ยังไงใต้พื้น เจอกันครับ

TESA Open Knowledge · © 2026 สมาคมสมองกลฝังตัวไทย (TESA) · ดัดแปลงจาก Edge AI Developer (รศ.วิรุฬห์ ศรีบริรักษ์, BUU) · CC BY-NC 4.0