home_automation.py

   1#!/usr/bin/env python3
   2"""
   3스마트홈 자동화 시스템 (Smart Home Automation System)
   4
   5통합 홈 자동화 시스템으로 조명 제어, 환경 모니터링, MQTT 기반 장치 통신을 제공합니다.
   6시뮬레이션 모드를 지원하여 실제 하드웨어 없이도 동작 가능합니다.
   7
   8주요 기능:
   9- 릴레이를 통한 조명/가전 제어 (시뮬레이션 모드 지원)
  10- 온습도 센서 모니터링 (시뮬레이션 데이터 생성)
  11- MQTT 기반 장치 통신 및 제어
  12- 자동화 규칙 엔진 (온도 기반, 모션 기반)
  13- 웹 대시보드 JSON API 제공
  14- 상태 관리 및 로깅
  15
  16사용법:
  17    # 시뮬레이션 모드 (하드웨어 불필요)
  18    python home_automation.py --simulate
  19
  20    # 실제 하드웨어 모드
  21    python home_automation.py
  22
  23    # MQTT 브로커 지정
  24    python home_automation.py --broker mqtt.example.com --simulate
  25"""
  26
  27import time
  28import json
  29import random
  30import threading
  31import logging
  32from datetime import datetime
  33from dataclasses import dataclass, asdict
  34from typing import Dict, List, Optional, Callable
  35from queue import Queue
  36import argparse
  37
  38# 시뮬레이션 모드 플래그
  39SIMULATION_MODE = True
  40
  41# 실제 하드웨어 라이브러리 (시뮬레이션 모드에서는 사용 안 함)
  42try:
  43    from gpiozero import OutputDevice
  44    import adafruit_dht
  45    import board
  46    HARDWARE_AVAILABLE = True
  47except ImportError:
  48    HARDWARE_AVAILABLE = False
  49    print("하드웨어 라이브러리 없음. 시뮬레이션 모드로 실행됩니다.")
  50
  51try:
  52    import paho.mqtt.client as mqtt
  53    MQTT_AVAILABLE = True
  54except ImportError:
  55    MQTT_AVAILABLE = False
  56    print("paho-mqtt 라이브러리가 없습니다. MQTT 기능이 비활성화됩니다.")
  57
  58
  59# ============================================================
  60# 데이터 모델
  61# ============================================================
  62
  63@dataclass
  64class Light:
  65    """조명 장치 데이터 클래스"""
  66    id: str
  67    name: str
  68    gpio_pin: int
  69    location: str
  70    is_on: bool = False
  71
  72    def to_dict(self) -> dict:
  73        """딕셔너리 변환"""
  74        return asdict(self)
  75
  76
  77@dataclass
  78class SensorReading:
  79    """센서 데이터 클래스"""
  80    sensor_id: str
  81    temperature: float
  82    humidity: float
  83    timestamp: datetime
  84
  85    def to_dict(self) -> dict:
  86        """딕셔너리 변환"""
  87        return {
  88            "sensor_id": self.sensor_id,
  89            "temperature": self.temperature,
  90            "humidity": self.humidity,
  91            "timestamp": self.timestamp.isoformat()
  92        }
  93
  94
  95# ============================================================
  96# 조명 제어기 (Light Controller)
  97# ============================================================
  98
  99class LightController:
 100    """
 101    조명 제어 클래스
 102
 103    릴레이를 통해 조명을 제어합니다.
 104    시뮬레이션 모드에서는 실제 GPIO 대신 상태만 관리합니다.
 105    """
 106
 107    def __init__(self, config: dict, simulate: bool = True):
 108        """
 109        초기화
 110
 111        Args:
 112            config: 조명 설정 (lights 리스트 포함)
 113            simulate: 시뮬레이션 모드 여부
 114        """
 115        self.simulate = simulate
 116        self.lights: Dict[str, Light] = {}
 117        self.relays: Dict[str, any] = {}
 118
 119        # 조명 설정
 120        for light_config in config.get('lights', []):
 121            light = Light(**light_config)
 122            self.lights[light.id] = light
 123
 124            # 릴레이 초기화
 125            if not simulate and HARDWARE_AVAILABLE:
 126                # 실제 하드웨어: gpiozero 사용
 127                relay = OutputDevice(
 128                    light.gpio_pin,
 129                    active_high=False,  # Active Low 릴레이
 130                    initial_value=False
 131                )
 132                self.relays[light.id] = relay
 133            else:
 134                # 시뮬레이션: None 저장
 135                self.relays[light.id] = None
 136
 137        logging.info(f"LightController 초기화 완료 (시뮬레이션={simulate}, 조명={len(self.lights)}개)")
 138
 139    def turn_on(self, light_id: str) -> bool:
 140        """
 141        조명 켜기
 142
 143        Args:
 144            light_id: 조명 ID
 145
 146        Returns:
 147            성공 여부
 148        """
 149        if light_id not in self.lights:
 150            logging.warning(f"조명 ID '{light_id}' 없음")
 151            return False
 152
 153        if self.simulate:
 154            # 시뮬레이션: 상태만 변경
 155            self.lights[light_id].is_on = True
 156            logging.info(f"[시뮬] 조명 ON: {self.lights[light_id].name}")
 157        else:
 158            # 실제 하드웨어: 릴레이 제어
 159            self.relays[light_id].on()
 160            self.lights[light_id].is_on = True
 161            logging.info(f"조명 ON: {self.lights[light_id].name}")
 162
 163        return True
 164
 165    def turn_off(self, light_id: str) -> bool:
 166        """
 167        조명 끄기
 168
 169        Args:
 170            light_id: 조명 ID
 171
 172        Returns:
 173            성공 여부
 174        """
 175        if light_id not in self.lights:
 176            logging.warning(f"조명 ID '{light_id}' 없음")
 177            return False
 178
 179        if self.simulate:
 180            # 시뮬레이션
 181            self.lights[light_id].is_on = False
 182            logging.info(f"[시뮬] 조명 OFF: {self.lights[light_id].name}")
 183        else:
 184            # 실제 하드웨어
 185            self.relays[light_id].off()
 186            self.lights[light_id].is_on = False
 187            logging.info(f"조명 OFF: {self.lights[light_id].name}")
 188
 189        return True
 190
 191    def toggle(self, light_id: str) -> bool:
 192        """
 193        조명 토글
 194
 195        Args:
 196            light_id: 조명 ID
 197
 198        Returns:
 199            성공 여부
 200        """
 201        if light_id not in self.lights:
 202            return False
 203
 204        if self.lights[light_id].is_on:
 205            return self.turn_off(light_id)
 206        else:
 207            return self.turn_on(light_id)
 208
 209    def get_status(self, light_id: str = None) -> Optional[dict]:
 210        """
 211        조명 상태 조회
 212
 213        Args:
 214            light_id: 조명 ID (None이면 전체)
 215
 216        Returns:
 217            상태 딕셔너리
 218        """
 219        if light_id:
 220            light = self.lights.get(light_id)
 221            if light:
 222                return light.to_dict()
 223            return None
 224
 225        # 전체 조명 상태
 226        return {
 227            "lights": [light.to_dict() for light in self.lights.values()]
 228        }
 229
 230    def all_off(self):
 231        """모든 조명 끄기"""
 232        for light_id in self.lights:
 233            self.turn_off(light_id)
 234        logging.info("모든 조명 OFF")
 235
 236    def all_on(self):
 237        """모든 조명 켜기"""
 238        for light_id in self.lights:
 239            self.turn_on(light_id)
 240        logging.info("모든 조명 ON")
 241
 242    def cleanup(self):
 243        """정리 (프로그램 종료 시 호출)"""
 244        if not self.simulate:
 245            for relay in self.relays.values():
 246                if relay:
 247                    relay.close()
 248        logging.info("LightController 정리 완료")
 249
 250
 251# ============================================================
 252# 환경 센서 모니터 (Environment Monitor)
 253# ============================================================
 254
 255class EnvironmentMonitor:
 256    """
 257    환경 센서 모니터링 클래스
 258
 259    DHT11 센서로 온도/습도를 주기적으로 읽고 히스토리를 저장합니다.
 260    시뮬레이션 모드에서는 랜덤 데이터를 생성합니다.
 261    """
 262
 263    def __init__(self, sensor_pin: int = 4, sensor_id: str = "env_01", simulate: bool = True):
 264        """
 265        초기화
 266
 267        Args:
 268            sensor_pin: GPIO 핀 번호
 269            sensor_id: 센서 ID
 270            simulate: 시뮬레이션 모드 여부
 271        """
 272        self.sensor_id = sensor_id
 273        self.sensor_pin = sensor_pin
 274        self.simulate = simulate
 275
 276        # DHT 센서 초기화
 277        self.dht = None
 278        if not simulate and HARDWARE_AVAILABLE:
 279            try:
 280                self.dht = adafruit_dht.DHT11(getattr(board, f"D{sensor_pin}"))
 281            except Exception as e:
 282                logging.error(f"DHT 센서 초기화 실패: {e}")
 283                self.simulate = True
 284
 285        # 데이터 큐 및 히스토리
 286        self.data_queue = Queue()
 287        self.latest_reading: Optional[SensorReading] = None
 288        self.readings_history: List[SensorReading] = []
 289        self.max_history = 1000
 290
 291        # 스레드 제어
 292        self.running = False
 293        self.thread = None
 294
 295        # 시뮬레이션용 현재 값
 296        self.sim_temperature = 25.0
 297        self.sim_humidity = 60.0
 298
 299        logging.info(f"EnvironmentMonitor 초기화 (시뮬레이션={simulate})")
 300
 301    def read_sensor(self) -> Optional[SensorReading]:
 302        """
 303        센서 읽기
 304
 305        Returns:
 306            센서 데이터 또는 None (실패 시)
 307        """
 308        if self.simulate:
 309            # 시뮬레이션: 랜덤 변화 생성
 310            self.sim_temperature += random.uniform(-0.5, 0.5)
 311            self.sim_temperature = max(10, min(40, self.sim_temperature))
 312
 313            self.sim_humidity += random.uniform(-2, 2)
 314            self.sim_humidity = max(30, min(90, self.sim_humidity))
 315
 316            reading = SensorReading(
 317                sensor_id=self.sensor_id,
 318                temperature=round(self.sim_temperature, 1),
 319                humidity=round(self.sim_humidity, 1),
 320                timestamp=datetime.now()
 321            )
 322            return reading
 323
 324        else:
 325            # 실제 센서
 326            try:
 327                temperature = self.dht.temperature
 328                humidity = self.dht.humidity
 329
 330                if temperature is not None and humidity is not None:
 331                    reading = SensorReading(
 332                        sensor_id=self.sensor_id,
 333                        temperature=temperature,
 334                        humidity=humidity,
 335                        timestamp=datetime.now()
 336                    )
 337                    return reading
 338
 339            except RuntimeError as e:
 340                # DHT 센서는 가끔 읽기 실패 (정상)
 341                logging.debug(f"센서 읽기 실패 (정상): {e}")
 342            except Exception as e:
 343                logging.error(f"센서 읽기 오류: {e}")
 344
 345            return None
 346
 347    def _monitor_loop(self, interval: int):
 348        """
 349        모니터링 루프 (백그라운드 스레드)
 350
 351        Args:
 352            interval: 읽기 간격 (초)
 353        """
 354        while self.running:
 355            reading = self.read_sensor()
 356
 357            if reading:
 358                # 최신 데이터 업데이트
 359                self.latest_reading = reading
 360
 361                # 히스토리 저장
 362                self.readings_history.append(reading)
 363                if len(self.readings_history) > self.max_history:
 364                    self.readings_history.pop(0)
 365
 366                # 큐에 추가 (외부 구독자용)
 367                self.data_queue.put(reading)
 368
 369            time.sleep(interval)
 370
 371    def start(self, interval: int = 5):
 372        """
 373        모니터링 시작
 374
 375        Args:
 376            interval: 읽기 간격 (초)
 377        """
 378        if self.running:
 379            logging.warning("이미 모니터링 중입니다")
 380            return
 381
 382        self.running = True
 383        self.thread = threading.Thread(
 384            target=self._monitor_loop,
 385            args=(interval,),
 386            daemon=True
 387        )
 388        self.thread.start()
 389        logging.info(f"환경 모니터링 시작 (간격={interval}초)")
 390
 391    def stop(self):
 392        """모니터링 중지"""
 393        if not self.running:
 394            return
 395
 396        self.running = False
 397        if self.thread:
 398            self.thread.join(timeout=5)
 399
 400        if self.dht and not self.simulate:
 401            self.dht.exit()
 402
 403        logging.info("환경 모니터링 중지")
 404
 405    def get_latest(self) -> Optional[dict]:
 406        """최신 센서 데이터 반환"""
 407        if self.latest_reading:
 408            return self.latest_reading.to_dict()
 409        return None
 410
 411    def get_stats(self) -> dict:
 412        """
 413        통계 데이터 반환
 414
 415        Returns:
 416            최소/최대/평균 통계
 417        """
 418        if not self.readings_history:
 419            return {}
 420
 421        temps = [r.temperature for r in self.readings_history]
 422        humids = [r.humidity for r in self.readings_history]
 423
 424        return {
 425            "count": len(self.readings_history),
 426            "temperature": {
 427                "min": round(min(temps), 1),
 428                "max": round(max(temps), 1),
 429                "avg": round(sum(temps) / len(temps), 1)
 430            },
 431            "humidity": {
 432                "min": round(min(humids), 1),
 433                "max": round(max(humids), 1),
 434                "avg": round(sum(humids) / len(humids), 1)
 435            }
 436        }
 437
 438
 439# ============================================================
 440# MQTT 핸들러
 441# ============================================================
 442
 443class SmartHomeMQTT:
 444    """
 445    MQTT 기반 스마트홈 제어 핸들러
 446
 447    MQTT 브로커를 통해 장치를 제어하고 센서 데이터를 발행합니다.
 448    """
 449
 450    TOPICS = {
 451        "light_command": "home/+/light/command",
 452        "light_status": "home/{}/light/status",
 453        "sensor_data": "home/sensor/{}",
 454        "motion": "home/motion/{}",
 455        "system": "home/system/status",
 456        "automation": "home/automation/event"
 457    }
 458
 459    def __init__(self, light_controller: LightController,
 460                 env_monitor: EnvironmentMonitor,
 461                 broker: str = "localhost",
 462                 port: int = 1883):
 463        """
 464        초기화
 465
 466        Args:
 467            light_controller: 조명 제어기
 468            env_monitor: 환경 모니터
 469            broker: MQTT 브로커 주소
 470            port: MQTT 포트
 471        """
 472        self.light_controller = light_controller
 473        self.env_monitor = env_monitor
 474        self.broker = broker
 475        self.port = port
 476
 477        if not MQTT_AVAILABLE:
 478            logging.warning("MQTT 라이브러리 없음. MQTT 기능 비활성화")
 479            self.client = None
 480            return
 481
 482        # MQTT 클라이언트 생성
 483        self.client = mqtt.Client(client_id="smart_home_gateway")
 484        self.client.on_connect = self._on_connect
 485        self.client.on_message = self._on_message
 486
 487        # Last Will and Testament (LWT) 설정
 488        self.client.will_set(
 489            self.TOPICS["system"],
 490            json.dumps({"status": "offline"}),
 491            qos=1,
 492            retain=True
 493        )
 494
 495        logging.info(f"MQTT 클라이언트 생성 (브로커={broker}:{port})")
 496
 497    def connect(self):
 498        """MQTT 브로커 연결"""
 499        if not self.client:
 500            return
 501
 502        try:
 503            self.client.connect(self.broker, self.port)
 504            logging.info("MQTT 브로커 연결 시도 중...")
 505        except Exception as e:
 506            logging.error(f"MQTT 연결 실패: {e}")
 507
 508    def _on_connect(self, client, userdata, flags, rc):
 509        """MQTT 연결 콜백"""
 510        if rc == 0:
 511            logging.info("MQTT 브로커 연결 성공")
 512
 513            # 토픽 구독
 514            client.subscribe(self.TOPICS["light_command"])
 515            logging.info(f"토픽 구독: {self.TOPICS['light_command']}")
 516
 517            # 온라인 상태 발행
 518            client.publish(
 519                self.TOPICS["system"],
 520                json.dumps({
 521                    "status": "online",
 522                    "timestamp": datetime.now().isoformat()
 523                }),
 524                qos=1,
 525                retain=True
 526            )
 527        else:
 528            logging.error(f"MQTT 연결 실패 (코드={rc})")
 529
 530    def _on_message(self, client, userdata, msg):
 531        """MQTT 메시지 수신 콜백"""
 532        try:
 533            topic = msg.topic
 534            payload = json.loads(msg.payload.decode())
 535
 536            logging.debug(f"MQTT 수신: {topic} = {payload}")
 537
 538            # 조명 명령 처리
 539            if "light/command" in topic:
 540                self._handle_light_command(topic, payload)
 541
 542        except json.JSONDecodeError:
 543            logging.error(f"잘못된 JSON: {msg.payload}")
 544        except Exception as e:
 545            logging.error(f"메시지 처리 오류: {e}")
 546
 547    def _handle_light_command(self, topic: str, payload: dict):
 548        """
 549        조명 명령 처리
 550
 551        Args:
 552            topic: MQTT 토픽 (home/{room}/light/command)
 553            payload: 명령 데이터 {"command": "on|off|toggle"}
 554        """
 555        # 토픽에서 방 ID 추출
 556        parts = topic.split('/')
 557        room = parts[1] if len(parts) >= 2 else None
 558
 559        if not room:
 560            logging.warning(f"방 ID 없음: {topic}")
 561            return
 562
 563        command = payload.get("command")
 564
 565        # 명령 실행
 566        result = False
 567        if command == "on":
 568            result = self.light_controller.turn_on(room)
 569        elif command == "off":
 570            result = self.light_controller.turn_off(room)
 571        elif command == "toggle":
 572            result = self.light_controller.toggle(room)
 573        else:
 574            logging.warning(f"알 수 없는 명령: {command}")
 575
 576        # 상태 발행
 577        if result:
 578            status = self.light_controller.get_status(room)
 579            if status:
 580                self.publish_light_status(room, status)
 581
 582    def publish_light_status(self, room: str, status: dict):
 583        """
 584        조명 상태 발행
 585
 586        Args:
 587            room: 방 ID
 588            status: 상태 데이터
 589        """
 590        if not self.client:
 591            return
 592
 593        topic = self.TOPICS["light_status"].format(room)
 594        self.client.publish(topic, json.dumps(status), qos=1, retain=True)
 595        logging.debug(f"조명 상태 발행: {topic}")
 596
 597    def publish_sensor_data(self, sensor_id: str, data: dict):
 598        """
 599        센서 데이터 발행
 600
 601        Args:
 602            sensor_id: 센서 ID
 603            data: 센서 데이터
 604        """
 605        if not self.client:
 606            return
 607
 608        topic = self.TOPICS["sensor_data"].format(sensor_id)
 609        self.client.publish(topic, json.dumps(data), qos=0)
 610
 611    def publish_motion(self, sensor_id: str, detected: bool):
 612        """
 613        모션 감지 발행
 614
 615        Args:
 616            sensor_id: 센서 ID
 617            detected: 감지 여부
 618        """
 619        if not self.client:
 620            return
 621
 622        topic = self.TOPICS["motion"].format(sensor_id)
 623        data = {
 624            "detected": detected,
 625            "timestamp": datetime.now().isoformat()
 626        }
 627        self.client.publish(topic, json.dumps(data), qos=1)
 628
 629    def publish_automation_event(self, event_type: str, details: dict):
 630        """
 631        자동화 이벤트 발행
 632
 633        Args:
 634            event_type: 이벤트 타입
 635            details: 상세 정보
 636        """
 637        if not self.client:
 638            return
 639
 640        data = {
 641            "event_type": event_type,
 642            "details": details,
 643            "timestamp": datetime.now().isoformat()
 644        }
 645        self.client.publish(self.TOPICS["automation"], json.dumps(data), qos=1)
 646
 647    def start(self):
 648        """MQTT 루프 시작"""
 649        if self.client:
 650            self.client.loop_start()
 651
 652    def stop(self):
 653        """MQTT 중지"""
 654        if not self.client:
 655            return
 656
 657        # 오프라인 상태 발행
 658        self.client.publish(
 659            self.TOPICS["system"],
 660            json.dumps({"status": "offline"}),
 661            qos=1,
 662            retain=True
 663        )
 664
 665        self.client.loop_stop()
 666        self.client.disconnect()
 667        logging.info("MQTT 연결 종료")
 668
 669
 670# ============================================================
 671# 자동화 규칙 엔진
 672# ============================================================
 673
 674class AutomationEngine:
 675    """
 676    자동화 규칙 엔진
 677
 678    센서 데이터 기반으로 조명/가전을 자동 제어합니다.
 679    """
 680
 681    def __init__(self, light_controller: LightController, mqtt_handler: SmartHomeMQTT):
 682        """
 683        초기화
 684
 685        Args:
 686            light_controller: 조명 제어기
 687            mqtt_handler: MQTT 핸들러
 688        """
 689        self.light_controller = light_controller
 690        self.mqtt_handler = mqtt_handler
 691        self.rules: List[dict] = []
 692
 693        logging.info("AutomationEngine 초기화")
 694
 695    def add_rule(self, name: str, condition: Callable, action: Callable):
 696        """
 697        규칙 추가
 698
 699        Args:
 700            name: 규칙 이름
 701            condition: 조건 함수 (True/False 반환)
 702            action: 실행 함수
 703        """
 704        self.rules.append({
 705            "name": name,
 706            "condition": condition,
 707            "action": action,
 708            "last_triggered": None
 709        })
 710        logging.info(f"자동화 규칙 추가: {name}")
 711
 712    def check_rules(self, sensor_data: dict):
 713        """
 714        규칙 검사 및 실행
 715
 716        Args:
 717            sensor_data: 센서 데이터
 718        """
 719        for rule in self.rules:
 720            try:
 721                if rule["condition"](sensor_data):
 722                    # 조건 만족 시 액션 실행
 723                    logging.info(f"자동화 규칙 트리거: {rule['name']}")
 724                    rule["action"](sensor_data)
 725                    rule["last_triggered"] = datetime.now()
 726
 727                    # MQTT 이벤트 발행
 728                    self.mqtt_handler.publish_automation_event(
 729                        event_type=rule["name"],
 730                        details=sensor_data
 731                    )
 732
 733            except Exception as e:
 734                logging.error(f"규칙 실행 오류 ({rule['name']}): {e}")
 735
 736    def get_rules_status(self) -> List[dict]:
 737        """규칙 상태 조회"""
 738        return [
 739            {
 740                "name": rule["name"],
 741                "last_triggered": rule["last_triggered"].isoformat() if rule["last_triggered"] else None
 742            }
 743            for rule in self.rules
 744        ]
 745
 746
 747# ============================================================
 748# 스마트홈 게이트웨이 (통합 시스템)
 749# ============================================================
 750
 751class SmartHomeGateway:
 752    """
 753    스마트홈 통합 게이트웨이
 754
 755    모든 컴포넌트를 통합하여 스마트홈 시스템을 운영합니다.
 756    """
 757
 758    def __init__(self, config: dict, simulate: bool = True):
 759        """
 760        초기화
 761
 762        Args:
 763            config: 설정 딕셔너리
 764            simulate: 시뮬레이션 모드 여부
 765        """
 766        self.config = config
 767        self.simulate = simulate
 768
 769        # 조명 제어기
 770        self.light_controller = LightController(config, simulate=simulate)
 771
 772        # 환경 모니터
 773        self.env_monitor = EnvironmentMonitor(
 774            sensor_pin=config.get('dht_pin', 4),
 775            sensor_id="env_01",
 776            simulate=simulate
 777        )
 778
 779        # MQTT 핸들러
 780        self.mqtt_handler = SmartHomeMQTT(
 781            self.light_controller,
 782            self.env_monitor,
 783            broker=config.get('mqtt_broker', 'localhost'),
 784            port=config.get('mqtt_port', 1883)
 785        )
 786
 787        # 자동화 엔진
 788        self.automation_engine = AutomationEngine(
 789            self.light_controller,
 790            self.mqtt_handler
 791        )
 792
 793        # 스레드 제어
 794        self.running = False
 795        self.threads = []
 796
 797        # 자동화 규칙 설정
 798        self._setup_automation_rules()
 799
 800        logging.info("SmartHomeGateway 초기화 완료")
 801
 802    def _setup_automation_rules(self):
 803        """자동화 규칙 설정"""
 804
 805        # 규칙 1: 온도가 30도 이상이면 거실 조명 켜기 (예: 에어컨 대신)
 806        def temp_high_condition(data):
 807            return data.get("temperature", 0) > 30
 808
 809        def temp_high_action(data):
 810            self.light_controller.turn_on("living_room")
 811            logging.info(f"[자동화] 고온 감지 ({data['temperature']}°C) - 거실 조명 ON")
 812
 813        self.automation_engine.add_rule(
 814            "high_temperature_alert",
 815            temp_high_condition,
 816            temp_high_action
 817        )
 818
 819        # 규칙 2: 온도가 20도 이하이면 모든 조명 끄기
 820        def temp_low_condition(data):
 821            return data.get("temperature", 100) < 20
 822
 823        def temp_low_action(data):
 824            self.light_controller.all_off()
 825            logging.info(f"[자동화] 저온 감지 ({data['temperature']}°C) - 모든 조명 OFF")
 826
 827        self.automation_engine.add_rule(
 828            "low_temperature_save",
 829            temp_low_condition,
 830            temp_low_action
 831        )
 832
 833        # 규칙 3: 습도가 80% 이상이면 욕실 조명 켜기
 834        def humidity_high_condition(data):
 835            return data.get("humidity", 0) > 80
 836
 837        def humidity_high_action(data):
 838            self.light_controller.turn_on("bathroom")
 839            logging.info(f"[자동화] 고습도 감지 ({data['humidity']}%) - 욕실 조명 ON")
 840
 841        self.automation_engine.add_rule(
 842            "high_humidity_ventilation",
 843            humidity_high_condition,
 844            humidity_high_action
 845        )
 846
 847    def _sensor_publish_loop(self, interval: int):
 848        """
 849        센서 데이터 발행 루프
 850
 851        Args:
 852            interval: 발행 간격 (초)
 853        """
 854        while self.running:
 855            data = self.env_monitor.get_latest()
 856            if data:
 857                # MQTT 발행
 858                self.mqtt_handler.publish_sensor_data("env_01", data)
 859
 860                # 자동화 규칙 검사
 861                self.automation_engine.check_rules(data)
 862
 863            time.sleep(interval)
 864
 865    def _status_report_loop(self, interval: int):
 866        """
 867        상태 리포트 루프
 868
 869        Args:
 870            interval: 리포트 간격 (초)
 871        """
 872        while self.running:
 873            # 시스템 상태 출력
 874            sensor_data = self.env_monitor.get_latest()
 875            light_status = self.light_controller.get_status()
 876
 877            logging.info("=" * 60)
 878            logging.info("시스템 상태 리포트")
 879            logging.info("-" * 60)
 880
 881            if sensor_data:
 882                logging.info(f"온도: {sensor_data['temperature']}°C, 습도: {sensor_data['humidity']}%")
 883
 884            if light_status:
 885                for light in light_status["lights"]:
 886                    status = "ON" if light["is_on"] else "OFF"
 887                    logging.info(f"{light['name']} ({light['location']}): {status}")
 888
 889            logging.info("=" * 60)
 890
 891            time.sleep(interval)
 892
 893    def start(self):
 894        """게이트웨이 시작"""
 895        if self.running:
 896            logging.warning("이미 실행 중입니다")
 897            return
 898
 899        logging.info("=" * 60)
 900        logging.info("스마트홈 게이트웨이 시작")
 901        logging.info(f"시뮬레이션 모드: {self.simulate}")
 902        logging.info("=" * 60)
 903
 904        self.running = True
 905
 906        # 환경 모니터링 시작
 907        self.env_monitor.start(interval=5)
 908
 909        # MQTT 연결 및 시작
 910        self.mqtt_handler.connect()
 911        self.mqtt_handler.start()
 912
 913        # 센서 데이터 발행 스레드
 914        sensor_thread = threading.Thread(
 915            target=self._sensor_publish_loop,
 916            args=(10,),
 917            daemon=True
 918        )
 919        sensor_thread.start()
 920        self.threads.append(sensor_thread)
 921
 922        # 상태 리포트 스레드
 923        status_thread = threading.Thread(
 924            target=self._status_report_loop,
 925            args=(30,),
 926            daemon=True
 927        )
 928        status_thread.start()
 929        self.threads.append(status_thread)
 930
 931        logging.info("게이트웨이 실행 중...")
 932
 933    def stop(self):
 934        """게이트웨이 중지"""
 935        if not self.running:
 936            return
 937
 938        logging.info("게이트웨이 중지 중...")
 939
 940        self.running = False
 941
 942        # 컴포넌트 정리
 943        self.env_monitor.stop()
 944        self.mqtt_handler.stop()
 945        self.light_controller.all_off()
 946        self.light_controller.cleanup()
 947
 948        # 스레드 종료 대기
 949        for thread in self.threads:
 950            thread.join(timeout=2)
 951
 952        logging.info("게이트웨이 중지 완료")
 953
 954    def run(self):
 955        """메인 실행 루프"""
 956        self.start()
 957
 958        try:
 959            while True:
 960                time.sleep(1)
 961        except KeyboardInterrupt:
 962            logging.info("\nKeyboardInterrupt 수신")
 963        finally:
 964            self.stop()
 965
 966    def get_dashboard_data(self) -> dict:
 967        """
 968        웹 대시보드용 JSON 데이터 제공
 969
 970        Returns:
 971            시스템 전체 상태
 972        """
 973        return {
 974            "timestamp": datetime.now().isoformat(),
 975            "lights": self.light_controller.get_status(),
 976            "sensor": {
 977                "latest": self.env_monitor.get_latest(),
 978                "stats": self.env_monitor.get_stats()
 979            },
 980            "automation": {
 981                "rules": self.automation_engine.get_rules_status()
 982            },
 983            "system": {
 984                "running": self.running,
 985                "simulation_mode": self.simulate
 986            }
 987        }
 988
 989
 990# ============================================================
 991# 메인 실행
 992# ============================================================
 993
 994def main():
 995    """메인 함수"""
 996
 997    # 명령행 인자 파싱
 998    parser = argparse.ArgumentParser(
 999        description="스마트홈 자동화 시스템",
1000        formatter_class=argparse.RawDescriptionHelpFormatter,
1001        epilog="""
1002예제:
1003  python home_automation.py --simulate
1004  python home_automation.py --broker mqtt.example.com --simulate
1005  python home_automation.py --loglevel DEBUG --simulate
1006        """
1007    )
1008
1009    parser.add_argument(
1010        "--simulate",
1011        action="store_true",
1012        help="시뮬레이션 모드 (하드웨어 불필요)"
1013    )
1014
1015    parser.add_argument(
1016        "--broker",
1017        type=str,
1018        default="localhost",
1019        help="MQTT 브로커 주소 (기본값: localhost)"
1020    )
1021
1022    parser.add_argument(
1023        "--port",
1024        type=int,
1025        default=1883,
1026        help="MQTT 포트 (기본값: 1883)"
1027    )
1028
1029    parser.add_argument(
1030        "--loglevel",
1031        type=str,
1032        default="INFO",
1033        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
1034        help="로그 레벨 (기본값: INFO)"
1035    )
1036
1037    args = parser.parse_args()
1038
1039    # 로깅 설정
1040    logging.basicConfig(
1041        level=getattr(logging, args.loglevel),
1042        format='%(asctime)s [%(levelname)s] %(message)s',
1043        datefmt='%Y-%m-%d %H:%M:%S'
1044    )
1045
1046    # 시뮬레이션 모드 설정
1047    simulate = args.simulate or not HARDWARE_AVAILABLE
1048
1049    if simulate:
1050        logging.info("시뮬레이션 모드로 실행합니다 (하드웨어 불필요)")
1051    else:
1052        logging.info("실제 하드웨어 모드로 실행합니다")
1053
1054    # 설정
1055    config = {
1056        "lights": [
1057            {
1058                "id": "living_room",
1059                "name": "거실 조명",
1060                "gpio_pin": 17,
1061                "location": "거실"
1062            },
1063            {
1064                "id": "bedroom",
1065                "name": "침실 조명",
1066                "gpio_pin": 27,
1067                "location": "침실"
1068            },
1069            {
1070                "id": "kitchen",
1071                "name": "주방 조명",
1072                "gpio_pin": 22,
1073                "location": "주방"
1074            },
1075            {
1076                "id": "bathroom",
1077                "name": "욕실 조명",
1078                "gpio_pin": 23,
1079                "location": "욕실"
1080            }
1081        ],
1082        "dht_pin": 4,
1083        "mqtt_broker": args.broker,
1084        "mqtt_port": args.port
1085    }
1086
1087    # 게이트웨이 생성 및 실행
1088    gateway = SmartHomeGateway(config, simulate=simulate)
1089
1090    # 데모: 5초 후 대시보드 데이터 출력
1091    def print_dashboard():
1092        time.sleep(5)
1093        dashboard_data = gateway.get_dashboard_data()
1094        logging.info("\n" + "=" * 60)
1095        logging.info("대시보드 데이터 (JSON API)")
1096        logging.info("=" * 60)
1097        print(json.dumps(dashboard_data, indent=2, ensure_ascii=False))
1098        logging.info("=" * 60)
1099
1100    demo_thread = threading.Thread(target=print_dashboard, daemon=True)
1101    demo_thread.start()
1102
1103    # 메인 루프 실행
1104    gateway.run()
1105
1106
1107if __name__ == "__main__":
1108    main()