Home IoT Hub Series · 25

[25편] AI 집사 Method C - 라즈베리파이 하나로 전부 통합하기

Method A는 ESP32 혼자 AI API를 불렀다. 위험했다. Method B는 고물 PC를 브로커로 썼다. 안전했지만 PC를 24시간 켜둬야 했다.

Method C는 라즈베리파이 하나로 끝낸다. 두뇌 + AI 브로커 + 텔레그램 + 디스플레이 + IR 제어를 전부 이 조그만 보드 하나에 넣는다. 전력 3W. 소음 0. 크기는 신용카드만 하다. 24시간 365일 돌려도 전기세 월 500원.

라즈베리파이 SPI LCD에 20개 홈 IoT 센서 슬롯 상태를 표시한 테스트 화면 — 라즈베리파이 SPI LCD에 20개 홈 IoT 슬롯 상태를 표시한 테스트 화면입니다. 주변에는 ESP8266/ESP32 보드와 DHT11 계열 온습도 센서, 릴레이 모듈을 함께 배치해 각 센서 노드가 중앙 허브로 값을 보내는 구조를 보여주었습니다.

ESP8266 보드와 DHT11 온습도 센서를 라즈베리파이 대시보드 옆에 배치한 근접 사진 — 온습도 노드 예시입니다. 사진 속 파란 센서는 DHT11 모듈이며, DHT22와 같은 DHT 계열이라 전원·데이터·GND를 연결해 온도와 습도 값을 허브 슬롯으로 보내는 흐름을 설명하는 데 사용할 수 있습니다.

이게 최종 목표다: AI와 대화로 집을 제어하는 흐름

25편까지 왔으면 이제 이런 일이 가능해진다:

1. Pi가 5분마다 센서값을 AI에게 보낸다
2. AI가 분석해서 텔레그램으로 메시지를 보낸다
   📱 "CO2 1050ppm입니다. 거실 환기가 필요해요.
       에어컨이 계속 돌고 있는데 문이 열려있어요.
       문 닫기를 실행할까요?"

3. 네가 텔레그램으로 답장한다
   👤 "문 닫아줘. 그리고 에어컨은 26도로 해줘"

4. AI가 네 메시지를 이해하고 Pi에게 명령을 보낸다
   → Front Door 경고 리셋, AC 온도 조정

5. Pi가 IR LED로 실제 에어컨에 명령을 쏜다
   → 에어컨 ON, 문 닫힘 확인

6. Pi가 텔레그램으로 완료 메시지를 보낸다
   📱 "문 닫기 확인했습니다. 에어컨 26℃로 설정했어요.
       현재 거실 온도 24.8℃입니다."

이게 Physical AI의 완성이다. 감지(센서) → 판단(AI) → 보고(텔레그램) → 지시(네 대화) → 실행(Pi가 IR로).

준비물과 비용

준비물	가격	설명
라즈베리파이 (Zero 2 W 이상)	20,000원~80,000원	Zero 2 W면 충분. Pi 4/5는 더 빠름
microSD 카드 16GB+	8,000원	OS 설치용
Pi용 전원 (5V 2.5A)	5,000원	스마트폰 충전기로도 가능
IR LED + 100Ω 저항 + 2N2222	500원	GPIO17에 연결 (17편과 동일 회로)
AI API 키	종량제	23편과 동일
텔레그램 계정	0원	24편에서 이미 만듦
추가 비용	약 33,500원~

센서 노드 19대는 그대로 Wemos D1 Mini를 쓴다. Pi는 그 데이터를 받는 두뇌 역할만 한다. 바꿀 필요 없다.

개념 정리

어려운 말	진짜 뜻
라즈베리파이	신용카드 크기의 작은 컴퓨터. 키보드·마우스·모니터 연결하면 진짜 PC처럼 쓸 수 있다
Raspberry Pi OS	Pi에 까는 운영체제. 윈도우나 맥OS 같은 거. 무료
GPIO	Pi의 다리(핀). 여기에 센서나 LED를 연결한다. 40개가 있다
pigpio	Pi에서 정밀한 적외선 신호를 만들 때 쓰는 라이브러리. 38kHz 반송파를 소프트웨어로 만든다

전체 구조: AI와의 대화 흐름

┌──────────────────────────────────────────────────────────┐
│                    라즈베리파이 (두뇌 + AI + IR)            │
│                                                          │
│  Flask 서버 ← 센서 노드 19대의 /set 요청 (데이터 수집)     │
│      │                                                   │
│      ▼ (5분마다)                                         │
│  AI 분석   → GPT/Claude에 20개 센서값 전송                │
│      │                                                   │
│      ├──→ 📱 텔레그램: AI 분석 결과 + 추천 행동            │
│      │       "CO2 1050ppm. 환기 필요."                    │
│      │                                                   │
│      ←── 📱 네 답장: "환기해줘. 에어컨 꺼."                │
│      │       (handle_telegram → pending_commands)        │
│      │                                                   │
│      ▼                                                   │
│  execute_commands(): 슬롯 업데이트 + IR LED 발사          │
│      │                                                   │
│      ├──→ GPIO17: IR LED로 에어컨 ON/OFF                  │
│      └──→ 📱 텔레그램: "에어컨 OFF, 환기 시작"             │
│                                                          │
└──────────────────────────────────────────────────────────┘

🚨 라즈베리파이 준비 — 딱 한 번만

OS 설치

1. raspberrypi.com/software 에서 Raspberry Pi Imager 다운로드 2. microSD 카드를 PC에 꽂는다 (젠더 필요할 수 있음) 3. Imager 실행 → Raspberry Pi OS (32-bit) → SD 카드 선택 → Write 4. 톱니바퀴(설정) → SSH 활성화 + Wi-Fi SSID/비번 입력 → Write 5. 10분 정도 걸린다

첫 부팅

1. SD 카드를 Pi에 꽂고 전원 연결 2. 1~2분 기다린다 (LED가 안정되면 준비 완료) 3. PC에서 접속: ssh pi@raspberrypi.local (비밀번호: raspberry) 4. sudo raspi-config → 비번 변경

이제 Pi에 모니터 연결할 필요 없다. PC에서 SSH로 모든 작업을 한다.

IR LED 연결 (선택: GPIO17)

Pi의 GPIO17(6번 핀)에 17편과 똑같은 IR 송신 회로를 연결하면 Pi 자체가 만능 리모컨이 된다.

Pi GPIO17 ── 100Ω ── IR LED(+) ── IR LED(-) ── 2N2222 Collector
                                             2N2222 Emitter ── GND
Pi GND ── 10kΩ ── 2N2222 Base

IR LED가 없거나 당장 연결 못 해도 괜찮다. IR_ENABLED = False로 두면 텔레그램 명령 수신 + 슬롯 업데이트까지만 실행된다. IR LED는 나중에 달아도 늦지 않다.

전체 코드 — `server.py` 하나로 끝

아래 코드를 Pi에 server.py로 저장한다. API 키, Bot 토큰, Chat ID 세 개만 바꾸면 된다.

[25편] AI 집사 Method C - 라즈베리파이 하나로 전부 통합하기 - 전체 코드

from flask import Flask, request, jsonify
import requests
import threading
import time
import json
import urllib.request
import urllib.parse

# =============================================
# 1. 설정 (직접 입력!)
# =============================================
OPENAI_API_KEY = "sk-your-api-key-here"
OPENAI_URL     = "https://api.openai.com/v1/chat/completions"
AI_MODEL       = "gpt-3.5-turbo"

TELEGRAM_BOT_TOKEN = "123456:ABC-DEF1234ghijk"
YOUR_CHAT_ID = 123456789

# =============================================
# 2. Flask 앱
# =============================================
app = Flask(__name__)

SLOT_COUNT = 20
slots = ["--"] * SLOT_COUNT

LABELS = [
    "Living Temp","Living Hum","Outside Temp","AC",
    "Boiler","Stove","Gas Leak","Fire",
    "Front Door","Door Lock","Living Light","Fridge Temp",
    "CO2","Dust PM2.5","Power","Water",
    "Hot Water","Rain","Bath Hum","Pet Bowl"
]

latest_data = {}
pending_commands = {}
last_ai_time = 0
AI_INTERVAL = 300

# =============================================
# 3. 센서값 수신 (ESP32의 /set 대체)
# =============================================
@app.route('/set', methods=['GET'])
def receive_sensor():
    """센서 노드들이 보내는 데이터를 받는다"""
    updated = 0
    for i in range(1, SLOT_COUNT + 1):
        key = f"no{i}"
        if key in request.args:
            slots[i - 1] = request.args[key]
            latest_data[key] = request.args[key]
            updated += 1
    if updated > 0:
        print(f"[DATA] Updated {updated} slots")
    return jsonify({"updated": updated})

@app.route('/status')
def status():
    return jsonify(latest_data)

@app.route('/')
def dashboard():
    """HTML 대시보드"""
    html = '<!DOCTYPE html><html lang="ko"><head>'
    html += '<meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1.0">'
    html += '<title>IoT Hub - Pi AI</title>'
    html += '<style>'
    html += '*{margin:0;padding:0;box-sizing:border-box;}'
    html += 'body{font-family:sans-serif;background:#1a1a2e;color:#eee;padding:16px;}'
    html += 'h1{text-align:center;font-size:1.3rem;color:#e94560;}'
    html += '.grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(180px,1fr));gap:10px;}'
    html += '.card{background:#16213e;padding:12px;border-radius:8px;border-left:4px solid #0f3460;}'
    html += '.card.warn{border-left-color:#e94560;background:#1c0e1a;}'
    html += '.card .label{font-size:.8rem;color:#a0a0b0;}'
    html += '.card .value{font-size:1.2rem;font-weight:700;}'
    html += '</style></head><body><h1>IoT Hub - Pi AI</h1><div class="grid">'

    for i in range(SLOT_COUNT):
        val = latest_data.get(f"no{i+1}", "--")
        is_warn = False
        if i in [6,7,8] and val not in ("0", "--"): is_warn = True
        if i == 19 and val != "--":
            try:
                if int(val) < 50: is_warn = True
            except: pass
        cls = ' class="warn"' if is_warn else ""
        html += f'<div{cls}><div class="label">#{i+1} {LABELS[i]}</div>'
        html += f'<div class="value">{val}</div></div>'

    html += '</div></body></html>'
    return html

# =============================================
# 4. 명령 실행 + Pi GPIO IR 제어
# =============================================
@app.route('/command', methods=['GET'])
def send_command():
    """24편 호환용. 실제 명령은 텔레그램 → execute_commands에서 처리"""
    return "{}"

def execute_commands(cmds):
    """텔레그램에서 받은 명령을 실제 액션으로 실행"""
    for key, val in cmds.items():
        print(f"[CMD] Executing: {key}={val}")

        # 슬롯 업데이트
        slot_num = int(key.replace("no", ""))
        if 1 <= slot_num <= SLOT_COUNT:
            slots[slot_num - 1] = val
            latest_data[key] = val

        # AC(4번) IR 송출
        if key == "no4" and IR_ENABLED:
            try:
                import pigpio
                pi = pigpio.pi()
                code = AC_ON_CODE if val == "ON" else AC_OFF_CODE
                if code != 0x0:
                    send_ir_nec(pi, code)
                    print(f"[IR] Sent {'ON' if val == 'ON' else 'OFF'} via GPIO{IR_PIN}")
                else:
                    print("[IR] AC code not set. Run 17편 to learn IR codes first.")
                pi.stop()
            except ImportError:
                print("[IR] pigpio not installed. Run: sudo apt install pigpio && sudo systemctl start pigpiod && pip3 install pigpio")
            except Exception as e:
                print(f"[IR] Error: {e}")
                send_telegram(YOUR_CHAT_ID, f"IR 송출 실패: {e}")

def send_ir_nec(pi, code, bits=32):
    """pigpio로 38kHz NEC 코드 송출"""
    # NEC 프로토콜: 리더(9ms ON + 4.5ms OFF) + 데이터 비트
    # 비트 0: 562μs ON + 562μs OFF
    # 비트 1: 562μs ON + 1687μs OFF
    carrier = 38000  # 38kHz
    duty_cycle = 0.33

    # 리더 코드
    pi.hardware_PWM(IR_PIN, carrier, int(duty_cycle * 1e6))
    time.sleep(0.009)  # 9ms
    pi.hardware_PWM(IR_PIN, 0, 0)
    time.sleep(0.0045) # 4.5ms

    # 데이터 비트
    for i in range(bits):
        pi.hardware_PWM(IR_PIN, carrier, int(duty_cycle * 1e6))
        time.sleep(0.000562) # 562μs ON
        pi.hardware_PWM(IR_PIN, 0, 0)
        if code & (1 << i):
            time.sleep(0.001687) # 비트 1: 1687μs
        else:
            time.sleep(0.000562) # 비트 0: 562μs

    pi.hardware_PWM(IR_PIN, carrier, int(duty_cycle * 1e6))
    time.sleep(0.000562) # Stop bit
    pi.hardware_PWM(IR_PIN, 0, 0)

# IR 설정
IR_PIN = 17        # GPIO17 (Pi 핀 11)
IR_ENABLED = False  # IR LED 연결 후 True로 변경
AC_ON_CODE  = 0x0   # 17편에서 학습한 ON 코드
AC_OFF_CODE = 0x0   # 17편에서 학습한 OFF 코드

# =============================================
# 5. 텔레그램 명령 처리
# =============================================
def handle_telegram(text):
    """네가 보낸 메시지를 분석해서 명령으로 변환"""
    global pending_commands
    t = text.lower().strip()

    if "에어컨" in text or "ac" in t:
        if "켜" in text or "on" in t:
            pending_commands["no4"] = "ON"
            return "에어컨 ON 완료"
        elif "꺼" in text or "off" in t:
            pending_commands["no4"] = "OFF"
            return "에어컨 OFF 완료"

    if "조명" in text or "불" in text or "light" in t:
        if "켜" in text or "on" in t:
            pending_commands["no11"] = "ON"
            return "조명 ON 완료"
        elif "꺼" in text or "off" in t:
            pending_commands["no11"] = "OFF"
            return "조명 OFF 완료"

    if "상태" in text or "집" in text or "status" in t:
        return build_status_report()

    # 직접 인식 못 한 메시지는 AI에게 위임
    # AI가 네 의도를 해석해서 적절한 행동을 결정한다
    try:
        ai_reply = ask_ai_about_command(text)
        return ai_reply
    except:
        return None

def ask_ai_about_command(user_text):
    """네 메시지를 AI가 해석해서 어떤 행동을 할지 결정"""
    data_text = "\n".join(
        f"{LABELS[i]}: {latest_data.get(f'no{i+1}', '--')}"
        for i in range(SLOT_COUNT)
    )
    resp = requests.post(
        OPENAI_URL,
        headers={"Authorization": f"Bearer {OPENAI_API_KEY}",
                 "Content-Type": "application/json"},
        json={"model": AI_MODEL,
              "messages": [
                  {"role": "system",
                   "content": "You are a smart home controller. The user will send "
                   "commands in Korean. Interpret their intent and decide what to do. "
                   "Available actions: turn AC ON/OFF (no4), turn light ON/OFF (no11). "
                   "Respond in Korean. Keep it under 2 sentences. "
                   "If the command can be executed, start with 'EXEC:' followed by "
                   "the JSON actions like {\"no4\":\"ON\"}. "
                   "If it's just a question, answer naturally."},
                  {"role": "user",
                   "content": f"Sensor data:\n{data_text}\n\nUser said: {user_text}"}
              ], "max_tokens": 200},
        timeout=30
    )
    if resp.status_code == 200:
        content = resp.json()["choices"][0]["message"]["content"]
        # AI가 EXEC: 명령을 줬으면 실행
        if "EXEC:" in content:
            try:
                exec_start = content.index("EXEC:") + 5
                exec_end = content.index("}", exec_start) + 1
                exec_json = content[exec_start:exec_end]
                cmds = json.loads(exec_json)
                execute_commands(cmds)
                # 실행했으면 남은 메시지만 반환
                remaining = content[exec_end:].strip()
                return remaining if remaining else "실행 완료했습니다"
            except:
                pass
        return content
    return "죄송합니다, 응답을 생성하지 못했습니다."

def build_status_report():
    if not latest_data:
        return "아직 센서 데이터가 없습니다."
    r = "집 상태\n\n"
    w = []
    for i in range(SLOT_COUNT):
        v = latest_data.get(f"no{i+1}", "--")
        r += f"  {LABELS[i]}: {v}\n"
        if i == 6 and v not in ("0", "--"): w.append("가스 누출!")
        if i == 7 and v not in ("0", "--"): w.append("화재 감지!")
        if i == 8 and v not in ("0", "--"): w.append("현관문 열림!")
        if i == 19 and v != "--":
            try:
                if int(v) < 50: w.append("반려견 밥 보충!")
            except: pass
    if w: r += "\n" + "\n".join(w)
    return r

# =============================================
# 6. AI 분석 → 텔레그램 보고 (5분 간격)
# =============================================
def ai_loop():
    global last_ai_time
    while True:
        time.sleep(30)
        if not latest_data: continue
        if time.time() - last_ai_time < AI_INTERVAL: continue
        last_ai_time = time.time()

        try:
            data_text = "\n".join(
                f"{LABELS[i]}: {latest_data.get(f'no{i+1}', '--')}"
                for i in range(SLOT_COUNT)
            )
            resp = requests.post(
                OPENAI_URL,
                headers={"Authorization": f"Bearer {OPENAI_API_KEY}",
                         "Content-Type": "application/json"},
                json={"model": AI_MODEL,
                      "messages": [
                          {"role": "system",
                           "content": "You are an AI home assistant. Analyze the sensor data "
                           "and send a Korean message to the homeowner. "
                           "1) Report any warnings immediately. "
                           "2) Point out anomalies (e.g., high CO2, door open too long). "
                           "3) Suggest concrete actions. "
                           "4) Ask if they want you to execute the action. "
                           "Keep it conversational, like a helpful butler. "
                           "Under 4 sentences."},
                          {"role": "user", "content": data_text}
                      ], "max_tokens": 250},
                timeout=30
            )
            if resp.status_code == 200:
                content = resp.json()["choices"][0]["message"]["content"]
                print(f"[AI] {content}")
                # AI 분석 결과를 텔레그램으로 전송
                send_telegram(YOUR_CHAT_ID, "[AI 분석]\n" + content)
            else:
                print(f"[AI] API error: {resp.status_code}")
        except Exception as e:
            print(f"[AI] Error: {e}")

# =============================================
# 7. 텔레그램 봇 (메시지 수신 + 답장)
# =============================================
def telegram_loop():
    base = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}"
    last_id = 0
    print("[TELEGRAM] Bot started. Send /start to begin.")

    while True:
        time.sleep(3)
        try:
            u = f"{base}/getUpdates?offset={last_id+1}&timeout=10"
            resp = requests.get(u, timeout=15).json()
            for upd in resp.get("result", []):
                last_id = upd["update_id"]
                if "message" not in upd: continue
                m = upd["message"]
                cid = m["chat"]["id"]
                txt = m.get("text", "")
                print(f"[TELEGRAM] {cid}: {txt}")

                if txt == "/start":
                    send_telegram(cid,
                        "AI 집사입니다.\n\n"
                        "'상태' — 집 상태 확인\n"
                        "'에어컨 켜줘/꺼줘' — 에어컨 제어\n"
                        "'조명 켜줘/꺼줘' — 조명 제어\n"
                        "그 외 질문도 자유롭게 물어보세요.\n"
                        f"Chat ID: {cid}")
                    continue

                # 먼저 간단한 명령어 체크
                reply = handle_telegram(txt)
                if reply:
                    send_telegram(cid, reply)
                    # 명령이 있었으면 execute_commands 호출
                    if pending_commands:
                        execute_commands(pending_commands)
                        pending_commands = {}
                else:
                    send_telegram(cid, "'상태', '에어컨 켜줘/꺼줘' 또는 자유롭게 물어보세요.")
        except Exception as e:
            print(f"[TELEGRAM] Error: {e}")

def send_telegram(cid, text):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    urllib.request.urlopen(url, urllib.parse.urlencode(
        {"chat_id": cid, "text": text}).encode())

# =============================================
# 8. 실행
# =============================================
if __name__ == "__main__":
    # __name__ == "__main__" — 이 파일을 직접 실행할 때만 아래 코드가 돈다.
    # 다른 파일에서 import 했을 때는 실행되지 않는다.
    # Python의 관습적인 주문 같은 거니까 신경 쓰지 마라.

    # 백그라운드에서 계속 도는 두 개의 작은 프로그램을 시작한다.
    threading.Thread(target=ai_loop, daemon=True).start()
    threading.Thread(target=telegram_loop, daemon=True).start()

    print("=" * 50)
    print("  IoT Hub - Pi AI Edition")
    print("  센서 노드 → http://Pi_IP:5000/set?no1=값")
    print("  텔레그램 봇 → @MyHomeAIBot")
    print("  AI 분석 → 5분 간격 자동 보고")
    print("=" * 50)

    app.run(host="0.0.0.0", port=5000, debug=False)

from flask import Flask, request, jsonify
import requests
import threading
import time
import json
import urllib.request
import urllib.parse

# =============================================
# 1. 설정 (직접 입력!)
# =============================================
OPENAI_API_KEY = "sk-your-api-key-here"
OPENAI_URL     = "https://api.openai.com/v1/chat/completions"
AI_MODEL       = "gpt-3.5-turbo"

TELEGRAM_BOT_TOKEN = "123456:ABC-DEF1234ghijk"
YOUR_CHAT_ID = 123456789

# =============================================
# 2. Flask 앱
# =============================================
app = Flask(__name__)

SLOT_COUNT = 20
slots = ["--"] * SLOT_COUNT

LABELS = [
    "Living Temp","Living Hum","Outside Temp","AC",
    "Boiler","Stove","Gas Leak","Fire",
    "Front Door","Door Lock","Living Light","Fridge Temp",
    "CO2","Dust PM2.5","Power","Water",
    "Hot Water","Rain","Bath Hum","Pet Bowl"
]

latest_data = {}
pending_commands = {}
last_ai_time = 0
AI_INTERVAL = 300

# =============================================
# 3. 센서값 수신 (ESP32의 /set 대체)
# =============================================
@app.route('/set', methods=['GET'])
def receive_sensor():
    """센서 노드들이 보내는 데이터를 받는다"""
    updated = 0
    for i in range(1, SLOT_COUNT + 1):
        key = f"no{i}"
        if key in request.args:
            slots[i - 1] = request.args[key]
            latest_data[key] = request.args[key]
            updated += 1
    if updated > 0:
        print(f"[DATA] Updated {updated} slots")
    return jsonify({"updated": updated})

@app.route('/status')
def status():
    return jsonify(latest_data)

@app.route('/')
def dashboard():
    """HTML 대시보드"""
    html = '<!DOCTYPE html><html lang="ko"><head>'
    html += '<meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1.0">'
    html += '<title>IoT Hub - Pi AI</title>'
    html += '<style>'
    html += '*{margin:0;padding:0;box-sizing:border-box;}'
    html += 'body{font-family:sans-serif;background:#1a1a2e;color:#eee;padding:16px;}'
    html += 'h1{text-align:center;font-size:1.3rem;color:#e94560;}'
    html += '.grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(180px,1fr));gap:10px;}'
    html += '.card{background:#16213e;padding:12px;border-radius:8px;border-left:4px solid #0f3460;}'
    html += '.card.warn{border-left-color:#e94560;background:#1c0e1a;}'
    html += '.card .label{font-size:.8rem;color:#a0a0b0;}'
    html += '.card .value{font-size:1.2rem;font-weight:700;}'
    html += '</style></head><body><h1>IoT Hub - Pi AI</h1><div class="grid">'

for i in range(SLOT_COUNT):
        val = latest_data.get(f"no{i+1}", "--")
        is_warn = False
        if i in [6,7,8] and val not in ("0", "--"): is_warn = True
        if i == 19 and val != "--":
            try:
                if int(val) < 50: is_warn = True
            except: pass
        cls = ' class="warn"' if is_warn else ""
        html += f'<div{cls}><div class="label">#{i+1} {LABELS[i]}</div>'
        html += f'<div class="value">{val}</div></div>'

html += '</div></body></html>'
    return html

# =============================================
# 4. 명령 실행 + Pi GPIO IR 제어
# =============================================
@app.route('/command', methods=['GET'])
def send_command():
    """24편 호환용. 실제 명령은 텔레그램 → execute_commands에서 처리"""
    return "{}"

def execute_commands(cmds):
    """텔레그램에서 받은 명령을 실제 액션으로 실행"""
    for key, val in cmds.items():
        print(f"[CMD] Executing: {key}={val}")

# 슬롯 업데이트
        slot_num = int(key.replace("no", ""))
        if 1 <= slot_num <= SLOT_COUNT:
            slots[slot_num - 1] = val
            latest_data[key] = val

# AC(4번) IR 송출
        if key == "no4" and IR_ENABLED:
            try:
                import pigpio
                pi = pigpio.pi()
                code = AC_ON_CODE if val == "ON" else AC_OFF_CODE
                if code != 0x0:
                    send_ir_nec(pi, code)
                    print(f"[IR] Sent {'ON' if val == 'ON' else 'OFF'} via GPIO{IR_PIN}")
                else:
                    print("[IR] AC code not set. Run 17편 to learn IR codes first.")
                pi.stop()
            except ImportError:
                print("[IR] pigpio not installed. Run: sudo apt install pigpio && sudo systemctl start pigpiod && pip3 install pigpio")
            except Exception as e:
                print(f"[IR] Error: {e}")
                send_telegram(YOUR_CHAT_ID, f"IR 송출 실패: {e}")

def send_ir_nec(pi, code, bits=32):
    """pigpio로 38kHz NEC 코드 송출"""
    # NEC 프로토콜: 리더(9ms ON + 4.5ms OFF) + 데이터 비트
    # 비트 0: 562μs ON + 562μs OFF
    # 비트 1: 562μs ON + 1687μs OFF
    carrier = 38000  # 38kHz
    duty_cycle = 0.33

# 리더 코드
    pi.hardware_PWM(IR_PIN, carrier, int(duty_cycle * 1e6))
    time.sleep(0.009)  # 9ms
    pi.hardware_PWM(IR_PIN, 0, 0)
    time.sleep(0.0045) # 4.5ms

# 데이터 비트
    for i in range(bits):
        pi.hardware_PWM(IR_PIN, carrier, int(duty_cycle * 1e6))
        time.sleep(0.000562) # 562μs ON
        pi.hardware_PWM(IR_PIN, 0, 0)
        if code & (1 << i):
            time.sleep(0.001687) # 비트 1: 1687μs
        else:
            time.sleep(0.000562) # 비트 0: 562μs

pi.hardware_PWM(IR_PIN, carrier, int(duty_cycle * 1e6))
    time.sleep(0.000562) # Stop bit
    pi.hardware_PWM(IR_PIN, 0, 0)

# IR 설정
IR_PIN = 17        # GPIO17 (Pi 핀 11)
IR_ENABLED = False  # IR LED 연결 후 True로 변경
AC_ON_CODE  = 0x0   # 17편에서 학습한 ON 코드
AC_OFF_CODE = 0x0   # 17편에서 학습한 OFF 코드

# =============================================
# 5. 텔레그램 명령 처리
# =============================================
def handle_telegram(text):
    """네가 보낸 메시지를 분석해서 명령으로 변환"""
    global pending_commands
    t = text.lower().strip()

if "에어컨" in text or "ac" in t:
        if "켜" in text or "on" in t:
            pending_commands["no4"] = "ON"
            return "에어컨 ON 완료"
        elif "꺼" in text or "off" in t:
            pending_commands["no4"] = "OFF"
            return "에어컨 OFF 완료"

if "조명" in text or "불" in text or "light" in t:
        if "켜" in text or "on" in t:
            pending_commands["no11"] = "ON"
            return "조명 ON 완료"
        elif "꺼" in text or "off" in t:
            pending_commands["no11"] = "OFF"
            return "조명 OFF 완료"

if "상태" in text or "집" in text or "status" in t:
        return build_status_report()

# 직접 인식 못 한 메시지는 AI에게 위임
    # AI가 네 의도를 해석해서 적절한 행동을 결정한다
    try:
        ai_reply = ask_ai_about_command(text)
        return ai_reply
    except:
        return None

def ask_ai_about_command(user_text):
    """네 메시지를 AI가 해석해서 어떤 행동을 할지 결정"""
    data_text = "\n".join(
        f"{LABELS[i]}: {latest_data.get(f'no{i+1}', '--')}"
        for i in range(SLOT_COUNT)
    )
    resp = requests.post(
        OPENAI_URL,
        headers={"Authorization": f"Bearer {OPENAI_API_KEY}",
                 "Content-Type": "application/json"},
        json={"model": AI_MODEL,
              "messages": [
                  {"role": "system",
                   "content": "You are a smart home controller. The user will send "
                   "commands in Korean. Interpret their intent and decide what to do. "
                   "Available actions: turn AC ON/OFF (no4), turn light ON/OFF (no11). "
                   "Respond in Korean. Keep it under 2 sentences. "
                   "If the command can be executed, start with 'EXEC:' followed by "
                   "the JSON actions like {\"no4\":\"ON\"}. "
                   "If it's just a question, answer naturally."},
                  {"role": "user",
                   "content": f"Sensor data:\n{data_text}\n\nUser said: {user_text}"}
              ], "max_tokens": 200},
        timeout=30
    )
    if resp.status_code == 200:
        content = resp.json()["choices"][0]["message"]["content"]
        # AI가 EXEC: 명령을 줬으면 실행
        if "EXEC:" in content:
            try:
                exec_start = content.index("EXEC:") + 5
                exec_end = content.index("}", exec_start) + 1
                exec_json = content[exec_start:exec_end]
                cmds = json.loads(exec_json)
                execute_commands(cmds)
                # 실행했으면 남은 메시지만 반환
                remaining = content[exec_end:].strip()
                return remaining if remaining else "실행 완료했습니다"
            except:
                pass
        return content
    return "죄송합니다, 응답을 생성하지 못했습니다."

def build_status_report():
    if not latest_data:
        return "아직 센서 데이터가 없습니다."
    r = "집 상태\n\n"
    w = []
    for i in range(SLOT_COUNT):
        v = latest_data.get(f"no{i+1}", "--")
        r += f"  {LABELS[i]}: {v}\n"
        if i == 6 and v not in ("0", "--"): w.append("가스 누출!")
        if i == 7 and v not in ("0", "--"): w.append("화재 감지!")
        if i == 8 and v not in ("0", "--"): w.append("현관문 열림!")
        if i == 19 and v != "--":
            try:
                if int(v) < 50: w.append("반려견 밥 보충!")
            except: pass
    if w: r += "\n" + "\n".join(w)
    return r

# =============================================
# 6. AI 분석 → 텔레그램 보고 (5분 간격)
# =============================================
def ai_loop():
    global last_ai_time
    while True:
        time.sleep(30)
        if not latest_data: continue
        if time.time() - last_ai_time < AI_INTERVAL: continue
        last_ai_time = time.time()

try:
            data_text = "\n".join(
                f"{LABELS[i]}: {latest_data.get(f'no{i+1}', '--')}"
                for i in range(SLOT_COUNT)
            )
            resp = requests.post(
                OPENAI_URL,
                headers={"Authorization": f"Bearer {OPENAI_API_KEY}",
                         "Content-Type": "application/json"},
                json={"model": AI_MODEL,
                      "messages": [
                          {"role": "system",
                           "content": "You are an AI home assistant. Analyze the sensor data "
                           "and send a Korean message to the homeowner. "
                           "1) Report any warnings immediately. "
                           "2) Point out anomalies (e.g., high CO2, door open too long). "
                           "3) Suggest concrete actions. "
                           "4) Ask if they want you to execute the action. "
                           "Keep it conversational, like a helpful butler. "
                           "Under 4 sentences."},
                          {"role": "user", "content": data_text}
                      ], "max_tokens": 250},
                timeout=30
            )
            if resp.status_code == 200:
                content = resp.json()["choices"][0]["message"]["content"]
                print(f"[AI] {content}")
                # AI 분석 결과를 텔레그램으로 전송
                send_telegram(YOUR_CHAT_ID, "[AI 분석]\n" + content)
            else:
                print(f"[AI] API error: {resp.status_code}")
        except Exception as e:
            print(f"[AI] Error: {e}")

# =============================================
# 7. 텔레그램 봇 (메시지 수신 + 답장)
# =============================================
def telegram_loop():
    base = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}"
    last_id = 0
    print("[TELEGRAM] Bot started. Send /start to begin.")

while True:
        time.sleep(3)
        try:
            u = f"{base}/getUpdates?offset={last_id+1}&timeout=10"
            resp = requests.get(u, timeout=15).json()
            for upd in resp.get("result", []):
                last_id = upd["update_id"]
                if "message" not in upd: continue
                m = upd["message"]
                cid = m["chat"]["id"]
                txt = m.get("text", "")
                print(f"[TELEGRAM] {cid}: {txt}")

if txt == "/start":
                    send_telegram(cid,
                        "AI 집사입니다.\n\n"
                        "'상태' — 집 상태 확인\n"
                        "'에어컨 켜줘/꺼줘' — 에어컨 제어\n"
                        "'조명 켜줘/꺼줘' — 조명 제어\n"
                        "그 외 질문도 자유롭게 물어보세요.\n"
                        f"Chat ID: {cid}")
                    continue

# 먼저 간단한 명령어 체크
                reply = handle_telegram(txt)
                if reply:
                    send_telegram(cid, reply)
                    # 명령이 있었으면 execute_commands 호출
                    if pending_commands:
                        execute_commands(pending_commands)
                        pending_commands = {}
                else:
                    send_telegram(cid, "'상태', '에어컨 켜줘/꺼줘' 또는 자유롭게 물어보세요.")
        except Exception as e:
            print(f"[TELEGRAM] Error: {e}")

def send_telegram(cid, text):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    urllib.request.urlopen(url, urllib.parse.urlencode(
        {"chat_id": cid, "text": text}).encode())

# =============================================
# 8. 실행
# =============================================
if __name__ == "__main__":
    # __name__ == "__main__" — 이 파일을 직접 실행할 때만 아래 코드가 돈다.
    # 다른 파일에서 import 했을 때는 실행되지 않는다.
    # Python의 관습적인 주문 같은 거니까 신경 쓰지 마라.

# 백그라운드에서 계속 도는 두 개의 작은 프로그램을 시작한다.
    threading.Thread(target=ai_loop, daemon=True).start()
    threading.Thread(target=telegram_loop, daemon=True).start()

print("=" * 50)
    print("  IoT Hub - Pi AI Edition")
    print("  센서 노드 → http://Pi_IP:5000/set?no1=값")
    print("  텔레그램 봇 → @MyHomeAIBot")
    print("  AI 분석 → 5분 간격 자동 보고")
    print("=" * 50)

app.run(host="0.0.0.0", port=5000, debug=False)

실행 방법

Pi에서 서버 실행

1. Pi에 SSH로 접속: ssh pi@raspberrypi.local 2. nano server.py → 위 코드 붙여넣기 → Ctrl+X → Y → Enter 3. 라이브러리 설치:

[25편] AI 집사 Method C - 라즈베리파이 하나로 전부 통합하기 - 전체 코드

pip3 install flask requests
# IR 제어를 쓰려면:
sudo apt install pigpio -y && sudo systemctl start pigpiod && pip3 install pigpio

4. 실행: python3 server.py 5. 아래 출력이 나오면 성공:

==================================================
  IoT Hub - Pi AI Edition
  센서 노드 → http://Pi_IP:5000/set?no1=값
  텔레그램 봇 → @MyHomeAIBot
  AI 분석 → 5분 간격 자동 보고
==================================================

자동 시작 (Pi 부팅 시)

[25편] AI 집사 Method C - 라즈베리파이 하나로 전부 통합하기 - 전체 코드

crontab -e

맨 아래 줄에:

@reboot cd /home/pi && python3 server.py &

확인할 것 — AI와의 대화가 실제로 어떻게 흐르는지

라즈베리파이 홈 IoT 허브가 텔레그램으로 센서 상태와 주의 알림을 보낸 테스트 화면 — 20개 센서 슬롯을 점검한 뒤 라즈베리파이 허브가 텔레그램으로 보낸 테스트 알림입니다. 센서값이 기준을 넘으면 주의 알림을 보내고, 네트워크·API·LCD 상태를 함께 확인하도록 구성했습니다.

시나리오 1: AI가 먼저 말을 건다

1. 센서 노드들을 켜고 5분 기다린다. 2. 텔레그램으로 이런 메시지가 온다:

[AI 분석]
현재 거실 온도 24.5℃, 습도 58%로 쾌적합니다.
다만 CO2가 880ppm으로 약간 높습니다. 30분간 창문을 열어두는 걸 추천합니다.
모든 경고 센서는 정상입니다.

시나리오 2: 네가 명령을 내린다

1. 텔레그램 봇에게 말을 건다:

👤: "에어컨 켜줘"
🤖: "에어컨 ON 완료"

2. 실제로 IR LED가 발사되고 에어컨이 켜진다. 3. Pi 콘솔에 [CMD] Executing: no4=ON, [IR] Sent ON via GPIO17 출력.

시나리오 3: AI가 네 의도를 해석해서 실행한다

1. 더 복잡한 요청을 해본다:

👤: "집이 너무 더운데 에어컨 좀 켜주고 CO2도 확인해줘"

2. handle_telegram이 "에어컨"을 인식 못 해도, ask_ai_about_command가 AI에게 물어본다. 3. AI가 현재 센서값을 보고 판단:

🤖: "거실 온도 27.3℃, CO2 720ppm입니다.
      에어컨을 켜고 창문은 닫는 게 좋겠네요. 실행할게요."

4. AI 응답에 포함된 EXEC:{"no4":"ON"}을 Pi가 파싱해서 실행. 5. 실제로 에어컨이 켜진다.

안 될 때 체크포인트

증상	원인	해결
Pi에 SSH 접속 안 됨	Wi-Fi 설정 오류	Imager에서 Wi-Fi SSID/비번 다시 설정 후 SD 재구성
`pip3 install` 실패	Python 미설치	`sudo apt update && sudo apt install python3-pip -y`
센서 노드 HTTP -1	Pi IP 바뀜	`hostname -I`로 Pi IP 확인. 공유기에서 고정 할당
AI 응답 없음	API 키 오류	`OPENAI_API_KEY` 확인
텔레그램 무응답	BotFather 토큰/ID 불일치	`@userinfobot`으로 Chat ID 재확인
IR 안 쏴짐	pigpio 미실행	`sudo systemctl start pigpiod`
"AC code not set"	리모컨 학습 안 됨	`IR_ENABLED = True` 전에 17편으로 IR 코드 먼저 학습

지금까지 뭘 한 거냐면

Raspberry Pi 하나로 두뇌 + AI + 텔레그램 + IR 제어를 전부 통합했다.
5분마다 AI가 집 상태를 분석해서 텔레그램으로 보고한다.
네가 텔레그램으로 "에어컨 켜줘"라고 말하면 Pi가 GPIO로 IR LED를 쏴서 실제 에어컨을 제어한다.
심지어 "집이 더운데" 같은 자연어도 AI가 해석해서 적절한 행동을 한다.
Physical AI 완성: 감지 → 판단 → 보고 → 대화 → 실행.

시리즈 완결

축하한다. 1편의 ESP32 두뇌 하나로 시작해서, 25편에서는 AI와 대화하는 집을 만들었다. 이제 네 집은 온도가 올라가면 네 폰에 "에어컨 켤까요?"라고 묻고, 네가 "응" 하면 실제로 켜는 집이다. Physical AI의 시작이다.

우리집 IoT HUB 시리즈[24편] AI 집사 Method B [25편] AI 집사 Method C 전체 시리즈 목록 보기

[25편] AI 집사 Method C - 라즈베리파이 하나로 전부 통합하기

이게 최종 목표다: AI와 대화로 집을 제어하는 흐름

준비물과 비용

개념 정리

전체 구조: AI와의 대화 흐름

🚨 라즈베리파이 준비 — 딱 한 번만

OS 설치

첫 부팅

IR LED 연결 (선택: GPIO17)

전체 코드 — server.py 하나로 끝

실행 방법

Pi에서 서버 실행

자동 시작 (Pi 부팅 시)

확인할 것 — AI와의 대화가 실제로 어떻게 흐르는지

시나리오 1: AI가 먼저 말을 건다

시나리오 2: 네가 명령을 내린다

시나리오 3: AI가 네 의도를 해석해서 실행한다

안 될 때 체크포인트

지금까지 뭘 한 거냐면

시리즈 완결

전체 코드 — `server.py` 하나로 끝