Raspberry Pi และ IoT ด้วย Python: 100 ตัวอย่างตั้งแต่พื้นฐานถึงระดับมืออาชีพ

Raspberry Pi คือคอมพิวเตอร์ขนาดเล็กที่อ่านเซนเซอร์ สั่งรีเลย์ และส่งข้อมูลผ่านเครือข่ายได้ จึงเป็นฐานที่นิยมของโปรเจกต์ IoT (อินเทอร์เน็ตของสรรพสิ่ง) บทความนี้พาคุณจากโปรแกรม Python แรกไปจนถึงไปป์ไลน์เซนเซอร์ที่สมบูรณ์ใน 100 ตัวอย่าง คุณจะได้เรียนพื้นฐาน Python การประมวลผลข้อมูลเซนเซอร์ การควบคุมขาด้วย GPIO การอ่านเซนเซอร์ผ่าน I2C, SPI และ 1-Wire การส่งข้อมูลด้วย HTTP และ MQTT การสร้างแดชบอร์ดเว็บขนาดเล็ก การเก็บค่าใน SQLite และการรันเป็นบริการที่เริ่มทำงานเองตอนเปิดเครื่อง หมายเหตุด้านความโปร่งใส: ตัวอย่างที่รันได้บนคอมพิวเตอร์ทั่วไปถูกรันจริงและแสดงผลลัพธ์จริงหลังโค้ด ส่วนตัวอย่างที่เหลือต้องใช้ Raspberry Pi บริการเครือข่าย หรือบริการของระบบ จึงตรวจเพียงไวยากรณ์ และไม่แสดงผลลัพธ์

สิ่งที่ต้องเตรียม

Raspberry Pi รุ่น 3, 4 หรือ 5 ที่รัน Raspberry Pi OS การ์ด microSD แหล่งจ่ายไฟ และการเชื่อมต่อเครือข่าย Python 3.10 ขึ้นไปมีอยู่แล้วใน Raspberry Pi OS สำหรับตัวอย่างฮาร์ดแวร์คุณต้องมี LED ตัวต้านทาน ปุ่มกด สายจัมเปอร์ และเซนเซอร์บางตัว ตัวอย่างส่วนใหญ่รันได้บนคอมพิวเตอร์ใดก็ได้ที่มี Python 3.10 ขึ้นไป ซึ่งเป็นวิธีที่ใช้ทดสอบตัวอย่างที่รันได้

วิธีรันตัวอย่าง

บันทึกตัวอย่างเป็นไฟล์ main.py แล้วรันด้วย python3 main.py ตัวอย่างที่ระบุว่ารันได้จะแสดงผลลัพธ์จริงจากการรันบนคอมพิวเตอร์ ตัวอย่าง bash เป็นไฟล์ตั้งค่าของ Pi และไม่ได้รันที่นี่

หัวข้อที่ 1: เริ่มต้นกับ Python บน Pi

Python เป็นภาษาหลักของโปรเจกต์ Raspberry Pi เพราะอ่านง่ายและมีไลบรารีสำหรับเซนเซอร์เกือบทุกชนิด ตัวอย่างแรกครอบคลุมการแสดงผล ตัวแปร ชนิดข้อมูล และการคำนวณ

รันโปรแกรมแรก

บันทึกไฟล์ที่มีนามสกุล .py แล้วรันด้วย python3 บน Pi print พิมพ์ข้อความออกทางเทอร์มินัล และ f-string ใส่ตัวแปรในข้อความด้วย {ชื่อ} Python อ่านไฟล์จากบนลงล่าง ลำดับของบรรทัดจึงสำคัญ

ชนิดข้อมูลและความหมาย

ตัวเลขเป็น int (จำนวนเต็ม) หรือ float (ทศนิยม) ข้อความเป็น str ค่า True และ False เป็น bool และ None หมายถึงไม่มีค่า ใช้ type() เพื่อตรวจชนิดของค่า ค่าที่อ่านจากเซนเซอร์มักเป็น float ส่วนหมายเลขขามักเป็น int

การคำนวณสำหรับงานเซนเซอร์

/ ให้ผลลัพธ์เป็น float เสมอ // หารแล้วตัดส่วนทศนิยมทิ้ง และ % ให้เศษ round() จำกัดจำนวนทศนิยมที่แสดง ใช้ตัวดำเนินการเหล่านี้เพื่อแปลงค่าดิบและจัดกลุ่มค่า

ภาพประกอบ

ระบบ IoT ทำงานอย่างไร: เซนเซอร์ป้อนข้อมูลให้ Raspberry Pi, Pi กรองและเก็บข้อมูล แล้วเผยแพร่ค่าผ่านเครือข่ายไปยังบรอกเกอร์หรือบริการเว็บที่ผู้คนดูได้

ตัวอย่างที่ 1: สวัสดีจาก Pi

โปรแกรม Python แรกของคุณบน Raspberry Pi OS

print("Hello from my Raspberry Pi!")

ผลลัพธ์:

Hello from my Raspberry Pi!

ตัวอย่างที่ 2: ตัวแปรและ f-string

เก็บค่าและแสดงในข้อความ

sensor_name = "DHT22"
temperature = 27.5
print(f"{sensor_name} reads {temperature} °C")

ผลลัพธ์:

DHT22 reads 27.5 °C

ตัวอย่างที่ 3: ชนิดข้อมูล

ชนิดพื้นฐานของ Python และชื่อของมัน

for value in (42, 3.14, "pi", True, None):
    print(repr(value), type(value).__name__)

ผลลัพธ์:

42 int
3.14 float
'pi' str
True bool
None NoneType

ตัวอย่างที่ 4: การคำนวณ การหารปัดลง และการปัดเศษ

การหาร เศษ และการปัดเศษสำหรับการคำนวณจากเซนเซอร์

readings = 17
per_group = 5
print(readings / per_group)
print(readings // per_group, readings % per_group)
print(round(3.14159, 2))

ผลลัพธ์:

3.4
3 2
3.14

ตัวอย่างที่ 5: การทำงานกับข้อความ

ทำความสะอาดชื่ออุปกรณ์และตรวจคำนำหน้า

status = "  ONLINE  "
clean = status.strip().lower()
print(clean, clean.startswith("on"), len(clean))

ผลลัพธ์:

online True 6

ตัวอย่างที่ 6: การแปลงหน่วย

แปลงเซลเซียสเป็นฟาเรนไฮต์ด้วยฟังก์ชัน

def celsius_to_fahrenheit(c):
    return c * 9 / 5 + 32

for c in (0, 25, 37.5):
    print(f"{c} C = {celsius_to_fahrenheit(c):.1f} F")

ผลลัพธ์:

0 C = 32.0 F
25 C = 77.0 F
37.5 C = 99.5 F

หัวข้อที่ 2: การตัดสินใจและการวนซ้ำ

โปรเจกต์ Pi ตอบสนองต่ออินพุตของมัน เงื่อนไขกำหนดว่าจะเกิดอะไรขึ้น และลูปทำงานซ้ำ เช่นการสุ่มค่าเซนเซอร์ทุกไม่กี่วินาที

if, elif และ else

เงื่อนไขถูกตรวจจากบนลงล่าง และสาขาแรกที่เป็นจริงจะทำงาน ควรวางการตรวจที่เจาะจงที่สุดไว้ก่อน การเปรียบเทียบต่อเนื่องเช่น 20 <= t <= 25 ตรวจทั้งช่วงในบรรทัดเดียว

for, while, break และ continue

for วนซ้ำตามช่วงหรือลำดับ while วนจนกว่าเงื่อนไขจะเปลี่ยน break ออกจากลูป และ continue ข้ามรอบปัจจุบัน ซึ่งมีประโยชน์เมื่อต้องข้ามค่าที่ผิดพลาดแล้วทำงานต่อ

ค่าจริง/เท็จและคำสั่ง match

ค่าว่าง เช่นสตริงว่างหรือลิสต์ว่าง นับเป็นเท็จ คำสั่ง match ตั้งแต่ Python 3.10 จับคู่คำสั่งกับหลายรูปแบบ เป็นวิธีที่สะอาดในการจัดการคำสั่งข้อความจากแอปในโทรศัพท์

ตัวอย่างที่ 7: if, elif และ else

เลือกการทำงานจากอุณหภูมิ

temperature = 31
if temperature >= 35:
    print("Too hot: stop the fan test")
elif temperature >= 28:
    print("Warm: run the fan")
else:
    print("Comfortable")

ผลลัพธ์:

Warm: run the fan

ตัวอย่างที่ 8: ลูป for กับ range

ทำซ้ำงานตามจำนวนที่กำหนด

for minute in range(0, 15, 5):
    print(f"minute {minute}")

ผลลัพธ์:

minute 0
minute 5
minute 10

ตัวอย่างที่ 9: ลูป while

ทำซ้ำจนกว่าเงื่อนไขจะเปลี่ยน

countdown = 3
while countdown > 0:
    print("sampling in", countdown)
    countdown -= 1
print("start")

ผลลัพธ์:

sampling in 3
sampling in 2
sampling in 1
start

ตัวอย่างที่ 10: break และ continue

ข้ามค่าที่ผิดพลาด และหยุดเมื่อพบข้อขัดข้อง

for reading in [21, None, 23, 99, 24]:
    if reading is None:
        continue
    if reading > 50:
        print("sensor fault, stop")
        break
    print("ok", reading)

ผลลัพธ์:

ok 21
ok 23
sensor fault, stop

ตัวอย่างที่ 11: ค่าจริงและค่าเท็จ

ค่าใดนับว่าเป็นจริงหรือเท็จ

for value in ("", "Pi", 0, 5, [], [1]):
    print(repr(value), "is true" if value else "is false")

ผลลัพธ์:

'' is false
'Pi' is true
0 is false
5 is true
[] is false
[1] is true

ตัวอย่างที่ 12: การเปรียบเทียบต่อเนื่อง

ตรวจว่าค่าอยู่ในช่วงที่กำหนด

humidity = 55
print(40 <= humidity <= 60)
print(humidity > 60 or humidity < 40)

ผลลัพธ์:

True
False

ตัวอย่างที่ 13: นิพจน์เงื่อนไข

if-else บรรทัดเดียวที่ให้ค่ากลับมา

reading = 12
state = "ON" if reading < 15 else "OFF"
print(state)

ผลลัพธ์:

ON

ตัวอย่างที่ 14: คำสั่ง match

จับคู่คำสั่งข้อความกับการทำงาน

def describe(command):
    match command:
        case "on" | "1":
            return "turn relay on"
        case "off" | "0":
            return "turn relay off"
        case _:
            return "unknown command"

for c in ("on", "0", "blink"):
    print(c, "->", describe(c))

ผลลัพธ์:

on -> turn relay on
0 -> turn relay off
blink -> unknown command

หัวข้อที่ 3: ฟังก์ชันและไลบรารีมาตรฐาน

ฟังก์ชันตั้งชื่อให้งานที่ทำซ้ำ ไลบรารีมาตรฐานมีฟังก์ชันคณิตศาสตร์ สถิติ และเวลาอยู่แล้ว จึงแทบไม่ต้องเขียนขึ้นเอง

การนิยามฟังก์ชัน

def สร้างฟังก์ชัน และ return ส่งค่ากลับ อาร์กิวเมนต์เริ่มต้นให้ค่าที่เหมาะสมเมื่อผู้เรียกไม่ส่งมา และอาร์กิวเมนต์แบบมีชื่อทำให้การเรียกอ่านง่าย เช่น sample(4, unit='volts')

การคืนและส่งค่าหลายค่า

ฟังก์ชันคืนค่าหลายค่าเป็นทูเพิลที่ผู้เรียกแตกค่าออกได้ *args รับค่าตำแหน่งได้กี่ค่าก็ได้ และ **kwargs รับค่าแบบมีชื่อได้กี่ค่าก็ได้ เหมาะกับตัวช่วยบันทึกล็อก

lambda, map, filter และการเรียกซ้ำ

lambda คือฟังก์ชันขนาดเล็กที่ไม่มีชื่อ ใช้กับ map และ filter เพื่อแปลงลิสต์ ฟังก์ชันเรียกซ้ำเรียกตัวเองและต้องมีเงื่อนไขหยุด แฟกทอเรียลคือตัวอย่างคลาสสิก

ตัวอย่างที่ 15: ฟังก์ชันที่มีค่าเริ่มต้น

กำหนดค่าเริ่มต้นที่เหมาะสมให้ฟังก์ชัน

def sample(pin, samples=3, unit="raw"):
    return f"read pin {pin} x{samples} ({unit})"

print(sample(4))
print(sample(4, 10, unit="volts"))

ผลลัพธ์:

read pin 4 x3 (raw)
read pin 4 x10 (volts)

ตัวอย่างที่ 16: ฟังก์ชันที่คืนค่าหลายค่า

คืนค่าต่ำสุดและสูงสุดพร้อมกัน

def min_max(values):
    return min(values), max(values)

low, high = min_max([19.5, 22.0, 18.2])
print(f"low {low}, high {high}")

ผลลัพธ์:

low 18.2, high 22.0

ตัวอย่างที่ 17: *args และ **kwargs

รับค่าได้หลายตัวทั้งแบบตำแหน่งและแบบมีชื่อ

def log_event(*parts, **fields):
    text = " ".join(str(p) for p in parts)
    extra = ", ".join(f"{k}={v}" for k, v in fields.items())
    print(text, "|", extra)

log_event("door", "opened", sensor="reed-1", room="kitchen")

ผลลัพธ์:

door opened | sensor=reed-1, room=kitchen

ตัวอย่างที่ 18: lambda, map และ filter

แปลงค่าดิบจาก ADC ในบรรทัดเดียว

raw = [512, 1023, 0, 768]
volts = list(map(lambda x: round(x * 3.3 / 1023, 2), raw))
nonzero = list(filter(lambda v: v > 0, volts))
print(volts)
print(nonzero)

ผลลัพธ์:

[1.65, 3.3, 0.0, 2.48]
[1.65, 3.3, 2.48]

ตัวอย่างที่ 19: ไลบรารีมาตรฐาน: math และ statistics

ใช้โมดูลคณิตศาสตร์และสถิติที่มากับ Python

import math
import statistics

samples = [2.0, 4.0, 4.5, 3.5]
print(round(math.sqrt(16), 2), math.pi > 3.14)
print(statistics.mean(samples), statistics.median(samples))

ผลลัพธ์:

4.0 True
3.5 3.75

ตัวอย่างที่ 20: การเรียกซ้ำ

ฟังก์ชันที่เรียกตัวเอง พร้อมเงื่อนไขหยุด

def factorial(n):
    if n <= 1:
        return 1
    return n * factorial(n - 1)

print(factorial(5), factorial(10))

ผลลัพธ์:

120 3628800

หัวข้อที่ 4: โครงสร้างข้อมูลสำหรับอุปกรณ์และค่าที่วัดได้

โปรเจกต์ Pi เก็บลิสต์ของค่าที่วัดได้ ดิกชันนารีของการตั้งค่าอุปกรณ์ และเซตของชื่อที่ไม่ซ้ำ การเลือกโครงสร้างที่เหมาะทำให้โค้ดสั้นและเร็ว

ลิสต์และ comprehension

ลิสต์เก็บลำดับที่เปลี่ยนแปลงได้ List comprehension สร้างลิสต์ใหม่จากลิสต์เดิมในบรรทัดเดียว เช่นการแปลงค่าเซลเซียสทั้งลิสต์เป็นฟาเรนไฮต์

ดิกชันนารีสำหรับค่าที่มีชื่อ

ดิกชันนารีจับคู่คีย์กับค่า เหมาะกับการตั้งค่าของอุปกรณ์หนึ่งตัวหรือการวัดหนึ่งครั้งพร้อมฟิลด์ ใช้ get() เพื่ออ่านคีย์ที่อาจไม่มี และใช้ sorted() เพื่อแสดงคีย์ตามลำดับ

ทูเพิล เซต และการตัดส่วน

ทูเพิลคือกลุ่มของค่าที่คงที่ เซตเก็บรายการที่ไม่ซ้ำและรองรับการรวมและการตัดกัน การตัดส่วนเช่น values[-2:] เอาบางส่วนของลิสต์ และ [::-1] กลับลำดับ

ตัวอย่างที่ 21: ลิสต์

เพิ่ม แทรก และลบรายการ

readings = [20.1, 20.4]
readings.append(20.9)
readings.insert(0, 19.8)
readings.remove(20.4)
print(readings, len(readings), readings[-1])

ผลลัพธ์:

[19.8, 20.1, 20.9] 3 20.9

ตัวอย่างที่ 22: List comprehension

สร้างลิสต์ใหม่จากลิสต์เดิม

celsius = [0, 10, 20, 30]
fahrenheit = [c * 9 / 5 + 32 for c in celsius if c != 10]
print(fahrenheit)

ผลลัพธ์:

[32.0, 68.0, 86.0]

ตัวอย่างที่ 23: ทูเพิลและการแตกค่า

กลุ่มของค่าที่แก้ไขไม่ได้

location = ("kitchen", 2, 3.5)
room, floor, height = location
print(room, floor, height)
try:
    location[0] = "garage"
except TypeError as e:
    print("tuples cannot change:", type(e).__name__)

ผลลัพธ์:

kitchen 2 3.5
tuples cannot change: TypeError

ตัวอย่างที่ 24: ดิกชันนารี

เก็บค่าที่มีชื่อของอุปกรณ์

device = {"name": "node-1", "pin": 17, "online": True}
device["pin"] = 22
device["room"] = "garage"
print(device.get("name"), device.get("temp", "n/a"), len(device))

ผลลัพธ์:

node-1 n/a 4

ตัวอย่างที่ 25: แสดงดิกชันนารีตามลำดับ

พิมพ์จำนวนของแต่ละเหตุการณ์ตามลำดับ

counts = {"motion": 4, "door": 1, "temp": 9}
for name in sorted(counts):
    print(f"{name:<8}{counts[name]:>3}")

ผลลัพธ์:

door      1
motion    4
temp      9

ตัวอย่างที่ 26: เซต

ค่าที่ไม่ซ้ำกันและการดำเนินการของเซต

seen = {"kitchen", "garage"}
seen.add("kitchen")
new = {"garage", "attic"}
print(len(seen), sorted(seen | new), sorted(seen & new))

ผลลัพธ์:

2 ['attic', 'garage', 'kitchen'] ['garage']

ตัวอย่างที่ 27: การตัดส่วนของลิสต์

หยิบบางส่วนของลิสต์ และกลับลำดับ

samples = [1, 2, 3, 4, 5, 6, 7]
print(samples[:3], samples[-2:], samples[::2])
print(samples[::-1])

ผลลัพธ์:

[1, 2, 3] [6, 7] [1, 3, 5, 7]
[7, 6, 5, 4, 3, 2, 1]

ตัวอย่างที่ 28: ลิสต์ของดิกชันนารี

กรองรายการอุปกรณ์ที่ออนไลน์

devices = [
    {"id": "a1", "online": True},
    {"id": "b2", "online": False},
    {"id": "c3", "online": True},
]
online = [d["id"] for d in devices if d["online"]]
print("online:", online)

ผลลัพธ์:

online: ['a1', 'c3']

หัวข้อที่ 5: ข้อความ ไฟล์ และรูปแบบข้อมูล

ข้อมูลเซนเซอร์ต้องถูกบันทึกและแบ่งปัน ไฟล์ข้อความ CSV และ JSON เป็นรูปแบบที่โปรเจกต์ IoT ส่วนใหญ่ใช้ และไลบรารีมาตรฐานอ่านเขียนได้ทั้งสามแบบ

การอ่านและเขียนไฟล์

เปิดไฟล์ด้วยบล็อก with เพื่อให้ปิดอัตโนมัติ และใช้ encoding='utf-8' สำหรับข้อความ อ็อบเจกต์ Path จาก pathlib ใช้ได้กับทุกระบบปฏิบัติการและต่อชื่อไฟล์ได้ง่าย

CSV และ JSON

โมดูล csv เขียนแถวพร้อมหัวตาราง ซึ่งโปรแกรมสเปรดชีตเข้าใจ โมดูล json แปลงดิกชันนารีเป็นข้อความสำหรับบริการเว็บและข้อความ MQTT และแปลงกลับได้

จัดรูปแบบตารางและวันที่

ความกว้างรูปแบบเช่น {name:<10} จัดคอลัมน์ให้ตรงในเทอร์มินัล datetime จัดรูปแบบเวลาด้วย strftime และบวกระยะเวลาด้วย timedelta จึงจัดกลุ่มค่าตามเวลาได้

ตัวอย่างที่ 29: จัดรูปแบบผลลัพธ์เป็นตาราง

จัดคอลัมน์ให้ตรงด้วยความกว้างของรูปแบบ

rows = [("Temp", 27.456), ("Humidity", 55.0), ("Pressure", 1013.2)]
for name, value in rows:
    print(f"{name:<10}{value:>10.1f}")

ผลลัพธ์:

Temp            27.5
Humidity        55.0
Pressure      1013.2

ตัวอย่างที่ 30: อ่านและเขียนไฟล์ข้อความ

บันทึกและโหลดค่าจากเซนเซอร์ลงไฟล์

from pathlib import Path

log = Path("sensor_log.txt")
with log.open("w", encoding="utf-8") as f:
    f.write("21.5\n22.0\n22.4\n")
with log.open("r", encoding="utf-8") as f:
    values = [float(line) for line in f]
log.unlink()
print("read", len(values), "values, last", values[-1])

ผลลัพธ์:

read 3 values, last 22.4

ตัวอย่างที่ 31: ไฟล์ CSV

เขียนและอ่านตารางของค่าที่วัดได้

import csv
from pathlib import Path

path = Path("readings.csv")
with path.open("w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerow(["time", "temp_c"])
    writer.writerows([["08:00", 21.5], ["08:05", 21.9]])
with path.open(newline="", encoding="utf-8") as f:
    for row in csv.DictReader(f):
        print(row["time"], float(row["temp_c"]) + 1)
path.unlink()

ผลลัพธ์:

08:00 22.5
08:05 22.9

ตัวอย่างที่ 32: JSON

แปลงข้อมูลเป็นข้อความ JSON และกลับกัน

import json

payload = {"device": "node-1", "readings": [21.5, 21.9], "ok": True}
text = json.dumps(payload, indent=2)
print(text)
print(json.loads(text)["readings"][1])

ผลลัพธ์:

{
  "device": "node-1",
  "readings": [
    21.5,
    21.9
  ],
  "ok": true
}
21.9

ตัวอย่างที่ 33: พาธ

จัดการชื่อไฟล์และโฟลเดอร์อย่างปลอดภัย

from pathlib import PurePosixPath

path = PurePosixPath("/home/pi/projects/iot/config.json")
print(path.name, path.suffix, path.parent)
print(path.parent / "logs")

ผลลัพธ์:

config.json .json /home/pi/projects/iot
/home/pi/projects/iot/logs

ตัวอย่างที่ 34: วันที่และเวลา

จัดรูปแบบและบวกค่าวันที่กับเวลา

from datetime import datetime, timedelta

start = datetime(2026, 1, 15, 8, 0)
later = start + timedelta(minutes=90)
print(start.strftime("%Y-%m-%d %H:%M"), "->", later.strftime("%H:%M"))

ผลลัพธ์:

2026-01-15 08:00 -> 09:30

หัวข้อที่ 6: ข้อผิดพลาด การบันทึกล็อก และการจับเวลา

อุปกรณ์ทำงานเป็นเดือนโดยไม่มีคนดู ต้องรับมือกับค่าที่ผิดพลาด บันทึกสิ่งที่เกิดขึ้น และวัดเวลาของตัวเอง

จัดการข้อผิดพลาดด้วย try และ except

ใส่โค้ดที่อาจล้มเหลวไว้ใน try จับข้อยกเว้นที่เจาะจง และทำงานต่อ finally ทำงานไม่ว่าจะเกิดข้อผิดพลาดหรือไม่ จึงเป็นที่เหมาะสำหรับปิดเอาต์พุตหรือปิดไฟล์

การบันทึกล็อกแทน print

โมดูล logging บันทึกข้อความพร้อมระดับ เช่น INFO หรือ WARNING และคุณเลือกระดับที่จะแสดงได้ ล็อกส่งลงไฟล์ได้ ซึ่งจำเป็นเมื่อ Pi ทำงานโดยไม่มีจอ

การตรวจเชิงป้องกันและการจับเวลา

assert ปฏิเสธค่าที่เป็นไปไม่ได้ตั้งแต่ระหว่างพัฒนา time.perf_counter วัดเวลาที่ผ่านไปอย่างแม่นยำ ซึ่งช่วยหาโค้ดที่ช้าบนคอมพิวเตอร์ขนาดเล็ก

ตัวอย่างที่ 35: try และ except

จัดการค่าที่ผิดโดยไม่ให้โปรแกรมหยุด

def parse_temperature(text):
    try:
        return float(text)
    except ValueError:
        print(f"bad reading: {text!r}")
        return None

print(parse_temperature("23.4"))
print(parse_temperature("23,4"))

ผลลัพธ์:

23.4
bad reading: '23,4'
None

ตัวอย่างที่ 36: finally และการเก็บกวาด

เก็บกวาดเสมอแม้เกิดข้อผิดพลาด

from pathlib import Path

target = Path("demo_state.txt")
try:
    with target.open("w", encoding="utf-8") as f:
        f.write("on")
    print("saved")
finally:
    target.unlink(missing_ok=True)
    print("cleaned up")

ผลลัพธ์:

saved
cleaned up

ตัวอย่างที่ 37: ข้อยกเว้นที่กำหนดเอง

แจ้งข้อผิดพลาดที่ชัดเจนสำหรับค่าเซนเซอร์ที่เป็นไปไม่ได้

class SensorError(Exception):
    """Raised when a sensor gives an impossible value."""

def check(value):
    if value < -40 or value > 85:
        raise SensorError(f"value {value} is outside the sensor range")
    return value

try:
    check(120)
except SensorError as e:
    print("sensor error:", e)

ผลลัพธ์:

sensor error: value 120 is outside the sensor range

ตัวอย่างที่ 38: การบันทึกล็อก

บันทึกข้อความพร้อมระดับ แทนการใช้ print

import logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("node-1")
log.info("started")
log.warning("battery at 18%")
log.debug("this line is hidden at INFO level")

ผลลัพธ์:

ตัวอย่างที่ 39: การตรวจสอบเชิงป้องกันด้วย assert

ปฏิเสธข้อมูลที่ไม่ถูกต้องตั้งแต่ต้น

def set_pwm(duty):
    assert 0 <= duty <= 100, "duty must be 0 to 100"
    return duty

print(set_pwm(50))
try:
    set_pwm(150)
except AssertionError as e:
    print("rejected:", e)

ผลลัพธ์:

50
rejected: duty must be 0 to 100

ตัวอย่างที่ 40: การวัดเวลาที่ใช้

จับเวลาโค้ดด้วยนาฬิกาที่แม่นยำ

import time

start = time.perf_counter()
total = sum(i * i for i in range(200_000))
elapsed = time.perf_counter() - start
print("sum of squares:", total)
print("took measurable time:", elapsed >= 0)

ผลลัพธ์:

sum of squares: 2666646666700000
took measurable time: True

หัวข้อที่ 7: การประมวลผลข้อมูลเซนเซอร์

ค่าดิบจากเซนเซอร์มีสัญญาณรบกวนและมักเป็นหน่วยที่ไม่ถูกต้อง ตัวอย่างเหล่านี้แปลงค่าดิบเป็นค่าที่ใช้ได้ คือปรับเทียบ ทำให้เรียบ ตรวจค่าผิดปกติ และเทียบกับเกณฑ์

การปรับเทียบและการแปลงหน่วย

ADC ให้ค่าดิบเป็นจำนวนนับ จุดอ้างอิงที่รู้ค่าสองจุดให้การปรับเทียบเป็นเส้นตรง ทำให้แปลงจำนวนนับเป็นโวลต์ได้ บันทึกค่าปรับเทียบไว้กับอุปกรณ์ เพราะเซนเซอร์แต่ละตัวต่างกัน

การทำให้เรียบและฮิสเทอรีซิส

ค่าเฉลี่ยเคลื่อนที่ของค่าล่าสุดไม่กี่ค่าช่วยลดค่าที่พุ่งขึ้นผิดปกติ ฮิสเทอรีซิสใช้เกณฑ์สองค่า ค่าหนึ่งสำหรับเปิดและอีกค่าสำหรับปิด เพื่อไม่ให้รีเลย์สั่นเมื่อค่าอยู่ใกล้เกณฑ์

สถิติและค่าผิดปกติ

ค่าเฉลี่ย มัธยฐาน และส่วนเบี่ยงเบนมาตรฐานอธิบายช่วงของการวัด มัธยฐานไม่สนใจค่าที่สุดโต่ง จึงเป็นวิธีที่ดีในการตัดค่าพุ่งที่ผิดก่อนหาค่าเฉลี่ย

ตัวอย่างที่ 41: การปรับเทียบค่า ADC ดิบ

แปลงค่าดิบเป็นโวลต์ด้วยจุดอ้างอิงสองจุด

RAW_MIN, RAW_MAX = 102, 921      # readings at 0 V and at full scale
VOLT_MAX = 3.3

def to_volts(raw):
    return (raw - RAW_MIN) * VOLT_MAX / (RAW_MAX - RAW_MIN)

for raw in (102, 512, 921):
    print(raw, "->", round(to_volts(raw), 3), "V")

ผลลัพธ์:

102 -> 0.0 V
512 -> 1.652 V
921 -> 3.3 V

ตัวอย่างที่ 42: ตัวกรองค่าเฉลี่ยเคลื่อนที่

ทำให้ค่าที่สั่นเรียบขึ้นในช่วงของการวัด

from collections import deque

window = deque(maxlen=3)
def smooth(value):
    window.append(value)
    return sum(window) / len(window)

for reading in [20.0, 25.0, 21.0, 40.0, 22.0]:
    print(f"raw {reading:5.1f}  smoothed {smooth(reading):6.2f}")

ผลลัพธ์:

raw  20.0  smoothed  20.00
raw  25.0  smoothed  22.50
raw  21.0  smoothed  22.00
raw  40.0  smoothed  28.67
raw  22.0  smoothed  27.67

ตัวอย่างที่ 43: ฮิสเทอรีซิสสำหรับรีเลย์พัดลม

ป้องกันรีเลย์เปิดปิดถี่เกินไป

def relay_state(temp, currently_on, on_at=24.0, off_at=22.0):
    if not currently_on and temp >= on_at:
        return True
    if currently_on and temp <= off_at:
        return False
    return currently_on

state = False
for t in [21.0, 23.0, 24.5, 23.5, 22.5, 21.9]:
    state = relay_state(t, state)
    print(t, "fan", "ON" if state else "OFF")

ผลลัพธ์:

21.0 fan OFF
23.0 fan OFF
24.5 fan ON
23.5 fan ON
22.5 fan ON
21.9 fan OFF

ตัวอย่างที่ 44: หน่วยความดันและการปัดเศษ

แปลง hPa เป็น inHg และปัดเศษให้ถูกต้อง

def hpa_to_inhg(hpa):
    return hpa * 0.02953

pressure = 1013.25
print(f"{pressure} hPa = {hpa_to_inhg(pressure):.2f} inHg")
print(round(2.5), round(3.5))

ผลลัพธ์:

1013.25 hPa = 29.92 inHg
2 4

ตัวอย่างที่ 45: สถิติในช่วงของการวัด

ค่าเฉลี่ย การกระจาย และช่วงของค่าล่าสุด

import statistics

window = [21.4, 21.6, 21.5, 21.9, 21.7]
print("mean", round(statistics.mean(window), 2))
print("stdev", round(statistics.pstdev(window), 3))
print("range", round(max(window) - min(window), 2))

ผลลัพธ์:

mean 21.62
stdev 0.172
range 0.5

ตัวอย่างที่ 46: ตัดค่าผิดปกติด้วยค่ามัธยฐาน

ไม่สนใจค่าที่ห่างจากค่าปกติมาก

import statistics

def remove_outliers(values, tolerance=3.0):
    med = statistics.median(values)
    return [v for v in values if abs(v - med) <= tolerance]

raw = [21.5, 21.7, 85.0, 21.6, -40.0, 21.8]
print(remove_outliers(raw))

ผลลัพธ์:

[21.5, 21.7, 21.6, 21.8]

ตัวอย่างที่ 47: อัตราการเปลี่ยนแปลง

ค่ากำลังเพิ่มขึ้นเท่าไรต่อนาที

def rate_per_minute(prev_value, prev_minute, value, minute):
    if minute == prev_minute:
        return 0.0
    return (value - prev_value) / (minute - prev_minute)

print(round(rate_per_minute(20.0, 0, 21.2, 4), 2), "C per minute")

ผลลัพธ์:

0.3 C per minute

ตัวอย่างที่ 48: ค่าที่วัดได้เป็นคลาสข้อมูล

อธิบายการวัดหนึ่งครั้งด้วยคลาสเล็ก ๆ

from dataclasses import dataclass, asdict

@dataclass
class Reading:
    device: str
    temperature_c: float
    humidity_pct: float

r = Reading("node-1", 21.5, 48.0)
print(r)
print(asdict(r))

ผลลัพธ์:

Reading(device='node-1', temperature_c=21.5, humidity_pct=48.0)
{'device': 'node-1', 'temperature_c': 21.5, 'humidity_pct': 48.0}

หัวข้อที่ 8: GPIO: ควบคุม LED ปุ่มกด มอเตอร์ และรีเลย์

ขา 40 ขาของ Pi ใช้สั่งเอาต์พุตและอ่านอินพุตได้ ตัวอย่างแสดง LED ปุ่มกด การหรี่ไฟด้วย PWM เซอร์โว เซนเซอร์ระยะทาง และรีเลย์ ตรวจการต่อสายกับแผนผังขาให้ถูกต้องก่อนจ่ายไฟเสมอ

เอาต์พุต อินพุต และตัวต้านทานแบบพูลอัพ

ตั้งขาเป็นเอาต์พุตเพื่อขับ LED ผ่านตัวต้านทาน หรือเป็นอินพุตเพื่ออ่านปุ่มกด ตัวต้านทานพูลอัพทำให้อินพุตเป็นสูงจนกว่าปุ่มจะต่อลงกราวด์ ขาจึงไม่ลอย

PWM และเซอร์โว

การมอดูเลตความกว้างพัลส์เปิดปิดขาอย่างรวดเร็ว สัดส่วนของเวลาที่เปิดกำหนดความสว่างหรือตำแหน่งของเซอร์โว ไลบรารี gpiozero ซ่อนงานการจับเวลาส่วนใหญ่ไว้

รีเลย์ เซนเซอร์ และการปิดอย่างปลอดภัย

รีเลย์สวิตช์อุปกรณ์กำลังสูง เช่นปั๊ม ดังนั้น Pi ขับเพียงขดลวดควบคุมเล็ก ๆ ปิดเอาต์พุตใน finally เสมอ และเรียก cleanup เพื่อให้เกิดข้อขัดข้องก็ไม่ทำให้มอเตอร์ทำงานค้าง

ตัวอย่างที่ 49: กระพริบ LED (GPIO)

เปิดปิด LED จากขา GPIO ของ Raspberry Pi

import time
import RPi.GPIO as GPIO

LED_PIN = 17
GPIO.setmode(GPIO.BCM)
GPIO.setup(LED_PIN, GPIO.OUT)

try:
    for _ in range(5):
        GPIO.output(LED_PIN, GPIO.HIGH)
        time.sleep(0.5)
        GPIO.output(LED_PIN, GPIO.LOW)
        time.sleep(0.5)
finally:
    GPIO.cleanup()

ตัวอย่างนี้ต้องใช้ Raspberry Pi บริการเครือข่าย หรือบริการระบบ จึงตรวจเพียงไวยากรณ์ ไม่แสดงผลลัพธ์

ตัวอย่างที่ 50: อ่านปุ่มกดด้วย pull-up

ตรวจการกดปุ่มอย่างปลอดภัยด้วยการหน่วงเวลาดีบาวน์

import time
import RPi.GPIO as GPIO

BUTTON_PIN = 27
GPIO.setmode(GPIO.BCM)
GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)

try:
    while True:
        if GPIO.input(BUTTON_PIN) == GPIO.LOW:   # pressed connects the pin to GND
            print("button pressed")
            time.sleep(0.2)                      # simple debounce
        time.sleep(0.01)
finally:
    GPIO.cleanup()

ตัวอย่างที่ 51: ปรับความสว่าง LED ด้วย PWM

เปลี่ยนความสว่างของ LED ด้วยการมอดูเลตความกว้างพัลส์

import time
import RPi.GPIO as GPIO

GPIO.setmode(GPIO.BCM)
GPIO.setup(18, GPIO.OUT)
pwm = GPIO.PWM(18, 1000)          # 1 kHz
pwm.start(0)
try:
    for duty in range(0, 101, 10):
        pwm.ChangeDutyCycle(duty)
        time.sleep(0.1)
finally:
    pwm.stop()
    GPIO.cleanup()

ตัวอย่างที่ 52: LED ด้วย gpiozero

ไลบรารี GPIO ที่ง่ายและปลอดภัยกว่าสำหรับ LED

from gpiozero import LED
from time import sleep

led = LED(17)
for _ in range(3):
    led.on()
    sleep(0.5)
    led.off()
    sleep(0.5)

ตัวอย่างที่ 53: ฟังก์ชันตอบกลับเมื่อกดปุ่ม

ตอบสนองการกดปุ่มโดยไม่ต้องวนตรวจ

from gpiozero import Button
from signal import pause

button = Button(27)
button.when_pressed = lambda: print("pressed")
button.when_released = lambda: print("released")
pause()

ตัวอย่างที่ 54: ควบคุมเซอร์โว

หมุนเซอร์โวไปยังตำแหน่งที่กำหนด

from gpiozero import Servo
from time import sleep

servo = Servo(12)
for position in (-1, 0, 1, 0):   # -1 = left, 0 = centre, 1 = right
    servo.value = position
    sleep(1)

ตัวอย่างที่ 55: เซนเซอร์วัดระยะอัลตราโซนิก

วัดระยะด้วยเซนเซอร์ HC-SR04

import time
import RPi.GPIO as GPIO

TRIG, ECHO = 23, 24
GPIO.setmode(GPIO.BCM)
GPIO.setup(TRIG, GPIO.OUT)
GPIO.setup(ECHO, GPIO.IN)

def distance_cm():
    GPIO.output(TRIG, False)
    time.sleep(0.05)
    GPIO.output(TRIG, True)
    time.sleep(0.00001)
    GPIO.output(TRIG, False)
    while GPIO.input(ECHO) == 0:
        start = time.time()
    while GPIO.input(ECHO) == 1:
        end = time.time()
    return (end - start) * 34300 / 2   # speed of sound, there and back

print(round(distance_cm(), 1), "cm")
GPIO.cleanup()

ตัวอย่างที่ 56: รีเลย์พร้อมค่าเริ่มต้นที่ปลอดภัย

สลับปั๊มอย่างปลอดภัยและปิดเสมอเมื่อจบ

from gpiozero import OutputDevice

relay = OutputDevice(22, active_high=True, initial_value=False)
relay.on()     # power the pump
relay.off()    # always end with the relay off

ตัวอย่างที่ 57: เสียงบัซเซอร์

เล่นเสียงโทนง่าย ๆ บนบัซเซอร์

from gpiozero import TonalBuzzer
from time import sleep

buzzer = TonalBuzzer(26)
for note in (0.0, 0.5, 1.0):
    buzzer.play(note)
    sleep(0.3)
buzzer.stop()

ตัวอย่างที่ 58: ปิด GPIO อย่างปลอดภัย

ทำให้เอาต์พุตปลอดภัยก่อนโปรแกรมจบ

import RPi.GPIO as GPIO

GPIO.setmode(GPIO.BCM)
GPIO.setup(17, GPIO.OUT)
try:
    GPIO.output(17, GPIO.HIGH)
    raise RuntimeError("simulated fault")
except RuntimeError as e:
    print("error:", e)
finally:
    GPIO.output(17, GPIO.LOW)   # make the output safe before cleanup
    GPIO.cleanup()

หัวข้อที่ 9: เซนเซอร์ผ่าน I2C, SPI, 1-Wire และกล้อง

เซนเซอร์หลายตัวสื่อสารกับ Pi ผ่านบัส I2C ใช้สองสายและที่อยู่ SPI ใช้สี่สายและสายเลือก 1-Wire ใช้สายข้อมูลหนึ่งสาย โมดูลกล้องเป็นกรณีพิเศษที่มีไลบรารีของตัวเอง

I2C: สแกนและอ่านรีจิสเตอร์

อุปกรณ์ I2C แต่ละตัวมีที่อยู่ สแกนบัสก่อนเพื่อหาว่ามีอะไรต่ออยู่ อ่านรีจิสเตอร์รหัสชิปเพื่อยืนยันเซนเซอร์ จากนั้นเขียนรีจิสเตอร์ควบคุมและอ่านไบต์ข้อมูล

SPI และอินพุตอนาล็อก

Pi ไม่มีอินพุตอนาล็อก ADC เช่น MCP3008 จึงแปลงแรงดันผ่าน SPI ส่งหมายเลขช่อง แล้วประกอบค่า 10 บิตจากไบต์ที่ตอบกลับ

1-Wire, เซนเซอร์ DHT และกล้อง

DS18B20 บน 1-Wire ปรากฏเป็นไฟล์ใน /sys/bus/w1 ต้องเปิดการใช้งานอินเทอร์เฟซใน raspi-config ก่อน เซนเซอร์ DHT บางครั้งล้มเหลว จึงควรลองใหม่ โมดูลกล้องถ่ายภาพนิ่งและเฟรมด้วย picamera2

ตัวอย่างที่ 59: สแกนบัส I2C

หาว่ามีเซนเซอร์ใดเชื่อมต่ออยู่

from smbus2 import SMBus

with SMBus(1) as bus:
    found = []
    for address in range(0x03, 0x78):
        try:
            bus.read_byte(address)
            found.append(hex(address))
        except OSError:
            pass
print("devices:", found)

ตัวอย่างที่ 60: อ่านเซนเซอร์ความดัน BMP280

ตรวจรหัสชิปและอ่านไบต์ความดันดิบผ่าน I2C

from smbus2 import SMBus

ADDRESS = 0x76
with SMBus(1) as bus:
    chip_id = bus.read_byte_data(ADDRESS, 0xD0)
    print("chip id", hex(chip_id), "(expect 0x58)")
    bus.write_byte_data(ADDRESS, 0xF4, 0b00100111)   # normal mode, oversampling x1
    raw = bus.read_i2c_block_data(ADDRESS, 0xF7, 6)
print("raw bytes", raw)

ตัวอย่างที่ 61: อุณหภูมิและความชื้นด้วย DHT22

อ่าน DHT22 และลองใหม่เมื่อยังไม่พร้อม

import time
import board
import adafruit_dht

dht = adafruit_dht.DHT22(board.D4)
for _ in range(5):
    try:
        print(f"{dht.temperature:.1f} C  {dht.humidity:.0f} %")
    except RuntimeError as e:
        print("sensor not ready:", e)   # DHT sensors fail sometimes; just retry
    time.sleep(2)

ตัวอย่างที่ 62: อินพุตอนาล็อกด้วย MCP3008

อ่านแรงดันอนาล็อกผ่าน SPI

import spidev

spi = spidev.SpiDev()
spi.open(0, 0)                  # bus 0, device 0
spi.max_speed_hz = 1_350_000

def read_channel(channel):
    cmd = [1, (8 + channel) << 4, 0]
    reply = spi.xfer2(cmd)
    return ((reply[1] & 3) << 8) | reply[2]   # 10-bit value, 0 to 1023

print("channel 0 raw:", read_channel(0))
spi.close()

ตัวอย่างที่ 63: อุณหภูมิ DS18B20 ผ่าน 1-Wire

อ่านเซนเซอร์อุณหภูมิกันน้ำจากเคอร์เนล

from pathlib import Path
import glob

device_dirs = glob.glob("/sys/bus/w1/devices/28-*")
if not device_dirs:
    raise SystemExit("no DS18B20 found: enable 1-Wire in raspi-config")
lines = (Path(device_dirs[0]) / "w1_slave").read_text().splitlines()
if lines[0].endswith("YES"):
    milli = int(lines[1].split("t=")[1])
    print(f"{milli / 1000:.3f} C")

ตัวอย่างที่ 64: ลองอ่านซ้ำสำหรับเซนเซอร์ที่ไม่เสถียร

อ่านใหม่อีกครั้งก่อนจะยอมแพ้

import time

def read_with_retries(read_fn, attempts=3, delay=1.0):
    for attempt in range(1, attempts + 1):
        try:
            return read_fn()
        except OSError as e:
            print(f"attempt {attempt} failed: {e}")
            time.sleep(delay)
    raise RuntimeError("sensor unavailable after retries")

ตัวอย่างที่ 65: บันทึกค่าปรับเทียบในไฟล์ JSON

เก็บค่าปรับเทียบไว้ในไฟล์และโหลดกลับมาได้

import json
from pathlib import Path

CONFIG = Path("calibration.json")
DEFAULTS = {"offset_c": 0.0, "scale": 1.0}

def load():
    if CONFIG.exists():
        return {**DEFAULTS, **json.loads(CONFIG.read_text(encoding="utf-8"))}
    return dict(DEFAULTS)

def save(cfg):
    CONFIG.write_text(json.dumps(cfg, indent=2), encoding="utf-8")

cfg = load()
cfg["offset_c"] = -0.4
save(cfg)
print("saved:", load())
CONFIG.unlink()

ผลลัพธ์:

saved: {'offset_c': -0.4, 'scale': 1.0}

ตัวอย่างที่ 66: ถ่ายภาพนิ่ง

บันทึกภาพด้วยโมดูลกล้องของ Pi

from picamera2 import Picamera2
from time import sleep

cam = Picamera2()
cam.configure(cam.create_still_configuration())
cam.start()
sleep(2)                      # let the auto-exposure settle
cam.capture_file("photo.jpg")
cam.stop()

ตัวอย่างที่ 67: ตรวจจับการเคลื่อนไหวจากเฟรมกล้อง

เปรียบเทียบเฟรมเพื่อดูการเคลื่อนไหว

import numpy as np
from picamera2 import Picamera2

cam = Picamera2()
cam.configure(cam.create_preview_configuration(main={"size": (320, 240), "format": "YUV420"}))
cam.start()
previous = None
for _ in range(20):
    frame = cam.capture_array()[:240].astype(np.int16)
    if previous is not None and np.mean(np.abs(frame - previous)) > 10:
        print("motion")
    previous = frame
cam.stop()

หัวข้อที่ 10: เครือข่ายด้วยซ็อกเก็ตและ HTTP

อุปกรณ์ IoT ต้องคุยกับคอมพิวเตอร์เครื่องอื่น ซ็อกเก็ตเป็นพื้นฐานระดับต่ำ HTTP คือวิธีที่บริการเว็บและ API ทำงาน และโค้ดที่แข็งแรงจัดการการหมดเวลา การลองใหม่ และช่วงที่ออฟไลน์

ซ็อกเก็ต TCP และ UDP

ซ็อกเก็ต TCP เป็นการเชื่อมต่อที่เชื่อถือได้ ส่วน UDP ส่งข้อความเดี่ยวโดยไม่มีการเชื่อมต่อ การ bind ที่พอร์ต 0 ให้ระบบปฏิบัติการเลือกพอร์ตว่างเอง ทำให้ตัวอย่างไม่ชนกัน

คำขอ HTTP พร้อมการหมดเวลา

requests ทำการเรียก HTTP ได้ในบรรทัดเดียว ควรตั้ง timeout เสมอ มิฉะนั้นเครือข่ายที่ใช้ไม่ได้จะทำให้โปรแกรมค้าง raise_for_status แปลงการตอบกลับที่เป็นข้อผิดพลาดให้เป็นข้อยกเว้นที่จัดการได้

การลองใหม่ การถอยหลัง และคิวออฟไลน์

ลองคำขอที่ล้มเหลวใหม่ด้วยการถอยหลังแบบทวีคูณ โดยรอนานขึ้นทุกครั้ง เพื่อไม่ให้ถล่มเซิร์ฟเวอร์ เมื่อเครือข่ายใช้ไม่ได้ ให้เก็บค่าไว้ในคิวและส่งเมื่อกลับมาใช้ได้

ตัวอย่างที่ 68: เซิร์ฟเวอร์และไคลเอ็นต์ TCP แบบ echo

ส่งไบต์ไปยังเซิร์ฟเวอร์และอ่านคำตอบ

import socket
import threading

def serve(server):
    conn, _ = server.accept()
    with conn:
        data = conn.recv(1024)
        conn.sendall(b"echo: " + data)

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(("127.0.0.1", 0))            # port 0: let the OS choose a free port
server.listen(1)
port = server.getsockname()[1]
thread = threading.Thread(target=serve, args=(server,))
thread.start()

with socket.create_connection(("127.0.0.1", port)) as client:
    client.sendall(b"temp=21.5")
    print(client.recv(1024).decode())
thread.join()
server.close()

ผลลัพธ์:

echo: temp=21.5

ตัวอย่างที่ 69: ข้อความ UDP บน loopback

ส่งดาต้าแกรมโดยไม่ต้องสร้างการเชื่อมต่อ

import socket

receiver = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
receiver.bind(("127.0.0.1", 0))
addr = receiver.getsockname()

sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sender.sendto(b"humidity=48", addr)
data, _ = receiver.recvfrom(1024)
print("received:", data.decode())
sender.close()
receiver.close()

ผลลัพธ์:

received: humidity=48

ตัวอย่างที่ 70: ส่งค่าไปยังเว็บเซอร์วิสด้วย POST

ส่งค่าเซนเซอร์เป็น JSON ผ่าน HTTPS

import requests

reading = {"device": "node-1", "temperature_c": 21.5}
response = requests.post(
    "https://api.example.com/readings",
    json=reading,
    timeout=5,
)
response.raise_for_status()
print("server said:", response.status_code)

ตัวอย่างที่ 71: ดึง JSON จาก API พร้อมกำหนดเวลาหมด

ดึงข้อมูลและจัดการความล้มเหลวโดยไม่ค้าง

import requests

try:
    r = requests.get("https://api.example.com/forecast", params={"city": "Bangkok"}, timeout=5)
    r.raise_for_status()
    data = r.json()
    print(data.get("temp_c"))
except requests.RequestException as e:
    print("request failed:", e)

ตัวอย่างที่ 72: ลองใหม่แบบเพิ่มเวลารอเป็นทวีคูณ

รอนานขึ้นทุกครั้งที่พยายามไม่สำเร็จ

import random
import time

def with_backoff(fn, tries=4, base=0.1):
    for n in range(tries):
        try:
            return fn()
        except ConnectionError:
            wait = base * (2 ** n)
            print(f"try {n + 1} failed, waiting {wait:.1f}s")
            time.sleep(wait)
    raise ConnectionError("gave up")

random.seed(3)
calls = {"n": 0}
def flaky():
    calls["n"] += 1
    if calls["n"] < 3:
        raise ConnectionError("network down")
    return "uploaded"

print(with_backoff(flaky))

ผลลัพธ์:

try 1 failed, waiting 0.1s
try 2 failed, waiting 0.2s
uploaded

ตัวอย่างที่ 73: ส่ง JSON ด้วย urllib

ส่งคำขอด้วยไลบรารีมาตรฐานเท่านั้น

import json
import urllib.request

body = json.dumps({"device": "node-1", "humidity_pct": 48}).encode()
req = urllib.request.Request(
    "https://api.example.com/readings",
    data=body,
    headers={"Content-Type": "application/json"},
    method="POST",
)
with urllib.request.urlopen(req, timeout=5) as resp:
    print("status", resp.status)

ตัวอย่างที่ 74: เว็บเซิร์ฟเวอร์สถานะ

เสิร์ฟหน้าสถานะเล็ก ๆ บน Pi

from http.server import BaseHTTPRequestHandler, HTTPServer
import json

class Status(BaseHTTPRequestHandler):
    def do_GET(self):
        body = json.dumps({"status": "ok", "uptime_s": 42}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(body)

HTTPServer(("0.0.0.0", 8000), Status).serve_forever()

ตัวอย่างที่ 75: ตรวจเครือข่ายก่อนส่งข้อมูล

พักข้อมูลไว้เมื่ออินเทอร์เน็ตใช้ไม่ได้

import socket

def internet_ok(host="1.1.1.1", port=53, timeout=3):
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False

print("online" if internet_ok() else "offline, queue the data")

หัวข้อที่ 11: MQTT: เผยแพร่และสมัครรับข้อความ

MQTT คือโปรโตคอลการส่งข้อความมาตรฐานของ IoT อุปกรณ์เผยแพร่ค่าไปยังหัวข้อ บรอกเกอร์ส่งต่อ และผู้สมัครรับกี่คนก็ได้จะได้รับ เบาพอสำหรับอุปกรณ์ขนาดเล็กและทนต่อเครือข่ายที่อ่อน

หัวข้อ การเผยแพร่ และการสมัครรับ

หัวข้อคือเส้นทาง เช่น home/kitchen/temperature การเผยแพร่ส่งข้อความไปยังหัวข้อผ่านบรอกเกอร์ การสมัครรับใช้ไวลด์การ์ดได้ + ตรงกับหนึ่งระดับ และ # ตรงกับทุกอย่างที่อยู่ข้างล่าง

ระดับการบริการและข้อความที่เก็บไว้

QoS 0 ส่งแล้วไม่รับประกัน QoS 1 รับประกันอย่างน้อยหนึ่งครั้ง และ QoS 2 ส่งได้พอดีหนึ่งครั้ง ข้อความที่เก็บไว้ถูกบันทึกโดยบรอกเกอร์ ผู้สมัครรับใหม่จึงได้ค่าสุดท้ายทันที

last will, TLS และการค้นพบ

ข้อความ last will ถูกเผยแพร่โดยบรอกเกอร์เมื่ออุปกรณ์หลุดการเชื่อมต่อโดยไม่คาดคิด ซึ่งใช้ตรวจอุปกรณ์ออฟไลน์ ใช้ TLS และรหัสผ่านจากสภาพแวดล้อมสำหรับบรอกเกอร์บนอินเทอร์เน็ต Home Assistant หาเซนเซอร์ได้จากข้อความการค้นพบที่ส่งไปยังหัวข้อพิเศษ

ตัวอย่างที่ 76: MQTT: ส่งข้อมูลเทเลเมทรี

ส่งค่าเซนเซอร์ไปยังบรอกเกอร์ตามหัวข้อ

import json
import time
import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="node-1")
client.connect("broker.local", 1883, keepalive=60)
client.loop_start()

payload = {"temp_c": 21.5, "ts": int(time.time())}
client.publish("home/kitchen/temperature", json.dumps(payload), qos=1)
time.sleep(1)
client.loop_stop()
client.disconnect()

ตัวอย่างที่ 77: MQTT: สมัครรับข้อความ

รับค่าจากอุปกรณ์หลายตัวด้วยไวลด์การ์ด

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe("home/+/temperature")

def on_message(client, userdata, msg):
    print(msg.topic, "->", msg.payload.decode())

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.local", 1883)
client.loop_forever()

ตัวอย่างที่ 78: MQTT: QoS และข้อความที่เก็บไว้

เลือกการรับประกันการส่งและเก็บค่าสุดท้ายไว้

import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.connect("broker.local", 1883)
client.loop_start()

# QoS 0 = at most once, 1 = at least once, 2 = exactly once
client.publish("home/door/state", "closed", qos=1, retain=True)   # last value is kept
client.loop_stop()
client.disconnect()

ตัวอย่างที่ 79: MQTT: last will สำหรับตรวจการหลุด

แจ้งทุกคนเมื่ออุปกรณ์หลุดการเชื่อมต่อ

import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="node-1")
client.will_set("home/node-1/status", payload="offline", qos=1, retain=True)
client.publish("home/node-1/status", "online", qos=1, retain=True)   # set on connect
client.connect("broker.local", 1883)
client.loop_forever()

ตัวอย่างที่ 80: MQTT ผ่าน TLS

เข้ารหัสการเชื่อมต่อไปยังบรอกเกอร์

import ssl
import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.tls_set(ca_certs="/etc/ssl/certs/ca-certificates.crt",
               tls_version=ssl.PROTOCOL_TLS_CLIENT)
client.username_pw_set("node-1", "use-a-secret-from-env-not-code")
client.connect("broker.example.com", 8883)
client.loop_forever()

ตัวอย่างที่ 81: ข้อความค้นพบของ Home Assistant

เผยแพร่เซนเซอร์เพื่อให้ Home Assistant พบเอง

import json

config = {
    "name": "Kitchen temperature",
    "unique_id": "node-1_kitchen_temp",
    "state_topic": "home/kitchen/temperature",
    "unit_of_measurement": "°C",
    "device_class": "temperature",
    "value_template": "{{ value_json.temp_c }}",
}
topic = "homeassistant/sensor/node-1_kitchen_temp/config"
print(topic)
print(json.dumps(config, ensure_ascii=False, indent=2))

ผลลัพธ์:

homeassistant/sensor/node-1_kitchen_temp/config
{
  "name": "Kitchen temperature",
  "unique_id": "node-1_kitchen_temp",
  "state_topic": "home/kitchen/temperature",
  "unit_of_measurement": "°C",
  "device_class": "temperature",
  "value_template": "{{ value_json.temp_c }}"
}

ตัวอย่างที่ 82: ไวลด์การ์ดของหัวข้อ MQTT

ตรวจว่าการสมัครรับตรงกับหัวข้อใดบ้าง

def topic_matches(pattern, topic):
    p_parts = pattern.split("/")
    t_parts = topic.split("/")
    for i, part in enumerate(p_parts):
        if part == "#":
            return True
        if i >= len(t_parts):
            return False
        if part != "+" and part != t_parts[i]:
            return False
    return len(p_parts) == len(t_parts)

tests = [
    ("home/+/temperature", "home/kitchen/temperature", True),
    ("home/#", "home/kitchen/humidity", True),
    ("home/+/temperature", "home/kitchen/humidity", False),
]
for pattern, topic, expected in tests:
    matched = topic_matches(pattern, topic)
    print(pattern, topic, "match" if matched else "no match", "(as expected)" if matched == expected else "(UNEXPECTED)")

ผลลัพธ์:

home/+/temperature home/kitchen/temperature match (as expected)
home/# home/kitchen/humidity match (as expected)
home/+/temperature home/kitchen/humidity no match (as expected)

หัวข้อที่ 12: แดชบอร์ดเว็บและ API ด้วย Flask

Flask คือเฟรมเวิร์กเว็บขนาดเล็กของ Python ด้วยโค้ดไม่กี่บรรทัดก็แสดงหน้าสถานะ คืนค่า JSON หรือรับค่าจากอุปกรณ์อื่นได้ จึงดู Pi ของคุณจากโทรศัพท์หรือเบราว์เซอร์ได้

เส้นทางและการตอบกลับ

ตัวตกแต่งเส้นทาง เช่น @app.get('/') ผูก URL กับฟังก์ชัน ฟังก์ชันคืนข้อความ หน้าเว็บ หรือ JSON ด้วย jsonify ใช้ที่อยู่ 0.0.0.0 เพื่อรับการเชื่อมต่อจากเครือข่ายภายในบ้าน

รับข้อมูลพร้อมการตรวจสอบ

เส้นทาง POST อ่าน JSON จากคำขอ ตรวจชนิดและช่วงก่อนเก็บค่าใด ๆ และคืนสถานะ 400 เมื่อข้อมูลผิด เพื่อให้ผู้ส่งรู้ว่าต้องแก้อะไร

เธรดเบื้องหลังและแดชบอร์ด

เธรดเบื้องหลังอ่านเซนเซอร์ขณะที่เว็บเซิร์ฟเวอร์ตอบคำขอ เก็บข้อมูลที่ใช้ร่วมกันให้เรียบง่ายและปลอดภัย เทมเพลต HTML ขนาดเล็กเปลี่ยนค่าล่าสุดเป็นหน้าที่ทุกคนในเครือข่ายดูได้

ตัวอย่างที่ 83: เอนด์พอยต์ Flask แบบ hello

เสิร์ฟเส้นทางเว็บแรกบน Pi

from flask import Flask

app = Flask(__name__)

@app.get("/")
def hello():
    return "Raspberry Pi is alive"

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

ตัวอย่างที่ 84: เอนด์พอยต์ JSON ด้วย Flask

คืนค่าล่าสุดเป็น JSON

from flask import Flask, jsonify

app = Flask(__name__)
latest = {"temp_c": 21.5, "humidity_pct": 48}

@app.get("/api/latest")
def get_latest():
    return jsonify(latest)

app.run(host="0.0.0.0", port=5000)

ตัวอย่างที่ 85: Flask POST พร้อมการตรวจสอบ

รับค่าและปฏิเสธค่าที่ไม่ถูกต้องด้วยรหัสข้อผิดพลาด

from flask import Flask, request, jsonify

app = Flask(__name__)
readings = []

@app.post("/api/readings")
def add_reading():
    data = request.get_json(silent=True) or {}
    temp = data.get("temp_c")
    if not isinstance(temp, (int, float)) or not -40 <= temp <= 85:
        return jsonify(error="temp_c must be a number from -40 to 85"), 400
    readings.append(temp)
    return jsonify(saved=len(readings)), 201

app.run(port=5000)

ตัวอย่างที่ 86: แดชบอร์ด HTML แบบง่าย

แสดงหน้าที่มีอุณหภูมิล่าสุด

from flask import Flask, render_template_string

app = Flask(__name__)
PAGE = """
<h1>Kitchen</h1>
<p>Temperature: {{ temp }} °C</p>
"""

@app.get("/")
def home():
    return render_template_string(PAGE, temp=21.5)

app.run(port=5000)

ตัวอย่างที่ 87: Flask กับเธรดเซนเซอร์เบื้องหลัง

อ่านเซนเซอร์อย่างต่อเนื่องขณะเว็บเซิร์ฟเวอร์ทำงาน

import threading
import time
from flask import Flask, jsonify

app = Flask(__name__)
state = {"temp_c": None}

def sensor_loop():
    while True:
        state["temp_c"] = 21.5          # replace with the real sensor read
        time.sleep(5)

threading.Thread(target=sensor_loop, daemon=True).start()

@app.get("/api/temp")
def temp():
    return jsonify(state)

app.run(port=5000)

หัวข้อที่ 13: เก็บและสรุปค่าที่วัดได้

อุปกรณ์ที่บันทึกข้อมูลทั้งปีต้องการที่เก็บที่เรียบง่ายและไม่ทำให้การ์ด SD เต็ม SQLite เป็นฐานข้อมูลไฟล์เดียวที่มากับ Python และไฟล์ CSV เหมาะกับประวัติอย่างง่าย

ตาราง SQLite และการแทรกข้อมูล

SQLite เก็บตารางไว้ในไฟล์เดียว ใช้ตัวยึด ? ในคำสั่ง INSERT ซึ่งป้องกัน SQL injection และจัดการการใส่เครื่องหมายคำพูดให้ เรียก commit หลังเขียนเพื่อให้ข้อมูลถูกบันทึก

ค่ารวมและการจัดกลุ่มตามเวลา

GROUP BY ร่วมกับ AVG และ COUNT สรุปค่าต่ออุปกรณ์ การจัดกลุ่มเวลาตามนาที เช่นใช้ห้าตัวอักษรแรกของเวลา ISO ให้ค่าเฉลี่ยรายนาทีแบบง่าย

การเก็บรักษาและไฟล์บันทึกแบบต่อท้าย

ลบแถวเก่าตามกำหนดเวลาเพื่อให้พื้นที่เก็บข้อมูลเล็กอยู่เสมอ ไฟล์ CSV แบบต่อท้ายเพิ่มหนึ่งบรรทัดต่อการวัด และเขียนหัวตารางเพียงครั้งเดียว ซึ่งเรียบง่ายและทนทาน

ตัวอย่างที่ 88: SQLite: สร้างตารางและเพิ่มข้อมูล

เก็บค่าที่วัดได้ในไฟล์ฐานข้อมูล

import sqlite3

conn = sqlite3.connect(":memory:")   # use "readings.db" to keep data on the Pi
conn.execute("""CREATE TABLE readings (
    ts TEXT NOT NULL, device TEXT NOT NULL, temp_c REAL NOT NULL)""")
rows = [("2026-01-15T08:00", "node-1", 21.5), ("2026-01-15T08:05", "node-1", 21.9)]
conn.executemany("INSERT INTO readings VALUES (?, ?, ?)", rows)
conn.commit()
print(conn.execute("SELECT COUNT(*) FROM readings").fetchone()[0], "rows stored")
conn.close()

ผลลัพธ์:

2 rows stored

ตัวอย่างที่ 89: SQLite: คิวรีแบบสรุปผล

หาค่าเฉลี่ยและนับค่าต่อหนึ่งอุปกรณ์

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (device TEXT, temp_c REAL)")
conn.executemany("INSERT INTO readings VALUES (?, ?)",
                 [("node-1", 21.5), ("node-1", 22.5), ("node-2", 19.0)])
for device, avg, n in conn.execute(
        "SELECT device, AVG(temp_c), COUNT(*) FROM readings GROUP BY device ORDER BY device"):
    print(f"{device}: average {avg:.2f} from {n} readings")
conn.close()

ผลลัพธ์:

node-1: average 22.00 from 2 readings
node-2: average 19.00 from 1 readings

ตัวอย่างที่ 90: ตัวบันทึกประวัติ CSV

ต่อท้ายค่าลงไฟล์ประวัติพร้อมหัวตาราง

import csv
from pathlib import Path

LOG = Path("sensor_history.csv")
FIELDS = ["ts", "temp_c"]

def log_reading(ts, temp_c):
    new_file = not LOG.exists()
    with LOG.open("a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        if new_file:
            writer.writeheader()
        writer.writerow({"ts": ts, "temp_c": temp_c})

log_reading("08:00", 21.5)
log_reading("08:05", 21.9)
print(LOG.read_text(encoding="utf-8"))
LOG.unlink()

ผลลัพธ์:

ts,temp_c
08:00,21.5
08:05,21.9

ตัวอย่างที่ 91: จัดกลุ่มค่าตามนาที

รวมค่าเป็นค่าเฉลี่ยรายนาที

from collections import defaultdict
from statistics import mean

readings = [("08:00:10", 21.4), ("08:00:40", 21.6), ("08:01:05", 21.9)]
buckets = defaultdict(list)
for ts, value in readings:
    buckets[ts[:5]].append(value)     # keep HH:MM
for minute in sorted(buckets):
    print(minute, round(mean(buckets[minute]), 2))

ผลลัพธ์:

08:00 21.5
08:01 21.9

ตัวอย่างที่ 92: การเก็บรักษาข้อมูล

ลบแถวเก่าเพื่อให้พื้นที่เก็บข้อมูลของ Pi ไม่เต็ม

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (day INTEGER, temp_c REAL)")
conn.executemany("INSERT INTO readings VALUES (?, ?)", [(d, 20.0 + d / 10) for d in range(1, 11)])
keep_from = 6
deleted = conn.execute("DELETE FROM readings WHERE day < ?", (keep_from,)).rowcount
conn.commit()
remaining = conn.execute("SELECT COUNT(*) FROM readings").fetchone()[0]
print("deleted", deleted, "kept", remaining)
conn.close()

ผลลัพธ์:

deleted 5 kept 5

หัวข้อที่ 14: ทำงานเป็นบริการและการตั้งค่า

โปรเจกต์จะมีประโยชน์ก็ต่อเมื่อเริ่มทำงานเองหลังไฟดับและทำงานต่อไปได้ systemd รันบริการตอนบูต cron รันงานตามกำหนดเวลา และตัวแปรสภาพแวดล้อมเก็บการตั้งค่าให้อยู่นอกโค้ด

บริการ systemd

ไฟล์ยูนิตอธิบายวิธีรันโปรแกรม โฟลเดอร์ทำงาน ผู้ใช้ และสิ่งที่ต้องทำเมื่อล้มเหลวด้วย Restart=on-failure เปิดใช้บริการเพื่อให้เริ่มตอนบูต และอ่านล็อกด้วย journalctl

งานตามกำหนดเวลาด้วย cron

บรรทัดในไฟล์ crontab รันคำสั่งในเวลาที่กำหนด เช่นทุกห้านาที ส่งเอาต์พุตไปยังไฟล์ล็อกเพื่อดูว่าเกิดอะไรขึ้นตอนที่คุณไม่อยู่

การตั้งค่าจากสภาพแวดล้อม

อ่านการตั้งค่า เช่นรหัสอุปกรณ์ ช่วงเวลารายงาน และที่อยู่บรอกเกอร์ จากตัวแปรสภาพแวดล้อมพร้อมค่าเริ่มต้นที่ปลอดภัย คุณเปลี่ยนค่าได้ในไฟล์บริการโดยไม่ต้องแก้โค้ด

ตัวอย่างที่ 93: บริการ systemd ตอนเปิดเครื่อง

รันตัวบันทึกโดยอัตโนมัติเมื่อ Pi เริ่มทำงาน

# /etc/systemd/system/sensor-logger.service
[Unit]
Description=Sensor logger
After=network-online.target
Wants=network-online.target

[Service]
ExecStart=/usr/bin/python3 /home/pi/iot/logger.py
WorkingDirectory=/home/pi/iot
Restart=on-failure
RestartSec=5
User=pi

[Install]
WantedBy=multi-user.target

# enable and start it:
# sudo systemctl daemon-reload
# sudo systemctl enable --now sensor-logger.service
# journalctl -u sensor-logger -f

ตัวอย่างที่ 94: งานตามกำหนดเวลาด้วย cron

รันสคริปต์ทุกห้านาที

# edit the schedule with: crontab -e
# every 5 minutes, record the time and the readings file size
*/5 * * * * /usr/bin/python3 /home/pi/iot/snapshot.py >> /home/pi/iot/cron.log 2>&1
# every day at 02:30, remove logs older than 30 days
30 2 * * * find /home/pi/iot/logs -name "*.csv" -mtime +30 -delete

ตัวอย่างที่ 95: ไฟล์ heartbeat สำหรับ watchdog

เขียนเวลาไว้เพื่อให้ watchdog รู้ว่าโปรแกรมยังทำงาน

import time
from pathlib import Path

HEARTBEAT = Path("heartbeat.txt")

def beat(count):
    HEARTBEAT.write_text(f"{count} {time.time():.0f}", encoding="utf-8")

for i in range(3):
    beat(i)
    print("beat", i, "written")
    time.sleep(0.05)
print("last:", HEARTBEAT.read_text(encoding="utf-8").split()[0])
HEARTBEAT.unlink()

ผลลัพธ์:

beat 0 written
beat 1 written
beat 2 written
last: 2

ตัวอย่างที่ 96: การตั้งค่าจากตัวแปรสภาพแวดล้อม

อ่านการตั้งค่าพร้อมค่าเริ่มต้นที่ปลอดภัย

import os

def get_config():
    return {
        "device_id": os.environ.get("DEVICE_ID", "node-1"),
        "interval_s": int(os.environ.get("INTERVAL_S", "60")),
        "broker": os.environ.get("MQTT_HOST", "localhost"),
    }

print(get_config())

ผลลัพธ์:

{'device_id': 'node-1', 'interval_s': 60, 'broker': 'localhost'}

หัวข้อที่ 15: การบันทึกล็อก ความปลอดภัย และไปป์ไลน์ที่สมบูรณ์

หัวข้อสุดท้ายรวมทุกอย่างเข้าด้วยกัน ได้แก่ล็อกแบบหมุนเวียนเพื่อไม่ให้การ์ดเต็ม ความลับที่อยู่นอกโค้ด การตรวจข้อมูลที่เข้ามา และไปป์ไลน์เต็มตั้งแต่เซนเซอร์ถึงการแจ้งเตือน

ล็อกหมุนเวียนและ heartbeat

ตัวจัดการไฟล์แบบหมุนเวียนจำกัดขนาดของแต่ละไฟล์ล็อกและเก็บไฟล์เก่าไว้ไม่กี่ไฟล์ ไฟล์ heartbeat ที่มีเวลาทำให้ watchdog ตรวจได้ว่าโปรแกรมยังทำงานอยู่

ความลับและการตรวจสอบ

เก็บรหัสผ่านและโทเค็นไว้ในตัวแปรสภาพแวดล้อม และอย่าพิมพ์ออกมาเด็ดขาด ตรวจทุกค่าที่เข้ามาทั้งชนิดและช่วงก่อนนำไปเก็บหรือใช้แจ้งเตือน เพราะเซนเซอร์และเครือข่ายส่งอะไรมาก็ได้

ประกอบไปป์ไลน์

ไปป์ไลน์อ่านเซนเซอร์ ทำให้ค่าเรียบด้วยหน้าต่าง เทียบกับเกณฑ์ ตั้งธงแจ้งเตือน และเก็บผลแต่ละขั้น แต่ละขั้นมีขนาดเล็กและทดสอบได้ ซึ่งเป็นวิธีจัดโค้ด IoT แบบมืออาชีพ

ตัวอย่างที่ 97: ไฟล์ล็อกแบบหมุนเวียน

ไม่ให้ล็อกเต็มการ์ด SD

import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path

path = Path("iot.log")
handler = RotatingFileHandler(path, maxBytes=200, backupCount=2, encoding="utf-8")
log = logging.getLogger("iot")
log.setLevel(logging.INFO)
log.addHandler(handler)
for i in range(20):
    log.info("reading %d: temp=21.%d", i, i % 10)
handler.close()
print("files:", sorted(p.name for p in Path(".").glob("iot.log*")))
for p in Path(".").glob("iot.log*"):
    p.unlink()

ผลลัพธ์:

files: ['iot.log', 'iot.log.1', 'iot.log.2']

ตัวอย่างที่ 98: ความลับจากตัวแปรสภาพแวดล้อม

โหลดรหัสผ่านจากสภาพแวดล้อมและไม่พิมพ์ออกมา

import os

def require_secret(name):
    value = os.environ.get(name)
    if not value:
        raise SystemExit(f"set {name} in the environment (do not write it in the code)")
    return value

def mask(secret):
    return secret[:2] + "*" * (len(secret) - 2)

os.environ["MQTT_PASSWORD"] = "example-secret"   # demo only
token = require_secret("MQTT_PASSWORD")
print("loaded secret:", mask(token))

ผลลัพธ์:

loaded secret: ex************

ตัวอย่างที่ 99: ตรวจข้อมูลเซนเซอร์ที่เข้ามา

ตรวจชนิดและช่วงของค่าก่อนเก็บ

from dataclasses import dataclass

@dataclass
class Sample:
    device: str
    temp_c: float

def validate(raw):
    errors = []
    if not isinstance(raw.get("device"), str) or not raw["device"]:
        errors.append("device is required")
    temp = raw.get("temp_c")
    if not isinstance(temp, (int, float)):
        errors.append("temp_c must be a number")
    elif not -40 <= temp <= 85:
        errors.append("temp_c out of sensor range")
    if errors:
        return None, errors
    return Sample(raw["device"], float(temp)), []

print(validate({"device": "node-1", "temp_c": 21.5}))
print(validate({"device": "", "temp_c": 300}))

ผลลัพธ์:

(Sample(device='node-1', temp_c=21.5), [])
(None, ['device is required', 'temp_c out of sensor range'])

ตัวอย่างที่ 100: ไปป์ไลน์เล็ก ๆ ที่สมบูรณ์

อ่าน ทำให้เรียบ แจ้งเตือน และเก็บผลลัพธ์

import sqlite3
from collections import deque

def fake_sensor(n):
    base = [21.0, 21.2, 21.1, 35.0, 21.3, 21.4]     # 35.0 is a spike
    return base[n % len(base)]

window = deque(maxlen=3)
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (tick INTEGER, raw REAL, smooth REAL, alarm INTEGER)")

for tick in range(6):
    raw = fake_sensor(tick)
    window.append(raw)
    smooth = sum(window) / len(window)
    alarm = 1 if smooth > 25 else 0
    conn.execute("INSERT INTO readings VALUES (?, ?, ?, ?)", (tick, raw, smooth, alarm))
    print(f"tick {tick}: raw {raw:5.1f} smooth {smooth:5.2f} alarm {alarm}")

alarms = conn.execute("SELECT COUNT(*) FROM readings WHERE alarm = 1").fetchone()[0]
print("alarm ticks:", alarms)
conn.close()

ผลลัพธ์:

tick 0: raw  21.0 smooth 21.00 alarm 0
tick 1: raw  21.2 smooth 21.10 alarm 0
tick 2: raw  21.1 smooth 21.10 alarm 0
tick 3: raw  35.0 smooth 25.77 alarm 1
tick 4: raw  21.3 smooth 25.80 alarm 1
tick 5: raw  21.4 smooth 25.90 alarm 1
alarm ticks: 3

โปรเจกต์ฝึก: เครื่องตรวจสอบห้อง

สร้างเครื่องตรวจสอบหนึ่งตัวที่ใช้แนวคิดจากทุกหัวข้อ เขียนเองก่อนแล้วเทียบกับตัวอย่าง ทดสอบส่วน Python บริสุทธิ์บนคอมพิวเตอร์ก่อนต่อฮาร์ดแวร์

1. เขียนฟังก์ชันที่แปลงค่าดิบเป็นอุณหภูมิโดยใช้จุดปรับเทียบสองจุด แล้วทดสอบด้วยค่าสามค่า

2. ทำให้ค่าเรียบด้วยค่าเฉลี่ยเคลื่อนที่ของห้าค่าล่าสุด และเพิ่มฮิสเทอรีซิสสำหรับพัดลมที่เกณฑ์ 24 และ 22

3. เก็บแต่ละค่าในตาราง SQLite และพิมพ์ค่าเฉลี่ยรายนาทีหลังจากอ่านสิบค่า

4. เผยแพร่ค่าล่าสุดไปยังหัวข้อ MQTT ทุก 60 วินาที และเมื่อบรอกเกอร์ออฟไลน์ให้เก็บค่าไว้ในคิว

5. สร้างบริการ systemd ที่เริ่มเครื่องตรวจสอบตอนบูต แล้วอ่านล็อกด้วย journalctl

ขั้นต่อไป

ขั้นต่อไป ให้เรียนแพลตฟอร์ม IoT เช่น Home Assistant หรือ Node-RED เพิ่มการระบุตัวตนของอุปกรณ์อย่างปลอดภัยและการอัปเดตผ่านอากาศ และศึกษา Raspberry Pi Pico สำหรับงานไมโครคอนโทรลเลอร์ ทดสอบตัวอย่างฮาร์ดแวร์ทุกตัวกับการต่อสายของคุณก่อน และตรวจหมายเลขขาของรุ่นที่คุณใช้ในเอกสารของ Raspberry Pi

สรุป

Raspberry Pi จะกลายเป็นอุปกรณ์ IoT จริงเมื่อมันอ่านเซนเซอร์ได้อย่างเชื่อถือได้ ประมวลผลข้อมูล เก็บข้อมูล ส่งข้อมูลผ่านเครือข่าย และทำงานได้เองโดยไม่ต้องมีคนช่วย ผ่าน 100 ตัวอย่าง บทความนี้ครอบคลุมพื้นฐาน Python การตัดสินใจและลูป ฟังก์ชัน โครงสร้างข้อมูล ไฟล์และรูปแบบข้อมูล การจัดการข้อผิดพลาดและการบันทึกล็อก การประมวลผลเซนเซอร์ GPIO I2C SPI 1-Wire และกล้อง ซ็อกเก็ตและ HTTP MQTT Flask SQLite systemd และ cron และไปป์ไลน์ที่สมบูรณ์ ลองสร้างเครื่องตรวจสอบห้องในแบบฝึก และเปลี่ยนทีละอย่างเพื่อเรียนรู้ว่าแต่ละส่วนทำงานอย่างไร