aa

main.py

@@ -1,6 +1,5 @@
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
-import uvicorn
 import os
 import joblib
 import pandas as pd
@@ -10,85 +9,15 @@ app = FastAPI()
 
 @app.get("/")
 def read_root():
-    return {"message": "Welcome to the Taxi Fare Prediction API. Use POST /predict to get predictions."}
-
-# --- 1. Define Input and Output Data Models ---
-class InferenceInput(BaseModel):
-    vendor_id: str
-    dist_meters: float
-    wait_sec: float
-    geodetic_dist: float
-    mean_velocity: float
-    is_rush_hour: bool
-    model_name: str  # "bog", "mex", or "uio"
-
-class InferenceOutput(BaseModel):
-    trip_duration: float
-    model_used: str
-    message: str
-
-# --- 2. Define ML Model Manager Class ---
-class MLModels:
-    def __init__(self):
-        model_dir = "models"
-        self.bog_pipeline = self.load_pipeline(os.path.join(model_dir, "bog_ridge_pipeline.pkl"))
-        self.mex_pipeline = self.load_pipeline(os.path.join(model_dir, "mex_ridge_pipeline.pkl"))
-        self.uio_pipeline = self.load_pipeline(os.path.join(model_dir, "uio_ridge_pipeline.pkl"))
-
-    def load_pipeline(self, path):
-        if os.path.exists(path):
-            return joblib.load(path)
-        else:
-            raise FileNotFoundError(f"Model file not found: {path}")
-
-    def predict_one(self, features: dict, model_name: str):
-        model_map = {
-            "bog": self.bog_pipeline,
-            "mex": self.mex_pipeline,
-            "uio": self.uio_pipeline
-        }
-        pipeline = model_map.get(model_name.lower())
-        if not pipeline:
-            raise ValueError(f"Model '{model_name}' not found. Choose from 'bog', 'mex', or 'uio'.")
-        X_df = pd.DataFrame([features])
-        pred = pipeline.predict(X_df)[0]
-        return float(pred)
-
-# --- 3. Initialize ML Model Manager ---
-ml_models = MLModels()
-
-# --- 4. Define Inference Endpoint ---
-@app.post("/predict", response_model=InferenceOutput)
-async def predict_inference(data: InferenceInput):
-    try:
-        features = data.dict()
-        model_name = features.pop("model_name")
-        trip_duration = ml_models.predict_one(features, model_name)
-        return InferenceOutput(
-            trip_duration=trip_duration,
-            model_used=model_name,
-            message=f"Inference successful using {model_name.upper()} model."
-        )
-    except Exception as e:
-        raise HTTPException(status_code=400, detail=f"Inference failed: {str(e)}")
-
-#rebuild
-from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel
-import os
-import joblib
-import pandas as pd
-
-app = FastAPI()
-
-@app.get("/")
-def read_root():
-    return {"message": "Welcome to the Taxi Fare Prediction API. Use POST /predict to get predictions."}
+    return {
+        "message": "Welcome to the Taxi Fare Prediction API. Use POST /predict to get predictions."
+    }
 
 @app.get("/health")
 def health():
     return {"status": "ok"}
 
+# --- 1. Define Input and Output Data Models ---
 class InferenceInput(BaseModel):
     vendor_id: str
     dist_meters: float
@@ -103,28 +32,43 @@ class InferenceOutput(BaseModel):
     model_used: str
     message: str
 
+# --- 2. Lazy-loading ML Model Manager ---
 class MLModels:
     def __init__(self):
-        model_dir = "models"
         self.models = {}
-        for name in ["bog", "mex", "uio"]:
-            path = os.path.join(model_dir, f"{name}_ridge_pipeline.pkl")
-            try:
-                self.models[name] = joblib.load(path)
-                print(f"Loaded model: {path}")
-            except Exception as e:
-                print(f"Could not load model {path}: {e}")
+        self.model_dir = "models"
+        self.valid_models = {"bog", "mex", "uio"}
+
+    def get_model(self, model_name: str):
+        model_name = model_name.lower()
+        if model_name not in self.valid_models:
+            raise ValueError(f"Invalid model name '{model_name}'. Choose from 'bog', 'mex', or 'uio'.")
+
+        if model_name in self.models:
+            return self.models[model_name]
+
+        model_path = os.path.join(self.model_dir, f"{model_name}_ridge_pipeline.pkl")
+        if not os.path.exists(model_path):
+            raise FileNotFoundError(f"Model file not found: {model_path}")
+
+        try:
+            model = joblib.load(model_path)
+            self.models[model_name] = model
+            print(f"✅ Loaded model '{model_name}' from {model_path}")
+            return model
+        except Exception as e:
+            raise RuntimeError(f"Error loading model '{model_name}': {e}")
 
     def predict_one(self, features: dict, model_name: str):
-        model = self.models.get(model_name.lower())
-        if not model:
-            raise ValueError(f"Model '{model_name}' not found. Choose from 'bog', 'mex', or 'uio'.")
+        model = self.get_model(model_name)
         X_df = pd.DataFrame([features])
         pred = model.predict(X_df)[0]
         return float(pred)
 
+# --- 3. Instantiate ML Model Manager ---
 ml_models = MLModels()
 
+# --- 4. Inference Endpoint ---
 @app.post("/predict", response_model=InferenceOutput)
 async def predict_inference(data: InferenceInput):
     try:
@@ -136,9 +80,11 @@ async def predict_inference(data: InferenceInput):
             model_used=model_name,
             message=f"Inference successful using {model_name.upper()} model."
         )
+    except FileNotFoundError as fnf:
+        raise HTTPException(status_code=404, detail=str(fnf))
+    except ValueError as ve:
+        raise HTTPException(status_code=400, detail=str(ve))
     except Exception as e:
-        raise HTTPException(status_code=400, detail=f"Inference failed: {str(e)}")
-      
-    
-if __name__ == "__main__":
-    uvicorn.run("your_module:app", host="0.0.0.0", port=8000, reload=True)
+        raise HTTPException(status_code=500, detail=f"Inference failed: {str(e)}")
+
+# --- 5. Run the app ---