Spaces:
Sleeping
Sleeping
File size: 3,364 Bytes
fbb2ad3 5d28477 d49b154 a645773 fbb2ad3 a645773 2c8ffe4 5d28477 d49b154 5d28477 89f83ba 5d28477 89f83ba 5d28477 89f83ba d49b154 5d28477 50b03b3 2c8ffe4 5d28477 50b03b3 057c9fe 6bf6502 5d28477 89f83ba 216e2b9 5d28477 216e2b9 5d28477 70e8a40 5d28477 89f83ba 5d28477 6bf6502 d49b154 3adb8c5 d49b154 916f777 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
from typing import Dict
from datetime import datetime
import uuid
import model # your existing model.py
import joblib
import pandas as pd
import tensorflow as tf # Import TensorFlow here
import asyncio
app = FastAPI()
# Load or train model at startup
ml_model = model.load_or_train_model()
SCALER_PATH = "/tmp/scaler.pkl"
# -------------------------
# Request & Response Schemas
# -------------------------
class FamilyInput(BaseModel):
adult_male: int
adult_female: int
child: int
class UserInput(BaseModel):
user_id: str
region: str
season: str
event: str
family: FamilyInput
stock: Dict[str, float] # product_name: quantity
class RetrainRequest(BaseModel):
user_id: str
# -------------------------
# API Routes
# -------------------------
# Define request body schema
class Item(BaseModel):
name: str
quantity: int
@app.get("/")
def read_root():
return {"message": "✅ GrocyGenie API is running."}
@app.post("/testpost")
def test_post(item: Item):
return {"message": f"Received item '{item.name}' with quantity {item.quantity}"}
@app.post("/predict")
def predict(input_data: UserInput):
try:
user_dict = input_data.dict()
user_id = user_dict["user_id"]
predictions = model.predict_user_input(user_dict)
model.store_predictions(user_id, predictions, user_dict)
feedback = pd.DataFrame([{
'date': datetime.today().strftime('%Y-%m-%d'),
'product': k,
'region': user_dict['region'],
'season': user_dict['season'],
'event': user_dict['event'],
'adult_male': user_dict['family']['adult_male'],
'adult_female': user_dict['family']['adult_female'],
'child': user_dict['family']['child'],
'consumption': v['predicted_consumption'],
'finish_error': v['predicted_finish_error'],
'finish_days': v['predicted_finish_days'],
'stock_quantity': user_dict['stock'][k] # <--- Added stock quantity here
} for k, v in predictions.items()])
model.insert_feedback(user_id, feedback)
return {
"user_id": user_id,
"predictions": predictions
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/retrain")
def retrain_model(request: RetrainRequest):
success = model.retrain_model_with_feedback(request.user_id)
if success:
# Reload model and scaler globally for future predictions
model.ml_model = tf.keras.models.load_model(model.MODEL_PATH)
model.scaler = joblib.load(SCALER_PATH)
return {"message": f"Model retrained using feedback for user {request.user_id}."}
else:
raise HTTPException(status_code=404, detail="No feedback found for retraining.")
@app.post("/train")
async def train_model_from_api():
try:
# Call your existing training function
model.ml_model = model.load_or_train_model()
# Reload the scaler as well
model.scaler = joblib.load(SCALER_PATH)
return {"message": "✅ Model retrained from scratch using latest data."}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
|