-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
71 lines (51 loc) · 1.7 KB
/
Copy pathmain.py
File metadata and controls
71 lines (51 loc) · 1.7 KB
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
from pathlib import Path
import joblib
import pandas as pd
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from schemas import SensorData
app = FastAPI(
title="Predictive Maintenance API",
description="API for predicting machine maintenance requirements",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
BASE_DIR = Path(__file__).resolve().parent
app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static")
MODEL_PATH = BASE_DIR / "models" / "maintenance_model.pkl"
PREPROCESSOR_PATH = BASE_DIR / "models" / "preprocessor.pkl"
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
model = joblib.load(MODEL_PATH)
preprocessor = joblib.load(PREPROCESSOR_PATH)
@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
return templates.TemplateResponse(
request=request,
name="index.html",
context={}
)
@app.post("/predict")
async def predict(data: SensorData):
input_df = pd.DataFrame([data.model_dump(by_alias=True)])
processed_data = preprocessor.transform(input_df)
prediction = int(model.predict(processed_data)[0])
failure_probability = round(
float(model.predict_proba(processed_data)[0][1] * 100),
4
)
status = (
"Maintenance required"
if prediction == 1
else "No maintenance required"
)
return {
"prediction": prediction,
"failure_probability": failure_probability,
"status": status
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)