Raspberry Pi dan IoT dengan Python: 100 contoh dari dasar hingga profesional
Raspberry Pi adalah komputer kecil yang dapat membaca sensor, mengendalikan relay, dan mengirim data melalui jaringan, sehingga sering dipakai sebagai dasar proyek IoT (Internet of Things). Artikel ini membawa Anda dari program Python pertama hingga jalur pemrosesan sensor yang lengkap dalam 100 contoh. Anda akan mempelajari dasar Python, mengolah data sensor, mengendalikan pin dengan GPIO, membaca sensor melalui I2C, SPI, dan 1-Wire, mengirim data dengan HTTP dan MQTT, membuat dasbor web kecil, menyimpan pembacaan di SQLite, dan menjalankan semuanya sebagai layanan yang aktif saat boot. Catatan kejujuran: contoh yang dapat berjalan di komputer biasa telah dijalankan dan keluarannya yang sebenarnya ditampilkan setelah kode. Contoh lainnya memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya dan keluarannya tidak ditampilkan.
Yang perlu disiapkan
Raspberry Pi 3, 4, atau 5 yang menjalankan Raspberry Pi OS, kartu microSD, catu daya, dan koneksi jaringan. Python 3.10 atau yang lebih baru sudah tersedia di Raspberry Pi OS. Untuk contoh perangkat keras Anda memerlukan LED, resistor, tombol tekan, kabel jumper, dan beberapa sensor. Sebagian besar contoh yang dapat dijalankan bekerja di komputer apa pun yang memiliki Python 3.10 atau yang lebih baru, dan begitulah contoh yang dapat dijalankan diuji.
Cara menjalankan contoh
Simpan contoh sebagai main.py lalu jalankan dengan python3 main.py. Contoh yang ditandai dapat dijalankan menampilkan keluaran sebenarnya dari eksekusi di komputer. Contoh bash adalah berkas konfigurasi Pi dan tidak dijalankan di sini.
Topik 1: Memulai Python di Pi
Python adalah bahasa yang lazim untuk proyek Raspberry Pi karena mudah dibaca dan memiliki pustaka untuk hampir setiap sensor. Contoh-contoh awal membahas keluaran, variabel, tipe, dan perhitungan.
Menjalankan program pertama
Simpan berkas dengan ekstensi .py lalu jalankan dengan python3 di Pi. print menulis teks ke terminal, dan f-string memasukkan variabel ke dalam teks dengan {nama}. Python membaca berkas dari atas ke bawah, jadi urutan baris itu penting.
Tipe dan maknanya
Angka dapat berupa int (bilangan bulat) atau float (desimal), teks adalah str, True dan False bertipe bool, dan None berarti tidak ada nilai. Gunakan type() untuk memeriksa tipe suatu nilai. Pembacaan sensor biasanya float, sedangkan nomor pin biasanya int.
Aritmetika untuk perhitungan sensor
/ selalu menghasilkan float, // membagi lalu membuang bagian pecahan, dan % memberi sisa bagi. round() membatasi jumlah angka desimal yang ditampilkan. Gunakan operator ini untuk mengonversi pembacaan mentah dan mengelompokkan nilai.
Ilustrasi
Contoh 1: Salam dari Pi
Program Python pertama Anda di Raspberry Pi OS.
print("Hello from my Raspberry Pi!")
Keluaran:
Hello from my Raspberry Pi!
Contoh 2: Variabel dan f-string
Menyimpan nilai dan menampilkannya di dalam teks.
sensor_name = "DHT22"
temperature = 27.5
print(f"{sensor_name} reads {temperature} °C")
Keluaran:
DHT22 reads 27.5 °C
Contoh 3: Tipe data
Tipe dasar Python dan namanya.
for value in (42, 3.14, "pi", True, None):
print(repr(value), type(value).__name__)
Keluaran:
42 int
3.14 float
'pi' str
True bool
None NoneType
Contoh 4: Aritmetika, pembagian bulat, dan pembulatan
Pembagian, sisa bagi, dan pembulatan untuk perhitungan sensor.
readings = 17
per_group = 5
print(readings / per_group)
print(readings // per_group, readings % per_group)
print(round(3.14159, 2))
Keluaran:
3.4
3 2
3.14
Contoh 5: Mengolah teks
Membersihkan nama perangkat dan memeriksa awalan.
status = " ONLINE "
clean = status.strip().lower()
print(clean, clean.startswith("on"), len(clean))
Keluaran:
online True 6
Contoh 6: Konversi satuan
Mengubah Celsius menjadi Fahrenheit dengan fungsi.
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")
Keluaran:
0 C = 32.0 F
25 C = 77.0 F
37.5 C = 99.5 F
Topik 2: Keputusan dan Perulangan
Proyek Pi bereaksi terhadap masukannya. Kondisi menentukan apa yang terjadi, dan perulangan mengulang pekerjaan, misalnya mengambil sampel sensor setiap beberapa detik.
if, elif, dan else
Kondisi diperiksa dari atas ke bawah, dan cabang pertama yang bernilai benar dijalankan. Letakkan pemeriksaan yang paling spesifik di depan. Perbandingan berantai seperti 20 <= t <= 25 memeriksa seluruh rentang dalam satu baris.
for, while, break, dan continue
for mengulang pada rentang atau urutan. while mengulang sampai kondisi berubah. break keluar dari perulangan, dan continue melewati putaran saat ini, yang berguna untuk mengabaikan pembacaan yang salah dan tetap berjalan.
Nilai benar-salah dan pernyataan match
Nilai kosong, seperti string kosong atau list kosong, dianggap salah. Pernyataan match, yang tersedia sejak Python 3.10, mencocokkan perintah dengan beberapa pola, cara yang rapi untuk menangani perintah teks dari aplikasi ponsel.
Contoh 7: if, elif, dan else
Memilih tindakan dari suhu.
temperature = 31
if temperature >= 35:
print("Too hot: stop the fan test")
elif temperature >= 28:
print("Warm: run the fan")
else:
print("Comfortable")
Keluaran:
Warm: run the fan
Contoh 8: Perulangan for dengan range
Mengulang pekerjaan sejumlah tertentu.
for minute in range(0, 15, 5):
print(f"minute {minute}")
Keluaran:
minute 0
minute 5
minute 10
Contoh 9: Perulangan while
Mengulang sampai kondisi berubah.
countdown = 3
while countdown > 0:
print("sampling in", countdown)
countdown -= 1
print("start")
Keluaran:
sampling in 3
sampling in 2
sampling in 1
start
Contoh 10: break dan continue
Melewati pembacaan yang salah dan berhenti saat ada gangguan.
for reading in [21, None, 23, 99, 24]:
if reading is None:
continue
if reading > 50:
print("sensor fault, stop")
break
print("ok", reading)
Keluaran:
ok 21
ok 23
sensor fault, stop
Contoh 11: Nilai benar dan salah
Nilai mana yang dianggap benar atau salah.
for value in ("", "Pi", 0, 5, [], [1]):
print(repr(value), "is true" if value else "is false")
Keluaran:
'' is false
'Pi' is true
0 is false
5 is true
[] is false
[1] is true
Contoh 12: Perbandingan berantai
Memeriksa apakah nilai berada dalam rentang.
humidity = 55
print(40 <= humidity <= 60)
print(humidity > 60 or humidity < 40)
Keluaran:
True
False
Contoh 13: Ekspresi kondisional
if-else satu baris yang menghasilkan nilai.
reading = 12
state = "ON" if reading < 15 else "OFF"
print(state)
Keluaran:
ON
Contoh 14: Pernyataan match
Mencocokkan perintah teks dengan tindakan.
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))
Keluaran:
on -> turn relay on
0 -> turn relay off
blink -> unknown command
Topik 3: Fungsi dan Pustaka Standar
Fungsi memberi nama pada tugas yang Anda ulang. Pustaka standar sudah berisi fungsi matematika, statistik, dan waktu, sehingga jarang perlu ditulis sendiri.
Mendefinisikan fungsi
def membuat fungsi, dan return mengembalikan nilai. Argumen default memberikan nilai yang masuk akal ketika pemanggil tidak menyebutkannya, dan argumen kata kunci membuat pemanggilan mudah dibaca, misalnya sample(4, unit='volts').
Mengembalikan dan menerima banyak nilai
Fungsi dapat mengembalikan beberapa nilai sebagai tuple yang dibongkar oleh pemanggil. *args menerima sejumlah argumen posisi, dan **kwargs menerima sejumlah argumen bernama, cocok untuk fungsi bantu pencatatan.
lambda, map, filter, dan rekursi
lambda adalah fungsi kecil tanpa nama, berguna bersama map dan filter untuk mengubah list. Fungsi rekursif memanggil dirinya sendiri dan harus memiliki kondisi berhenti; faktorial adalah contoh klasiknya.
Contoh 15: Fungsi dengan argumen default
Memberi fungsi nilai awal yang masuk akal.
def sample(pin, samples=3, unit="raw"):
return f"read pin {pin} x{samples} ({unit})"
print(sample(4))
print(sample(4, 10, unit="volts"))
Keluaran:
read pin 4 x3 (raw)
read pin 4 x10 (volts)
Contoh 16: Mengembalikan beberapa nilai
Mengembalikan nilai minimum dan maksimum sekaligus.
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}")
Keluaran:
low 18.2, high 22.0
Contoh 17: *args dan **kwargs
Menerima jumlah nilai posisional dan bernama yang bebas.
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")
Keluaran:
door opened | sensor=reed-1, room=kitchen
Contoh 18: lambda, map, dan filter
Mengubah nilai ADC mentah dalam satu baris.
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)
Keluaran:
[1.65, 3.3, 0.0, 2.48]
[1.65, 3.3, 2.48]
Contoh 19: Pustaka standar: math dan statistics
Memakai modul matematika dan statistik bawaan.
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))
Keluaran:
4.0 True
3.5 3.75
Contoh 20: Rekursi
Fungsi yang memanggil dirinya sendiri, dengan kondisi berhenti.
def factorial(n):
if n <= 1:
return 1
return n * factorial(n - 1)
print(factorial(5), factorial(10))
Keluaran:
120 3628800
Topik 4: Struktur Data untuk Perangkat dan Pembacaan
Proyek Pi menyimpan list pembacaan, dictionary pengaturan perangkat, dan set nama yang unik. Memilih struktur yang tepat membuat kode lebih pendek dan lebih cepat.
List dan comprehension
List adalah urutan yang terurut dan dapat diubah. List comprehension membangun list baru dari list lama dalam satu baris, misalnya mengonversi daftar nilai Celsius menjadi Fahrenheit.
Dictionary untuk nilai bernama
Dictionary memetakan kunci ke nilai, ideal untuk pengaturan satu perangkat atau satu pembacaan beserta fieldnya. Gunakan get() untuk membaca kunci yang mungkin tidak ada, dan sorted() untuk mencetak kunci secara berurutan.
Tuple, set, dan slicing
Tuple adalah kelompok nilai tetap. Set menyimpan item unik dan mendukung gabungan dan irisan. Slice seperti values[-2:] mengambil sebagian list, dan [::-1] membaliknya.
Contoh 21: List
Menambah, menyisipkan, dan menghapus item.
readings = [20.1, 20.4]
readings.append(20.9)
readings.insert(0, 19.8)
readings.remove(20.4)
print(readings, len(readings), readings[-1])
Keluaran:
[19.8, 20.1, 20.9] 3 20.9
Contoh 22: List comprehension
Membuat list baru dari list lama.
celsius = [0, 10, 20, 30]
fahrenheit = [c * 9 / 5 + 32 for c in celsius if c != 10]
print(fahrenheit)
Keluaran:
[32.0, 68.0, 86.0]
Contoh 23: Tuple dan unpacking
Kelompok nilai tetap yang tidak bisa diubah.
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__)
Keluaran:
kitchen 2 3.5
tuples cannot change: TypeError
Contoh 24: Dictionary
Menyimpan nilai bernama untuk sebuah perangkat.
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))
Keluaran:
node-1 n/a 4
Contoh 25: Keluaran dictionary terurut
Mencetak jumlah tiap kejadian secara berurutan.
counts = {"motion": 4, "door": 1, "temp": 9}
for name in sorted(counts):
print(f"{name:<8}{counts[name]:>3}")
Keluaran:
door 1
motion 4
temp 9
Contoh 26: Set
Nilai unik dan operasi himpunan.
seen = {"kitchen", "garage"}
seen.add("kitchen")
new = {"garage", "attic"}
print(len(seen), sorted(seen | new), sorted(seen & new))
Keluaran:
2 ['attic', 'garage', 'kitchen'] ['garage']
Contoh 27: Slicing
Mengambil sebagian list, dan membaliknya.
samples = [1, 2, 3, 4, 5, 6, 7]
print(samples[:3], samples[-2:], samples[::2])
print(samples[::-1])
Keluaran:
[1, 2, 3] [6, 7] [1, 3, 5, 7]
[7, 6, 5, 4, 3, 2, 1]
Contoh 28: List berisi dictionary
Menyaring daftar perangkat yang daring.
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)
Keluaran:
online: ['a1', 'c3']
Topik 5: Teks, Berkas, dan Format Data
Data sensor harus disimpan dan dibagikan. Berkas teks, CSV, dan JSON adalah format yang paling banyak dipakai proyek IoT, dan pustaka standar dapat membaca serta menulis ketiganya.
Membaca dan menulis berkas
Buka berkas dengan blok with agar tertutup otomatis, dan gunakan encoding='utf-8' untuk teks. Objek Path dari pathlib bekerja di sistem operasi apa pun dan memudahkan penggabungan nama berkas.
CSV dan JSON
Modul csv menulis baris dengan header yang dipahami spreadsheet dan alat data. Modul json mengubah dictionary menjadi teks untuk layanan web dan pesan MQTT, dan sebaliknya.
Memformat tabel dan tanggal
Lebar format seperti {nama:<10} merapikan kolom di terminal. datetime memformat cap waktu dengan strftime dan menambah durasi dengan timedelta, sehingga pembacaan dapat dikelompokkan menurut waktu.
Contoh 29: Format keluaran tabel
Menyejajarkan kolom dengan lebar format.
rows = [("Temp", 27.456), ("Humidity", 55.0), ("Pressure", 1013.2)]
for name, value in rows:
print(f"{name:<10}{value:>10.1f}")
Keluaran:
Temp 27.5
Humidity 55.0
Pressure 1013.2
Contoh 30: Membaca dan menulis file teks
Menyimpan dan memuat nilai sensor ke file.
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])
Keluaran:
read 3 values, last 22.4
Contoh 31: File CSV
Menulis dan membaca tabel pembacaan.
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()
Keluaran:
08:00 22.5
08:05 22.9
Contoh 32: JSON
Mengubah data ke dan dari teks 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])
Keluaran:
{
"device": "node-1",
"readings": [
21.5,
21.9
],
"ok": true
}
21.9
Contoh 33: Path
Mengelola nama file dan folder dengan aman.
from pathlib import PurePosixPath
path = PurePosixPath("/home/pi/projects/iot/config.json")
print(path.name, path.suffix, path.parent)
print(path.parent / "logs")
Keluaran:
config.json .json /home/pi/projects/iot
/home/pi/projects/iot/logs
Contoh 34: Tanggal dan waktu
Memformat dan menambah nilai tanggal serta waktu.
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"))
Keluaran:
2026-01-15 08:00 -> 09:30
Topik 6: Kesalahan, Pencatatan, dan Pengukuran Waktu
Sebuah perangkat berjalan berbulan-bulan tanpa diawasi orang. Ia harus tetap berjalan saat ada pembacaan yang buruk, mencatat apa yang terjadi, dan mengukur waktunya sendiri.
Menangani kesalahan dengan try dan except
Letakkan kode yang bisa gagal di dalam try, tangkap pengecualian yang spesifik, dan tetap berjalan. finally selalu dijalankan, baik ada kesalahan maupun tidak, sehingga tepat untuk mematikan keluaran atau menutup berkas.
Logging sebagai ganti print
Modul logging mencatat pesan dengan level seperti INFO atau WARNING, dan Anda memilih level mana yang ditampilkan. Log dapat dikirim ke berkas, yang diperlukan ketika Pi berjalan tanpa layar.
Pemeriksaan defensif dan pengukuran waktu
assert menolak nilai yang mustahil sejak awal saat pengembangan. time.perf_counter mengukur waktu yang berlalu dengan tepat, membantu Anda menemukan kode yang lambat di komputer kecil.
Contoh 35: try dan except
Menangani pembacaan yang salah tanpa menghentikan program.
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"))
Keluaran:
23.4
bad reading: '23,4'
None
Contoh 36: finally dan pembersihan
Selalu membersihkan, meski terjadi error.
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")
Keluaran:
saved
cleaned up
Contoh 37: Exception kustom
Memunculkan error yang jelas untuk nilai sensor yang mustahil.
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)
Keluaran:
sensor error: value 120 is outside the sensor range
Contoh 38: Logging
Mencatat pesan dengan level, bukan 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")
Keluaran:
Contoh 39: Pemeriksaan defensif dengan assert
Menolak masukan yang tidak valid sejak awal.
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)
Keluaran:
50
rejected: duty must be 0 to 100
Contoh 40: Mengukur waktu berjalan
Mengukur waktu kode dengan jam yang presisi.
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)
Keluaran:
sum of squares: 2666646666700000
took measurable time: True
Topik 7: Mengolah Data Sensor
Nilai sensor mentah biasanya berisik dan sering memakai satuan yang salah. Contoh-contoh ini mengubah angka mentah menjadi pembacaan yang berguna: dikalibrasi, dihaluskan, diperiksa dari pencilan, dan dibandingkan dengan ambang.
Kalibrasi dan konversi satuan
ADC mengembalikan hitungan mentah. Dua titik referensi yang diketahui memberi kalibrasi garis lurus, sehingga hitungan dapat diubah menjadi volt. Simpan kalibrasi bersama perangkat, karena setiap sensor berbeda.
Penghalusan dan histeresis
Rata-rata bergerak dari beberapa pembacaan terakhir menghilangkan lonjakan sesaat. Histeresis memakai dua ambang, satu untuk menyalakan dan satu untuk mematikan, sehingga relay tidak bergetar saat nilai berada dekat salah satu batas.
Statistik dan pencilan
Rata-rata, median, dan simpangan baku menggambarkan satu rentang pembacaan. Median tidak terpengaruh nilai ekstrem, sehingga baik untuk membuang lonjakan yang salah sebelum menghitung rata-rata.
Contoh 41: Kalibrasi nilai ADC mentah
Mengubah nilai mentah menjadi volt dengan dua titik acuan.
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")
Keluaran:
102 -> 0.0 V
512 -> 1.652 V
921 -> 3.3 V
Contoh 42: Filter rata-rata bergerak
Menghaluskan pembacaan yang berisik dalam satu jendela.
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}")
Keluaran:
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
Contoh 43: Histeresis untuk relay kipas
Mencegah relay menyala dan mati terlalu cepat.
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")
Keluaran:
21.0 fan OFF
23.0 fan OFF
24.5 fan ON
23.5 fan ON
22.5 fan ON
21.9 fan OFF
Contoh 44: Satuan tekanan dan pembulatan
Mengubah hPa menjadi inHg dan membulatkan dengan benar.
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))
Keluaran:
1013.25 hPa = 29.92 inHg
2 4
Contoh 45: Statistik dalam jendela
Rata-rata, sebaran, dan rentang pembacaan terbaru.
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))
Keluaran:
mean 21.62
stdev 0.172
range 0.5
Contoh 46: Membuang pencilan dengan median
Mengabaikan nilai yang jauh dari pembacaan wajar.
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))
Keluaran:
[21.5, 21.7, 21.6, 21.8]
Contoh 47: Laju perubahan
Seberapa cepat nilai naik per menit.
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")
Keluaran:
0.3 C per minute
Contoh 48: Pembacaan sebagai data class
Menggambarkan satu pengukuran dengan kelas kecil.
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))
Keluaran:
Reading(device='node-1', temperature_c=21.5, humidity_pct=48.0)
{'device': 'node-1', 'temperature_c': 21.5, 'humidity_pct': 48.0}
Topik 8: GPIO: Mengendalikan LED, Tombol, Motor, dan Relay
Header 40 pin memungkinkan Pi menyalakan keluaran dan membaca masukan. Contoh-contohnya meliputi LED, tombol, peredupan PWM, servo, sensor jarak, dan relay. Selalu periksa kabel Anda dengan diagram pin sebelum memberi daya.
Keluaran, masukan, dan resistor pull-up
Atur pin sebagai keluaran untuk menyalakan LED melalui resistor, atau sebagai masukan untuk membaca tombol. Resistor pull-up menjaga masukan tetap tinggi sampai tombol menghubungkannya ke ground, sehingga pin tidak mengambang.
PWM dan servo
Modulasi lebar pulsa menyalakan dan mematikan pin dengan cepat; proporsi waktu nyala menentukan kecerahan atau posisi servo. Pustaka gpiozero menyembunyikan sebagian besar pekerjaan pengaturan waktu.
Relay, sensor, dan mematikan dengan aman
Relay menyalakan perangkat berdaya lebih besar, seperti pompa, sehingga Pi hanya menggerakkan koil kontrol yang kecil. Matikan keluaran di dalam blok finally dan panggil cleanup, supaya kerusakan program tidak membuat motor terus berjalan.
Contoh 49: Menyalakan LED (GPIO)
Menyalakan dan mematikan LED dari pin 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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 50: Membaca tombol dengan pull-up
Mendeteksi penekanan tombol dengan aman dan jeda debounce.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 51: Meredupkan LED dengan PWM
Mengubah kecerahan LED dengan modulasi lebar pulsa.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 52: LED dengan gpiozero
Pustaka GPIO yang lebih sederhana dan aman untuk 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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 53: Callback tombol
Bereaksi terhadap penekanan tombol tanpa perulangan polling.
from gpiozero import Button
from signal import pause
button = Button(27)
button.when_pressed = lambda: print("pressed")
button.when_released = lambda: print("released")
pause()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 54: Kendali servo
Menggerakkan servo ke posisi yang ditentukan.
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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 55: Sensor jarak ultrasonik
Mengukur jarak dengan sensor 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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 56: Relay dengan nilai awal yang aman
Menyalakan pompa dengan aman dan selalu mematikannya.
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
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 57: Nada buzzer
Memainkan nada sederhana pada buzzer.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 58: Mematikan GPIO dengan aman
Membuat keluaran aman sebelum program berhenti.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Topik 9: Sensor melalui I2C, SPI, 1-Wire, dan Kamera
Banyak sensor berkomunikasi dengan Pi melalui bus. I2C memakai dua kabel dan sebuah alamat, SPI memakai empat kabel dan chip select, dan 1-Wire memakai satu kabel data. Modul kamera adalah kasus khusus dengan pustakanya sendiri.
I2C: memindai dan membaca register
Setiap perangkat I2C memiliki alamat. Pindai bus dulu untuk mengetahui apa yang terhubung. Baca register ID chip untuk memastikan sensor, lalu tulis register kontrol dan baca byte datanya.
SPI dan masukan analog
Pi tidak memiliki masukan analog, sehingga ADC seperti MCP3008 mengonversi tegangan melalui SPI. Kirim nomor kanal, lalu susun nilai 10 bit dari byte balasan.
1-Wire, sensor DHT, dan kamera
DS18B20 pada 1-Wire muncul sebagai berkas di /sys/bus/w1; aktifkan antarmukanya di raspi-config dulu. Sensor DHT kadang gagal, jadi coba lagi. Modul kamera mengambil gambar diam dan bingkai dengan picamera2.
Contoh 59: Memindai bus I2C
Mencari sensor apa saja yang terhubung.
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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 60: Membaca sensor tekanan BMP280
Memeriksa ID chip dan membaca byte tekanan mentah melalui 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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 61: Suhu dan kelembapan DHT22
Membaca DHT22 dan mencoba lagi jika belum siap.
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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 62: Masukan analog dengan MCP3008
Membaca tegangan analog melalui 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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 63: Suhu DS18B20 melalui 1-Wire
Membaca sensor suhu tahan air dari kernel.
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")
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 64: Mencoba lagi untuk sensor yang tidak stabil
Membaca ulang sebelum menyerah.
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")
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 65: Kalibrasi disimpan dalam JSON
Menyimpan nilai kalibrasi dalam file dan memuatnya kembali.
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()
Keluaran:
saved: {'offset_c': -0.4, 'scale': 1.0}
Contoh 66: Mengambil foto diam
Mengambil gambar dengan modul kamera 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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 67: Deteksi gerakan dari bingkai kamera
Membandingkan bingkai untuk mendeteksi gerakan.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Topik 10: Jaringan dengan Soket dan HTTP
Perangkat IoT harus berkomunikasi dengan komputer lain. Soket adalah dasar tingkat rendah, HTTP adalah cara layanan web dan API bekerja, dan kode yang tangguh menangani batas waktu, percobaan ulang, dan masa offline.
Soket TCP dan UDP
Soket TCP adalah koneksi yang andal, sedangkan UDP mengirim pesan tunggal tanpa koneksi. Mengikat ke port 0 membuat sistem operasi memilih port yang kosong, sehingga contoh-contoh tidak saling bentrok.
Permintaan HTTP dengan batas waktu
requests melakukan panggilan HTTP dalam satu baris. Selalu atur batas waktu, atau jaringan yang tidak merespons akan membuat program macet. raise_for_status mengubah respons galat menjadi pengecualian yang dapat ditangani.
Percobaan ulang, backoff, dan antrean offline
Coba ulang permintaan yang gagal dengan exponential backoff, menunggu lebih lama setiap kali, agar server yang sibuk tidak kebanjiran. Saat jaringan mati, simpan pembacaan dalam antrean dan kirim ketika jaringan kembali.
Contoh 68: Server dan klien TCP echo
Mengirim byte ke server dan membaca balasannya.
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()
Keluaran:
echo: temp=21.5
Contoh 69: Pesan UDP pada loopback
Mengirim datagram tanpa koneksi.
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()
Keluaran:
received: humidity=48
Contoh 70: POST pembacaan ke layanan web
Mengirim pembacaan sensor sebagai JSON melalui 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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 71: GET JSON dari API dengan batas waktu
Mengambil data dan menangani kegagalan tanpa macet.
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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 72: Mencoba lagi dengan exponential backoff
Menunggu lebih lama setelah setiap percobaan gagal.
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))
Keluaran:
try 1 failed, waiting 0.1s
try 2 failed, waiting 0.2s
uploaded
Contoh 73: Mengirim JSON dengan urllib
Mengirim permintaan hanya dengan pustaka standar.
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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 74: Server web status
Menyajikan halaman status kecil di 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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 75: Memeriksa jaringan sebelum mengirim
Menyimpan data dalam antrean saat internet mati.
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")
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Topik 11: MQTT: Publish dan Subscribe
MQTT adalah protokol pesan standar untuk IoT. Perangkat mempublikasikan pembacaan ke topik, broker meneruskannya, dan berapa pun pelanggan dapat menerimanya. Protokol ini cukup ringan untuk perangkat kecil dan tahan terhadap jaringan yang lemah.
Topik, publish, dan subscribe
Topik adalah jalur seperti home/kitchen/temperature. Publish mengirim pesan ke sebuah topik melalui broker. Subscribe dapat memakai wildcard: + cocok dengan satu tingkat dan # cocok dengan semua yang ada di bawahnya.
Kualitas layanan dan pesan tersimpan
QoS 0 adalah kirim-dan-lupakan, QoS 1 menjamin pengiriman setidaknya sekali, dan QoS 2 mengirim tepat sekali. Pesan tersimpan (retained) disimpan oleh broker, sehingga pelanggan baru langsung menerima nilai terakhir.
Last will, TLS, dan discovery
Pesan last will dipublikasikan oleh broker ketika perangkat terputus secara tidak terduga, sehingga perangkat yang offline dapat terdeteksi. Gunakan TLS dan kata sandi dari variabel lingkungan untuk broker di internet. Home Assistant dapat menemukan sensor melalui pesan discovery yang dikirim ke topik khusus.
Contoh 76: MQTT: mempublikasikan telemetri
Mengirim pembacaan ke broker pada sebuah topik.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 77: MQTT: berlangganan pesan
Menerima pembacaan dari banyak perangkat dengan wildcard.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 78: MQTT: QoS dan pesan tersimpan
Memilih jaminan pengiriman dan menyimpan nilai terakhir.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 79: MQTT: last will untuk mendeteksi offline
Memberi tahu semua pihak saat perangkat terputus.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 80: MQTT melalui TLS
Mengenkripsi koneksi ke broker.
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()
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 81: Pesan discovery Home Assistant
Mempublikasikan sensor agar Home Assistant menemukannya.
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))
Keluaran:
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 }}"
}
Contoh 82: Wildcard topik MQTT
Memeriksa topik mana yang cocok dengan langganan.
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)")
Keluaran:
home/+/temperature home/kitchen/temperature match (as expected)
home/# home/kitchen/humidity match (as expected)
home/+/temperature home/kitchen/humidity no match (as expected)
Topik 12: Dasbor Web dan API dengan Flask
Flask adalah kerangka kerja web Python yang kecil. Dengan beberapa baris, ia menyajikan halaman status, mengembalikan JSON, atau menerima pembacaan dari perangkat lain, sehingga Anda dapat melihat Pi dari ponsel atau peramban.
Rute dan respons
Dekorator rute seperti @app.get('/') menghubungkan URL dengan sebuah fungsi. Fungsi mengembalikan teks, halaman, atau JSON dengan jsonify. Gunakan alamat 0.0.0.0 untuk menerima koneksi dari jaringan lokal.
Menerima data dengan validasi
Rute POST membaca JSON dari permintaan. Periksa tipe dan rentang sebelum menyimpan apa pun, dan kembalikan status 400 saat data salah, agar pengirim tahu apa yang harus diperbaiki.
Thread latar belakang dan dasbor
Thread latar belakang membaca sensor sementara server web menjawab permintaan. Jaga data bersama tetap sederhana dan aman. Sebuah templat HTML kecil mengubah pembacaan terbaru menjadi halaman yang dapat dilihat siapa pun di jaringan.
Contoh 83: Endpoint hello dengan Flask
Menyajikan rute web pertama di 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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 84: Endpoint JSON dengan Flask
Mengembalikan pembacaan terbaru sebagai 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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 85: POST Flask dengan validasi
Menerima pembacaan dan menolak nilai buruk dengan kode error.
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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 86: Dasbor HTML sederhana
Menampilkan halaman dengan suhu terbaru.
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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 87: Flask dengan thread sensor di latar belakang
Membaca sensor terus-menerus saat server web berjalan.
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)
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Topik 13: Menyimpan dan Meringkas Pembacaan
Perangkat yang mencatat selama setahun memerlukan penyimpanan yang sederhana dan tidak memenuhi kartu SD. SQLite adalah basis data satu berkas yang sudah ada di Python, dan berkas CSV cocok untuk riwayat sederhana.
Tabel dan penyisipan SQLite
SQLite menyimpan tabel dalam satu berkas. Gunakan placeholder ? dalam pernyataan INSERT, yang melindungi dari SQL injection dan mengurus tanda kutip. Panggil commit setelah menulis agar data tersimpan.
Agregasi dan keranjang waktu
GROUP BY dengan AVG dan COUNT merangkum pembacaan per perangkat. Mengelompokkan cap waktu per menit, misalnya mengambil lima karakter pertama dari waktu ISO, memberi rata-rata satu menit yang sederhana.
Retensi dan log hanya-tambah
Hapus baris lama sesuai jadwal agar penyimpanan tetap kecil. Berkas CSV hanya-tambah menambahkan satu baris per pembacaan dan menulis header hanya sekali, sederhana dan tangguh.
Contoh 88: SQLite: membuat tabel dan menyisipkan
Menyimpan pembacaan dalam file basis data.
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()
Keluaran:
2 rows stored
Contoh 89: SQLite: kueri agregat
Menghitung rata-rata dan jumlah pembacaan per perangkat.
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()
Keluaran:
node-1: average 22.00 from 2 readings
node-2: average 19.00 from 1 readings
Contoh 90: Pencatat riwayat CSV
Menambahkan pembacaan ke file riwayat dengan header.
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()
Keluaran:
ts,temp_c
08:00,21.5
08:05,21.9
Contoh 91: Mengelompokkan pembacaan per menit
Mengelompokkan pembacaan menjadi rata-rata per menit.
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))
Keluaran:
08:00 21.5
08:01 21.9
Contoh 92: Retensi data
Menghapus baris lama agar penyimpanan Pi tetap kecil.
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()
Keluaran:
deleted 5 kept 5
Topik 14: Berjalan sebagai Layanan dan Konfigurasi
Proyek hanya berguna jika ia menyala sendiri setelah listrik padam dan terus berjalan. systemd menjalankan layanan saat boot, cron menjalankan tugas berdasarkan jadwal, dan variabel lingkungan menjaga pengaturan tetap di luar kode.
Layanan systemd
Berkas unit menjelaskan cara menjalankan program, folder kerja, pengguna, dan apa yang dilakukan saat gagal dengan Restart=on-failure. Aktifkan layanan agar mulai saat boot, dan baca log-nya dengan journalctl.
Tugas terjadwal dengan cron
Satu baris crontab menjalankan perintah pada waktu yang tetap, misalnya setiap lima menit. Alihkan keluaran ke berkas log agar Anda tahu apa yang terjadi saat Anda pergi.
Konfigurasi dari lingkungan
Baca pengaturan seperti ID perangkat, interval pelaporan, dan alamat broker dari variabel lingkungan dengan nilai default yang aman. Anda dapat mengubahnya di berkas layanan tanpa mengedit kode.
Contoh 93: Layanan systemd saat boot
Menjalankan pencatat secara otomatis saat Pi menyala.
# /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
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 94: Tugas terjadwal dengan cron
Menjalankan skrip setiap lima menit.
# 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
Contoh ini memerlukan Raspberry Pi, layanan jaringan, atau layanan sistem, sehingga hanya diperiksa sintaksnya. Keluaran tidak ditampilkan.
Contoh 95: File heartbeat untuk watchdog
Menulis cap waktu agar watchdog tahu program masih hidup.
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()
Keluaran:
beat 0 written
beat 1 written
beat 2 written
last: 2
Contoh 96: Konfigurasi dari variabel lingkungan
Membaca pengaturan dengan nilai awal yang aman.
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())
Keluaran:
{'device_id': 'node-1', 'interval_s': 60, 'broker': 'localhost'}
Topik 15: Log, Keamanan, dan Jalur Pemrosesan Lengkap
Topik terakhir menyatukan semuanya: log yang dirotasi agar kartu SD tidak penuh, rahasia yang dijaga di luar kode, validasi data masuk, dan jalur pemrosesan lengkap dari sensor hingga alarm.
Log yang dirotasi dan heartbeat
Penangan berkas yang dirotasi membatasi ukuran setiap berkas log dan menyimpan beberapa berkas lama. Berkas heartbeat dengan cap waktu memungkinkan watchdog memeriksa bahwa program masih hidup.
Rahasia dan validasi
Simpan kata sandi dan token dalam variabel lingkungan, dan jangan pernah mencetaknya. Validasi setiap nilai masuk dari sisi tipe dan rentang sebelum masuk ke penyimpanan atau alarm, karena sensor dan jaringan dapat mengirim apa saja.
Menyusun jalur pemrosesan
Jalur pemrosesan membaca sensor, menghaluskan nilai dengan jendela, membandingkannya dengan ambang, menyalakan penanda alarm, dan menyimpan setiap langkah. Setiap tahap kecil dan dapat diuji, dan begitulah kode IoT profesional disusun.
Contoh 97: File log bergilir
Mencegah log memenuhi kartu 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()
Keluaran:
files: ['iot.log', 'iot.log.1', 'iot.log.2']
Contoh 98: Rahasia dari variabel lingkungan
Memuat kata sandi dari lingkungan dan tidak pernah mencetaknya.
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))
Keluaran:
loaded secret: ex************
Contoh 99: Memvalidasi data sensor yang masuk
Memeriksa tipe dan rentang sebelum menyimpan.
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}))
Keluaran:
(Sample(device='node-1', temp_c=21.5), [])
(None, ['device is required', 'temp_c out of sensor range'])
Contoh 100: Pipeline kecil yang lengkap
Membaca, menghaluskan, mendeteksi alarm, dan menyimpan hasil.
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()
Keluaran:
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
Proyek Latihan: Pemantau Ruangan
Bangun satu pemantau yang memakai gagasan dari setiap bagian. Tulis sendiri dulu, lalu bandingkan dengan contohnya. Uji bagian Python murni di komputer Anda sebelum menghubungkan perangkat keras.
1. Tulis fungsi yang mengonversi pembacaan mentah menjadi suhu menggunakan dua titik kalibrasi, lalu uji dengan tiga nilai.
2. Haluskan pembacaan dengan rata-rata bergerak dari lima nilai terakhir, dan tambahkan histeresis untuk kipas dengan ambang 24 dan 22.
3. Simpan setiap pembacaan dalam tabel SQLite, dan cetak rata-rata per menit setelah sepuluh pembacaan.
4. Publikasikan pembacaan terbaru ke topik MQTT setiap 60 detik, dan saat broker offline, masukkan pembacaan ke antrean.
5. Buat layanan systemd yang menjalankan pemantau saat boot, lalu baca log-nya dengan journalctl.
Langkah berikutnya
Berikutnya, pelajari platform IoT seperti Home Assistant atau Node-RED, tambahkan identitas perangkat yang aman dan pembaruan over-the-air, serta pelajari Raspberry Pi Pico untuk pekerjaan mikrokontroler. Uji setiap contoh perangkat keras dengan kabel Anda sendiri terlebih dahulu, dan periksa dokumentasi resmi Raspberry Pi untuk nomor pin model Anda.
Ringkasan
Raspberry Pi menjadi perangkat IoT yang sesungguhnya ketika ia membaca sensor dengan andal, mengolah data, menyimpannya, mengirimkannya melalui jaringan, dan berjalan tanpa bantuan. Melalui 100 contoh, artikel ini membahas dasar Python, keputusan dan perulangan, fungsi, struktur data, berkas dan format, penanganan kesalahan dan logging, pemrosesan sensor, GPIO, I2C, SPI, 1-Wire dan kamera, soket dan HTTP, MQTT, Flask, SQLite, systemd dan cron, serta jalur pemrosesan lengkap. Bangunlah pemantau ruangan pada latihan, dan ubah satu hal setiap kali untuk mempelajari cara setiap bagian bekerja.