Koding Indonesia

Pendahuluan — Kenapa Subscriber Python?

Di broker Mosquitto (#16), ESP32 sudah publish JSON sensor ke topic MQTT. Di artikel NTP (#34), setiap payload punya field timestamp dan unix — tapi data itu masih lewat kecuali ada proses yang menangkap dan menyimpan ke database.

Artikel Jalur B ini melengkapi stack: script Python berjalan di Raspberry Pi / VPS, subscribe topic sensor, lalu INSERT ke MySQL. Hasilnya: histori suhu & kelembaban bisa di-query, diekspor, atau divisualisasikan di InfluxDB + Grafana (#19).

Prasyarat: Broker Mosquitto #16 jalan, ESP32 publish JSON dengan timestamp (#34), paham dasar MQTT (#7). Opsi production: transport TLS dari #17.

Arsitektur: MQTT → Python → MySQL

  [ ESP32 + DHT22 ]
        | publish JSON (suhu, kelembaban, timestamp, unix)
        v
  [ Mosquitto #16 ]  topic: kodingindonesia/esp32/dht22/data
        |
        | subscribe (user terpisah: kindo_subscriber)
        v
  [ Python + paho-mqtt ]  on_message → parse JSON → INSERT
        v
  [ MySQL ]  tabel sensor_readings
        |
        +-- Query SQL / phpMyAdmin
        +-- Grafana (#19) / export CSV

Payload contoh (dari ESP32 #34):

{"suhu":28.5,"kelembaban":65.2,"timestamp":"2026-07-02T14:30:00","unix":1782977400}

Yang Kamu Butuhkan

Raspberry Pi / VPS — tempat broker (#16) dan MySQL (bisa satu mesin)
MySQL 8 atau MariaDB 10.6+
Python 3.10+ dengan venv
Library: paho-mqtt, mysql-connector-python
ESP32 publish ke topic kodingindonesia/esp32/dht22/data (broker 192.168.1.50:1883)

Estimasi biaya: Rp 0 jika pakai Pi/VPS yang sudah ada untuk Mosquitto; MySQL open source.

Skema Database MySQL

Buat database dan tabel untuk histori sensor:

CREATE DATABASE IF NOT EXISTS kindo_iot
  CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

CREATE USER IF NOT EXISTS 'kindo_iot'@'localhost' IDENTIFIED BY 'GANTI_PASSWORD_MYSQL';
GRANT ALL PRIVILEGES ON kindo_iot.* TO 'kindo_iot'@'localhost';
FLUSH PRIVILEGES;

USE kindo_iot;

CREATE TABLE IF NOT EXISTS sensor_readings (
  id           BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  device_id    VARCHAR(64)  NOT NULL DEFAULT 'esp32-dht22',
  suhu         DECIMAL(5,2) NULL,
  kelembaban   DECIMAL(5,2) NULL,
  recorded_at  DATETIME     NOT NULL,
  unix_ts      INT UNSIGNED NULL,
  topic        VARCHAR(128) NOT NULL,
  created_at   TIMESTAMP    DEFAULT CURRENT_TIMESTAMP,
  INDEX idx_recorded_at (recorded_at),
  INDEX idx_device (device_id)
) ENGINE=InnoDB;

Kolom recorded_at diisi dari field unix payload (UTC epoch) — lebih andal daripada string ISO tanpa timezone.

User MQTT Subscriber di Mosquitto

Pisahkan user subscriber dari publisher ESP32 (kindo_esp32 / GANTI_PASSWORD_MQTT di #16):

sudo mosquitto_passwd /etc/mosquitto/passwd kindo_subscriber
# masukkan password — jangan sama dengan kindo_esp32

sudo systemctl restart mosquitto

Pro tip: Beri ACL Mosquitto agar kindo_subscriber hanya read topic sensor, sedangkan kindo_esp32 hanya write — mencegah subscriber ikut publish data palsu.

Setup Python (venv + dependensi)

sudo apt update
sudo apt install -y python3-venv python3-pip mysql-server

mkdir -p ~/kindo-mqtt-mysql && cd ~/kindo-mqtt-mysql
python3 -m venv .venv
source .venv/bin/activate
pip install paho-mqtt mysql-connector-python

Simpan kredensial di file env (jangan commit ke Git):

cat > .env <<'EOF'
MQTT_HOST=192.168.1.50
MQTT_PORT=1883
MQTT_USER=kindo_subscriber
MQTT_PASS=GANTI_PASSWORD_SUBSCRIBER
MQTT_TOPIC=kodingindonesia/esp32/dht22/data
# Untuk TLS (#17): MQTT_PORT=8883 + tls_set() di script

MYSQL_HOST=127.0.0.1
MYSQL_USER=kindo_iot
MYSQL_PASS=GANTI_PASSWORD_MYSQL
MYSQL_DB=kindo_iot
EOF
chmod 600 .env

Script Python: Subscriber + INSERT MySQL

File subscriber_mqtt_mysql.py:

#!/usr/bin/env python3
"""Subscriber MQTT → MySQL untuk sensor ESP32 (Seri 2 Koding Indonesia)."""

import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

import mysql.connector
import paho.mqtt.client as mqtt

# Muat .env sederhana (tanpa library tambahan)
env_path = Path(__file__).with_name(".env")
if env_path.exists():
    for line in env_path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line and not line.startswith("#") and "=" in line:
            k, v = line.split("=", 1)
            os.environ.setdefault(k.strip(), v.strip())

MQTT_HOST = os.getenv("MQTT_HOST", "192.168.1.50")
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
MQTT_USER = os.getenv("MQTT_USER", "kindo_subscriber")
MQTT_PASS = os.getenv("MQTT_PASS", "GANTI_PASSWORD_SUBSCRIBER")
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "kodingindonesia/esp32/dht22/data")

MYSQL_CFG = {
    "host": os.getenv("MYSQL_HOST", "127.0.0.1"),
    "user": os.getenv("MYSQL_USER", "kindo_iot"),
    "password": os.getenv("MYSQL_PASS", "GANTI_PASSWORD_MYSQL"),
    "database": os.getenv("MYSQL_DB", "kindo_iot"),
}


def koneksi_mysql():
    return mysql.connector.connect(**MYSQL_CFG)


def parse_waktu(payload: dict) -> datetime:
    """Prioritas: unix epoch → fallback ISO timestamp lokal."""
    if "unix" in payload and payload["unix"]:
        return datetime.fromtimestamp(int(payload["unix"]), tz=timezone.utc).replace(tzinfo=None)
    if "timestamp" in payload and payload["timestamp"]:
        return datetime.strptime(payload["timestamp"], "%Y-%m-%dT%H:%M:%S")
    return datetime.utcnow()


def simpan_ke_mysql(topic: str, payload: dict) -> None:
    suhu = payload.get("suhu")
    rh = payload.get("kelembaban")
    if suhu is None and rh is None:
        return

    recorded = parse_waktu(payload)
    unix_ts = payload.get("unix")

    sql = """
        INSERT INTO sensor_readings
          (device_id, suhu, kelembaban, recorded_at, unix_ts, topic)
        VALUES (%s, %s, %s, %s, %s, %s)
    """
    vals = ("esp32-dht22", suhu, rh, recorded, unix_ts, topic)

    conn = koneksi_mysql()
    try:
        cur = conn.cursor()
        cur.execute(sql, vals)
        conn.commit()
        print(f"OK INSERT {recorded} suhu={suhu} rh={rh}", flush=True)
    finally:
        conn.close()


def on_connect(client, userdata, flags, reason_code, properties=None):
    if reason_code == 0:
        print(f"MQTT connected → subscribe {MQTT_TOPIC}", flush=True)
        client.subscribe(MQTT_TOPIC, qos=1)
    else:
        print(f"MQTT connect gagal rc={reason_code}", flush=True)


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload.decode("utf-8"))
    except json.JSONDecodeError as e:
        print(f"JSON invalid: {e}", flush=True)
        return
    try:
        simpan_ke_mysql(msg.topic, data)
    except mysql.connector.Error as e:
        print(f"MySQL error: {e}", flush=True)


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="kindo-python-subscriber")
    client.username_pw_set(MQTT_USER, MQTT_PASS)
    client.on_connect = on_connect
    client.on_message = on_message

    print(f"Connect {MQTT_HOST}:{MQTT_PORT} ...", flush=True)
    client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
    client.loop_forever()


if __name__ == "__main__":
    main()

Penjelasan Bagian Kritis

parse_waktu() — pakai unix dari ESP32 (#34) agar urutan waktu konsisten di MySQL.
User terpisah — kindo_subscriber vs kindo_esp32; jangan pakai credential publisher untuk script server.
QoS 1 subscribe — pesan sensor tidak hilang saat subscriber offline sebentar (broker retain QoS1 queue).
mysql.connector — koneksi per INSERT sederhana untuk tutorial; production bisa pakai connection pool.
Topic — harus sama dengan ESP32: kodingindonesia/esp32/dht22/data.
Tanpa timestamp di payload — script fallback ke utcnow(), tapi sebaiknya ESP32 sudah ikut #34.

Jalankan sebagai Layanan (systemd)

Agar subscriber jalan otomatis setelah reboot Pi:

# /etc/systemd/system/kindo-mqtt-mysql.service
[Unit]
Description=Koding Indonesia MQTT subscriber to MySQL
After=network.target mosquitto.service mysql.service

[Service]
Type=simple
User=pi
WorkingDirectory=/home/pi/kindo-mqtt-mysql
EnvironmentFile=/home/pi/kindo-mqtt-mysql/.env
ExecStart=/home/pi/kindo-mqtt-mysql/.venv/bin/python subscriber_mqtt_mysql.py
Restart=on-failure
RestartSec=5

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now kindo-mqtt-mysql
sudo systemctl status kindo-mqtt-mysql
journalctl -u kindo-mqtt-mysql -f

Opsi TLS (Port 8883)

Jika broker sudah pakai TLS dari artikel #17, tambahkan sebelum connect():

import ssl

MQTT_PORT = 8883
client.tls_set(
    ca_certs="/path/to/ca.crt",
    certfile=None,
    keyfile=None,
    cert_reqs=ssl.CERT_REQUIRED,
    tls_version=ssl.PROTOCOL_TLS_CLIENT,
)
# client.tls_insecure_set(True)  # JANGAN di produksi

Subscriber CLI dengan TLS:

mosquitto_sub -h 192.168.1.50 -p 8883 \
  --cafile ca.crt \
  -u kindo_subscriber -P 'GANTI_PASSWORD_SUBSCRIBER' \
  -t "kodingindonesia/esp32/dht22/data" -v

Verifikasi & Query SQL

Setelah ESP32 publish dan script jalan, cek data masuk:

SELECT id, device_id, suhu, kelembaban, recorded_at, unix_ts, topic
FROM sensor_readings
ORDER BY recorded_at DESC
LIMIT 10;

-- Rata-rata suhu per jam (contoh analitik sederhana)
SELECT DATE_FORMAT(recorded_at, '%Y-%m-%d %H:00') AS jam,
       ROUND(AVG(suhu), 1) AS avg_suhu,
       COUNT(*) AS jumlah
FROM sensor_readings
GROUP BY jam
ORDER BY jam DESC
LIMIT 24;

Uji Coba (Checklist)

MySQL: tabel sensor_readings ada, user kindo_iot bisa INSERT
Mosquitto: user kindo_subscriber bisa subscribe topic sensor
Jalankan script → log MQTT connected → subscribe ...
ESP32 publish (atau mosquitto_pub dengan JSON + unix) → log OK INSERT ...
Query SQL menampilkan baris baru dengan recorded_at masuk akal
Restart Pi → systemd auto-start subscriber
Opsional: aktifkan TLS (#17) dan uji port 8883

Tips & Troubleshooting

rc=5 MQTT: User/password salah — cek .env vs /etc/mosquitto/passwd
MySQL Access denied: Grant belum jalan atau password beda dengan .env
JSON invalid: Payload bukan UTF-8 JSON — cek ESP32 serializeJson
recorded_at tahun 1970: Field unix kosong — pastikan ESP32 sudah NTP (#34)
Duplikat banyak: Normal jika ESP32 publish tiap 10 detik; untuk dedup pakai UNIQUE key opsional
Connection refused MySQL: bind-address di mysqld.cnf — untuk lokal pakai 127.0.0.1
Script mati setelah logout: Pakai systemd, bukan terminal SSH saja

Keamanan & Produksi

Jangan commit file .env atau password ke GitHub
Plain MQTT (1883) hanya untuk LAN — internet wajib TLS #17
Backup database MySQL rutin (mysqldump kindo_iot)
Batasi user MySQL: hanya INSERT/SELECT pada tabel sensor, bukan root

Langkah Selanjutnya (Seri 2)

InfluxDB + Grafana (#19) — dashboard grafik time-series
REST API vs MQTT (#20) — kapan pakai HTTP vs push MQTT
NTP & timestamp (#34) — pastikan setiap node kirim unix akurat
MQTT TLS (#17) — amankan subscriber di jaringan publik
Broker Mosquitto (#16) — fondasi infrastruktur
PIR (#24) — tambah tabel/event gerak dengan pola subscriber sama
BME280 (#13) — perluas kolom tekanan udara di skema MySQL

Dengan pipeline MQTT → Python → MySQL, data sensor ESP32 akhirnya tersimpan sebagai histori — langkah berikutnya visualisasi di InfluxDB + Grafana (#19). Lanjutkan Seri 2 di halaman artikel Koding Indonesia.

Subscriber Python: Simpan Data MQTT ke MySQL untuk Histori Sensor ESP32