Otomasi IoT dengan Python: MQTT, GPIO & Scheduling

"""
Pembacaan Sensor DHT11/DHT22 dengan Python
BeebaneLabs - https://beebanelabs.pages.dev

Rangkaian:
  DHT11 Pin 1 (VCC) → 3.3V Raspberry Pi
  DHT11 Pin 2 (DATA) → GPIO4 (Pin 7) + Resistor 10kΩ ke 3.3V
  DHT11 Pin 4 (GND) → GND Raspberry Pi
"""

import time
import board
import adafruit_dht

# --- Inisialisasi Sensor ---
# Untuk DHT11 gunakan adafruit_dht.DHT11(board.D4)
# Untuk DHT22 gunakan adafruit_dht.DHT22(board.D4)
sensor = adafruit_dht.DHT11(board.D4)

# Interval pembacaan (detik)
interval = 5


def baca_sensor():
    """Membaca data dari sensor DHT dan mengembalikan dict."""
    try:
        suhu = sensor.temperature       # Celsius
        kelembaban = sensor.humidity     # Persen

        if suhu is None or kelembaban is None:
            return None

        return {
            "suhu": round(suhu, 1),
            "kelembaban": round(kelembaban, 1),
            "timestamp": int(time.time())
        }

    except RuntimeError as e:
        # Error umum DHT: CRC check failed
        print(f"[WARN] Sensor error: {e}")
        return None
    except Exception as e:
        print(f"[ERROR] Error tak terduga: {e}")
        return None


def main():
    print("=== Pembaca Sensor DHT11 ===")
    print(f"Membaca setiap {interval} detik...\n")

    try:
        while True:
            data = baca_sensor()

            if data:
                print(f"🌡️  Suhu       : {data['suhu']} °C")
                print(f"💧 Kelembaban : {data['kelembaban']} %")
                print(f"⏰ Timestamp  : {data['timestamp']}")
                print("---")
            else:
                print("[SKIP] Pembacaan gagal, coba lagi...\n")

            time.sleep(interval)

    except KeyboardInterrupt:
        print("\n[INFO] Program dihentikan.")
    finally:
        sensor.exit()


if __name__ == "__main__":
    main()

"""
Pembacaan Sensor BME280 via I2C dengan smbus2
BeebaneLabs - https://beebanelabs.pages.dev

Rangkaian:
  BME280 VCC → 3.3V
  BME280 GND → GND
  BME280 SDA → GPIO2 (Pin 3)
  BME280 SCL → GPIO3 (Pin 5)
  BME280 SDO → GND (Alamat I2C: 0x76)
"""

import time
import struct
from smbus2 import SMBus

# Alamat default BME280 (jika SDO ke GND = 0x76, ke VCC = 0x77)
BME280_ADDR = 0x76

# Register BME280
REG_ID          = 0xD0
REG_RESET       = 0xE0
REG_CTRL_HUM    = 0xF2
REG_CTRL_MEAS   = 0xF4
REG_CONFIG      = 0xF5
REG_DATA_START  = 0xF7

# Kalibrasi data register
CALIB_26_41     = 0xE1
CALIB_00_25     = 0x88
CALIB_00_25_LEN = 26
CALIB_26_41_LEN = 7


class BME280:
    """Driver sederhana untuk sensor BME280."""

    def __init__(self, bus_number=1, address=BME280_ADDR):
        self.bus = SMBus(bus_number)
        self.addr = address

        # Verifikasi ID chip (harus 0x60)
        chip_id = self.bus.read_byte_data(self.addr, REG_ID)
        if chip_id != 0x60:
            raise RuntimeError(f"BME280 tidak terdeteksi! Chip ID: {chip_id:#x}")

        # Baca data kalibrasi
        self._load_calibration()

        # Konfigurasi: oversampling, mode normal
        self.bus.write_byte_data(self.addr, REG_CTRL_HUM, 0x05)    # Humidity x16
        self.bus.write_byte_data(self.addr, REG_CTRL_MEAS, 0xB7)   # Temp x16, Press x16, Normal mode
        self.bus.write_byte_data(self.addr, REG_CONFIG, 0xA0)       # Standby 1000ms

    def _load_calibration(self):
        """Baca register kalibrasi untuk kompensasi suhu & tekanan."""
        cal26 = self.bus.read_i2c_block_data(self.addr, CALIB_26_41, CALIB_26_41_LEN)
        cal00 = self.bus.read_i2c_block_data(self.addr, CALIB_00_25, CALIB_00_25_LEN)

        # Kompensasi suhu
        self.dig_T1 = struct.unpack(' 2047:
            self.dig_H4 -= 4096
        self.dig_H5 = (cal26[6] << 4) | ((cal26[5] >> 4) & 0x0F)
        if self.dig_H5 > 2047:
            self.dig_H5 -= 4096
        self.dig_H6 = struct.unpack('> 4)
        raw_temp  = (data[3] << 12) | (data[4] << 4) | (data[5] >> 4)
        raw_hum   = (data[6] << 8)  | data[7]
        return raw_temp, raw_press, raw_hum

    def read_compensated(self):
        """Baca dan kompensasi data suhu, tekanan, kelembaban."""
        raw_temp, raw_press, raw_hum = self.read_raw()

        # Kompensasi suhu (dari datasheet BME280)
        var1 = (raw_temp / 16384.0 - self.dig_T1 / 1024.0) * self.dig_T2
        var2 = ((raw_temp / 131072.0 - self.dig_T1 / 8192.0) ** 2) * self.dig_T3
        t_fine = var1 + var2
        suhu = t_fine / 5120.0

        # Kompensasi kelembaban
        h = t_fine - 76800.0
        h = (raw_hum - (self.dig_H4 * 64.0 + self.dig_H5 / 16384.0 * h)) * \
            (self.dig_H2 / 65536.0 * (1.0 + self.dig_H6 / 67108864.0 * h *
            (1.0 + self.dig_H3 / 67108864.0 * h)))
        kelembaban = max(0.0, min(h, 100.0))

        return round(suhu, 2), round(kelembaban, 2)


# --- Program Utama ---
if __name__ == "__main__":
    sensor = BME280()
    print("=== BME280 Sensor Reader ===\n")

    try:
        while True:
            suhu, kelembaban = sensor.read_compensated()
            print(f"🌡️  Suhu       : {suhu} °C")
            print(f"💧 Kelembaban : {kelembaban} %")
            print("---")
            time.sleep(3)

    except KeyboardInterrupt:
        print("\n[INFO] Selesai.")