반응형
오늘은 AI Agent가 현실 생활에 쓸만해졌다면, 내 생활에 바로 적용할 수 있는 방법은 없을까 생각해 보았다.
그래서 Claude AI에게 '오픈소스 AI와 프레임워크를 사용하고 센서와 액추에이터를 결합해 실제 정보를 바탕으로 행동하는 AI Agent를 만들꺼야. 이번에는 매일 아침에 날씨를 물어보면 확인하고 날씨 상황에 따라 준비해야 할 사항을 스피커로 알려주는 방법을 알려줘'라고 물어보았다.
아래 코드는 검증이 된 것이 아니라 틀린 부분도 있을 것이고, 지역별 날씨 정보를 가져오는 것이 아니라 추가해야 할 코드도 있을 것이라 생각한다. 실제 제작은 Cursor AI로 진행해서 라즈베리 파이에 넣어 스크린과 스피커로 출력하도록 만들어 보면 재미있을 것 같다. 이건 하나 하나 도전해 보도록 하겠다.
날씨 정보 기반 음성 비서 에이전트 개발하기
실제 날씨 정보를 센서처럼 활용하고 음성 출력을 액추에이터로 사용하는 AI 에이전트를 개발해 보겠습니다. 이 에이전트는 매일 아침 날씨를 확인하고 적절한 준비 사항을 스피커로 알려주는 기능을 수행합니다.
1. 필요한 라이브러리 설치
먼저 필요한 라이브러리를 설치합니다:
pip install requests schedule gtts playsound transformers torch pyttsx32. 날씨 API와 음성 출력을 결합한 에이전트 구현
import requests
import json
import schedule
import time
import pyttsx3
import datetime
from transformers import pipeline
class WeatherAssistant:
def __init__(self, api_key, city="Seoul", language="ko"):
# OpenWeatherMap API 설정
self.api_key = api_key
self.city = city
self.language = language
self.base_url = "http://api.openweathermap.org/data/2.5/weather"
# 음성 엔진 초기화
self.engine = pyttsx3.init()
self.engine.setProperty('rate', 150) # 말하기 속도 설정
# 추천 생성을 위한 Hugging Face 모델 초기화
self.text_generator = pipeline('text-generation', model='gpt2')
# 날씨 상태에 따른 추천 사항 정의
self.recommendations = {
"Clear": "맑은 날씨입니다. 자외선 차단제를 바르고 선글라스를 챙기세요.",
"Clouds": "구름이 많습니다. 평소와 같이 준비하시면 됩니다.",
"Rain": "비가 오고 있습니다. 우산을 꼭 챙기시고 방수 신발을 신으세요.",
"Drizzle": "이슬비가 내리고 있습니다. 우산이나 비옷을 준비하세요.",
"Thunderstorm": "천둥번개가 치고 있습니다. 우산과 비옷을 준비하고 야외 활동은 자제하세요.",
"Snow": "눈이 오고 있습니다. 미끄럼 방지 신발과 따뜻한 옷을 입으세요.",
"Mist": "안개가 있습니다. 운전 시 주의하세요.",
"Fog": "짙은 안개가 있습니다. 운전 시 각별한 주의가 필요합니다.",
"Haze": "연무가 있습니다. 마스크를 착용하세요."
}
print("날씨 비서가 초기화되었습니다.")
def sense_weather(self):
"""날씨 API로부터 현재 날씨 정보를 가져옴 (센서 역할)"""
params = {
"q": self.city,
"appid": self.api_key,
"units": "metric", # 섭씨 온도
"lang": self.language
}
try:
response = requests.get(self.base_url, params=params)
weather_data = response.json()
if response.status_code == 200:
self.weather_condition = weather_data["weather"][0]["main"]
self.weather_description = weather_data["weather"][0]["description"]
self.temperature = weather_data["main"]["temp"]
self.feels_like = weather_data["main"]["feels_like"]
self.humidity = weather_data["main"]["humidity"]
self.wind_speed = weather_data["wind"]["speed"]
print(f"날씨 정보 수집 완료: {self.weather_condition}, {self.temperature}°C")
return True
else:
print(f"날씨 정보 수집 실패: {weather_data.get('message', '알 수 없는 오류')}")
return False
except Exception as e:
print(f"날씨 정보 요청 중 오류 발생: {e}")
return False
def decide_recommendations(self):
"""날씨 상태에 따라 추천 사항 결정"""
# 기본 추천 사항 가져오기
self.recommendation = self.recommendations.get(self.weather_condition, "오늘도 좋은 하루 되세요.")
# 온도에 따른 추가 추천
if self.temperature < 5:
self.recommendation += " 오늘은 매우 춥습니다. 두꺼운 코트와 장갑, 목도리를 착용하세요."
elif 5 <= self.temperature < 10:
self.recommendation += " 오늘은 쌀쌀합니다. 가벼운 코트나 자켓을 챙기세요."
elif 10 <= self.temperature < 20:
self.recommendation += " 오늘은 선선합니다. 얇은 겉옷을 챙기세요."
elif 20 <= self.temperature < 25:
self.recommendation += " 오늘은 따뜻합니다. 가벼운 복장이 좋겠습니다."
elif 25 <= self.temperature < 30:
self.recommendation += " 오늘은 덥습니다. 시원한 옷을 입고 수분을 충분히 섭취하세요."
else:
self.recommendation += " 오늘은 매우 덥습니다. 야외 활동은 자제하고 수분을 충분히 섭취하세요."
# 습도가 높은 경우 추가 안내
if self.humidity > 80:
self.recommendation += " 습도가 높으니 불쾌지수가 높을 수 있습니다."
# 바람이 강한 경우 추가 안내
if self.wind_speed > 10:
self.recommendation += " 바람이 강하게 불어 체감온도가 낮을 수 있으니 주의하세요."
print("추천 사항 결정 완료")
def act_voice_output(self):
"""결정된 추천 사항을 음성으로 출력 (액추에이터 역할)"""
current_time = datetime.datetime.now().strftime("%H시 %M분")
# 출력할 전체 텍스트 구성
output_text = f"좋은 아침입니다. 현재 시각은 {current_time}입니다. "
output_text += f"{self.city}의 현재 날씨는 {self.weather_description}이며, 기온은 {self.temperature:.1f}도입니다. "
output_text += f"체감 온도는 {self.feels_like:.1f}도, 습도는 {self.humidity}%입니다. "
output_text += self.recommendation
print("음성 출력: " + output_text)
# 텍스트를 음성으로 변환하여 출력
self.engine.say(output_text)
self.engine.runAndWait()
def morning_briefing(self):
"""아침 브리핑 전체 과정 실행"""
print("아침 브리핑을 시작합니다...")
if self.sense_weather(): # 날씨 정보 수집 (센서)
self.decide_recommendations() # 추천 사항 결정 (의사결정)
self.act_voice_output() # 음성 출력 (액추에이터)
else:
# 날씨 정보를 가져오지 못한 경우 기본 메시지 출력
self.engine.say("죄송합니다. 날씨 정보를 가져오는 데 문제가 발생했습니다. 인터넷 연결을 확인해주세요.")
self.engine.runAndWait()
def schedule_morning_briefing(self, time="07:00"):
"""지정된 시간에 아침 브리핑이 실행되도록 스케줄링"""
schedule.every().day.at(time).do(self.morning_briefing)
print(f"아침 브리핑이 매일 {time}에 실행되도록 설정되었습니다.")
def run(self):
"""스케줄러 실행"""
print("날씨 비서가 실행 중입니다. Ctrl+C로 종료할 수 있습니다.")
try:
while True:
schedule.run_pending()
time.sleep(1)
except KeyboardInterrupt:
print("날씨 비서를 종료합니다.")
def test_run(self):
"""테스트를 위해 즉시 브리핑 실행"""
self.morning_briefing()
# 에이전트 사용 예시
if __name__ == "__main__":
# OpenWeatherMap API 키 (실제 사용 시 본인의 API 키로 대체)
API_KEY = "your_openweathermap_api_key"
# 날씨 비서 초기화 (원하는 도시 설정 가능)
assistant = WeatherAssistant(api_key=API_KEY, city="Seoul")
# 방법 1: 지정된.시간에 실행되도록 스케줄링
assistant.schedule_morning_briefing(time="07:00")
assistant.run()
# 방법 2: 테스트를 위해 즉시 실행
# assistant.test_run()3. 오프라인 음성인식 기능 추가
날씨 정보를 요청하면 응답하는 대화형 기능을 추가해 보겠습니다.
import speech_recognition as sr
from vosk import Model, KaldiRecognizer
import pyaudio
import json
import threading
class WeatherAssistantWithVoiceInput(WeatherAssistant):
def __init__(self, api_key, city="Seoul", language="ko"):
super().__init__(api_key, city, language)
# 음성 인식 설정 (Vosk 오프라인 음성인식 모델)
self.model = Model(model_path="vosk-model-ko-0.22") # 한국어 모델 다운로드 필요
self.recognizer = KaldiRecognizer(self.model, 16000)
# 음성 인식 스레드 플래그
self.listening = False
print("음성 인식 모듈이 초기화되었습니다.")
def listen_for_commands(self):
"""음성 명령 인식을 위한 리스너 스레드"""
self.listening = True
# 오디오 스트림 설정
p = pyaudio.PyAudio()
stream = p.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True, frames_per_buffer=8000)
print("음성 인식 시작. '날씨 알려줘'와 같은 명령을 말해보세요...")
try:
while self.listening:
data = stream.read(4000)
if len(data) == 0:
break
if self.recognizer.AcceptWaveform(data):
result = json.loads(self.recognizer.Result())
text = result.get("text", "").lower()
print(f"인식된 텍스트: {text}")
# 날씨 관련 명령 처리
if "날씨" in text and ("알려" in text or "어때" in text or "어떠" in text):
print("날씨 정보 요청 감지")
self.morning_briefing()
# 종료 명령 처리
elif "종료" in text or "그만" in text:
print("종료 명령 감지")
self.engine.say("날씨 비서를 종료합니다.")
self.engine.runAndWait()
self.listening = False
break
finally:
stream.stop_stream()
stream.close()
p.terminate()
print("음성 인식 종료")
def run_with_voice_commands(self):
"""스케줄러와 음성 명령을 동시에 실행"""
# 스케줄러 스레드 시작
scheduler_thread = threading.Thread(target=self.run)
scheduler_thread.daemon = True
scheduler_thread.start()
# 음성 인식 시작
self.listen_for_commands()
# 확장된 에이전트 사용 예시
if __name__ == "__main__":
API_KEY = "your_openweathermap_api_key"
# 음성 명령이 가능한 날씨 비서 초기화
voice_assistant = WeatherAssistantWithVoiceInput(api_key=API_KEY, city="Seoul")
# 아침 브리핑 스케줄링
voice_assistant.schedule_morning_briefing(time="07:00")
# 음성 명령과 함께 실행
voice_assistant.run_with_voice_commands()반응형
4. 여러 날의 날씨 예보와 준비물 추천 기능 확장
7일 예보를 가져와서 오늘뿐만 아니라 앞으로의 날씨 변화에 대한 정보도 제공하는 기능을 추가합니다.
class AdvancedWeatherAssistant(WeatherAssistantWithVoiceInput):
def __init__(self, api_key, city="Seoul", language="ko"):
super().__init__(api_key, city, language)
self.forecast_url = "http://api.openweathermap.org/data/2.5/forecast"
# 일정 데이터베이스 (실제로는 캘린더 API 등과 연동 가능)
self.schedule_db = {
"2025-04-09": "오전 9시 회의",
"2025-04-12": "야외 활동 계획"
}
# 의류 추천 데이터베이스
self.clothing_recommendations = {
"very_cold": ["두꺼운 패딩", "목도리", "장갑", "기모 바지"],
"cold": ["코트", "가벼운 장갑", "목도리"],
"cool": ["자켓", "가디건", "얇은 니트"],
"mild": ["얇은 자켓", "긴팔 셔츠", "가벼운 바지"],
"warm": ["반팔 셔츠", "얇은 긴팔", "면바지"],
"hot": ["반팔", "반바지", "민소매", "린넨 소재 옷"]
}
print("고급 날씨 비서가 초기화되었습니다.")
def get_temperature_category(self, temp):
"""온도에 따른 카테고리 반환"""
if temp < 5:
return "very_cold"
elif 5 <= temp < 12:
return "cold"
elif 12 <= temp < 18:
return "cool"
elif 18 <= temp < 23:
return "mild"
elif 23 <= temp < 28:
return "warm"
else:
return "hot"
def sense_forecast(self):
"""5일 날씨 예보 정보를 가져옴"""
params = {
"q": self.city,
"appid": self.api_key,
"units": "metric",
"lang": self.language
}
try:
response = requests.get(self.forecast_url, params=params)
forecast_data = response.json()
if response.status_code == 200:
# 5일 예보 데이터 처리
self.forecast = []
for item in forecast_data["list"]:
# 3시간 간격 데이터 중 정오(12:00) 데이터만 선택
if "12:00" in item["dt_txt"]:
date = item["dt_txt"].split()[0]
temp = item["main"]["temp"]
weather = item["weather"][0]["main"]
description = item["weather"][0]["description"]
self.forecast.append({
"date": date,
"temperature": temp,
"weather": weather,
"description": description
})
print(f"5일 예보 정보 수집 완료: {len(self.forecast)}일 데이터")
return True
else:
print(f"예보 정보 수집 실패: {forecast_data.get('message', '알 수 없는 오류')}")
return False
except Exception as e:
print(f"예보 정보 요청 중 오류 발생: {e}")
return False
def decide_weekly_recommendations(self):
"""주간 날씨에 따른 추천 사항 결정"""
self.weekly_recommendation = "이번 주 날씨 예보입니다. "
for day in self.forecast:
date = day["date"]
temp = day["temperature"]
weather = day["weather"]
desc = day["description"]
# 날짜 형식 변환 (YYYY-MM-DD -> MM월 DD일)
date_parts = date.split("-")
formatted_date = f"{int(date_parts[1])}월 {int(date_parts[2])}일"
# 기본 예보 정보
self.weekly_recommendation += f"{formatted_date}은 {desc}, 기온은 {temp:.1f}도 예상됩니다. "
# 일정이 있는 경우 추가 정보
if date in self.schedule_db:
event = self.schedule_db[date]
self.weekly_recommendation += f"이날은 {event} 일정이 있습니다. "
# 야외 활동이 있는 날인 경우 특별 안내
if "야외" in event and ("Rain" in weather or "Snow" in weather):
self.weekly_recommendation += f"야외 활동이 있으나 비 또는 눈이 예상되니 일정 변경을 고려하거나 우산을 준비하세요. "
# 의류 추천
temp_category = self.get_temperature_category(temp)
clothes = self.clothing_recommendations[temp_category]
self.weekly_recommendation += f"추천 의류는 {', '.join(clothes[:2])} 등입니다. "
print("주간 추천 사항 결정 완료")
def morning_briefing(self):
"""향상된 아침 브리핑 실행"""
print("향상된 아침 브리핑을 시작합니다...")
# 현재 날씨 정보 수집
current_weather_success = self.sense_weather()
# 예보 정보 수집
forecast_success = self.sense_forecast()
if current_weather_success:
# 현재 날씨 기반 추천 사항 결정
self.decide_recommendations()
# 주간 날씨 기반 추천 사항 결정
if forecast_success:
self.decide_weekly_recommendations()
# 음성 출력 (현재 날씨 + 주간 예보)
current_time = datetime.datetime.now().strftime("%H시 %M분")
output_text = f"좋은 아침입니다. 현재 시각은 {current_time}입니다. "
output_text += f"{self.city}의 현재 날씨는 {self.weather_description}이며, 기온은 {self.temperature:.1f}도입니다. "
output_text += self.recommendation
if forecast_success:
output_text += " " + self.weekly_recommendation
print("음성 출력: " + output_text[:100] + "...") # 너무 길어서 일부만 출력
# 텍스트를 음성으로 변환하여 출력
self.engine.say(output_text)
self.engine.runAndWait()
else:
# 날씨 정보를 가져오지 못한 경우 기본 메시지 출력
self.engine.say("죄송합니다. 날씨 정보를 가져오는 데 문제가 발생했습니다. 인터넷 연결을 확인해주세요.")
self.engine.runAndWait()
# 고급 날씨 비서 사용 예시
if __name__ == "__main__":
API_KEY = "your_openweathermap_api_key"
# 고급 날씨 비서 초기화
advanced_assistant = AdvancedWeatherAssistant(api_key=API_KEY, city="Seoul")
# 아침 브리핑 스케줄링
advanced_assistant.schedule_morning_briefing(time="07:00")
# 음성 명령과 함께 실행
advanced_assistant.run_with_voice_commands()5. 라즈베리 파이에서 실행하기 위한 하드웨어 통합
라즈베리 파이와 같은 소형 컴퓨터에서 실행하여 실제 홈 어시스턴트로 활용할 수 있도록 하드웨어 센서와 통합하는 코드를 추가합니다.
# 라즈베리 파이에서만 사용할 수 있는 라이브러리
# 다른 환경에서는 주석 처리 필요
try:
import Adafruit_DHT # 온습도 센서
import RPi.GPIO as GPIO # GPIO 제어
HARDWARE_AVAILABLE = True
except ImportError:
HARDWARE_AVAILABLE = False
print("하드웨어 라이브러리를 가져올 수 없습니다. 하드웨어 기능이 비활성화됩니다.")
class HardwareIntegratedAssistant(AdvancedWeatherAssistant):
def __init__(self, api_key, city="Seoul", language="ko"):
super().__init__(api_key, city, language)
self.hardware_enabled = HARDWARE_AVAILABLE
if self.hardware_enabled:
# 하드웨어 설정
self.dht_sensor = Adafruit_DHT.DHT22
self.dht_pin = 4 # GPIO 핀 번호
# LED 핀 설정
GPIO.setmode(GPIO.BCM)
self.led_pins = {
"sunny": 17, # 노란색 LED
"rainy": 18, # 파란색 LED
"cold": 27, # 파란색 LED
"hot": 22 # 빨간색 LED
}
for pin in self.led_pins.values():
GPIO.setup(pin, GPIO.OUT)
GPIO.output(pin, GPIO.LOW) # 초기화 시 모든 LED 끄기
print("하드웨어 모듈이 초기화되었습니다.")
def sense_local_conditions(self):
"""로컬 온습도 센서에서 데이터 읽기"""
if not self.hardware_enabled:
return False
try:
humidity, temperature = Adafruit_DHT.read_retry(self.dht_sensor, self.dht_pin)
if humidity is not None and temperature is not None:
self.local_humidity = humidity
self.local_temperature = temperature
print(f"로컬 센서 데이터: 온도 {temperature:.1f}°C, 습도 {humidity:.1f}%")
return True
else:
print("센서에서 데이터를 읽을 수 없습니다.")
return False
except Exception as e:
print(f"센서 읽기 오류: {e}")
return False
def act_led_output(self):
"""날씨 상태에 따라 LED 표시 (액추에이터)"""
if not self.hardware_enabled:
return
# 모든 LED 초기화
for pin in self.led_pins.values():
GPIO.output(pin, GPIO.LOW)
# 날씨 상태에 따른 LED 켜기
if self.weather_condition in ["Clear", "Clouds"]:
GPIO.output(self.led_pins["sunny"], GPIO.HIGH)
if self.weather_condition in ["Rain", "Drizzle", "Thunderstorm"]:
GPIO.output(self.led_pins["rainy"], GPIO.HIGH)
# 온도에 따른 LED 켜기
if self.temperature < 10:
GPIO.output(self.led_pins["cold"], GPIO.HIGH)
elif self.temperature > 25:
GPIO.output(self.led_pins["hot"], GPIO.HIGH)
def morning_briefing(self):
"""하드웨어 통합된 아침 브리핑 실행"""
print("하드웨어 통합 아침 브리핑을 시작합니다...")
# 로컬 센서 데이터 수집
local_sensor_success = self.sense_local_conditions()
# API에서 현재 날씨 정보 수집
current_weather_success = self.sense_weather()
# 예보 정보 수집
forecast_success = self.sense_forecast()
if current_weather_success:
# LED 표시 (액추에이터)
self.act_led_output()
# 현재 날씨 기반 추천 사항 결정
self.decide_recommendations()
# 주간 날씨 기반 추천 사항 결정
if forecast_success:
self.decide_weekly_recommendations()
# 음성 출력 구성 (현재 날씨 + 로컬 센서 + 주간 예보)
current_time = datetime.datetime.now().strftime("%H시 %M분")
output_text = f"좋은 아침입니다. 현재 시각은 {current_time}입니다. "
output_text += f"{self.city}의 현재 날씨는 {self.weather_description}이며, 기온은 {self.temperature:.1f}도입니다. "
# 로컬 센서 데이터가 있는 경우 추가
if local_sensor_success:
output_text += f"실내 온도는 {self.local_temperature:.1f}도, 습도는 {self.local_humidity:.1f}%입니다. "
# 실내외 온도 차이가 큰 경우 안내
temp_diff = abs(self.local_temperature - self.temperature)
if temp_diff > 10:
output_text += f"실내외 온도 차이가 {temp_diff:.1f}도로 매우 큽니다. 외출 시 체온 변화에 주의하세요. "
output_text += self.recommendation
if forecast_success:
output_text += " " + self.weekly_recommendation
print("음성 출력: " + output_text[:100] + "...") # 너무 길어서 일부만 출력
# 텍스트를 음성으로 변환하여 출력
self.engine.say(output_text)
self.engine.runAndWait()
else:
# 날씨 정보를 가져오지 못한 경우 기본 메시지 출력
self.engine.say("죄송합니다. 날씨 정보를 가져오는 데 문제가 발생했습니다. 인터넷 연결을 확인해주세요.")
self.engine.runAndWait()
def cleanup(self):
"""GPIO 리소스 정리"""
if self.hardware_enabled:
GPIO.cleanup()
print("GPIO 리소스가 정리되었습니다")
def run_with_voice_commands(self):
"""스케줄러와 음성 명령을 동시에 실행 (종료 시 리소스 정리)"""
try:
# 스케줄러 스레드 시작
scheduler_thread = threading.Thread(target=self.run)
scheduler_thread.daemon = True
scheduler_thread.start()
# 음성 인식 시작
self.listen_for_commands()
finally:
# 프로그램 종료 시 GPIO 리소스 정리
if self.hardware_enabled:
self.cleanup()
# 하드웨어 통합 날씨 비서 사용 예시
if __name__ == "__main__":
API_KEY = "your_openweathermap_api_key"
# 하드웨어 통합 날씨 비서 초기화
hw_assistant = HardwareIntegratedAssistant(api_key=API_KEY, city="Seoul")
# 아침 브리핑 스케줄링
hw_assistant.schedule_morning_briefing(time="07:00")
# 음성 명령과 함께 실행
hw_assistant.run_with_voice_commands()6. 자동 부팅 시 실행을 위한 서비스 설정
라즈베리 파이에서 부팅 시 자동으로 날씨 비서를 실행하려면 시스템 서비스로 등록해야 합니다. 다음은 systemd 서비스 파일 예시입니다:
# /etc/systemd/system/weather-assistant.service 파일 내용
[Unit]
Description=Weather Assistant Service
After=network.target
[Service]
ExecStart=/usr/bin/python3 /home/pi/weather_assistant/main.py
WorkingDirectory=/home/pi/weather_assistant
StandardOutput=inherit
StandardError=inherit
Restart=always
User=pi
[Install]
WantedBy=multi-user.target7. 웹 인터페이스 추가
날씨 비서의 원격 관리를 위한 간단한 웹 인터페이스를 추가해보겠습니다:
from flask import Flask, render_template, request, jsonify
import threading
import json
class WebEnabledAssistant(HardwareIntegratedAssistant):
def __init__(self, api_key, city="Seoul", language="ko", web_port=5000):
super().__init__(api_key, city, language)
# 웹 서버 설정
self.app = Flask(__name__)
self.web_port = web_port
# 웹 서버 라우트 설정
@self.app.route('/')
def home():
return render_template('index.html',
city=self.city,
schedule_time=self.scheduled_time)
@self.app.route('/api/status')
def status():
return jsonify({
'city': self.city,
'scheduled_time': self.scheduled_time,
'current_weather': getattr(self, 'weather_condition', None),
'current_temperature': getattr(self, 'temperature', None),
'local_temperature': getattr(self, 'local_temperature', None),
'local_humidity': getattr(self, 'local_humidity', None)
})
@self.app.route('/api/update_settings', methods=['POST'])
def update_settings():
data = request.json
if 'city' in data:
self.city = data['city']
if 'scheduled_time' in data:
# 이전 스케줄 삭제
schedule.clear()
# 새 스케줄 설정
self.schedule_morning_briefing(data['scheduled_time'])
return jsonify({'status': 'success', 'message': '설정이 업데이트되었습니다.'})
@self.app.route('/api/test_briefing', methods=['POST'])
def test_briefing():
# 별도 스레드에서 브리핑 실행
threading.Thread(target=self.morning_briefing).start()
return jsonify({'status': 'success', 'message': '브리핑이 실행됩니다.'})
self.scheduled_time = "07:00" # 기본값
print(f"웹 인터페이스가 초기화되었습니다. 포트: {web_port}")
def schedule_morning_briefing(self, time="07:00"):
"""지정된 시간에 아침 브리핑이 실행되도록 스케줄링"""
self.scheduled_time = time
schedule.every().day.at(time).do(self.morning_briefing)
print(f"아침 브리핑이 매일 {time}에 실행되도록 설정되었습니다.")
def run_with_web_interface(self):
"""웹 인터페이스와 스케줄러, 음성 명령을 동시에 실행"""
try:
# 스케줄러 스레드 시작
scheduler_thread = threading.Thread(target=self.run)
scheduler_thread.daemon = True
scheduler_thread.start()
# 음성 인식 스레드 시작
voice_thread = threading.Thread(target=self.listen_for_commands)
voice_thread.daemon = True
voice_thread.start()
# 웹 서버 시작
self.app.run(host='0.0.0.0', port=self.web_port)
finally:
# 프로그램 종료 시 GPIO 리소스 정리
if self.hardware_enabled:
self.cleanup()
# 웹 인터페이스 파일 생성 (templates/index.html)
"""
<!DOCTYPE html>
<html>
<head>
<title>날씨 비서 관리</title>
<meta name="viewport" content="width=device-width, initial-scale=1">
<style>
body { font-family: Arial, sans-serif; margin: 0; padding: 20px; }
.container { max-width: 800px; margin: 0 auto; }
.card { border: 1px solid #ddd; border-radius: 8px; padding: 20px; margin-bottom: 20px; }
.button { background-color: #4CAF50; border: none; color: white; padding: 10px 20px;
text-align: center; text-decoration: none; display: inline-block;
font-size: 16px; margin: 4px 2px; cursor: pointer; border-radius: 4px; }
.status { display: flex; justify-content: space-between; flex-wrap: wrap; }
.status-item { flex: 1; min-width: 200px; margin: 10px; padding: 10px; background-color: #f9f9f9; border-radius: 4px; }
</style>
</head>
<body>
<div class="container">
<h1>날씨 비서 관리 시스템</h1>
<div class="card">
<h2>현재 상태</h2>
<div class="status" id="status-container">
<div class="status-item">
<h3>도시</h3>
<p id="city">{{ city }}</p>
</div>
<div class="status-item">
<h3>알림 시간</h3>
<p id="scheduled-time">{{ schedule_time }}</p>
</div>
<div class="status-item">
<h3>현재 날씨</h3>
<p id="current-weather">로딩 중...</p>
</div>
<div class="status-item">
<h3>현재 온도</h3>
<p id="current-temperature">로딩 중...</p>
</div>
<div class="status-item">
<h3>실내 온도</h3>
<p id="local-temperature">로딩 중...</p>
</div>
<div class="status-item">
<h3>실내 습도</h3>
<p id="local-humidity">로딩 중...</p>
</div>
</div>
</div>
<div class="card">
<h2>설정</h2>
<form id="settings-form">
<div>
<label for="city-input">도시:</label>
<input type="text" id="city-input" value="{{ city }}">
</div>
<div>
<label for="time-input">알림 시간:</label>
<input type="time" id="time-input" value="{{ schedule_time }}">
</div>
<button type="submit" class="button">설정 저장</button>
</form>
</div>
<div class="card">
<h2>테스트</h2>
<button id="test-button" class="button">지금 브리핑 실행</button>
</div>
</div>
<script>
// 상태 정보 업데이트
function updateStatus() {
fetch('/api/status')
.then(response => response.json())
.then(data => {
document.getElementById('city').textContent = data.city || '-';
document.getElementById('scheduled-time').textContent = data.scheduled_time || '-';
document.getElementById('current-weather').textContent = data.current_weather || '-';
document.getElementById('current-temperature').textContent =
data.current_temperature ? `${data.current_temperature}°C` : '-';
document.getElementById('local-temperature').textContent =
data.local_temperature ? `${data.local_temperature}°C` : '-';
document.getElementById('local-humidity').textContent =
data.local_humidity ? `${data.local_humidity}%` : '-';
})
.catch(error => console.error('상태 업데이트 오류:', error));
}
// 페이지 로드 시 상태 업데이트
updateStatus();
// 1분마다 상태 업데이트
setInterval(updateStatus, 60000);
// 설정 저장
document.getElementById('settings-form').addEventListener('submit', function(e) {
e.preventDefault();
const city = document.getElementById('city-input').value;
const time = document.getElementById('time-input').value;
fetch('/api/update_settings', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
city: city,
scheduled_time: time
}),
})
.then(response => response.json())
.then(data => {
alert(data.message);
updateStatus();
})
.catch(error => console.error('설정 업데이트 오류:', error));
});
// 브리핑 테스트
document.getElementById('test-button').addEventListener('click', function() {
fetch('/api/test_briefing', {
method: 'POST',
})
.then(response => response.json())
.then(data => {
alert(data.message);
})
.catch(error => console.error('브리핑 테스트 오류:', error));
});
</script>
</body>
</html>
"""
# 웹 인터페이스 통합 날씨 비서 사용 예시
if __name__ == "__main__":
API_KEY = "your_openweathermap_api_key"
# 웹 인터페이스 통합 날씨 비서 초기화
web_assistant = WebEnabledAssistant(api_key=API_KEY, city="Seoul", web_port=5000)
# 아침 브리핑 스케줄링
web_assistant.schedule_morning_briefing(time="07:00")
# 웹 인터페이스, 스케줄러, 음성 명령 함께 실행
web_assistant.run_with_web_interface()8. 알림 및 이메일 기능 통합
중요한 날씨 변화가 예상될 때 이메일이나 다른 알림 수단을 통해 사용자에게 알려주는 기능을 추가합니다.
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
class NotificationEnabledAssistant(WebEnabledAssistant):
def __init__(self, api_key, city="Seoul", language="ko", web_port=5000,
email_user=None, email_password=None, recipient_email=None):
super().__init__(api_key, city, language, web_port)
# 이메일 설정
self.email_user = email_user
self.email_password = email_password
self.recipient_email = recipient_email
self.email_enabled = (email_user is not None and email_password is not None and recipient_email is not None)
# 중요 날씨 경고 기준
self.severe_weather_types = ["Thunderstorm", "Tornado", "Hurricane", "Flood", "Snow"]
self.extreme_temp_threshold_high = 35 # 35도 이상
self.extreme_temp_threshold_low = -10 # -10도 이하
# 이메일 전송 기록 (중복 방지)
self.sent_notifications = {}
if self.email_enabled:
print("알림 기능이 활성화되었습니다.")
else:
print("이메일 설정이 완료되지 않아 알림 기능이 비활성화되었습니다.")
def check_severe_conditions(self):
"""심각한 날씨 조건 확인"""
alerts = []
# 현재 날씨에서 심각한 조건 확인
if hasattr(self, 'weather_condition') and self.weather_condition in self.severe_weather_types:
alerts.append(f"현재 심각한 날씨 상태: {self.weather_condition}")
if hasattr(self, 'temperature'):
if self.temperature >= self.extreme_temp_threshold_high:
alerts.append(f"현재 매우 높은 온도: {self.temperature}°C")
elif self.temperature <= self.extreme_temp_threshold_low:
alerts.append(f"현재 매우 낮은 온도: {self.temperature}°C")
# 예보에서 심각한 조건 확인
if hasattr(self, 'forecast'):
for day in self.forecast:
date = day["date"]
# 이미 알림을 보낸 날짜는 건너뛰기
if date in self.sent_notifications:
continue
weather = day["weather"]
temp = day["temperature"]
if weather in self.severe_weather_types:
alerts.append(f"{date}에 예상되는 심각한 날씨: {weather}")
self.sent_notifications[date] = True
if temp >= self.extreme_temp_threshold_high:
alerts.append(f"{date}에 예상되는 매우 높은 온도: {temp}°C")
self.sent_notifications[date] = True
elif temp <= self.extreme_temp_threshold_low:
alerts.append(f"{date}에 예상되는 매우 낮은 온도: {temp}°C")
self.sent_notifications[date] = True
return alerts
def send_email_notification(self, subject, message):
"""이메일 알림 전송"""
if not self.email_enabled:
print("이메일 설정이 완료되지 않아 알림을 보낼 수 없습니다.")
return False
try:
msg = MIMEMultipart()
msg['From'] = self.email_user
msg['To'] = self.recipient_email
msg['Subject'] = subject
msg.attach(MIMEText(message, 'plain'))
server = smtplib.SMTP('smtp.gmail.com', 587)
server.starttls()
server.login(self.email_user, self.email_password)
text = msg.as_string()
server.sendmail(self.email_user, self.recipient_email, text)
server.quit()
print(f"이메일 알림이 전송되었습니다: {subject}")
return True
except Exception as e:
print(f"이메일 전송 중 오류 발생: {e}")
return False
def morning_briefing(self):
"""알림 기능이 통합된 아침 브리핑"""
# 기존 브리핑 실행
super().morning_briefing()
# 심각한 날씨 조건 확인
alerts = self.check_severe_conditions()
# 알림이 필요한 경우 이메일 전송
if alerts and self.email_enabled:
subject = f"[날씨 비서] 중요 날씨 알림 - {datetime.datetime.now().strftime('%Y-%m-%d')}"
message = "다음과 같은 중요한 날씨 상황이 감지되었습니다:\n\n"
message += "\n".join(f"- {alert}" for alert in alerts)
message += "\n\n날씨 비서 드림"
self.send_email_notification(subject, message)
# 알림 기능 통합 날씨 비서 사용 예시
if __name__ == "__main__":
API_KEY = "your_openweathermap_api_key"
# 이메일 설정 (Gmail 사용 시)
EMAIL_USER = "your_email@gmail.com"
EMAIL_PASSWORD = "your_app_password" # Gmail의 경우 앱 비밀번호 필요
RECIPIENT_EMAIL = "recipient@example.com"
# 알림 기능 통합 날씨 비서 초기화
notification_assistant = NotificationEnabledAssistant(
api_key=API_KEY,
city="Seoul",
web_port=5000,
email_user=EMAIL_USER,
email_password=EMAIL_PASSWORD,
recipient_email=RECIPIENT_EMAIL
)
# 아침 브리핑 스케줄링
notification_assistant.schedule_morning_briefing(time="07:00")
# 웹 인터페이스, 스케줄러, 음성 명령 함께 실행
notification_assistant.run_with_web_interface()9. 결론
이제 우리는 다음과 같은 기능을 갖춘 고급 날씨 비서 에이전트를 개발했습니다:
- 센서(정보 수집)
- 온라인 날씨 API를 통한 현재 날씨 및 예보 정보 수집
- 라즈베리 파이의 온습도 센서를 통한 실내 환경 데이터 수집
- 음성 인식을 통한 사용자 명령 수집
- 의사결정 시스템
- 날씨 상태에 따른 맞춤 추천 사항 생성
- 온도 범주에 따른 의류 추천
- 일정과 연동된 상황별 조언
- 심각한 날씨 상태 감지 및 알림 결정
- 액추에이터(출력)
- 음성 합성을 통한 정보 전달
- LED 표시를 통한 시각적 날씨 상태 표현
- 이메일 알림 발송
- 웹 인터페이스를 통한 정보 시각화
이 에이전트는 오픈소스 AI 기술과 파이썬 라이브러리를 활용하여 실제 환경 데이터를 수집하고 유용한 정보를 제공하는 실용적인 시스템입니다. 라즈베리 파이와 같은 저비용 하드웨어에서 실행할 수 있으며, 다양한 센서와 액추에이터를 추가하여 더 확장할 수 있습니다.
이러한 접근 방식은 다른 종류의 스마트 홈 에이전트나 IoT 시스템 개발에도 유사하게 적용할 수 있습니다. 센서로부터 데이터를 수집하고, AI 알고리즘으로 처리한 후, 의미 있는 행동을 취하는 원리는 다양한 실생활 문제 해결에 활용할 수 있습니다.
우리는 라즈베리파이와 같은 소형 머신과 API, 다양한 센서, 액추에이터를 주변에서 쉽게 구할 수 있고, 3D 프린터로 원하는 부품들을 만들어서 AI Agent와 연결할 수 있는 시대에 살고 있다.
거기에 덧붙혀서 소프트웨어도 예전에는 상상하지 못한 기술로 쉽게 만들어 낼 수 있기도 하다.
정말 잘 동작하도록 만들기 위해서는 여러 가지 시도와 도전이 필요하지만 예전에는 꿈도 꿀 수 없는 것을 이제는 꿈꿀 수 있게 되었다고 본다.
물론 상용 제품처럼 더 많은 기능을 가진 소프트웨어 개발을 위해서는 내가 직접 구조도를 기획하고, 그에 따른 코드를 읽어보면서 일정 부분을 AI에게 요청해서 수정하는 것이 필요할 것이다.
아직 전체를 다 만들어 달라고 하면 제대로 못하고 코드를 이상하게 만들어 버려서 멘붕에 빠지게 되는 경험을 하게 될 것이 분명하다. :)
그렇기 때문에 복잡한 시스템 구축이 필요할 경우에는 전체 계획을 세우고 그 계획에 대한 PRD(Product Requirements Document)와 같은 개발 요구사항을 직접 짜는게 필요하다고 생각한다.
언젠가는 말만 하면 처음부터 끝까지 다 만들어 주고 세세한 수정사항도 다 반영해서 고쳐줄 수 있겠지만, 아직은 코드를 볼 수 있는 눈이 필요하다는 것이다. AI Agent와 함께라면 시간을 쏟은 만큼 얻을 수 있는 시대가 된 것은 아닐까?
그럼, 나처럼 직접 코딩하는 것은 어려워 하지만 뭔가 나만의 비서를 만들어보고 싶은 분들은 도전해 보시길 바라며 글을 마친다.
반응형
'파이썬으로 할 수 있는 일 > 실사용' 카테고리의 다른 글
| AI 에이전트의 핵심 기술 MCP 이해하기: 비개발자를 위한 쉬운 설명 (5) | 2025.04.16 |
|---|---|
| 여행지 검색에 Claude AI 활용(+MCP를 사용해 최신 데이타 검색) (20) | 2025.04.09 |
| Claude MCP 가이드: 실시간 정보로 AI 성능 높이는 Model Context Protocol 설정 및 활용 2025 (4) | 2025.04.04 |
| 파이썬을 이용한 PDF 텍스트 추출: PyPDF2 라이브러리 활용 가이드 (0) | 2023.10.30 |
| FinanceDataReader를 사용해 FRED 경기선행 지수 데이터 확인하기 (0) | 2023.10.25 |
