텔레그램 API를 활용한 스마트 화분 프로젝트(3) – 명령어 설정 및 코드, 데모 비디오

# -*- coding: utf-8 -*-
import logging
from grove.adc import ADC
import RPi.GPIO as GPIO
from telegram import Update
from telegram.ext import Application, CommandHandler, CallbackContext
from datetime import datetime
import time
import threading
import schedule

# 텔레그램 봇의 API 키와 Chat id(-포함)를 입력하세요
TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN'
GROUP_CHAT_ID = -'YOUR_GROUP_CHAT_ID'

# GPIO 핀 설정
relay_pin = 17
last_watered_file = "/home/pi/last_watered.txt"
log_file = "/home/pi/watering_log.txt"

# ADC 설정 (Grove Hat의 A0 포트를 사용)
adc = ADC()
soil_sensor_channel = 0  # A0 포트 사용

# GPIO 핀 모드 설정
GPIO.setmode(GPIO.BCM)
GPIO.setup(relay_pin, GPIO.OUT)

# 로깅 설정
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)

# 토양 수분 임계값 설정
SOIL_MOISTURE_THRESHOLD = 300

# 명령어 처리 함수들
def status(update: Update, context: CallbackContext) -> None:
    if update.effective_chat.id != GROUP_CHAT_ID:
        return

    # 토양 수분 데이터를 읽어옵니다.
    soil_moisture = adc.read(soil_sensor_channel)
    soil_moisture_percentage = (soil_moisture / 950.0) * 100
    update.message.reply_text(f"Soil moisture level: {soil_moisture_percentage:.1f}%")

# 물 주기 함수
def water(update: Update = None, context: CallbackContext = None) -> None:
    if update and update.effective_chat.id != GROUP_CHAT_ID:
        return

    try:
        # 물 주기 시작
        GPIO.output(relay_pin, GPIO.HIGH)
        if update:
            update.message.reply_text("Watering the plant...")
        time.sleep(10)  # 10초 동안 물 주기 (필요시 조정)
        GPIO.output(relay_pin, GPIO.LOW)
        if update:
            update.message.reply_text("Watering complete.")

        # 현재 시간을 last_watered_file에 기록
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with open(last_watered_file, "w") as f:
            f.write(now)

        # 물 주기 기록 로그 파일에 기록
        with open(log_file, "a") as log:
            log.write(f"Watered on: {now}\n")

    except IOError:
        if update:
            update.message.reply_text("Error controlling the water pump")

# 마지막 물 준 시간 확인 함수
def lastwatered(update: Update, context: CallbackContext) -> None:
    if update.effective_chat.id != GROUP_CHAT_ID:
        return

    try:
        with open(last_watered_file, "r") as f:
            last_watered_time = f.read()
            update.message.reply_text(f"Last watered on: {last_watered_time}")
    except FileNotFoundError:
        update.message.reply_text("No record of last watering found.")

# 자동 물 주기 체크 함수
def auto_water_check():
    soil_moisture = adc.read(soil_sensor_channel)
    if soil_moisture < SOIL_MOISTURE_THRESHOLD:
        water()

# 자동 물 주기 스케줄링 함수
def schedule_daily_check():
    schedule.every().day.at("09:00").do(auto_water_check)  # 매일 오전 9시에 체크
    while True:
        schedule.run_pending()
        time.sleep(1)

# 봇 시작 함수
def main():
    # Application을 생성하고, Dispatcher를 가져옵니다.
    application = Application.builder().token(TOKEN).build()

    # 핸들러들 등록
    application.add_handler(CommandHandler("status", status))
    application.add_handler(CommandHandler("water", water))
    application.add_handler(CommandHandler("lastwatered", lastwatered))

    # 자동 물 주기 체크 스레드 시작
    threading.Thread(target=schedule_daily_check, daemon=True).start()

    # 봇 시작
    application.run_polling()

# 메인 함수 실행
if __name__ == "__main__":
    main()