# import time # import pickle # import numpy as np # from fastapi import FastAPI, Request, status, HTTPException, Depends # from fastapi.security.api_key import APIKeyHeader # from pydantic import BaseModel # from typing import List # import os # import joblib # app = FastAPI( # title="ML Music Classifier API", # description="Predict 'liked' or not from audio features using 8 ML models.", # version="1.0.0" # ) # DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0" # api_key_header = APIKeyHeader(name="X-Token", auto_error=False) # @app.middleware("http") # async def add_process_time_header(request: Request, call_next): # start_time = time.perf_counter() # response = await call_next(request) # process_time = time.perf_counter() - start_time # response.headers["X-Process-Time"] = str(process_time) # return response # def verify_token(x_token: str = Depends(api_key_header)): # if x_token != DEFAULT_TOKEN: # raise HTTPException( # status_code=status.HTTP_401_UNAUTHORIZED, # detail="Invalid or missing token. Use correct 'X-Token' in headers." # ) # class SongFeatures(BaseModel): # danceability: float # energy: float # key: int # loudness: float # mode: int # speechiness: float # acousticness: float # instrumentalness: float # liveness: float # valence: float # tempo: float # duration_ms: int # time_signature: int # MODEL_DIR = "models" # model_names = [ # "ANN_model", "knn_model", "logistic_regression_model", # "neural_model", "Naive_Bayes_model", # "random_forest_model", "svm_model", "XGBoost_model" # ] # models = {} # for name in model_names: # path = os.path.join(MODEL_DIR, f"{name}.pkl") # try: # models[name] = joblib.load(path) # except Exception as e: # print(f"Error loading {name}: {e}") # @app.post("/predict/{model_name}", dependencies=[Depends(verify_token)]) # def predict(model_name: str, features: SongFeatures): # if model_name not in models: # raise HTTPException(status_code=404, detail="Model not found") # model = models[model_name] # input_array = np.array([[getattr(features, field) for field in features.model_fields]]) # try: # prediction = model.predict(input_array) # return { # "model": model_name, # "input": features, # "prediction": int(prediction[0]), # "prediction_label": "liked" if int(prediction[0]) == 1 else "not_liked" # } # except Exception as e: # raise HTTPException(status_code=500, detail=f"Prediction error: {str(e)}") # import time # import pickle # import numpy as np # from fastapi import FastAPI, Request, status, HTTPException, Depends # from fastapi.security.api_key import APIKeyHeader # from pydantic import BaseModel # from typing import List # import os # import joblib # app = FastAPI( # title="ML Music Classifier API", # description="Predict 'liked' or not from audio features using 8 ML models.", # version="1.0.0" # ) # DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0" # api_key_header = APIKeyHeader(name="X-Token", auto_error=False) # @app.middleware("http") # async def add_process_time_header(request: Request, call_next): # start_time = time.perf_counter() # response = await call_next(request) # process_time = time.perf_counter() - start_time # response.headers["X-Process-Time"] = str(process_time) # return response # def verify_token(x_token: str = Depends(api_key_header)): # if x_token != DEFAULT_TOKEN: # raise HTTPException( # status_code=status.HTTP_401_UNAUTHORIZED, # detail="Invalid or missing token. Use correct 'X-Token' in headers." # ) # class SongFeatures(BaseModel): # danceability: float # energy: float # key: int # loudness: float # mode: int # speechiness: float # acousticness: float # instrumentalness: float # liveness: float # valence: float # tempo: float # duration_ms: int # time_signature: int # MODEL_DIR = "models" # model_names = [ # "ANN_model", "knn_model", "logistic_regression_model", # "neural_model", "Naive_Bayes_model", # "random_forest_model", "svm_model", "XGBoost_model" # ] # models = {} # for name in model_names: # path = os.path.join(MODEL_DIR, f"{name}.pkl") # try: # models[name] = joblib.load(path) # print(f"✅ Successfully loaded {name}") # except Exception as e: # print(f"❌ Error loading {name}: {e}") # @app.get("/") # def root(): # return { # "message": "ML Music Classifier API is running!", # "loaded_models": len(models), # "available_models": list(models.keys()), # "endpoints": { # "predict": "/predict/{model_name}", # "docs": "/docs", # "health": "/health" # } # } # @app.get("/health") # def health_check(): # return { # "status": "healthy", # "loaded_models": len(models), # "available_models": list(models.keys()) # } # @app.post("/predict/{model_name}", dependencies=[Depends(verify_token)]) # def predict(model_name: str, features: SongFeatures): # if model_name not in models: # raise HTTPException(status_code=404, detail="Model not found") # model = models[model_name] # input_array = np.array([[getattr(features, field) for field in features.model_fields]]) # try: # prediction = model.predict(input_array) # return { # "model": model_name, # "input": features.dict(), # "prediction": int(prediction[0]), # "prediction_label": "liked" if int(prediction[0]) == 1 else "not_liked" # } # except Exception as e: # raise HTTPException(status_code=500, detail=f"Prediction error: {str(e)}") # if __name__ == "__main__": # import uvicorn # uvicorn.run(app, host="0.0.0.0", port=7860) import time import pickle import numpy as np from fastapi import FastAPI, Request, status, HTTPException, Depends from fastapi.security.api_key import APIKeyHeader from fastapi.templating import Jinja2Templates from fastapi.responses import HTMLResponse from pydantic import BaseModel from typing import List import os import joblib app = FastAPI( title="ML Music Classifier API", description="Predict 'liked' or not from audio features using 8 ML models.", version="1.0.0" ) templates = Jinja2Templates(directory="templates") DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0" api_key_header = APIKeyHeader(name="X-Token", auto_error=False) @app.middleware("http") async def add_process_time_header(request: Request, call_next): start_time = time.perf_counter() response = await call_next(request) process_time = time.perf_counter() - start_time response.headers["X-Process-Time"] = str(process_time) return response def verify_token(x_token: str = Depends(api_key_header)): if x_token != DEFAULT_TOKEN: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or missing token. Use correct 'X-Token' in headers." ) class SongFeatures(BaseModel): danceability: float energy: float key: int loudness: float mode: int speechiness: float acousticness: float instrumentalness: float liveness: float valence: float tempo: float duration_ms: int time_signature: int MODEL_DIR = "models" model_names = [ "ANN_model", "knn_model", "logistic_regression_model", "neural_model", "Naive_Bayes_model", "random_forest_model", "svm_model", "XGBoost_model" ] models = {} for name in model_names: path = os.path.join(MODEL_DIR, f"{name}.pkl") try: models[name] = joblib.load(path) print(f"✅ Successfully loaded {name}") except Exception as e: print(f"❌ Error loading {name}: {e}") @app.get("/", response_class=HTMLResponse) def home(request: Request): return templates.TemplateResponse("index.html", { "request": request, "models": list(models.keys()) }) @app.get("/health") def health_check(): return { "status": "healthy", "loaded_models": len(models), "available_models": list(models.keys()) } @app.post("/predict/{model_name}", dependencies=[Depends(verify_token)]) def predict(model_name: str, features: SongFeatures): if model_name not in models: raise HTTPException(status_code=404, detail="Model not found") model = models[model_name] input_array = np.array([[getattr(features, field) for field in features.model_fields]]) try: prediction = model.predict(input_array) return { "model": model_name, "input": features.dict(), "prediction": int(prediction[0]), "prediction_label": "liked" if int(prediction[0]) == 1 else "Not liked" } except Exception as e: raise HTTPException(status_code=500, detail=f"Prediction error: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)