솔직히 말하면, IoT(Internet of Things)는 이미 우리 일상 깊숙이 들어와 있어. 스마트폰으로 집 에어컨 켜고, 냉장고가 유통기한 알려주고, 공장 기계가 스스로 이상 신호를 보내는 것들 — 전부 IoT야. 근데 이걸 응용프로그램 개발자 입장에서 보면 어떨까?

IoT는 크게 세 계층으로 나눌 수 있어:

2024년 기준으로 전 세계 IoT 기기 수는 약 170억 개를 넘어섰고, 2030년에는 300억 개를 돌파할 것으로 예측돼. 이 엄청난 기기들이 전부 어딘가의 서버나 앱과 연결되어 있다는 뜻이야. 그러니까 IoT 연동 개발 능력은 이제 선택이 아니라 필수가 된 거지.

IoT 개발에서 가장 먼저 부딪히는 벽이 바로 프로토콜이야. "어떤 방식으로 기기랑 대화할 거야?" 하는 문제인데, 선택지가 꽤 많아서 처음엔 헷갈릴 수 있어. 각각의 특징을 제대로 알면 상황에 맞는 걸 골라 쓸 수 있어.

MQTT(Message Queuing Telemetry Transport)는 IoT 세계에서 가장 널리 쓰이는 프로토콜이야. 1999년 IBM이 원유 파이프라인 모니터링을 위해 만든 건데, 지금은 IoT의 사실상 표준이 됐어.

핵심 개념은 발행/구독(Pub/Sub) 모델이야. 기기(Publisher)가 특정 토픽에 메시지를 발행하면, 그 토픽을 구독(Subscribe)한 앱이 메시지를 받는 구조야. 중간에 브로커(Broker)가 메시지를 중계해줘.

MQTT의 QoS(Quality of Service) 레벨도 중요해:

웹 개발을 해봤다면 HTTP는 이미 잘 알겠지? IoT에서도 HTTP/REST를 많이 써. 특히 클라우드 IoT 플랫폼들이 REST API를 제공하는 경우가 많아서, 앱에서 HTTP 요청으로 기기 상태를 조회하거나 제어 명령을 보낼 수 있어.

단점은 오버헤드가 크다는 거야. HTTP 헤더만 해도 수백 바이트인데, 배터리로 동작하는 소형 IoT 기기에는 부담이 될 수 있어. 그래서 실시간성이 중요하거나 기기 수가 많을 때는 MQTT를 선호해.

WebSocket은 HTTP 연결을 업그레이드해서 지속적인 양방향 통신 채널을 만드는 방식이야. 한 번 연결하면 서버와 클라이언트가 자유롭게 데이터를 주고받을 수 있어서, 실시간 대시보드나 모니터링 앱에 딱이야.

IoT 기기 자체보다는 IoT 데이터를 보여주는 프론트엔드 앱에서 많이 써. 서버에서 센서 데이터가 업데이트될 때마다 WebSocket으로 브라우저에 실시간 푸시하는 패턴이 대표적이야.

자, 이제 진짜 핵심이야. 응용프로그램에서 IoT 기기를 제어하는 방법은 크게 두 가지 접근법이 있어:

IoT 기기에 명령을 보낼 때는 보통 JSON 형식의 페이로드를 사용해. 예를 들어 스마트 전구를 제어하는 명령은 이런 식이야:

// 스마트 전구 ON/OFF 제어 명령 (MQTT 페이로드)
{
  "device_id": "bulb_living_001",
  "command": "set_state",
  "params": {
    "power": "on",
    "brightness": 80,
    "color_temp": 4000,
    "color": {
      "r": 255,
      "g": 200,
      "b": 100
    }
  },
  "timestamp": "2024-11-15T10:30:00Z",
  "request_id": "req_abc123"
}

기기는 이 명령을 받아서 처리한 뒤, 응답 토픽으로 결과를 발행해:

// 기기 응답 (MQTT 페이로드)
{
  "device_id": "bulb_living_001",
  "request_id": "req_abc123",
  "status": "success",
  "current_state": {
    "power": "on",
    "brightness": 80,
    "color_temp": 4000
  },
  "timestamp": "2024-11-15T10:30:00.250Z"
}

IoT에서 데이터 흐름을 제대로 설계하는 게 정말 중요해. 잘못 설계하면 데이터 손실, 지연, 서버 과부하 같은 문제가 생기거든. 실무에서 검증된 패턴들을 알아보자.

IoT 데이터는 기본적으로 시계열 데이터(Time-Series Data)야. 온도가 10초마다 기록되고, 전력 사용량이 1분마다 쌓이는 식이지. 이런 데이터는 일반 RDBMS보다 시계열 DB가 훨씬 효율적이야.

IoT 데이터를 수집하는 방식에는 두 가지 주요 패턴이 있어:

이제 진짜 코드를 써볼 시간이야! Python은 IoT 백엔드 개발에서 가장 많이 쓰이는 언어 중 하나야. paho-mqtt 라이브러리를 사용하면 MQTT 통신을 쉽게 구현할 수 있어.

# 필요한 라이브러리 설치
pip install paho-mqtt
pip install influxdb-client  # 시계열 DB 연동용
pip install python-dotenv    # 환경변수 관리

import paho.mqtt.client as mqtt
import json
import time
import uuid
from datetime import datetime

class IoTController:
    """IoT 기기 제어를 위한 MQTT 클라이언트 클래스"""
    
    def __init__(self, broker_host, broker_port=1883, 
                 username=None, password=None):
        self.broker_host = broker_host
        self.broker_port = broker_port
        self.client = mqtt.Client(client_id=f"app_{uuid.uuid4().hex[:8]}")
        
        # 인증 설정 (있을 경우)
        if username and password:
            self.client.username_pw_set(username, password)
        
        # TLS 설정 (보안 연결)
        # self.client.tls_set()  # 실제 운영 환경에서는 반드시 사용!
        
        # 콜백 함수 등록
        self.client.on_connect = self._on_connect
        self.client.on_message = self._on_message
        self.client.on_disconnect = self._on_disconnect
        
        # 응답 대기용 딕셔너리
        self.pending_responses = {}
        
    def _on_connect(self, client, userdata, flags, rc):
        """브로커 연결 성공 시 호출"""
        if rc == 0:
            print(f"✅ MQTT 브로커 연결 성공: {self.broker_host}")
            # 응답 토픽 구독
            self.client.subscribe("devices/+/response", qos=1)
            self.client.subscribe("devices/+/telemetry", qos=0)
            print("📡 토픽 구독 완료")
        else:
            print(f"❌ 연결 실패, 코드: {rc}")
    
    def _on_message(self, client, userdata, msg):
        """메시지 수신 시 호출"""
        try:
            topic = msg.topic
            payload = json.loads(msg.payload.decode('utf-8'))
            
            # 응답 메시지 처리
            if '/response' in topic:
                request_id = payload.get('request_id')
                if request_id and request_id in self.pending_responses:
                    self.pending_responses[request_id] = payload
                    
            # 텔레메트리(센서 데이터) 처리
            elif '/telemetry' in topic:
                device_id = topic.split('/')[1]
                self._process_telemetry(device_id, payload)
                
        except json.JSONDecodeError as e:
            print(f"⚠️ JSON 파싱 오류: {e}")
    
    def _on_disconnect(self, client, userdata, rc):
        """연결 해제 시 호출"""
        if rc != 0:
            print(f"⚠️ 예상치 못한 연결 해제 (코드: {rc}), 재연결 시도...")
    
    def _process_telemetry(self, device_id, data):
        """센서 데이터 처리 (DB 저장, 알림 등)"""
        print(f"📊 [{device_id}] 텔레메트리 수신: {data}")
        # 여기서 DB 저장 로직 추가
        # self.db.write(device_id, data)
    
    def connect(self):
        """브로커에 연결"""
        self.client.connect(self.broker_host, self.broker_port, keepalive=60)
        self.client.loop_start()  # 백그라운드 스레드로 루프 시작
        time.sleep(1)  # 연결 대기
    
    def control_device(self, device_id, command, params, timeout=5):
        """
        기기에 제어 명령 전송
        
        Args:
            device_id: 기기 ID
            command: 명령어 (예: 'set_state', 'reboot')
            params: 명령 파라미터 딕셔너리
            timeout: 응답 대기 시간 (초)
        
        Returns:
            응답 딕셔너리 또는 None (타임아웃 시)
        """
        request_id = f"req_{uuid.uuid4().hex[:12]}"
        
        payload = {
            "device_id": device_id,
            "command": command,
            "params": params,
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "request_id": request_id
        }
        
        # 응답 대기 등록
        self.pending_responses[request_id] = None
        
        # 명령 발행 (QoS 1: 최소 1회 전달 보장)
        topic = f"devices/{device_id}/command"
        result = self.client.publish(
            topic, 
            json.dumps(payload), 
            qos=1
        )
        
        print(f"📤 명령 전송: {topic} → {command}")
        
        # 응답 대기
        start_time = time.time()
        while time.time() - start_time < timeout:
            if self.pending_responses[request_id] is not None:
                response = self.pending_responses.pop(request_id)
                return response
            time.sleep(0.1)
        
        # 타임아웃
        self.pending_responses.pop(request_id, None)
        print(f"⏱️ 응답 타임아웃: {device_id}")
        return None
    
    def disconnect(self):
        """연결 해제"""
        self.client.loop_stop()
        self.client.disconnect()
        print("🔌 MQTT 연결 해제")


# ===== 실제 사용 예시 =====
if __name__ == "__main__":
    # 컨트롤러 초기화
    controller = IoTController(
        broker_host="mqtt.example.com",
        broker_port=1883,
        username="iot_app",
        password="secure_password"
    )
    
    controller.connect()
    
    # 스마트 전구 켜기
    response = controller.control_device(
        device_id="bulb_living_001",
        command="set_state",
        params={
            "power": "on",
            "brightness": 80,
            "color_temp": 4000
        }
    )
    
    if response and response.get('status') == 'success':
        print(f"✅ 전구 제어 성공: {response['current_state']}")
    else:
        print("❌ 전구 제어 실패")
    
    # 온도 센서 최신값 요청
    response = controller.control_device(
        device_id="temp_sensor_001",
        command="get_reading",
        params={}
    )
    
    if response:
        print(f"🌡️ 현재 온도: {response.get('temperature')}°C")
    
    time.sleep(10)  # 텔레메트리 수신 대기
    controller.disconnect()

import paho.mqtt.client as mqtt
import json
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
from datetime import datetime

class SensorDataCollector:
    """센서 데이터를 수집하여 InfluxDB에 저장하는 클래스"""
    
    def __init__(self, mqtt_broker, influx_url, influx_token, 
                 influx_org, influx_bucket):
        # MQTT 설정
        self.mqtt_client = mqtt.Client()
        self.mqtt_client.on_connect = self._on_connect
        self.mqtt_client.on_message = self._on_message
        self.mqtt_broker = mqtt_broker
        
        # InfluxDB 설정
        self.influx_client = InfluxDBClient(
            url=influx_url, 
            token=influx_token, 
            org=influx_org
        )
        self.write_api = self.influx_client.write_api(
            write_options=SYNCHRONOUS
        )
        self.influx_bucket = influx_bucket
        self.influx_org = influx_org
    
    def _on_connect(self, client, userdata, flags, rc):
        print(f"✅ 데이터 수집기 연결 완료")
        # 모든 센서 텔레메트리 구독
        client.subscribe("devices/+/telemetry", qos=0)
        # 특정 센서 타입만 구독하려면:
        # client.subscribe("devices/temp_+/telemetry", qos=0)
    
    def _on_message(self, client, userdata, msg):
        """수신된 센서 데이터를 InfluxDB에 저장"""
        try:
            topic_parts = msg.topic.split('/')
            device_id = topic_parts[1]  # devices/{device_id}/telemetry
            
            data = json.loads(msg.payload.decode('utf-8'))
            self._save_to_influxdb(device_id, data)
            
        except Exception as e:
            print(f"⚠️ 데이터 처리 오류: {e}")
    
    def _save_to_influxdb(self, device_id, data):
        """InfluxDB에 시계열 데이터 저장"""
        try:
            # 데이터 포인트 생성
            point = Point("sensor_data") \
                .tag("device_id", device_id) \
                .tag("location", data.get("location", "unknown")) \
                .tag("device_type", data.get("device_type", "unknown"))
            
            # 측정값 추가 (있는 필드만)
            if "temperature" in data:
                point = point.field("temperature", float(data["temperature"]))
            if "humidity" in data:
                point = point.field("humidity", float(data["humidity"]))
            if "pressure" in data:
                point = point.field("pressure", float(data["pressure"]))
            if "battery_level" in data:
                point = point.field("battery_level", 
                                    float(data["battery_level"]))
            
            # 타임스탬프 설정
            if "timestamp" in data:
                point = point.time(data["timestamp"])
            
            # InfluxDB에 쓰기
            self.write_api.write(
                bucket=self.influx_bucket, 
                org=self.influx_org, 
                record=point
            )
            
            print(f"💾 [{device_id}] 데이터 저장 완료: "
                  f"temp={data.get('temperature')}°C, "
                  f"humidity={data.get('humidity')}%")
                  
        except Exception as e:
            print(f"❌ DB 저장 오류: {e}")
    
    def start(self):
        """수집기 시작"""
        self.mqtt_client.connect(self.mqtt_broker, 1883, 60)
        print("🚀 센서 데이터 수집기 시작!")
        self.mqtt_client.loop_forever()  # 블로킹 루프


# 실행
collector = SensorDataCollector(
    mqtt_broker="mqtt.example.com",
    influx_url="http://localhost:8086",
    influx_token="your_influxdb_token",
    influx_org="your_org",
    influx_bucket="iot_data"
)
collector.start()

MQTT만이 답은 아니야. 많은 상용 IoT 기기들은 자체 HTTP API를 제공해. 필립스 휴(Hue) 스마트 전구, 샤오미 기기들, 각종 스마트홈 허브들이 REST API를 통해 제어 가능해. 이런 기기들을 응용프로그램에서 제어하는 방법을 알아보자.

import requests
import json
from typing import Optional, Dict, Any
import asyncio
import aiohttp  # 비동기 HTTP 클라이언트

class SmartHomeAPIClient:
    """
    스마트홈 기기 REST API 클라이언트
    (Philips Hue, 스마트 허브 등 HTTP API 기기용)
    """
    
    def __init__(self, hub_ip: str, api_key: str):
        self.base_url = f"http://{hub_ip}/api/{api_key}"
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        })
    
    def get_all_lights(self) -> Dict:
        """모든 조명 기기 상태 조회"""
        response = self.session.get(f"{self.base_url}/lights")
        response.raise_for_status()
        return response.json()
    
    def get_light_state(self, light_id: str) -> Dict:
        """특정 조명 상태 조회"""
        response = self.session.get(
            f"{self.base_url}/lights/{light_id}"
        )
        response.raise_for_status()
        return response.json()
    
    def set_light_state(self, light_id: str, 
                        state: Dict[str, Any]) -> Dict:
        """
        조명 상태 변경
        
        Args:
            light_id: 조명 ID
            state: 변경할 상태 딕셔너리
                - on: bool (켜기/끄기)
                - bri: int (밝기, 0-254)
                - hue: int (색조, 0-65535)
                - sat: int (채도, 0-254)
                - ct: int (색온도, 153-500 미레드)
                - transitiontime: int (전환 시간, 100ms 단위)
        """
        response = self.session.put(
            f"{self.base_url}/lights/{light_id}/state",
            data=json.dumps(state)
        )
        response.raise_for_status()
        return response.json()
    
    def create_scene(self, name: str, light_ids: list, 
                     states: Dict) -> str:
        """씬(Scene) 생성 — 여러 조명을 한 번에 설정"""
        payload = {
            "name": name,
            "lights": light_ids,
            "lightstates": states
        }
        response = self.session.post(
            f"{self.base_url}/scenes",
            data=json.dumps(payload)
        )
        response.raise_for_status()
        result = response.json()
        return result[0].get('success', {}).get('id', '')
    
    def get_sensors(self) -> Dict:
        """모든 센서 데이터 조회"""
        response = self.session.get(f"{self.base_url}/sensors")
        response.raise_for_status()
        return response.json()


# ===== 비동기 버전 (대량 기기 동시 제어) =====
class AsyncSmartHomeClient:
    """비동기 스마트홈 클라이언트 — 여러 기기 동시 제어"""
    
    def __init__(self, hub_ip: str, api_key: str):
        self.base_url = f"http://{hub_ip}/api/{api_key}"
    
    async def set_multiple_lights(self, 
                                   light_commands: list) -> list:
        """
        여러 조명을 동시에 제어 (비동기)
        
        Args:
            light_commands: [{"id": "1", "state": {...}}, ...]
        """
        async with aiohttp.ClientSession() as session:
            tasks = []
            for cmd in light_commands:
                task = self._set_light(
                    session, cmd['id'], cmd['state']
                )
                tasks.append(task)
            
            # 모든 요청을 동시에 실행
            results = await asyncio.gather(*tasks, 
                                           return_exceptions=True)
            return results
    
    async def _set_light(self, session, light_id, state):
        """단일 조명 비동기 제어"""
        url = f"{self.base_url}/lights/{light_id}/state"
        async with session.put(url, json=state) as response:
            return await response.json()


# ===== 사용 예시 =====
def main():
    client = SmartHomeAPIClient(
        hub_ip="192.168.1.100",
        api_key="your_hue_api_key"
    )
    
    # 모든 조명 상태 확인
    lights = client.get_all_lights()
    print(f"🏠 연결된 조명 수: {len(lights)}개")
    
    # 거실 조명 켜기 (따뜻한 색온도)
    result = client.set_light_state("1", {
        "on": True,
        "bri": 200,
        "ct": 370,          # 따뜻한 흰색 (약 2700K)
        "transitiontime": 10  # 1초에 걸쳐 전환
    })
    print(f"💡 거실 조명 설정: {result}")
    
    # 영화 모드 씬 생성
    scene_id = client.create_scene(
        name="영화 모드",
        light_ids=["1", "2", "3"],
        states={
            "1": {"on": True, "bri": 50, "ct": 500},
            "2": {"on": False},
            "3": {"on": True, "bri": 30, "ct": 500}
        }
    )
    print(f"🎬 씬 생성 완료: {scene_id}")
    
    # 비동기 버전으로 10개 조명 동시 제어
    async_client = AsyncSmartHomeClient(
        hub_ip="192.168.1.100",
        api_key="your_hue_api_key"
    )
    
    commands = [
        {"id": str(i), "state": {"on": True, "bri": 150}}
        for i in range(1, 11)
    ]
    
    results = asyncio.run(
        async_client.set_multiple_lights(commands)
    )
    print(f"✅ 10개 조명 동시 제어 완료: {len(results)}개 응답")

main()

# 폴링 예시
import time
import requests

def polling_sensor_data(interval=30):
    while True:
        data = requests.get(
            "http://device/api/sensor"
        ).json()
        process_data(data)
        time.sleep(interval)

# 웹훅 수신 서버 (Flask)
from flask import Flask, request
app = Flask(__name__)

@app.route('/webhook/iot', methods=['POST'])
def receive_iot_event():
    data = request.json
    process_iot_event(data)
    return {'status': 'ok'}, 200

IoT 보안은 진짜 심각한 문제야. 2016년 미라이(Mirai) 봇넷 공격은 수십만 개의 IoT 기기를 해킹해서 역대 최대 규모의 DDoS 공격을 일으켰어. 기본 비밀번호를 바꾸지 않은 CCTV, 공유기들이 좀비가 된 거야. 개발자로서 이런 문제를 예방하는 코드를 짜는 게 중요해.

import paho.mqtt.client as mqtt
import ssl

def create_secure_mqtt_client(broker_host, broker_port=8883,
                               ca_cert_path, client_cert_path, 
                               client_key_path):
    """
    TLS 인증서를 사용한 보안 MQTT 연결
    (실제 운영 환경에서는 반드시 TLS 사용!)
    """
    client = mqtt.Client()
    
    # TLS 설정
    client.tls_set(
        ca_certs=ca_cert_path,          # CA 인증서
        certfile=client_cert_path,       # 클라이언트 인증서
        keyfile=client_key_path,         # 클라이언트 개인키
        tls_version=ssl.PROTOCOL_TLS,
        cert_reqs=ssl.CERT_REQUIRED      # 서버 인증서 검증 필수
    )
    
    # 호스트명 검증 활성화
    client.tls_insecure_set(False)
    
    return client

import jwt
import time
from datetime import datetime, timedelta

class IoTDeviceAuthenticator:
    """IoT 기기 JWT 인증 처리"""
    
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
    
    def generate_device_token(self, device_id: str, 
                               device_type: str,
                               expires_hours: int = 24) -> str:
        """기기용 JWT 토큰 생성"""
        payload = {
            "device_id": device_id,
            "device_type": device_type,
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + timedelta(hours=expires_hours),
            "iss": "iot-platform",
            # 기기 권한 정의
            "permissions": {
                "topics": {
                    "publish": [
                        f"devices/{device_id}/telemetry",
                        f"devices/{device_id}/response"
                    ],
                    "subscribe": [
                        f"devices/{device_id}/command"
                    ]
                }
            }
        }
        
        token = jwt.encode(payload, self.secret_key, 
                           algorithm='HS256')
        return token
    
    def verify_device_token(self, token: str) -> dict:
        """토큰 검증 및 페이로드 반환"""
        try:
            payload = jwt.decode(
                token, 
                self.secret_key, 
                algorithms=['HS256']
            )
            return {"valid": True, "payload": payload}
        except jwt.ExpiredSignatureError:
            return {"valid": False, "error": "토큰 만료"}
        except jwt.InvalidTokenError as e:
            return {"valid": False, "error": f"유효하지 않은 토큰: {e}"}
    
    def check_topic_permission(self, token: str, 
                                topic: str, 
                                action: str) -> bool:
        """
        기기가 특정 토픽에 접근 권한이 있는지 확인
        
        Args:
            action: 'publish' 또는 'subscribe'
        """
        result = self.verify_device_token(token)
        if not result["valid"]:
            return False
        
        permissions = result["payload"].get("permissions", {})
        allowed_topics = permissions.get("topics", {}).get(action, [])
        
        return topic in allowed_topics

처음부터 모든 인프라를 직접 구축하는 건 엄청난 공수가 들어. 그래서 실무에서는 클라우드 IoT 플랫폼을 많이 활용해. 주요 플랫폼들의 특징을 비교해볼게.

import boto3
import json
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient

class AWSIoTConnector:
    """AWS IoT Core 연동 클라이언트"""
    
    def __init__(self, endpoint, client_id, 
                 root_ca, private_key, certificate):
        self.client = AWSIoTMQTTClient(client_id)
        self.client.configureEndpoint(endpoint, 8883)
        self.client.configureCredentials(
            root_ca, private_key, certificate
        )
        
        # 연결 설정
        self.client.configureAutoReconnectBackoffTime(1, 32, 20)
        self.client.configureOfflinePublishQueueing(-1)
        self.client.configureDrainingFrequency(2)
        self.client.configureConnectDisconnectTimeout(10)
        self.client.configureMQTTOperationTimeout(5)
    
    def connect(self):
        self.client.connect()
        print("✅ AWS IoT Core 연결 완료")
    
    def publish_telemetry(self, device_id: str, data: dict):
        """텔레메트리 데이터 발행"""
        topic = f"devices/{device_id}/telemetry"
        payload = json.dumps(data)
        self.client.publish(topic, payload, 1)
    
    def update_device_shadow(self, device_id: str, 
                              desired_state: dict):
        """
        AWS IoT Device Shadow 업데이트
        Shadow: 기기의 원하는 상태(desired)와 
                실제 상태(reported)를 관리하는 기능
        """
        shadow_topic = (
            f"$aws/things/{device_id}/shadow/update"
        )
        shadow_payload = {
            "state": {
                "desired": desired_state
            }
        }
        self.client.publish(
            shadow_topic, 
            json.dumps(shadow_payload), 
            1
        )
        print(f"📱 Shadow 업데이트: {device_id} → {desired_state}")
    
    def get_device_shadow(self, device_id: str):
        """Device Shadow 조회 (boto3 사용)"""
        iot_data = boto3.client('iot-data')
        response = iot_data.get_thing_shadow(
            thingName=device_id
        )
        shadow = json.loads(
            response['payload'].read().decode('utf-8')
        )
        return shadow

재능넷 같은 플랫폼에서 IoT 관련 개발 재능을 거래할 때도, 클라우드 IoT 플랫폼 경험이 있는 개발자를 찾는 수요가 꾸준히 늘고 있어. AWS IoT Core나 Azure IoT Hub 경험은 확실히 강점이 돼.

모든 데이터를 클라우드로 보내는 건 비효율적이야. 공장에서 초당 1000개의 센서 데이터가 발생한다면? 전부 클라우드로 보내면 네트워크 비용도 어마어마하고, 지연도 생겨. 이걸 해결하는 게 엣지 컴퓨팅(Edge Computing)이야.

import statistics
from collections import deque
from datetime import datetime, timedelta
import threading

class EdgeDataProcessor:
    """
    엣지 노드(Raspberry Pi, 산업용 PC 등)에서 실행되는
    실시간 데이터 전처리 클래스
    """
    
    def __init__(self, window_size=60, 
                 upload_interval=300):
        """
        Args:
            window_size: 슬라이딩 윈도우 크기 (초)
            upload_interval: 클라우드 업로드 간격 (초)
        """
        self.window_size = window_size
        self.upload_interval = upload_interval
        
        # 센서별 데이터 버퍼
        self.buffers = {}
        self.lock = threading.Lock()
        
        # 이상 감지 임계값
        self.thresholds = {
            "temperature": {"min": -10, "max": 80},
            "humidity": {"min": 0, "max": 100},
            "vibration": {"max": 50}
        }
    
    def add_reading(self, sensor_id: str, 
                    sensor_type: str, value: float):
        """센서 데이터 추가 및 실시간 이상 감지"""
        timestamp = datetime.utcnow()
        
        with self.lock:
            if sensor_id not in self.buffers:
                self.buffers[sensor_id] = {
                    "type": sensor_type,
                    "data": deque(maxlen=self.window_size * 10),
                    "anomaly_count": 0
                }
            
            self.buffers[sensor_id]["data"].append({
                "value": value,
                "timestamp": timestamp
            })
        
        # 실시간 이상 감지 (엣지에서 즉시 처리)
        if self._is_anomaly(sensor_type, value):
            self._handle_anomaly(sensor_id, sensor_type, value)
    
    def _is_anomaly(self, sensor_type: str, value: float) -> bool:
        """이상값 감지"""
        if sensor_type not in self.thresholds:
            return False
        
        threshold = self.thresholds[sensor_type]
        
        if "min" in threshold and value < threshold["min"]:
            return True
        if "max" in threshold and value > threshold["max"]:
            return True
        
        return False
    
    def _handle_anomaly(self, sensor_id, sensor_type, value):
        """이상 감지 시 즉시 처리 (엣지에서 로컬 알림)"""
        print(f"🚨 이상 감지! [{sensor_id}] {sensor_type}: {value}")
        # 로컬 알림 (사이렌, LED 등)
        # 즉시 클라우드 전송 (이상 데이터는 즉시 업로드)
        self._urgent_upload(sensor_id, sensor_type, value)
    
    def get_aggregated_data(self, sensor_id: str) -> dict:
        """
        집계 데이터 생성 (클라우드 업로드용)
        원시 데이터 대신 통계값만 전송하여 트래픽 절감
        """
        with self.lock:
            if sensor_id not in self.buffers:
                return {}
            
            buffer = self.buffers[sensor_id]
            values = [d["value"] for d in buffer["data"]]
            
            if not values:
                return {}
            
            return {
                "sensor_id": sensor_id,
                "sensor_type": buffer["type"],
                "period_start": buffer["data"][0]["timestamp"].isoformat(),
                "period_end": buffer["data"][-1]["timestamp"].isoformat(),
                "count": len(values),
                "mean": round(statistics.mean(values), 2),
                "median": round(statistics.median(values), 2),
                "std_dev": round(statistics.stdev(values), 3) 
                           if len(values) > 1 else 0,
                "min": min(values),
                "max": max(values),
                "anomaly_count": buffer["anomaly_count"]
            }
    
    def _urgent_upload(self, sensor_id, sensor_type, value):
        """이상 데이터 즉시 클라우드 전송"""
        # MQTT로 즉시 발행
        pass


# 엣지 처리 효과 비교
print("📊 엣지 처리 효과:")
print(f"  원시 데이터: 초당 100개 × 86400초 = 8,640,000개/일")
print(f"  집계 후 전송: 5분마다 1개 × 288회 = 288개/일")
print(f"  데이터 절감률: 약 99.997%")

IoT 기기는 네트워크가 불안정한 환경에서 동작하는 경우가 많아. 연결이 끊겼다가 다시 연결되는 상황을 우아하게 처리하는 코드가 필수야.

import paho.mqtt.client as mqtt
import time
import json
from collections import deque

class ResilientMQTTClient:
    """
    재연결 및 오프라인 메시지 큐잉을 지원하는
    견고한 MQTT 클라이언트
    """
    
    def __init__(self, broker_host, broker_port=1883):
        self.broker_host = broker_host
        self.broker_port = broker_port
        self.is_connected = False
        
        # 오프라인 중 발생한 메시지 큐 (최대 1000개)
        self.offline_queue = deque(maxlen=1000)
        
        self.client = mqtt.Client()
        self.client.on_connect = self._on_connect
        self.client.on_disconnect = self._on_disconnect
        
        # 지수 백오프 재연결 설정
        self.client.reconnect_delay_set(
            min_delay=1,    # 최소 1초
            max_delay=120   # 최대 2분
        )
    
    def _on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            self.is_connected = True
            print("✅ 연결 성공")
            # 오프라인 큐에 쌓인 메시지 전송
            self._flush_offline_queue()
        
    def _on_disconnect(self, client, userdata, rc):
        self.is_connected = False
        if rc != 0:
            print(f"⚠️ 연결 해제 (코드: {rc}), 자동 재연결 대기...")
    
    def _flush_offline_queue(self):
        """오프라인 중 쌓인 메시지 전송"""
        flushed = 0
        while self.offline_queue:
            topic, payload, qos = self.offline_queue.popleft()
            self.client.publish(topic, payload, qos)
            flushed += 1
        
        if flushed > 0:
            print(f"📤 오프라인 큐 플러시: {flushed}개 메시지 전송")
    
    def safe_publish(self, topic: str, payload: dict, qos=1):
        """
        안전한 메시지 발행
        오프라인 상태면 큐에 저장, 온라인이면 즉시 전송
        """
        payload_str = json.dumps(payload)
        
        if self.is_connected:
            self.client.publish(topic, payload_str, qos)
        else:
            # 오프라인 큐에 저장
            self.offline_queue.append((topic, payload_str, qos))
            print(f"📦 오프라인 큐 저장: {topic} "
                  f"(큐 크기: {len(self.offline_queue)})")
    
    def connect_with_retry(self, max_retries=None):
        """재시도 로직이 포함된 연결"""
        attempt = 0
        while max_retries is None or attempt < max_retries:
            try:
                self.client.connect(
                    self.broker_host, 
                    self.broker_port, 
                    keepalive=60
                )
                self.client.loop_start()
                return True
            except Exception as e:
                attempt += 1
                wait_time = min(2 ** attempt, 60)
                print(f"❌ 연결 실패 (시도 {attempt}): {e}")
                print(f"⏳ {wait_time}초 후 재시도...")
                time.sleep(wait_time)
        
        return False

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import asyncio
import json
from typing import List

app = FastAPI()

class ConnectionManager:
    """WebSocket 연결 관리자"""
    
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
        print(f"🔌 새 클라이언트 연결 "
              f"(총 {len(self.active_connections)}개)")
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
        print(f"🔌 클라이언트 연결 해제 "
              f"(총 {len(self.active_connections)}개)")
    
    async def broadcast(self, data: dict):
        """모든 연결된 클라이언트에게 데이터 브로드캐스트"""
        if not self.active_connections:
            return
        
        message = json.dumps(data, ensure_ascii=False)
        disconnected = []
        
        for connection in self.active_connections:
            try:
                await connection.send_text(message)
            except Exception:
                disconnected.append(connection)
        
        # 끊어진 연결 정리
        for conn in disconnected:
            self.active_connections.remove(conn)


manager = ConnectionManager()

@app.websocket("/ws/iot-dashboard")
async def websocket_endpoint(websocket: WebSocket):
    """IoT 실시간 대시보드 WebSocket 엔드포인트"""
    await manager.connect(websocket)
    try:
        while True:
            # 클라이언트로부터 메시지 수신 (제어 명령 등)
            data = await websocket.receive_text()
            command = json.loads(data)
            
            if command.get("type") == "control":
                # IoT 기기 제어 명령 처리
                await handle_control_command(command)
                
    except WebSocketDisconnect:
        manager.disconnect(websocket)


async def handle_control_command(command: dict):
    """제어 명령 처리 및 결과 브로드캐스트"""
    device_id = command.get("device_id")
    action = command.get("action")
    
    # 실제 기기 제어 로직 (MQTT 발행 등)
    # result = await mqtt_controller.control(device_id, action)
    
    # 결과를 모든 대시보드 클라이언트에게 전송
    await manager.broadcast({
        "type": "control_result",
        "device_id": device_id,
        "action": action,
        "status": "success",
        "timestamp": "2024-11-15T10:30:00Z"
    })


# MQTT에서 받은 센서 데이터를 WebSocket으로 브로드캐스트
async def forward_sensor_data_to_dashboard(sensor_data: dict):
    """
    MQTT 수신 데이터를 WebSocket 대시보드로 전달
    (MQTT 콜백에서 이 함수를 호출)
    """
    await manager.broadcast({
        "type": "sensor_update",
        "data": sensor_data
    })

여기까지 왔다면 IoT 응용프로그래밍의 핵심을 꽤 많이 파악한 거야! 정리해보면:

IoT 개발은 하드웨어와 소프트웨어의 경계에 있는 흥미로운 분야야. 처음엔 낯선 개념들이 많지만, 실제로 센서 데이터가 앱에 실시간으로 표시되는 걸 보면 정말 짜릿해! 🎉

만약 IoT 프로젝트를 시작하려는데 혼자 하기 어렵다면, 재능넷에서 IoT 개발 경험이 있는 개발자를 찾아보는 것도 좋은 방법이야. MQTT 브로커 설정부터 클라우드 연동까지 다양한 전문가들이 도움을 줄 수 있거든.

IoT는 계속 발전하는 분야야. Matter 프로토콜(스마트홈 통합 표준), 5G NB-IoT, TinyML(엣지 AI) 같은 새로운 기술들이 계속 나오고 있어. 기본기를 탄탄히 다지면서 새로운 트렌드도 꾸준히 팔로우해봐!