251118 [ Day 90 ] - Project (10)

시작하며

오늘은 팀원들과 의논을 하여 RandomForest를 사용하기로 결정을 하고, 모델 예측 API를 작성하였다.

모델 예측 API

# 모델 예측 API
from app.db import get_connection, close_connection
from flask import Blueprint, jsonify
import traceback, pickle
import pandas as pd

# lux -> 일사량 변환
def lux_to_insolation(lux):
    wm2 = lux * 0.0079
    mj = wm2 * 0.0036
    return mj

# 모델 불러오기
MODEL_PATH = "../models/rf_best_model.pkl"

with open(MODEL_PATH, "rb") as f:
    model = pickle.load(f)

# Blueprint 생성
energy_bp = Blueprint("energy", __name__)

# 모델 예측 API
@energy_bp.route("/api/energy/predict", methods=["post"])
def predict_generation():
    """
    1h, 2h, 3h 뒤 발전량 예측
    """
    # DB에서 데이터 조회
    conn, cursor = None, None
    try:
        # DB 연결
        conn, cursor = get_connection()
        if conn is None:
            return jsonify({"message": "DB 연결 실패"}), 500
        
        sql = """
            SELECT 
                YEAR(timestamp) as year,
                MONTH(timestamp) as month,
                DAY(timestamp) as day,
                HOUR(timestamp) as hour,
                generation_amount as generation,
                LAG(generation_amount, 1) OVER (ORDER BY timestamp) as prev_generation,
                LAG(generation_amount, 24) OVER (ORDER BY timestamp) as yesterday_generation,
                lux
            FROM sun_data
            ORDER BY timestamp DESC
            LIMIT 1
        """
        cursor.execute(sql)
        latest_data = cursor.fetchone()

        # 일사량 데이터로 변환
        latest_data["insolation"] = lux_to_insolation(latest_data["lux"])

        if not latest_data:
            return jsonify({"message": "No data found"}), 404
        
        # DataFrame으로 변환
        X = pd.DataFrame([latest_data])

        # 모델 예측
        y = model.predict(X)

        # 결과 출력 딕셔너리 생성
        result = {
            "1h": round(float(y[0]), 2),
            "2h": round(float(y[1]), 2),
            "3h": round(float(y[2]), 2)
        }

        # 결과 반환
        return jsonify(result), 200

    except Exception as e:
        print(f"Error in check_available_channels: {e}")
        traceback.print_exc()
        return jsonify({"message": "Internal Server Error"}), 500
    
    finally:
        close_connection(conn, cursor)

마치며

이제 DB와 서버 구축은 거의 마무리 단계인 것 같아서 아두이노와 서버 연동 테스트를 해봐야 할 것 같다.