%%{init: {"flowchart": {"diagramPadding": 40}}}%%
flowchart LR
A["Sensor<br/>suhu, pH, oksigen"] --> B["Perangkat<br/>MicroPython / Raspberry Pi"]
B --> C["Pengirim<br/>MQTT atau HTTP"]
C --> D["Broker / server"]
D --> E["Gudang data<br/>Parquet (Bab 50)"]
E --> F["Dasbor dan peringatan"]
Bab 52 — Perangkat Lapangan, IoT, dan Drone: Python Turun ke Kebun
Seri A2 — Data Skala Besar dan Geospasial Lanjut
Semua skrip bab ini ada di folder kode/bab52/. Di versi web, Anda juga bisa mengunduh semuanya sekaligus (zip). lapangan.py adalah modul bantu yang dipakai beberapa skrip.
Sampai sekarang, data di buku ini datang dari berkas atau dari internet. Padahal sumber data yang sebenarnya ada di lapangan: termometer di kolam, pengukur kelembapan di kebun, sensor suhu di kandang. IoT (Internet of Things, “internet untuk benda”) berarti perangkat kecil di lapangan yang membaca sensor dan mengirim datanya.
Bab ini jujur soal batasnya: perangkat keras tidak tersedia di komputer penulis. Kode yang harus berjalan di papan mikrokontroler ditampilkan dan diberi keterangan tidak dijalankan. Namun bagian yang paling banyak menentukan mutu sistem, yaitu logika, dibuat netral perangkat sehingga bisa diuji penuh di PC dengan sensor simulasi. Angka ambang di bab ini adalah angka contoh, bukan anjuran budidaya.
Perangkat: MicroPython dan Raspberry Pi
Ada dua keluarga perangkat yang umum:
- Papan mikrokontroler (misalnya jenis ESP32 atau Raspberry Pi Pico): murah, hemat daya, tanpa sistem operasi lengkap. Memakai MicroPython, versi Python yang dipangkas untuk papan kecil.
- Komputer papan tunggal (Raspberry Pi): sebuah komputer mini dengan Linux, sehingga Python biasa berjalan di dalamnya.
Kode MicroPython membaca sensor suhu-kelembapan DHT22 dan sebuah sensor analog. Menurut dokumentasi MicroPython, dht.DHT22(machine.Pin(...)) dibuat dengan nomor pin, measure() mengukur, lalu temperature() dan humidity() membaca hasilnya; DHT22 sebaiknya diukur paling cepat sekali tiap dua detik. machine.ADC(pin).read_u16() mengembalikan bilangan bulat 0 sampai 65535.
01_micropython_perangkat.py
# Bab 52 - Kode MicroPython untuk papan sensor kandang: DHT22 dan sebuah sensor analog
import time
import dht
import machine
sensor_suhu = dht.DHT22(machine.Pin(4)) # nomor pin bergantung papan dan kabel Anda
sensor_analog = machine.ADC(machine.Pin(34)) # pin ADC juga bergantung papan
while True:
sensor_suhu.measure() # DHT22: paling cepat sekali tiap 2 detik
suhu = sensor_suhu.temperature()
lembap = sensor_suhu.humidity()
mentah = sensor_analog.read_u16() # angka 0-65535
print(suhu, lembap, mentah)
time.sleep(10)Kode ini tidak dijalankan saat buku dibuat karena perangkat tidak tersedia (kode ini berjalan di papan MicroPython, bukan di PC), jadi keluarannya tidak ditampilkan. Nomor pin (4 dan 34) hanya contoh; sesuaikan dengan papan dan kabel Anda.
Pada Raspberry Pi, sensor analog butuh konverter analog-ke-digital karena Pi tidak punya masukan analog. Pustaka gpiozero menyediakan kelas MCP3008 (konverter 10 bit dengan 8 kanal) yang memberi nilai berskala 0 sampai 1.
02_raspberrypi_perangkat.py
# Bab 52 - Raspberry Pi: membaca sensor analog lewat MCP3008 dengan pustaka gpiozero
import time
from gpiozero import MCP3008
kanal0 = MCP3008(channel=0) # sensor di kanal 0 dari 8 kanal (0-7)
while True:
nilai = kanal0.value # skala 0 sampai 1
print("nilai:", round(nilai, 3), "-> tegangan:", round(nilai * 3.3, 2), "V")
time.sleep(5)Kode ini tidak dijalankan saat buku dibuat karena perangkat tidak tersedia (butuh Raspberry Pi dengan konverter analog MCP3008 yang terpasang), jadi keluarannya tidak ditampilkan.
CatatanINGAT
Kode perangkat hanya “tugas mengambil angka”. Jangan taruh logika penting (kalibrasi, ambang, penyaringan) di dalam kode yang hanya bisa diuji di papan. Taruh di fungsi biasa yang bisa diuji di PC, seperti lapangan.py berikut.
Sensor simulasi dan logika netral perangkat
Kita buat kelas SensorSimulasi yang punya antarmuka yang sama dengan sensor asli: satu metode baca(). Dengan begitu logger, pengirim, dan dasbor bisa diuji tanpa perangkat. Kelas ini membuat nilai harian yang naik-turun (oksigen terlarut tertinggi siang hari, terendah dini hari) ditambah derau acak dengan benih tetap, dan kadang “macet” (mengembalikan None).
Contoh logika netral perangkat lain: kalibrasi dua titik sensor pH. Probe pH menghasilkan tegangan; dengan dua larutan penyangga (pH 7 dan pH 4) kita dapat garis lurus yang mengubah tegangan menjadi pH. Fungsi u16_ke_volt dan volt_ke_ph bisa dipakai persis sama di papan.
03_sensor_simulasi.py
# Bab 52 - Sensor simulasi + logika netral perangkat (kalibrasi pH dari angka ADC)
from lapangan import SensorSimulasi, u16_ke_volt, volt_ke_ph
# Tiga sensor kolam tiruan. Oksigen terlarut tertinggi siang hari (puncak jam 14), terendah dini hari.
suhu = SensorSimulasi("suhu_c", rata=28.5, simpang=0.1, amplitudo_harian=1.2, benih=1)
oksigen = SensorSimulasi("oksigen_mgl", rata=5.8, simpang=0.1, amplitudo_harian=1.5, benih=2)
print("Jam suhu_c oksigen_mgl")
for jam in (2, 8, 14, 20):
print(f"{jam:3d} {suhu.baca(jam):6.2f} {oksigen.baca(jam):6.2f}")
# Logika yang sama berjalan di papan sungguhan: ADC memberi 0-65535, kita ubah ke volt lalu ke pH.
# Angka kalibrasi contoh: probe membaca 2,50 V di larutan pH 7 dan 3,00 V di larutan pH 4.
print()
for mentah in (49648, 54613, 59578):
volt = u16_ke_volt(mentah)
print(f"mentah={mentah} volt={volt:.3f} pH={volt_ke_ph(volt, 2.50, 3.00):.2f}")Jam suhu_c oksigen_mgl
2 27.40 4.33
8 28.36 5.79
14 29.56 7.29
20 28.55 5.65
mentah=49648 volt=2.500 pH=7.00
mentah=54613 volt=2.750 pH=5.50
mentah=59578 volt=3.000 pH=4.00Suhu naik dari 27,40 °C pukul 2 menjadi 29,56 °C pukul 14. Oksigen terlarut turun ke 4,33 mg/L menjelang pagi dan naik ke 7,29 mg/L siang hari, pola yang wajar karena tumbuhan air berfotosintesis pada siang hari (angka dibuat oleh simulasi, bukan pengukuran). Bagian kedua: angka ADC 49.648 setara 2,500 V dan pH 7,00; 54.613 setara 2,750 V dan pH 5,50; 59.578 setara 3,000 V dan pH 4,00. Tiga titik itu tepat berada pada garis kalibrasi.
Logger lapangan: tahan gangguan
Di lapangan, jaringan putus dan sensor macet. Logger yang baik tidak boleh kehilangan data dan tidak boleh mengarang angka. Contoh berikut memakai penyangga (buffer): bacaan dikumpulkan di memori dan ditulis ke berkas dalam kelompok, lebih hemat daripada menulis satu baris setiap kali. Bacaan yang macet disimpan sebagai sel kosong, dan alarm dicatat bila nilainya di bawah ambang.
04_logger_lapangan.py
# Bab 52 - Logger lapangan: penyangga (buffer), tulis berkala ke CSV, tangani sensor macet, alarm
import csv
from pathlib import Path
from lapangan import SensorSimulasi
tmp = Path(__file__).parent / "_tmp"
tmp.mkdir(exist_ok=True)
class LoggerLapangan:
def __init__(self, jalur, kolom, ukuran_buffer=6, batas_alarm=None):
self.jalur = jalur
self.kolom = kolom
self.ukuran_buffer = ukuran_buffer
self.batas_alarm = batas_alarm or {}
self.buffer = []
self.alarm = []
self.macet = 0
self.ditulis = 0
with open(jalur, "w", newline="", encoding="utf-8") as f:
csv.writer(f).writerow(["jam", *kolom])
def catat(self, jam, nilai):
for nama, v in nilai.items():
if v is None:
self.macet += 1 # simpan sel kosong, jangan mengarang angka
elif nama in self.batas_alarm and v < self.batas_alarm[nama]:
self.alarm.append((jam, nama, v))
self.buffer.append([jam, *[nilai[k] if nilai[k] is not None else "" for k in self.kolom]])
if len(self.buffer) >= self.ukuran_buffer:
self.tulis()
def tulis(self):
with open(self.jalur, "a", newline="", encoding="utf-8") as f:
csv.writer(f).writerows(self.buffer)
self.ditulis += len(self.buffer)
self.buffer = []
suhu = SensorSimulasi("suhu_c", 28.5, 0.2, 1.2, benih=1, peluang_macet=0.1)
oksigen = SensorSimulasi("oksigen_mgl", 5.2, 0.2, 1.8, benih=2, peluang_macet=0.1)
logger = LoggerLapangan(tmp / "kolam_a.csv", ["suhu_c", "oksigen_mgl"],
ukuran_buffer=6, batas_alarm={"oksigen_mgl": 4.0}) # ambang contoh
for langkah in range(48): # 24 jam, tiap 30 menit
jam = langkah / 2
logger.catat(jam, {"suhu_c": suhu.baca(jam), "oksigen_mgl": oksigen.baca(jam)})
logger.tulis() # kuras sisa penyangga sebelum selesai
print("Baris tertulis:", logger.ditulis, "| bacaan macet:", logger.macet)
print("Alarm oksigen < 4,0 mg/L:", len(logger.alarm), "kali")
for jam, nama, v in logger.alarm[:3]:
print(f" jam {jam:4.1f} {nama} = {v}")
print("5 baris pertama berkas:")
print("".join((tmp / "kolam_a.csv").read_text(encoding="utf-8").splitlines(keepends=True)[:5]), end="")Baris tertulis: 48 | bacaan macet: 10
Alarm oksigen < 4,0 mg/L: 11 kali
jam 0.0 oksigen_mgl = 3.71
jam 1.0 oksigen_mgl = 3.44
jam 1.5 oksigen_mgl = 3.33
5 baris pertama berkas:
jam,suhu_c,oksigen_mgl
0.0,27.66,3.71
0.5,27.11,
1.0,27.06,3.44
1.5,27.4,3.33Selama 24 jam (48 bacaan tiap 30 menit), logger menulis 48 baris, dan 10 bacaan macet dari dua sensor. Alarm oksigen di bawah 4,0 mg/L (ambang contoh) berbunyi 11 kali, terutama dini hari. Lima baris pertama berkas menunjukkan baris pukul 0,5 yang oksigennya kosong (sensor macet), bukan angka karangan.
TipTIPS
Selalu panggil flush() terakhir sebelum program berhenti, atau isi penyangga yang belum penuh hilang. Pada perangkat sungguhan, tulis juga ke penyimpanan tak-mudah-hilang secara berkala supaya listrik padam tidak menghapus data.
GPS: membaca kalimat NMEA
Modul GPS umumnya mengirim teks bernama NMEA 0183: kalimat berawalan $ yang berisi kolom dipisah koma, diakhiri * dan dua digit checksum. Menurut dokumentasi NMEA yang dipakai proyek gpsd, checksum adalah XOR 8 bit semua karakter di antara $ dan *, ditulis dua digit heksadesimal. Kalimat GGA berisi: waktu UTC, lintang (ddmm.mmm) dan belahan bumi, bujur (dddmm.mmm) dan belahan, kualitas fix, jumlah satelit, HDOP, dan ketinggian.
05_gps_nmea.py
# Bab 52 - GPS: membaca kalimat NMEA 0183 (tiruan), memeriksa checksum, mengubah ke derajat desimal
from functools import reduce
def checksum(isi):
"""XOR semua karakter di antara '$' dan '*', ditulis sebagai dua digit heksadesimal."""
return f"{reduce(lambda a, b: a ^ ord(b), isi, 0):02X}"
def buat_kalimat(isi):
return f"${isi}*{checksum(isi)}"
def valid(kalimat):
if not kalimat.startswith("$") or "*" not in kalimat:
return False
isi, _, tanda = kalimat[1:].partition("*")
return checksum(isi) == tanda.strip().upper()
def ke_derajat(nilai, belahan):
"""'0636.000' (derajat+menit: ddmm.mmm atau dddmm.mmm) -> derajat desimal; S dan W negatif."""
titik = nilai.index(".")
derajat = int(nilai[: titik - 2])
menit = float(nilai[titik - 2:])
hasil = derajat + menit / 60
return -hasil if belahan in ("S", "W") else hasil
def baca_gga(kalimat):
if not valid(kalimat):
raise ValueError("checksum salah, kalimat dibuang")
f = kalimat[1:].split("*")[0].split(",")
if not f[0].endswith("GGA"):
raise ValueError("bukan kalimat GGA")
return {
"waktu_utc": f"{f[1][0:2]}:{f[1][2:4]}:{f[1][4:6]}",
"lintang": round(ke_derajat(f[2], f[3]), 6),
"bujur": round(ke_derajat(f[4], f[5]), 6),
"kualitas_fix": int(f[6]),
"satelit": int(f[7]),
"tinggi_m": float(f[9]),
}
# Contoh klasik dari banyak buku panduan NMEA: checksum harus 47
klasik = "$GPGGA,123519,4807.038,N,01131.000,E,1,08,0.9,545.4,M,46.9,M,,*47"
print("Contoh klasik valid:", valid(klasik))
# Kalimat tiruan di sekitar Bogor (6 derajat 36 menit LS, 106 derajat 48 menit BT)
gga = buat_kalimat("GPGGA,031530,0636.000,S,10648.000,E,1,09,0.9,265.4,M,0.0,M,,")
print(gga)
print(baca_gga(gga))
# Satu karakter rusak karena gangguan kabel: checksum menolaknya
rusak = gga.replace("0636", "0637", 1)
print("Kalimat rusak valid:", valid(rusak))
try:
baca_gga(rusak)
except ValueError as galat:
print("Ditolak:", galat)Contoh klasik valid: True
$GPGGA,031530,0636.000,S,10648.000,E,1,09,0.9,265.4,M,0.0,M,,*66
{'waktu_utc': '03:15:30', 'lintang': -6.6, 'bujur': 106.8, 'kualitas_fix': 1, 'satelit': 9, 'tinggi_m': 265.4}
Kalimat rusak valid: False
Ditolak: checksum salah, kalimat dibuang- Contoh klasik dari banyak panduan NMEA lolos pemeriksaan checksum (
True): fungsichecksumkita benar. - Kalimat tiruan di sekitar Bogor dibuat dengan
buat_kalimat, lalu dibaca: pukul 03:15:30 UTC, lintang -6,6 (selatan bernilai negatif), bujur 106,8, fix GPS kualitas 1, 9 satelit, ketinggian 265,4 m. - Satu karakter diubah (
0636menjadi0637) meniru gangguan kabel: checksum menolaknya, dan program membuang kalimat itu. Tanpa checksum, Anda mungkin mencatat lokasi pohon yang meleset sekitar 1,8 km.
Ketinggian pada GGA adalah di atas permukaan laut rata-rata (geoid), yang berbeda dari ketinggian elipsoid; catat ini bila nanti menggabungkan GPS dengan data LiDAR di Bab 51.
Mengirim data: MQTT dan HTTP
Ada dua cara yang paling umum:
- MQTT: protokol ringan berbasis terbit-langganan (publish/subscribe). Perangkat menerbitkan pesan ke sebuah topik (misalnya
kebun/blokA/suhu) pada sebuah broker; program lain berlangganan topik itu dan menerima pesannya. Perangkat dan penerima tak saling kenal. Tanda+pada topik berarti “satu tingkat apa saja”. MQTT punya tiga tingkat jaminan pengiriman (QoS 0, 1, dan 2). - HTTP: perangkat memanggil server dengan
POSTberisi JSON. Lebih sederhana dan cocok bila Anda sudah punya API (Bab 36).
Untuk MQTT, kita memakai amqtt, broker dan klien MQTT yang ditulis dengan Python murni di atas asyncio (Bab 35). Dokumentasinya menyebut dukungan MQTT 3.1.1 dan Python 3.10 sampai 3.14 pada seri 0.12. Broker dijalankan di komputer sendiri (127.0.0.1, port bebas) khusus untuk uji.
06_mqtt_lokal.py
# Bab 52 - MQTT: broker lokal pure-Python (amqtt) + penerbit (publish) dan pelanggan (subscribe)
import asyncio
import json
import socket
from amqtt.broker import Broker
from amqtt.client import MQTTClient
from amqtt.mqtt.constants import QOS_1
def port_bebas():
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
async def main():
port = port_bebas()
konfigurasi = {
"listeners": {"default": {"type": "tcp", "bind": f"127.0.0.1:{port}"}},
"plugins": { # hanya untuk uji lokal: tanpa kata sandi
"amqtt.plugins.authentication.AnonymousAuthPlugin": {"allow_anonymous": True},
},
}
broker = Broker(konfigurasi)
await broker.start()
pelanggan = MQTTClient(client_id="dasbor")
await pelanggan.connect(f"mqtt://127.0.0.1:{port}/")
await pelanggan.subscribe([("kebun/+/suhu", QOS_1)]) # + = satu tingkat topik, bebas isinya
penerbit = MQTTClient(client_id="stasiun-kebun")
await penerbit.connect(f"mqtt://127.0.0.1:{port}/")
await penerbit.publish("kebun/blokA/suhu", json.dumps({"c": 27.4}).encode(), qos=QOS_1)
await penerbit.publish("kebun/blokB/suhu", json.dumps({"c": 28.1}).encode(), qos=QOS_1)
await penerbit.publish("kolam/A/ph", json.dumps({"ph": 7.1}).encode(), qos=QOS_1) # tidak cocok langganan
for _ in range(2):
pesan = await asyncio.wait_for(pelanggan.deliver_message(), timeout=5)
print(pesan.topic, "->", json.loads(pesan.data.decode()))
try:
await asyncio.wait_for(pelanggan.deliver_message(), timeout=1)
except asyncio.TimeoutError:
print("Pesan 'kolam/A/ph' tidak sampai: topiknya tidak cocok dengan langganan")
await penerbit.disconnect()
await pelanggan.disconnect()
await broker.shutdown()
asyncio.run(main())kebun/blokA/suhu -> {'c': 27.4}
kebun/blokB/suhu -> {'c': 28.1}
Pesan 'kolam/A/ph' tidak sampai: topiknya tidak cocok dengan langgananPelanggan dasbor berlangganan kebun/+/suhu. Penerbit mengirim tiga pesan: dua cocok (kebun/blokA/suhu dan kebun/blokB/suhu) dan tiba; satu lagi (kolam/A/ph) tidak cocok sehingga tidak sampai, dan program mencetak penjelasannya setelah menunggu satu detik.
PeringatanAWAS
Pada contoh ini, broker menerima sambungan tanpa kata sandi (allow_anonymous) dan hanya mendengarkan 127.0.0.1. Itu hanya untuk uji lokal. Broker sungguhan wajib memakai nama pengguna dan kata sandi serta enkripsi TLS (Bab 49).
Sekarang HTTP, dengan server uji lokal yang sengaja menolak permintaan pertama (status 503, “server sibuk”) supaya kita melatih ulang otomatis. Satu permintaan membawa banyak bacaan (batch): jauh lebih hemat daripada mengirim satu per satu.
07_http_kirim.py
# Bab 52 - Mengirim data lewat HTTP: paket JSON, server uji lokal, ulang saat gagal
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
import requests
diterima = []
hitung = {"permintaan": 0}
class ServerUji(BaseHTTPRequestHandler):
def log_message(self, *argumen):
pass
def do_POST(self):
hitung["permintaan"] += 1
panjang = int(self.headers["Content-Length"])
isi = json.loads(self.rfile.read(panjang))
if hitung["permintaan"] == 1: # permintaan pertama sengaja digagalkan
self.send_response(503)
self.end_headers()
return
diterima.extend(isi["bacaan"])
self.send_response(201)
self.end_headers()
server = HTTPServer(("127.0.0.1", 0), ServerUji)
threading.Thread(target=server.serve_forever, daemon=True).start()
alamat = f"http://127.0.0.1:{server.server_port}/bacaan"
def kirim(paket, maks_percobaan=4):
"""Kirim satu paket; ulangi bila server sibuk (503). Jeda dibuat nol agar contoh cepat."""
for percobaan in range(1, maks_percobaan + 1):
r = requests.post(alamat, json=paket, timeout=5)
if r.status_code == 201:
return percobaan
print(f" percobaan {percobaan}: status {r.status_code}, coba lagi")
raise RuntimeError("gagal terkirim; simpan di logger dan coba nanti")
# Satu permintaan membawa BANYAK bacaan (batch): jauh lebih hemat daripada satu per satu
paket = {"stasiun": "kolam-A", "bacaan": [
{"jam": 6.0, "suhu_c": 27.5, "ph": 7.1},
{"jam": 6.5, "suhu_c": 27.6, "ph": 7.1},
{"jam": 7.0, "suhu_c": 27.9, "ph": 7.2},
]}
print("Berhasil pada percobaan ke-", kirim(paket), sep="")
server.shutdown()
print("Bacaan diterima server:", len(diterima), "| permintaan total:", hitung["permintaan"]) percobaan 1: status 503, coba lagi
Berhasil pada percobaan ke-2
Bacaan diterima server: 3 | permintaan total: 2Percobaan pertama menerima 503; percobaan kedua berhasil (201). Server menerima 3 bacaan dalam 2 permintaan total. Pada sistem nyata, beri jeda yang makin lama di antara percobaan, dan bila semua gagal, simpan di logger dan kirim nanti.
Data dan citra drone
Foto dari drone membawa EXIF: metadata di dalam berkas JPEG, termasuk merek kamera, waktu, panjang fokus, dan posisi GPS. Pillow bisa menulis dan membaca EXIF. Kita membuat foto tiruan, menyisipkan EXIF buatan sendiri, lalu membacanya kembali dan menghitung GSD (ground sample distance), yaitu ukuran satu piksel di tanah: tinggi terbang x lebar sensor / (panjang fokus x lebar citra).
08_exif_drone.py
# Bab 52 - Citra drone: menulis dan membaca EXIF (posisi GPS, tinggi terbang) dengan Pillow
from pathlib import Path
from PIL import ExifTags, Image
from PIL.ExifTags import GPS, IFD
tmp = Path(__file__).parent / "_tmp"
tmp.mkdir(exist_ok=True)
def ke_dms(derajat):
d = int(derajat)
m = int((derajat - d) * 60)
s = round(((derajat - d) * 60 - m) * 60, 4)
return (d, m, s)
def dari_dms(dms, referensi):
d, m, s = (float(v) for v in dms)
nilai = d + m / 60 + s / 3600
return -nilai if referensi in ("S", "W") else nilai
# 1. Foto tiruan 'dari drone': gambar polos 400x300 + metadata EXIF yang kita tulis sendiri
foto = Image.new("RGB", (400, 300), (90, 140, 70))
exif = Image.Exif()
exif[ExifTags.Base.Make] = "DroneContoh"
exif[ExifTags.Base.Model] = "DC-1"
exif[ExifTags.Base.DateTimeOriginal] = "2026:03:01 09:30:15"
exif[ExifTags.Base.FocalLength] = 8.8 # mm
gps = exif.get_ifd(IFD.GPSInfo)
gps[GPS.GPSLatitudeRef] = "S"
gps[GPS.GPSLatitude] = ke_dms(6.6012)
gps[GPS.GPSLongitudeRef] = "E"
gps[GPS.GPSLongitude] = ke_dms(106.8034)
gps[GPS.GPSAltitude] = 120.0 # tinggi terbang (m)
foto.save(tmp / "drone_001.jpg", exif=exif)
# 2. Membaca kembali, seperti yang akan Anda lakukan pada ratusan foto drone nyata
baca = Image.open(tmp / "drone_001.jpg")
data = baca.getexif()
info_gps = data.get_ifd(IFD.GPSInfo)
lintang = dari_dms(info_gps[GPS.GPSLatitude], info_gps[GPS.GPSLatitudeRef])
bujur = dari_dms(info_gps[GPS.GPSLongitude], info_gps[GPS.GPSLongitudeRef])
tinggi = float(info_gps[GPS.GPSAltitude])
print("Kamera:", data[ExifTags.Base.Make], data[ExifTags.Base.Model])
print("Waktu :", data[ExifTags.Base.DateTimeOriginal])
print(f"Posisi: {lintang:.4f}, {bujur:.4f} tinggi {tinggi:.0f} m")
# 3. Ukuran piksel di tanah (GSD) = tinggi x lebar sensor / (fokus x lebar citra)
lebar_sensor_mm = 13.2 # angka contoh untuk kamera fiktif ini
fokus_mm = float(data[ExifTags.Base.FocalLength])
gsd_cm = tinggi * lebar_sensor_mm / (fokus_mm * baca.width) * 100
print(f"GSD (ukuran piksel di tanah): {gsd_cm:.1f} cm")
print(f"Cakupan satu foto: {gsd_cm * baca.width / 100:.0f} m x {gsd_cm * baca.height / 100:.0f} m")Kamera: DroneContoh DC-1
Waktu : 2026:03:01 09:30:15
Posisi: -6.6012, 106.8034 tinggi 120 m
GSD (ukuran piksel di tanah): 45.0 cm
Cakupan satu foto: 180 m x 135 mPosisi (-6,6012; 106,8034) dan tinggi terbang 120 m terbaca kembali dengan benar. Karena fotonya hanya 400 piksel lebar, GSD-nya 45 cm dan satu foto menutup 180 x 135 m. Drone sungguhan memotret ribuan piksel lebar, sehingga GSD-nya hanya beberapa sentimeter. Angka lebar sensor 13,2 mm dan fokus 8,8 mm adalah angka contoh untuk kamera fiktif.
Dari ratusan foto drone, Anda bisa menggabungkannya menjadi peta ortofoto dengan perangkat fotogrametri (misalnya OpenDroneMap), tetapi itu di luar cakupan bab ini dan tidak dicoba di sini.
Proyek Bersambung A2: stasiun lapangan mini
Proyek A2 punya tiga sumber data. Bab 50 membuat gudang Parquet; Bab 51 menangani citra dan LiDAR; sekarang kita membuat stasiun mini yang membaca sensor kebun, kolam, dan kandang selama 24 jam lalu mencatat hasilnya dalam satu tabel panjang.
09_proyek_stasiun.py
# Bab 52 - Proyek bersambung A2: stasiun lapangan mini (kebun, kolam, kandang) dengan sensor simulasi
from pathlib import Path
import polars as pl
from lapangan import SensorSimulasi
tmp = Path(__file__).parent / "_tmp"
tmp.mkdir(exist_ok=True)
# Satu stasiun = beberapa sensor. Angka rata-rata dan ambang adalah angka contoh, bukan anjuran budidaya.
stasiun = {
"kebun": {"suhu_c": SensorSimulasi("suhu_c", 26.0, 0.3, 3.0, benih=11),
"kelembapan_tanah_pct": SensorSimulasi("kelembapan_tanah_pct", 38.0, 0.8, 0.0, benih=12)},
"kolam": {"suhu_c": SensorSimulasi("suhu_c", 28.5, 0.1, 1.2, benih=21),
"ph": SensorSimulasi("ph", 7.2, 0.05, 0.3, benih=22),
"oksigen_mgl": SensorSimulasi("oksigen_mgl", 5.2, 0.15, 1.8, benih=23, peluang_macet=0.05)},
"kandang": {"suhu_c": SensorSimulasi("suhu_c", 29.0, 0.3, 2.5, benih=31),
"kelembapan_pct": SensorSimulasi("kelembapan_pct", 72.0, 1.0, -6.0, benih=32)},
}
ambang_rendah = {("kolam", "oksigen_mgl"): 4.0} # contoh
ambang_tinggi = {("kandang", "suhu_c"): 31.0} # contoh
# Tiap 30 menit selama 24 jam: baca semua sensor, simpan dalam format panjang
baris = []
for langkah in range(48):
jam = langkah / 2
for unit, sensor in stasiun.items():
for besaran, s in sensor.items():
baris.append((unit, besaran, jam, s.baca(jam)))
df = pl.DataFrame(baris, schema=["unit", "besaran", "jam", "nilai"], orient="row")
df.write_csv(tmp / "stasiun_24jam.csv")
print("Bacaan:", df.height, "| sensor macet (kosong):", df["nilai"].null_count())
ringkas = (df.group_by("unit", "besaran", maintain_order=True)
.agg(pl.col("nilai").mean().round(2).alias("rata"),
pl.col("nilai").min().alias("min"), pl.col("nilai").max().alias("maks")))
print(ringkas)
for (unit, besaran), batas in ambang_rendah.items():
n = df.filter((pl.col("unit") == unit) & (pl.col("besaran") == besaran) & (pl.col("nilai") < batas)).height
print(f"Peringatan {unit}/{besaran} < {batas}: {n} bacaan")
for (unit, besaran), batas in ambang_tinggi.items():
n = df.filter((pl.col("unit") == unit) & (pl.col("besaran") == besaran) & (pl.col("nilai") > batas)).height
print(f"Peringatan {unit}/{besaran} > {batas}: {n} bacaan")Bacaan: 336 | sensor macet (kosong): 2
shape: (7, 5)
┌─────────┬──────────────────────┬───────┬───────┬───────┐
│ unit ┆ besaran ┆ rata ┆ min ┆ maks │
│ --- ┆ --- ┆ --- ┆ --- ┆ --- │
│ str ┆ str ┆ f64 ┆ f64 ┆ f64 │
╞═════════╪══════════════════════╪═══════╪═══════╪═══════╡
│ kebun ┆ suhu_c ┆ 25.97 ┆ 22.77 ┆ 29.44 │
│ kebun ┆ kelembapan_tanah_pct ┆ 37.92 ┆ 36.55 ┆ 39.26 │
│ kolam ┆ suhu_c ┆ 28.51 ┆ 27.14 ┆ 29.9 │
│ kolam ┆ ph ┆ 7.2 ┆ 6.89 ┆ 7.52 │
│ kolam ┆ oksigen_mgl ┆ 5.24 ┆ 3.3 ┆ 7.23 │
│ kandang ┆ suhu_c ┆ 28.99 ┆ 26.08 ┆ 31.77 │
│ kandang ┆ kelembapan_pct ┆ 72.18 ┆ 65.45 ┆ 79.47 │
└─────────┴──────────────────────┴───────┴───────┴───────┘
Peringatan kolam/oksigen_mgl < 4.0: 13 bacaan
Peringatan kandang/suhu_c > 31.0: 9 bacaanAda 336 bacaan (7 sensor x 48), dengan 2 yang kosong karena sensor oksigen macet. Ringkasan menunjukkan rata-rata kolam: suhu 28,51 °C, pH 7,2, oksigen 5,24 mg/L (minimum 3,3). Dua peringatan contoh menyala: oksigen kolam di bawah 4,0 sebanyak 13 bacaan, dan suhu kandang di atas 31,0 sebanyak 9 bacaan. Berkas CSV hasilnya siap dimasukkan ke gudang Parquet milik Bab 50.
Latihan
- Pada
03_sensor_simulasi.py, tambahkan sensor kelembapan kandang (rata-rata 72, amplitudo harian -6, puncak jam 14) dan cetak nilainya pukul 2, 8, 14, dan 20. - Ubah
peluang_macetdi04_logger_lapangan.pymenjadi 0,3. Berapa sel kosong yang muncul? Apakah logger tetap berjalan sampai 48 baris? - Buat satu kalimat GGA baru dengan
buat_kalimat, lalu baca denganbaca_gga. Pastikan checksumnya valid. - Tantangan: ubah
06_mqtt_lokal.pyagar berlangganan+/+/suhu(dua tingkat), lalu terbitkan jugakolam/A/suhudankandang/K1/suhu. Pesan mana saja yang sampai?
Ringkasan Bab
- Sistem IoT: sensor, perangkat, pengirim, broker/server, gudang data, dasbor.
- MicroPython (papan mikrokontroler) dan Raspberry Pi (komputer mini) membaca sensor; kodenya hanya bisa diuji di perangkat, jadi logika dipisahkan dan diuji dengan sensor simulasi.
- Logger yang baik memakai penyangga, menyimpan sel kosong untuk bacaan macet, dan memanggil
flush()di akhir. - NMEA dari GPS divalidasi dengan checksum sebelum dipakai.
- MQTT (terbit-langganan, topik,
+) dan HTTP (POSTJSON, ulang bila gagal) mengirim data; broker uji hanya untuk lokal. - EXIF foto drone memuat posisi dan tinggi terbang; GSD = tinggi x lebar sensor / (fokus x lebar citra).
Selanjutnya, di Bab 53: data sudah mengalir dan menumpuk. Saat tabel dan citra tak lagi muat di memori, kita belajar memecahnya dengan Dask.