Spaces:
Sleeping
Sleeping
File size: 2,892 Bytes
3e9afea 04201e7 3e9afea 2824280 3e9afea 9ecea8c 2824280 da7ba77 8edafa7 3e9afea 04201e7 9ecea8c 3e9afea 1a01f13 9ecea8c 3e9afea 1a01f13 9ecea8c 3e9afea 9ecea8c 1a01f13 3e9afea 1a01f13 3e9afea 1a01f13 3e9afea 1a01f13 3e9afea 1a01f13 3e9afea 1a01f13 3e9afea | 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 | import os
import joblib
import numpy as np
from flask import Flask, request, jsonify
from supabase import create_client
SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_KEY = os.getenv("SUPABASE_KEY")
supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
# -----------------------------
# Load Model
# -----------------------------
model = joblib.load("predictive_model.pkl")
# -----------------------------
# Flask App
# -----------------------------
app = Flask(__name__)
# -----------------------------
# Health Check
# -----------------------------
@app.route("/", methods=["GET"])
def home():
return jsonify({
"status": "API is running",
"message": "Predictive Maintenance Backend Active"
})
# -----------------------------
# Prediction Endpoint
# -----------------------------
@app.route("/predict", methods=["POST"])
def predict():
try:
data = request.get_json()
# Extract features (must match training order)
features = [
data["Air_temperature"],
data["Process_temperature"],
data["Rotational_speed"],
data["Torque"],
data["Tool_wear"],
data["Type_L"],
data["Type_M"]
]
features_array = np.array([features])
prediction = model.predict(features_array)
probability = model.predict_proba(features_array)[0][1]
status_text = "Failure" if prediction[0] == 1 else "No Failure"
# -----------------------------
# Insert into Supabase
# -----------------------------
supabase.table("machine_logs").insert({
"air_temperature": data["Air_temperature"],
"process_temperature": data["Process_temperature"],
"rotational_speed": data["Rotational_speed"],
"torque": data["Torque"],
"tool_wear": data["Tool_wear"],
"type_l": data["Type_L"],
"type_m": data["Type_M"],
"prediction": int(prediction[0])
}).execute()
return jsonify({
"prediction": int(prediction[0]),
"status": status_text,
"failure_probability": float(round(probability, 4))
})
except Exception as e:
return jsonify({"error": str(e)}), 500
# -----------------------------
# Get Latest Logs (Dashboard)
# -----------------------------
@app.route("/logs", methods=["GET"])
def get_logs():
try:
response = supabase.table("machine_logs") \
.select("*") \
.order("created_at", desc=True) \
.limit(10) \
.execute()
return jsonify(response.data)
except Exception as e:
return jsonify({"error": str(e)}), 500
# -----------------------------
# Run App (for local testing)
# -----------------------------
if __name__ == "__main__":
app.run(host="0.0.0.0", port=7860) |