07. HTTP/REST for IoT

Previous 06. MQTT ν”„λ‘œν† μ½œ
πŸ“‹ Topic list
Next 08. Edge AI - TensorFlow Lite

07. HTTP/REST for IoT

ν•™μŠ΅ λͺ©ν‘œ

  • Flaskλ₯Ό μ΄μš©ν•œ IoT μ„œλ²„ ꡬ좕
  • μ„Όμ„œ 데이터 μˆ˜μ§‘ API 섀계
  • RESTful API 섀계 원칙 이해
  • JSON 데이터 처리 및 검증

1. Flask IoT μ„œλ²„

1.1 κΈ°λ³Έ μ„€μ •

# νŒ¨ν‚€μ§€ μ„€μΉ˜
pip install flask flask-cors

# μΆ”κ°€ μœ ν‹Έλ¦¬ν‹°
pip install python-dotenv

#!/usr/bin/env python3
"""Flask IoT μ„œλ²„ κΈ°λ³Έ ꡬ쑰"""

from flask import Flask, jsonify, request
from flask_cors import CORS
from datetime import datetime

app = Flask(__name__)
CORS(app)  # CORS ν™œμ„±ν™”

# λ©”λͺ¨λ¦¬ μ €μž₯μ†Œ (μ‹€μ œ ν”„λ‘œμ νŠΈμ—μ„œλŠ” DB μ‚¬μš©)
sensor_data_store = []
devices = {}

@app.route('/')
def index():
    """API 정보"""
    return jsonify({
        "name": "IoT API Server",
        "version": "1.0",
        "endpoints": {
            "/api/sensors": "GET, POST",
            "/api/sensors/<id>": "GET",
            "/api/devices": "GET, POST",
            "/api/devices/<id>": "GET, PUT, DELETE"
        }
    })

@app.route('/health')
def health():
    """ν—¬μŠ€ 체크"""
    return jsonify({
        "status": "healthy",
        "timestamp": datetime.now().isoformat()
    })

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=5000, debug=True)

1.2 ν”„λ‘œμ νŠΈ ꡬ쑰

iot_server/
β”œβ”€β”€ app.py              # 메인 μ• ν”Œλ¦¬μΌ€μ΄μ…˜
β”œβ”€β”€ config.py           # μ„€μ •
β”œβ”€β”€ requirements.txt    # μ˜μ‘΄μ„±
β”œβ”€β”€ routes/             # 라우트 λͺ¨λ“ˆ
β”‚   β”œβ”€β”€ __init__.py
β”‚   β”œβ”€β”€ sensors.py
β”‚   └── devices.py
β”œβ”€β”€ models/             # 데이터 λͺ¨λΈ
β”‚   β”œβ”€β”€ __init__.py
β”‚   └── sensor.py
└── utils/              # μœ ν‹Έλ¦¬ν‹°
    β”œβ”€β”€ __init__.py
    └── validators.py

1.3 μ„€μ • 관리

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    """μ• ν”Œλ¦¬μΌ€μ΄μ…˜ μ„€μ •"""
    SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key')
    DEBUG = os.getenv('DEBUG', 'True').lower() == 'true'
    HOST = os.getenv('HOST', '0.0.0.0')
    PORT = int(os.getenv('PORT', 5000))

    # λ°μ΄ν„°λ² μ΄μŠ€ (SQLite μ˜ˆμ‹œ)
    SQLALCHEMY_DATABASE_URI = os.getenv(
        'DATABASE_URL',
        'sqlite:///iot_data.db'
    )

    # MQTT μ„€μ •
    MQTT_BROKER = os.getenv('MQTT_BROKER', 'localhost')
    MQTT_PORT = int(os.getenv('MQTT_PORT', 1883))

2. μ„Όμ„œ 데이터 API

2.1 μ„Όμ„œ 데이터 CRUD

#!/usr/bin/env python3
"""μ„Όμ„œ 데이터 API"""

from flask import Blueprint, jsonify, request
from datetime import datetime
import uuid

sensors_bp = Blueprint('sensors', __name__)

# λ©”λͺ¨λ¦¬ μ €μž₯μ†Œ
sensor_readings = []
sensors_registry = {}

# === μ„Όμ„œ 등둝 ===
@sensors_bp.route('/sensors', methods=['GET'])
def list_sensors():
    """λ“±λ‘λœ μ„Όμ„œ λͺ©λ‘"""
    return jsonify({
        "sensors": list(sensors_registry.values()),
        "count": len(sensors_registry)
    })

@sensors_bp.route('/sensors', methods=['POST'])
def register_sensor():
    """μƒˆ μ„Όμ„œ 등둝"""
    data = request.get_json()

    if not data or 'name' not in data:
        return jsonify({"error": "name is required"}), 400

    sensor_id = str(uuid.uuid4())[:8]
    sensor = {
        "id": sensor_id,
        "name": data['name'],
        "type": data.get('type', 'generic'),
        "location": data.get('location', 'unknown'),
        "registered_at": datetime.now().isoformat(),
        "status": "active"
    }

    sensors_registry[sensor_id] = sensor

    return jsonify(sensor), 201

@sensors_bp.route('/sensors/<sensor_id>', methods=['GET'])
def get_sensor(sensor_id):
    """μ„Όμ„œ 정보 쑰회"""
    sensor = sensors_registry.get(sensor_id)

    if not sensor:
        return jsonify({"error": "Sensor not found"}), 404

    return jsonify(sensor)

# === μ„Όμ„œ 데이터 ===
@sensors_bp.route('/sensors/<sensor_id>/data', methods=['POST'])
def post_sensor_data(sensor_id):
    """μ„Όμ„œ 데이터 μˆ˜μ‹ """
    if sensor_id not in sensors_registry:
        # μžλ™ 등둝 (μ˜΅μ…˜)
        sensors_registry[sensor_id] = {
            "id": sensor_id,
            "name": f"auto_{sensor_id}",
            "registered_at": datetime.now().isoformat(),
            "status": "active"
        }

    data = request.get_json()

    if not data:
        return jsonify({"error": "No data provided"}), 400

    reading = {
        "id": str(uuid.uuid4()),
        "sensor_id": sensor_id,
        "data": data,
        "timestamp": datetime.now().isoformat()
    }

    sensor_readings.append(reading)

    # 졜근 1000개만 μœ μ§€
    if len(sensor_readings) > 1000:
        sensor_readings.pop(0)

    return jsonify({"status": "ok", "reading_id": reading["id"]}), 201

@sensors_bp.route('/sensors/<sensor_id>/data', methods=['GET'])
def get_sensor_data(sensor_id):
    """μ„Όμ„œ 데이터 쑰회"""
    # 쿼리 νŒŒλΌλ―Έν„°
    limit = request.args.get('limit', 100, type=int)
    since = request.args.get('since', None)  # ISO timestamp

    # 필터링
    readings = [r for r in sensor_readings if r['sensor_id'] == sensor_id]

    if since:
        readings = [r for r in readings if r['timestamp'] > since]

    # μ΅œμ‹ μˆœ μ •λ ¬ 및 μ œν•œ
    readings = sorted(readings, key=lambda x: x['timestamp'], reverse=True)[:limit]

    return jsonify({
        "sensor_id": sensor_id,
        "readings": readings,
        "count": len(readings)
    })

@sensors_bp.route('/sensors/<sensor_id>/latest', methods=['GET'])
def get_latest_reading(sensor_id):
    """μ΅œμ‹  μ„Όμ„œ 데이터"""
    readings = [r for r in sensor_readings if r['sensor_id'] == sensor_id]

    if not readings:
        return jsonify({"error": "No data found"}), 404

    latest = max(readings, key=lambda x: x['timestamp'])
    return jsonify(latest)

# λΈ”λ£¨ν”„λ¦°νŠΈ 등둝
# app.pyμ—μ„œ: app.register_blueprint(sensors_bp, url_prefix='/api')

2.2 집계 API

@sensors_bp.route('/sensors/<sensor_id>/stats', methods=['GET'])
def get_sensor_stats(sensor_id):
    """μ„Όμ„œ 데이터 톡계"""
    readings = [r for r in sensor_readings if r['sensor_id'] == sensor_id]

    if not readings:
        return jsonify({"error": "No data found"}), 404

    # 숫자 데이터 μΆ”μΆœ (예: temperature)
    field = request.args.get('field', 'temperature')

    values = []
    for r in readings:
        if field in r.get('data', {}):
            try:
                values.append(float(r['data'][field]))
            except (ValueError, TypeError):
                pass

    if not values:
        return jsonify({"error": f"No numeric data for field: {field}"}), 404

    stats = {
        "sensor_id": sensor_id,
        "field": field,
        "count": len(values),
        "min": min(values),
        "max": max(values),
        "avg": sum(values) / len(values),
        "latest": values[-1] if values else None
    }

    return jsonify(stats)

3. REST API 섀계

3.1 RESTful 원칙

β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
β”‚                    RESTful API 원칙                          β”‚
β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€
β”‚                                                              β”‚
β”‚  1. λ¦¬μ†ŒμŠ€ 쀑심 섀계                                         β”‚
β”‚     β€’ URL은 λͺ…사 μ‚¬μš©: /sensors, /devices                   β”‚
β”‚     β€’ λ™μ‚¬λŠ” HTTP λ©”μ„œλ“œλ‘œ ν‘œν˜„                              β”‚
β”‚                                                              β”‚
β”‚  2. HTTP λ©”μ„œλ“œ                                              β”‚
β”‚     β€’ GET: 쑰회 (λ©±λ“±, μ•ˆμ „)                                β”‚
β”‚     β€’ POST: 생성                                            β”‚
β”‚     β€’ PUT: 전체 μˆ˜μ • (λ©±λ“±)                                 β”‚
β”‚     β€’ PATCH: λΆ€λΆ„ μˆ˜μ •                                      β”‚
β”‚     β€’ DELETE: μ‚­μ œ (λ©±λ“±)                                   β”‚
β”‚                                                              β”‚
β”‚  3. μƒνƒœ μ½”λ“œ                                                β”‚
β”‚     β€’ 200: 성곡                                             β”‚
β”‚     β€’ 201: 생성됨                                           β”‚
β”‚     β€’ 204: λ‚΄μš© μ—†μŒ (μ‚­μ œ)                                 β”‚
β”‚     β€’ 400: 잘λͺ»λœ μš”μ²­                                      β”‚
β”‚     β€’ 401: 인증 ν•„μš”                                        β”‚
β”‚     β€’ 404: λ¦¬μ†ŒμŠ€ μ—†μŒ                                      β”‚
β”‚     β€’ 500: μ„œλ²„ 였λ₯˜                                        β”‚
β”‚                                                              β”‚
β”‚  4. 버저닝                                                   β”‚
β”‚     β€’ URL: /api/v1/sensors                                  β”‚
β”‚     β€’ 헀더: Accept: application/vnd.api.v1+json             β”‚
β”‚                                                              β”‚
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜

3.2 IoT API 섀계 μ˜ˆμ‹œ

# routes/devices.py
"""μž₯치 관리 API"""

from flask import Blueprint, jsonify, request
from datetime import datetime

devices_bp = Blueprint('devices', __name__)

# μ €μž₯μ†Œ
devices = {}

@devices_bp.route('/devices', methods=['GET'])
def list_devices():
    """μž₯치 λͺ©λ‘ 쑰회"""
    # 필터링
    device_type = request.args.get('type')
    status = request.args.get('status')

    result = list(devices.values())

    if device_type:
        result = [d for d in result if d.get('type') == device_type]
    if status:
        result = [d for d in result if d.get('status') == status]

    return jsonify({
        "devices": result,
        "total": len(result)
    })

@devices_bp.route('/devices', methods=['POST'])
def create_device():
    """μž₯치 등둝"""
    data = request.get_json()

    required_fields = ['id', 'name', 'type']
    for field in required_fields:
        if field not in data:
            return jsonify({"error": f"Missing field: {field}"}), 400

    device_id = data['id']
    if device_id in devices:
        return jsonify({"error": "Device already exists"}), 409

    device = {
        **data,
        "status": "offline",
        "created_at": datetime.now().isoformat(),
        "last_seen": None
    }

    devices[device_id] = device
    return jsonify(device), 201

@devices_bp.route('/devices/<device_id>', methods=['GET'])
def get_device(device_id):
    """μž₯치 정보 쑰회"""
    device = devices.get(device_id)
    if not device:
        return jsonify({"error": "Device not found"}), 404
    return jsonify(device)

@devices_bp.route('/devices/<device_id>', methods=['PUT'])
def update_device(device_id):
    """μž₯치 정보 전체 μˆ˜μ •"""
    if device_id not in devices:
        return jsonify({"error": "Device not found"}), 404

    data = request.get_json()
    data['id'] = device_id  # ID μœ μ§€

    devices[device_id] = {
        **data,
        "updated_at": datetime.now().isoformat()
    }

    return jsonify(devices[device_id])

@devices_bp.route('/devices/<device_id>', methods=['PATCH'])
def patch_device(device_id):
    """μž₯치 정보 λΆ€λΆ„ μˆ˜μ •"""
    if device_id not in devices:
        return jsonify({"error": "Device not found"}), 404

    data = request.get_json()

    # 일뢀 ν•„λ“œλ§Œ μ—…λ°μ΄νŠΈ
    devices[device_id].update(data)
    devices[device_id]['updated_at'] = datetime.now().isoformat()

    return jsonify(devices[device_id])

@devices_bp.route('/devices/<device_id>', methods=['DELETE'])
def delete_device(device_id):
    """μž₯치 μ‚­μ œ"""
    if device_id not in devices:
        return jsonify({"error": "Device not found"}), 404

    del devices[device_id]
    return '', 204

# === μž₯치 μ œμ–΄ ===
@devices_bp.route('/devices/<device_id>/commands', methods=['POST'])
def send_command(device_id):
    """μž₯μΉ˜μ— λͺ…λ Ή 전솑"""
    if device_id not in devices:
        return jsonify({"error": "Device not found"}), 404

    data = request.get_json()

    if 'command' not in data:
        return jsonify({"error": "Command required"}), 400

    # λͺ…λ Ή 처리 (μ‹€μ œλ‘œλŠ” MQTT λ°œν–‰ λ“±)
    command = {
        "device_id": device_id,
        "command": data['command'],
        "params": data.get('params', {}),
        "sent_at": datetime.now().isoformat()
    }

    # μ—¬κΈ°μ„œ MQTT λ°œν–‰ λ˜λŠ” 직접 μ œμ–΄
    print(f"Command sent: {command}")

    return jsonify({
        "status": "sent",
        "command": command
    }), 202

3.3 νŽ˜μ΄μ§€λ„€μ΄μ…˜

@sensors_bp.route('/readings', methods=['GET'])
def list_all_readings():
    """λͺ¨λ“  μ„Όμ„œ 데이터 (νŽ˜μ΄μ§€λ„€μ΄μ…˜)"""
    # 쿼리 νŒŒλΌλ―Έν„°
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    per_page = min(per_page, 100)  # μ΅œλŒ€ 100개

    # μ •λ ¬
    sort_by = request.args.get('sort', 'timestamp')
    order = request.args.get('order', 'desc')

    # 데이터 μ •λ ¬
    readings = sorted(
        sensor_readings,
        key=lambda x: x.get(sort_by, ''),
        reverse=(order == 'desc')
    )

    # νŽ˜μ΄μ§€λ„€μ΄μ…˜
    total = len(readings)
    start = (page - 1) * per_page
    end = start + per_page
    page_data = readings[start:end]

    return jsonify({
        "readings": page_data,
        "pagination": {
            "page": page,
            "per_page": per_page,
            "total": total,
            "pages": (total + per_page - 1) // per_page,
            "has_next": end < total,
            "has_prev": page > 1
        }
    })

4. JSON 데이터 처리

4.1 μš”μ²­ 검증

# utils/validators.py
"""μš”μ²­ 데이터 검증"""

from functools import wraps
from flask import request, jsonify

def validate_json(required_fields: list = None, optional_fields: list = None):
    """JSON μš”μ²­ 검증 λ°μ½”λ ˆμ΄ν„°"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # JSON 확인
            if not request.is_json:
                return jsonify({"error": "Content-Type must be application/json"}), 400

            data = request.get_json()

            if data is None:
                return jsonify({"error": "Invalid JSON"}), 400

            # ν•„μˆ˜ ν•„λ“œ 확인
            if required_fields:
                missing = [f for f in required_fields if f not in data]
                if missing:
                    return jsonify({
                        "error": "Missing required fields",
                        "fields": missing
                    }), 400

            return f(*args, **kwargs)
        return decorated_function
    return decorator

# μ‚¬μš© 예
@app.route('/api/sensors', methods=['POST'])
@validate_json(required_fields=['name', 'type'])
def create_sensor():
    data = request.get_json()
    # ... 처리

4.2 Pydantic을 μ΄μš©ν•œ 검증

# models/sensor.py
"""Pydantic λͺ¨λΈλ‘œ 데이터 검증"""

from pydantic import BaseModel, Field, validator
from typing import Optional
from datetime import datetime

class SensorReading(BaseModel):
    """μ„Όμ„œ 데이터 λͺ¨λΈ"""
    temperature: Optional[float] = Field(None, ge=-50, le=100)
    humidity: Optional[float] = Field(None, ge=0, le=100)
    pressure: Optional[float] = Field(None, ge=800, le=1200)
    timestamp: datetime = Field(default_factory=datetime.now)

    @validator('temperature', 'humidity', 'pressure', pre=True)
    def round_values(cls, v):
        if v is not None:
            return round(v, 2)
        return v

class SensorCreate(BaseModel):
    """μ„Όμ„œ 생성 μš”μ²­"""
    name: str = Field(..., min_length=1, max_length=100)
    type: str = Field(..., pattern=r'^(temperature|humidity|motion|generic)$')
    location: Optional[str] = None

class DeviceCommand(BaseModel):
    """μž₯치 λͺ…λ Ή"""
    command: str = Field(..., pattern=r'^(on|off|toggle|set)$')
    params: Optional[dict] = None

# Flaskμ—μ„œ μ‚¬μš©
from pydantic import ValidationError

@app.route('/api/sensors/<sensor_id>/data', methods=['POST'])
def post_data(sensor_id):
    try:
        reading = SensorReading(**request.get_json())
    except ValidationError as e:
        return jsonify({"error": e.errors()}), 400

    # κ²€μ¦λœ 데이터 μ‚¬μš©
    data = reading.dict()
    # ...

4.3 응닡 ν¬λ§·νŒ…

# utils/response.py
"""응닡 헬퍼"""

from flask import jsonify
from datetime import datetime
from functools import wraps

def api_response(data=None, message=None, status_code=200):
    """ν‘œμ€€ API 응닡 생성"""
    response = {
        "success": 200 <= status_code < 300,
        "timestamp": datetime.now().isoformat()
    }

    if message:
        response["message"] = message
    if data is not None:
        response["data"] = data

    return jsonify(response), status_code

def error_response(message: str, status_code: int = 400, details=None):
    """μ—λŸ¬ 응닡 생성"""
    response = {
        "success": False,
        "error": {
            "message": message,
            "code": status_code
        },
        "timestamp": datetime.now().isoformat()
    }

    if details:
        response["error"]["details"] = details

    return jsonify(response), status_code

# μ—λŸ¬ ν•Έλ“€λŸ¬
@app.errorhandler(404)
def not_found(error):
    return error_response("Resource not found", 404)

@app.errorhandler(500)
def internal_error(error):
    return error_response("Internal server error", 500)

5. μ’…ν•© 예제: IoT κ²Œμ΄νŠΈμ›¨μ΄ μ„œλ²„

#!/usr/bin/env python3
"""IoT κ²Œμ΄νŠΈμ›¨μ΄ μ„œλ²„"""

from flask import Flask, jsonify, request
from flask_cors import CORS
from datetime import datetime
import paho.mqtt.client as mqtt
import threading
import json

app = Flask(__name__)
CORS(app)

# 데이터 μ €μž₯μ†Œ
class DataStore:
    def __init__(self):
        self.sensors = {}
        self.readings = []
        self.devices = {}
        self.commands = []

    def add_reading(self, sensor_id: str, data: dict):
        reading = {
            "sensor_id": sensor_id,
            "data": data,
            "timestamp": datetime.now().isoformat()
        }
        self.readings.append(reading)

        # μ„Όμ„œ μžλ™ 등둝
        if sensor_id not in self.sensors:
            self.sensors[sensor_id] = {
                "id": sensor_id,
                "last_seen": reading["timestamp"]
            }
        else:
            self.sensors[sensor_id]["last_seen"] = reading["timestamp"]

        # μ΅œλŒ€ 10000개 μœ μ§€
        if len(self.readings) > 10000:
            self.readings = self.readings[-10000:]

store = DataStore()

# === MQTT ν΄λΌμ΄μ–ΈνŠΈ ===
mqtt_client = mqtt.Client()

def on_mqtt_connect(client, userdata, flags, rc):
    if rc == 0:
        print("MQTT 브둜컀 연결됨")
        client.subscribe("sensor/#")

def on_mqtt_message(client, userdata, msg):
    try:
        payload = json.loads(msg.payload.decode())
        # topic: sensor/{sensor_id}/data
        parts = msg.topic.split('/')
        if len(parts) >= 2:
            sensor_id = parts[1]
            store.add_reading(sensor_id, payload)
            print(f"[MQTT] {sensor_id}: {payload}")
    except Exception as e:
        print(f"MQTT λ©”μ‹œμ§€ 처리 였λ₯˜: {e}")

mqtt_client.on_connect = on_mqtt_connect
mqtt_client.on_message = on_mqtt_message

def start_mqtt():
    try:
        mqtt_client.connect("localhost", 1883)
        mqtt_client.loop_start()
    except Exception as e:
        print(f"MQTT μ—°κ²° μ‹€νŒ¨: {e}")

# === HTTP API ===
@app.route('/api/sensors', methods=['GET'])
def list_sensors():
    return jsonify({
        "sensors": list(store.sensors.values()),
        "count": len(store.sensors)
    })

@app.route('/api/sensors/<sensor_id>/data', methods=['GET'])
def get_sensor_data(sensor_id):
    limit = request.args.get('limit', 100, type=int)

    readings = [r for r in store.readings if r['sensor_id'] == sensor_id]
    readings = readings[-limit:]

    return jsonify({
        "sensor_id": sensor_id,
        "readings": readings,
        "count": len(readings)
    })

@app.route('/api/sensors/<sensor_id>/data', methods=['POST'])
def post_sensor_data(sensor_id):
    data = request.get_json()
    if not data:
        return jsonify({"error": "No data"}), 400

    store.add_reading(sensor_id, data)

    # MQTTλ‘œλ„ λ°œν–‰ (λ‹€λ₯Έ κ΅¬λ…μžμ—κ²Œ)
    mqtt_client.publish(f"sensor/{sensor_id}/data", json.dumps(data))

    return jsonify({"status": "ok"}), 201

@app.route('/api/devices/<device_id>/command', methods=['POST'])
def send_device_command(device_id):
    data = request.get_json()
    if not data or 'command' not in data:
        return jsonify({"error": "Command required"}), 400

    command = {
        "device_id": device_id,
        "command": data['command'],
        "params": data.get('params', {}),
        "timestamp": datetime.now().isoformat()
    }

    # MQTT둜 λͺ…λ Ή λ°œν–‰
    mqtt_client.publish(
        f"device/{device_id}/command",
        json.dumps(command)
    )

    store.commands.append(command)

    return jsonify({"status": "sent", "command": command}), 202

@app.route('/api/stats', methods=['GET'])
def get_stats():
    return jsonify({
        "sensors_count": len(store.sensors),
        "readings_count": len(store.readings),
        "devices_count": len(store.devices),
        "commands_count": len(store.commands)
    })

if __name__ == "__main__":
    # MQTT μŠ€λ ˆλ“œ μ‹œμž‘
    mqtt_thread = threading.Thread(target=start_mqtt, daemon=True)
    mqtt_thread.start()

    # Flask μ„œλ²„ μ‹œμž‘
    app.run(host='0.0.0.0', port=5000, debug=True)

μ—°μŠ΅ 문제

문제 1: μ„Όμ„œ API

  1. μ„Όμ„œ CRUD APIλ₯Ό κ΅¬ν˜„ν•˜μ„Έμš”.
  2. 데이터 검증을 μΆ”κ°€ν•˜μ„Έμš”.

문제 2: νŽ˜μ΄μ§€λ„€μ΄μ…˜

  1. μ„Όμ„œ 데이터 λͺ©λ‘μ— νŽ˜μ΄μ§€λ„€μ΄μ…˜μ„ κ΅¬ν˜„ν•˜μ„Έμš”.
  2. λ‚ μ§œ λ²”μœ„ 필터링을 μΆ”κ°€ν•˜μ„Έμš”.

문제 3: MQTT 연동

  1. HTTP POST둜 받은 데이터λ₯Ό MQTT둜 λ°œν–‰ν•˜μ„Έμš”.
  2. MQTT둜 받은 데이터λ₯Ό HTTP GET으둜 μ‘°νšŒν•˜μ„Έμš”.

λ‹€μŒ 단계

μ΅œμ’… μ—…λ°μ΄νŠΈ: 2026-02-01

Previous 06. MQTT ν”„λ‘œν† μ½œ
πŸ“‹ Topic list
Next 08. Edge AI - TensorFlow Lite
← → to navigate between lessons