🌈 멀티 센서 Publisher 예제

import paho.mqtt.client as mqtt
import Adafruit_DHT
import time
import json
import RPi.GPIO as GPIO

# GPIO 설정
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)

# 센서 핀 설정
DHT_PIN = 4          # 온습도 센서
PIR_PIN = 17         # 모션 센서
LIGHT_PIN = 27       # 조도 센서 (디지털)

# PIR 센서 입력 설정
GPIO.setup(PIR_PIN, GPIO.IN)
GPIO.setup(LIGHT_PIN, GPIO.IN)

# MQTT 설정
BROKER = "localhost"
PORT = 1883
BASE_TOPIC = "home/livingroom"

client = mqtt.Client("MultiSensor_Publisher")
client.username_pw_set("username", "password")
client.connect(BROKER, PORT, 60)
client.loop_start()

print("🏠 스마트홈 센서 시스템 시작!")

try:
    while True:
        sensor_readings = {}
        
        # 1. 온습도 센서 읽기
        humidity, temperature = Adafruit_DHT.read_retry(Adafruit_DHT.DHT22, DHT_PIN)
        if humidity and temperature:
            sensor_readings['temperature'] = round(temperature, 2)
            sensor_readings['humidity'] = round(humidity, 2)
            
            client.publish(f"{BASE_TOPIC}/temperature", 
                         json.dumps({"value": sensor_readings['temperature'], "unit": "°C"}))
            client.publish(f"{BASE_TOPIC}/humidity", 
                         json.dumps({"value": sensor_readings['humidity'], "unit": "%"}))
        
        # 2. 모션 센서 읽기
        motion_detected = GPIO.input(PIR_PIN)
        sensor_readings['motion'] = bool(motion_detected)
        client.publish(f"{BASE_TOPIC}/motion", 
                      json.dumps({"detected": sensor_readings['motion']}))
        
        if motion_detected:
            print("🚶 움직임 감지!")
        
        # 3. 조도 센서 읽기
        light_status = GPIO.input(LIGHT_PIN)
        sensor_readings['light'] = "bright" if light_status else "dark"
        client.publish(f"{BASE_TOPIC}/light", 
                      json.dumps({"status": sensor_readings['light']}))
        
        # 통합 데이터 발행
        sensor_readings['timestamp'] = time.time()
        client.publish(f"{BASE_TOPIC}/all", json.dumps(sensor_readings))
        
        print(f"📊 온도: {sensor_readings.get('temperature', 'N/A')}°C | "
              f"습도: {sensor_readings.get('humidity', 'N/A')}% | "
              f"움직임: {sensor_readings['motion']} | "
              f"조도: {sensor_readings['light']}")
        
        time.sleep(5)
        
except KeyboardInterrupt:
    print("\n⏹️ 시스템 종료")
    GPIO.cleanup()
    client.loop_stop()
    client.disconnect()

이 코드는 여러 센서의 데이터를 동시에 수집해서 각각의 토픽으로 발행해. 또한 모든 센서 데이터를 통합한 메시지도 함께 보내서, 구독자가 선택적으로 받을 수 있도록 했어. 🎯

🔑 인증서 생성

# OpenSSL로 자체 서명 인증서 생성
# CA 키 생성
openssl genrsa -out ca.key 2048

# CA 인증서 생성
openssl req -new -x509 -days 3650 -key ca.key -out ca.crt

# 서버 키 생성
openssl genrsa -out server.key 2048

# 서버 인증서 요청 생성
openssl req -new -key server.key -out server.csr

# 서버 인증서 서명
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 3650

⚙️ Mosquitto TLS 설정

# /etc/mosquitto/mosquitto.conf 수정
listener 8883
cafile /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key
require_certificate false
allow_anonymous false
password_file /etc/mosquitto/passwd

🔒 Python 클라이언트 TLS 적용

import paho.mqtt.client as mqtt
import ssl

client = mqtt.Client("Secure_Publisher")
client.username_pw_set("username", "password")

# TLS 설정
client.tls_set(ca_certs="/path/to/ca.crt",
               certfile=None,
               keyfile=None,
               cert_reqs=ssl.CERT_REQUIRED,
               tls_version=ssl.PROTOCOL_TLSv1_2)

# 8883 포트로 연결 (TLS 기본 포트)
client.connect("broker_address", 8883, 60)

이렇게 하면 데이터가 암호화되어 전송돼서 중간에 누가 가로채도 내용을 볼 수 없어. 특히 외부 네트워크를 통해 데이터를 전송할 때는 필수야! 🔐

🚀 HiveMQ Cloud 연결 예제

import paho.mqtt.client as mqtt
import ssl

# HiveMQ Cloud 설정
BROKER = "your-cluster.s1.eu.hivemq.cloud"
PORT = 8883
USERNAME = "your_username"
PASSWORD = "your_password"

client = mqtt.Client("RaspberryPi_Cloud")
client.username_pw_set(USERNAME, PASSWORD)

# TLS 설정 (HiveMQ Cloud는 TLS 필수)
client.tls_set(cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2)

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("✅ HiveMQ Cloud 연결 성공!")
        client.subscribe("sensor/#")
    else:
        print(f"❌ 연결 실패: {rc}")

client.on_connect = on_connect
client.connect(BROKER, PORT, 60)
client.loop_forever()

클라우드 브로커를 사용하면 인터넷만 연결되어 있으면 전 세계 어디서든 센서 데이터에 접근할 수 있어. 정말 편리하지? 🌍

💾 메모리 효율적인 코드 예제

import paho.mqtt.client as mqtt
import struct
import time

# 바이너리 포맷으로 데이터 전송 (메시지 크기 최소화)
def pack_sensor_data(temperature, humidity):
    # float 2개를 8바이트로 압축
    return struct.pack('ff', temperature, humidity)

def unpack_sensor_data(payload):
    # 바이너리 데이터를 다시 float로 변환
    return struct.unpack('ff', payload)

# 사용 예
client = mqtt.Client()
client.connect("localhost", 1883, 60)

# JSON: {"temperature": 25.5, "humidity": 60.2} → 약 40바이트
# Binary: struct.pack('ff', 25.5, 60.2) → 8바이트!

binary_data = pack_sensor_data(25.5, 60.2)
client.publish("sensor/data", binary_data, qos=0)

print(f"📦 전송 크기: {len(binary_data)} 바이트")

바이너리 포맷을 사용하면 메시지 크기를 80% 이상 줄일 수 있어. 특히 배터리로 동작하는 디바이스에서는 데이터 전송량이 곧 배터리 수명이니까 중요하지! 🔋

🔍 디버깅 도구

# Mosquitto 클라이언트로 메시지 모니터링
mosquitto_sub -h localhost -t "#" -v -u username -P password

# 특정 토픽 구독
mosquitto_sub -h localhost -t "home/sensor/temperature" -u username -P password

# 메시지 발행 테스트
mosquitto_pub -h localhost -t "test/topic" -m "Hello MQTT" -u username -P password

# 브로커 로그 확인
sudo tail -f /var/log/mosquitto/mosquitto.log

이런 명령어들로 문제를 빠르게 찾을 수 있어. 특히 mosquitto_sub -t "#"는 모든 메시지를 볼 수 있어서 디버깅할 때 정말 유용해! 🔎

🔄 자동 재연결 구현

import paho.mqtt.client as mqtt
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RobustMQTTClient:
    def __init__(self, broker, port, username, password):
        self.broker = broker
        self.port = port
        self.username = username
        self.password = password
        self.client = None
        self.connected = False
        
    def on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            self.connected = True
            logger.info("✅ 브로커 연결 성공")
            # 재연결 시 구독 복원
            client.subscribe("sensor/#", qos=1)
        else:
            self.connected = False
            logger.error(f"❌ 연결 실패: {rc}")
    
    def on_disconnect(self, client, userdata, rc):
        self.connected = False
        logger.warning(f"⚠️ 연결 끊김: {rc}")
        if rc != 0:
            logger.info("🔄 재연결 시도 중...")
            self.reconnect()
    
    def reconnect(self):
        retry_count = 0
        max_retries = 10
        
        while retry_count < max_retries and not self.connected:
            try:
                logger.info(f"재연결 시도 {retry_count + 1}/{max_retries}")
                self.client.reconnect()
                time.sleep(5)
                retry_count += 1
            except Exception as e:
                logger.error(f"재연결 실패: {e}")
                time.sleep(10)
        
        if not self.connected:
            logger.error("❌ 최대 재연결 시도 횟수 초과")
    
    def connect(self):
        self.client = mqtt.Client()
        self.client.username_pw_set(self.username, self.password)
        self.client.on_connect = self.on_connect
        self.client.on_disconnect = self.on_disconnect
        
        try:
            self.client.connect(self.broker, self.port, 60)
            self.client.loop_start()
        except Exception as e:
            logger.error(f"초기 연결 실패: {e}")
            self.reconnect()
    
    def publish(self, topic, payload, qos=1):
        if self.connected:
            self.client.publish(topic, payload, qos=qos)
        else:
            logger.warning("⚠️ 연결되지 않음. 메시지 대기 중...")
            # 여기서 메시지를 큐에 저장했다가 재연결 시 전송 가능

# 사용 예
robust_client = RobustMQTTClient("broker", 1883, "user", "pass")
robust_client.connect()

while True:
    robust_client.publish("sensor/data", "test")
    time.sleep(10)

이런 식으로 견고한 클라이언트를 만들면 네트워크가 불안정해도 안정적으로 동작해! 🛡️