src/main.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
import pandas as pd
import pickle, uvicorn, os, logging

app = FastAPI()

# Configure logging
logging.basicConfig(level=logging.INFO)

# Define filepath for ml_components.pkl
ML_COMPONENTS_FILEPATH = os.path.join("assets", "ml", "ml_components.pkl")

# Load machine learning model and other components
with open(ML_COMPONENTS_FILEPATH, "rb") as file:
    ml_components = pickle.load(file)

# preprocessor = ml_components["preprocessor"]
pipeline = ml_components["pipeline"]


class DeviceSpecs(BaseModel):
    """
    Device specifications.

    - battery_power: Total energy a battery can store in one time measured in mAh
    - blue: Has Bluetooth or not (0 for False, 1 for True)
    - clock_speed: The speed at which the microprocessor executes instructions
    - dual_sim: Has dual sim support or not (0 for False, 1 for True)
    - fc: Front Camera megapixels
    - four_g: Has 4G or not (0 for False, 1 for True)
    - int_memory: Internal Memory in Gigabytes
    - m_dep: Mobile Depth in cm
    - mobile_wt: Weight of mobile phone
    - n_cores: Number of cores of the processor
    - pc: Primary Camera megapixels
    - px_height: Pixel Resolution Height
    - px_width: Pixel Resolution Width
    - ram: Random Access Memory in Megabytes
    - sc_h: Screen Height of mobile in cm
    - sc_w: Screen Width of mobile in cm
    - talk_time: longest time that a single battery charge will last when you are
    - three_g: Has 3G or not (0 for False, 1 for True)
    - touch_screen: Has touch screen or not (0 for False, 1 for True)
    - wifi: Has wifi or not (0 for False, 1 for True)
    """

    battery_power: float
    blue: int
    clock_speed: float
    dual_sim: int
    fc: float
    four_g: int
    int_memory: float
    m_dep: float
    mobile_wt: float
    n_cores: float
    pc: float
    px_height: float
    px_width: float
    ram: float
    sc_h: float
    sc_w: float
    talk_time: float
    three_g: int
    touch_screen: int
    wifi: int

    @validator("blue", "dual_sim", "four_g", "three_g", "touch_screen", "wifi")
    def validate_boolean(cls, v):
        # Ensure the values are either 0 or 1
        if v not in (0, 1):
            raise ValueError("Value must be 0 or 1")
        return v


@app.post("/predict/{device_id}")
async def predict_price(device_id: int, specs: DeviceSpecs):
    """
    Predict the price of a device based on its specifications.

    Args:
        device_id (int): The ID of the device.
        specs (DeviceSpecs): The device specifications.

    Returns:
        dict: A dictionary containing the input data and predicted price.
    """
    try:
        logging.info(f"Input request received...")

        # Preprocess the data
        data = pd.DataFrame([{"device_id": device_id, **specs.dict()}])
        logging.info(f"Input as a dataframe\n{data.to_markdown()}\n")

        # Predict price
        data["predicted_price"] = pipeline.predict(data)

        logging.info(
            f"Predictions made\n{data[['device_id', 'predicted_price']].to_markdown()}\n"
        )

        # Return input data and predicted price
        return data.to_dict("records")
    except Exception as e:
        logging.error(
            f"An error occurred while processing prediction for device ID {device_id}: {str(e)}"
        )
        raise HTTPException(status_code=500, detail=str(e))


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000, reload=True)