Upload 12 files

Electricity Cost Prediction API

Project Overview ->
This Space hosts a Machine Learning API for predicting electricity costs, built with FastAPI and deployed using Docker.
It provides a robust API endpoint to estimate the total electricity cost based on various facility and operational parameters. It's designed to help understand and predict energy expenditure by leveraging a trained machine learning model.

Key Features->

Predictive API : Offers a `/predict` endpoint to receive input data and return an estimated electricity cost
Data Preprocessing : Handles data cleaning (imputation), categorical encoding (Label Encoding for `structure_type`), and numerical scaling (StandardScaler) automatically before prediction
Dockerized Deployment : Packaged as a Docker container for consistent and reproducible deployment across different environments
FastAPI Framework : Built on FastAPI, providing high performance, easy-to-use API development, and automatic interactive documentation (Swagger UI)

Input Features->

The API expects a JSON payload with the following parameters :

* `site_area` (float): Area of the site in square units.
* `structure_type` (string): Type of structure (e.g., "residential", "commercial").
* `water_consumption` (float): Daily/monthly water consumption.
* `recycling_rate` (float): Percentage of waste recycled.
* `utilisation_rate` (float): Rate of facility utilization.
* `air_qality_index` (float): Air quality index.
* `issue_reolution_time` (float): Time taken to resolve issues (e.g., in hours).
* `resident_count` (integer): Number of residents/occupants.

How to Use the API?
You can interact with the API directly from the automatically generated Swagger UI.

1. Navigate to your Space's URL (e.g., `https://huggingface.co/spaces/<your-username>/<your-space-name>`).
2. Append `/docs` to the URL to access the interactive API documentation :
`https://huggingface.co/spaces/<your-username>/<your-space-name>/docs`
3. Expand the `/predict` endpoint
4. Click "Try it out"
5. Enter a JSON payload in the "Request body" field with sample data->
json
{
"site_area": 1850.0,
"structure_type": "residential",
"water_consumption": 15300.0,
"recycling_rate": 52.3,
"utilisation_rate": 81.6,
"air_qality_index": 39.0,
"issue_reolution_time": 2.7,
"resident_count": 320
}
6. Click "Execute" to get a prediction.

Files in this Space ->

`Dockerfile`: Defines the Docker image for the FastAPI application
`main.py`: The core FastAPI application logic
`preprocessing.py`: Contains the data preprocessing functions and loads the fitted transformers
`train_and_save_model.py`: Script used to train the ML model and save all necessary preprocessing transformers
`requirements.txt`: Lists all Python dependencies
`.dockerignore`: Specifies files to exclude from the Docker build
`model.pkl`: The saved trained machine learning model
`numerical_imputer.pkl`: Saved `SimpleImputer` for numerical features
`categorical_imputer.pkl`: Saved `SimpleImputer` for categorical features
`label_encoder_structure_type.pkl`: Saved `LabelEncoder` for `structure_type`
`scaler.pkl`: Saved `StandardScaler` for numerical features
`electricity_cost_dataset.csv.xlsx`: The Kaggle dataset used for training

.dockerignore

@@ -0,0 +1,7 @@
+.git             
+.venv            
+__pycache__      
+*.pyc            
+*.ipynb          
+.DS_Store        
+*.log            

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+electricity_cost_dataset.csv.xlsx filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.9-slim-buster
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY . .
+EXPOSE 8000
+CMD ["gunicorn", "main:app", "--workers", "4", "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

Preprocessing.py

@@ -0,0 +1,144 @@
+#Now, comes another important task that is Preprocessing the data
+import pandas as pd
+import joblib
+import os
+from sklearn.impute import SimpleImputer
+from sklearn.preprocessing import LabelEncoder, StandardScaler
+
+NUM_IMPUTER_PATH = "numerical_imputer.pkl"
+CAT_IMPUTER_PATH = "categorical_imputer.pkl"
+LE_STRUCTURE_TYPE_PATH = "label_encoder_structure_type.pkl"
+SCALER_PATH = "scaler.pkl"
+
+numerical_imputer = None
+categorical_imputer = None
+le_structure_type = None
+scaler = None
+#I have done this to set them as a placeholder in this file....therefore no discrepancies related to it will occur
+
+try:
+    numerical_imputer = joblib.load(NUM_IMPUTER_PATH)
+    print(f"Loaded {NUM_IMPUTER_PATH}. Expected features: {getattr(numerical_imputer, 'feature_names_in_', 'N/A')}")
+except FileNotFoundError :
+    print(f"Warning : {NUM_IMPUTER_PATH} not found")
+except Exception as e :
+    print(f"Error loading {NUM_IMPUTER_PATH}: {e}")
+
+try:
+    categorical_imputer = joblib.load(CAT_IMPUTER_PATH)
+    print(f"Loaded {CAT_IMPUTER_PATH}. Expected features: {getattr(categorical_imputer, 'feature_names_in_', 'N/A')}")
+except FileNotFoundError :
+    print(f"Warning: {CAT_IMPUTER_PATH} not found")
+except Exception as e :
+    print(f"Error loading {CAT_IMPUTER_PATH}: {e}")
+
+try:
+    le_structure_type = joblib.load(LE_STRUCTURE_TYPE_PATH)
+    print(f"Loaded {LE_STRUCTURE_TYPE_PATH}")
+except FileNotFoundError :
+    print(f"Warning: {LE_STRUCTURE_TYPE_PATH} not found")
+except Exception as e :
+    print(f"Error loading {LE_STRUCTURE_TYPE_PATH}: {e}")
+
+try:
+    scaler = joblib.load(SCALER_PATH)
+    print(f"Loaded {SCALER_PATH}. Expected features: {getattr(scaler, 'feature_names_in_', 'N/A')}")
+except FileNotFoundError :
+    print(f"Warning: {SCALER_PATH} not found")
+except Exception as e :
+    print(f"Error loading {SCALER_PATH}: {e}")
+
+#You can see that I've used the try and except model for loading the data so that if error occurs I'm completely aware of it
+
+NUMERICAL_FEATURES = [
+    'site_area', 'water_consumption', 'recycling_rate', 'utilisation_rate', 'air_qality_index', 'issue_reolution_time', 'resident_count'
+]
+CATEGORICAL_FEATURES = ['structure_type']
+
+FINAL_MODEL_EXPECTED_FEATURES = [
+    'site_area', 'structure_type', 'water_consumption', 'recycling_rate', 'utilisation_rate', 'air_qality_index', 'issue_reolution_time', 'resident_count'
+] 
+#Final model expected features contains the list of the final output of the trained data
+
+#Now, our input will most likely be a dictionary...but for MLOps we would be needing a Pandas datframe so I converted this input dictionary into a dataframe and then returned it to my function after performing operation ->
+
+def preprocess_input(input_data: dict) -> pd.DataFrame:
+    df_processed = pd.DataFrame([input_data])
+    print(f"DataFrame after initial creation (df_processed)-> \n{df_processed}")
+
+    if 'structure_type' in df_processed.columns:
+        df_processed['structure_type'] = df_processed['structure_type'].astype(str).str.lower().str.strip()
+        print(f"'structure_type' standardized to: '{df_processed['structure_type'].iloc[0]}'")
+    
+    if numerical_imputer is not None and NUMERICAL_FEATURES:
+        missing_input = [col for col in NUMERICAL_FEATURES if col not in df_processed.columns] 
+        
+        if missing_input:
+            raise ValueError(f"Error : Numerical features {missing_input} are missing from input DataFrame!")
+        #This is only to verify...It will give us the missing columns which should  be present while doing numerical imputation....basically, I'm trying to handle all the errors possible
+        
+        try:
+            df_processed[NUMERICAL_FEATURES] = numerical_imputer.transform(df_processed[NUMERICAL_FEATURES])
+
+        except Exception as e:
+            raise RuntimeError(
+                f"Error during numerical imputation\n"
+                f"Error : {e}"
+            )
+            #raise functions are best here because as soon as the error occurs....it will stop the function
+
+    if categorical_imputer is not None and CATEGORICAL_FEATURES:
+        missing_input = [col for col in CATEGORICAL_FEATURES if col not in df_processed.columns]
+        
+        if missing_input:
+            raise ValueError(f"Error : Categorical features {missing_input} are missing from input DataFrame!")
+        try:
+            df_processed[CATEGORICAL_FEATURES] = categorical_imputer.transform(df_processed[CATEGORICAL_FEATURES])
+    
+        except Exception as e:
+            raise RuntimeError(
+                f"Error during categorical imputation\n"
+                f"Error : {e}"
+            )
+
+    if le_structure_type is not None and 'structure_type' in df_processed.columns:
+        try:
+            df_processed['structure_type'] = le_structure_type.transform(df_processed['structure_type'])
+        except ValueError as e:
+            raise ValueError(
+                f"Unknown category -> '{df_processed['structure_type'].iloc[0]}' in column 'structure_type'\n"
+                f"Error : {e}"
+            )
+        except Exception as e:
+            raise RuntimeError(f"Error during Label Encoding for 'structure_type'...Error: {e}")
+
+    if scaler is not None and NUMERICAL_FEATURES:
+        missing_input = [col for col in NUMERICAL_FEATURES if col not in df_processed.columns]
+        
+        if missing_input:
+            raise ValueError(f"Error : Numerical features {missing_input} are missing from input DataFrame")
+        
+        try:
+            df_processed[NUMERICAL_FEATURES] = scaler.transform(df_processed[NUMERICAL_FEATURES])
+            
+        except Exception as e:
+            raise RuntimeError(
+                f"Error during scaling\n"
+                f"Error: {e}"
+            )
+
+    print(f"Current df_processed columns before final reorder: {df_processed.columns.tolist()}")
+    #Checkpoint
+
+    for col in FINAL_MODEL_EXPECTED_FEATURES:
+        if col not in df_processed.columns:
+            print(f"Adding missing column: '{col}' with value 0.")
+            df_processed[col] = 0
+
+    df_final = df_processed[FINAL_MODEL_EXPECTED_FEATURES] 
+    print(f"Final DataFrame for prediction: \n{df_final}")
+
+    return df_final
+
+#The function I created above was based upon the numerical and categorical imputation, label encoding, scaling or basically all the data preprocessing that should be done after training all the models.....
+#I have show all the error messages in my coding lines because I got stuck in this process many time and to highlight the mistakes I have created some checkpoints also in between....Therefore, now all the data operations are done and the next thing is DEPLOYMENT-> creation of FastAPI and deployment on AWS etc.

categorical_imputer.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:ad3f6b2271f26471c7eb5abc2e80538142161a45503c29d0258c2c79021c0f5f
+size 874

electricity_cost_dataset.csv.xlsx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:36ed22e74eed1f3efa0e8633b8a84ff898f05581078f1f4d9079c3dc728c518f
+size 539799

label_encoder_structure_type.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:51543e74e42d53fb0c16cd39c1abd249c678b30409387c217660c3f47c110cac
+size 525

main.py

@@ -0,0 +1,64 @@
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+import joblib
+import pandas as pd
+import os
+
+from Preprocessing import preprocess_input
+
+app = FastAPI(
+    title="Electricity Cost Prediction API",
+    description="Predicts electricity cost based on facility and operational parameters"
+)
+
+MODEL_PATH = "model.pkl"
+
+if not os.path.exists(MODEL_PATH):
+    raise FileNotFoundError(
+        "Model file not found"
+    )
+try:
+    model = joblib.load(MODEL_PATH)
+except Exception as e:
+    raise RuntimeError(f"Error loading model from {MODEL_PATH}: {e}")
+
+class ElectricityInput(BaseModel):
+    site_area: float = Field(..., description="Area of the site in square units")
+    structure_type: str = Field(..., description="Type of structure (e.g., 'residential', 'commercial')")
+    water_consumption: float = Field(..., description="Daily/monthly water consumption")
+    recycling_rate: float = Field(..., description="Percentage of waste recycled")
+    utilisation_rate: float = Field(..., description="Rate of facility utilization")
+    air_qality_index: float = Field(..., description="Air quality index")
+    issue_reolution_time: float = Field(..., description="Time taken to resolve issues (e.g., in hours)")
+    resident_count: int = Field(..., description="Number of residents/occupants")
+
+#Basically all these inputs for the base model will be converted into pydantic object after checking, optimizing and safety assurance 
+
+#Moving to main API code-> Using the predict model for my task
+@app.post("/predict")
+async def predict_electricity_cost(data: ElectricityInput):
+    print("Predicts the total electricity cost based on the provided input features")
+    
+    try:
+        input_data_dict = data.model_dump() 
+        processed_df = preprocess_input(input_data_dict)
+        prediction = model.predict(processed_df)[0]
+        predicted_cost = round(float(prediction), 2)
+        return {"predicted_electricity_cost": predicted_cost}
+
+    except Exception as e:
+        print(f"An unexpected error occurred during prediction: {e}")
+        raise HTTPException(
+            status_code=500, #Internal server error....Basically giving error coverups
+            detail=f"An internal server error occurred during prediction. Error : {e}"
+        )
+
+#My overall work in this API File ->
+#User input : JSON -> Pydantic object -> dictionary -> DataFrame -> model.predict()
+
+@app.get("/health")
+async def health_check():  
+    #Using asyn function, because it can pause wherever the function needs to run a different block of code and  then restart again
+    return {"status": "ok", "message": "Electricity Cost Prediction API is running accurately!"}
+
+#So, this was the overall API implementation.....Also, I've created the docker and .dockerignore files in this folder to package my work and deploy it....basically storing it in a container...You can see that I've marked up all error points as much as possible to resolve all the incoming issues fast

model.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:490f135e4a7667db37a9cc7aa351f49fc47c78dc19b71652081d314723d3bc76
+size 1081

numerical_imputer.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:cbfc3ab70a663e882e5365dee33ebce4df6bc5024e55fc5beaffd0773134387b
+size 911

requirements.txt

@@ -0,0 +1,8 @@
+fastapi
+uvicorn
+pandas
+scikit-learn 
+joblib
+pydantic
+gunicorn
+openpyxl 

scaler.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:002fa9d924b015f36be4d9d1b34d407183f47641afd48623be69ee5134122a1d
+size 1151

train_and_save_model.py

@@ -0,0 +1,128 @@
+#Next Task -> Training the dataset
+#In this file I've done Training and encoding the dataset
+#Now as I've already done the EDA...the next task is to train and save the data for preprocessing
+
+import pandas as pd
+from sklearn.impute import SimpleImputer
+from sklearn.preprocessing import LabelEncoder, StandardScaler
+from sklearn.model_selection import train_test_split
+from sklearn.linear_model import LinearRegression
+import joblib
+import os  
+import re 
+
+DATASET_PATH = "C:/Users/kavya/Documents/GDG_Files_Kavya/electricity_predictor_API/electricity_cost_dataset.csv.xlsx"
+MODEL_OUTPUT_DIR = "."  
+
+os.makedirs(MODEL_OUTPUT_DIR, exist_ok=True)
+
+def RenamingColumns(Column_Name):
+    Column_Name = re.sub(r'\s+', '_', Column_Name) 
+    Column_Name = re.sub(r'[^\w_]', '', Column_Name) 
+    return Column_Name.lower()   
+
+try:
+    df = pd.read_excel(DATASET_PATH)
+    print("Original columns ->\n")
+    print(df.columns.tolist())
+
+    new_columns = []
+    
+    #As I've to rename the columns....I'm using a for loop to do this->
+    #If, the column names given as an input in the FastAPI are not same as the column names in the dataset...an error will be occured on the fastAPI application
+    
+    for col in df.columns:
+        new_col = RenamingColumns(col)
+        new_columns.append(new_col)
+
+    df.columns = new_columns
+
+    print("Renamed Columns ->\n")
+    print(df.columns.tolist())
+
+except FileNotFoundError:
+    print(f"Error: Dataset not found! Please ensure the file is in the same directory")
+    exit()
+    
+except Exception as e:
+    print(f"Error : {e}")
+    exit()
+    
+#I used try and except blocks for ERROR HANDLING
+#Now, all the names have been changed and I've converted same as the datset ones...Therefor from here, I've used new names
+
+TARGET_COL = 'electricity_cost' 
+
+if TARGET_COL not in df.columns:
+    print(f"Error: Target column '{TARGET_COL}' not found!")
+    exit()
+
+features_df = df.drop(columns=[TARGET_COL]) 
+#Using .drop, I removed the feature which will not be used in calculation
+y = df[TARGET_COL]
+
+NUMERICAL_FEATURES = [
+    'site_area', 'water_consumption', 'recycling_rate', 'utilisation_rate',
+    'air_qality_index', 'issue_reolution_time', 'resident_count'
+]
+CATEGORICAL_FEATURES = ['structure_type']
+
+all_expected_features = NUMERICAL_FEATURES + CATEGORICAL_FEATURES
+missing_features = [col for col in all_expected_features if col not in features_df.columns]
+
+if missing_features:
+    print(f"Error: The following expected features are missing from the data after renaming: {missing_features}")
+    exit()
+#The above steps were only for the safety purpose...to recheck if there is any missing features.
+#Actually, I did it just because I was facing many errors...therefore just to check I added some checkpoints.
+
+numerical_imputer = SimpleImputer(strategy='mean')
+if NUMERICAL_FEATURES:
+    features_df[NUMERICAL_FEATURES] = numerical_imputer.fit_transform(features_df[NUMERICAL_FEATURES])
+    joblib.dump(numerical_imputer, os.path.join(MODEL_OUTPUT_DIR, 'numerical_imputer.pkl'))
+    print("Numerical imputer fitted and saved")
+else:
+    print("No numerical columns to impute")
+
+categorical_imputer = SimpleImputer(strategy='most_frequent')
+if CATEGORICAL_FEATURES:
+    features_df[CATEGORICAL_FEATURES] = categorical_imputer.fit_transform(features_df[CATEGORICAL_FEATURES])
+    joblib.dump(categorical_imputer, os.path.join(MODEL_OUTPUT_DIR, 'categorical_imputer.pkl'))
+    print("Categorical imputer fitted and saved")
+else:
+    print("No categorical columns to impute")
+#I used joblib because I wanted to use this data later as well...therefore, whenever I will be in need of it I will load this with joblib.load()
+
+if 'structure_type' in features_df.columns:
+    features_df['structure_type'] = features_df['structure_type'].astype(str).str.lower().str.strip()
+    le_structure_type = LabelEncoder()
+    features_df['structure_type'] = le_structure_type.fit_transform(features_df['structure_type'])
+    joblib.dump(le_structure_type, os.path.join(MODEL_OUTPUT_DIR, 'label_encoder_structure_type.pkl'))
+    print("LabelEncoder for 'structure_type' fitted and saved.")
+else:
+    print("structure_type column not found or not categorical, skipping LabelEncoder.")
+
+if NUMERICAL_FEATURES:
+    scaler = StandardScaler()
+    features_df[NUMERICAL_FEATURES] = scaler.fit_transform(features_df[NUMERICAL_FEATURES])
+    joblib.dump(scaler, os.path.join(MODEL_OUTPUT_DIR, 'scaler.pkl'))
+    print("StandardScaler fitted and saved.")
+else:
+    print("No numerical columns to scale.")
+
+#You can see that, I've used joblib.dump to create a separate directory for each imputer and encoder made
+
+X = features_df
+y = df[TARGET_COL]
+
+X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
+
+model = LinearRegression() 
+model.fit(X_train, y_train)
+joblib.dump(model, os.path.join(MODEL_OUTPUT_DIR, 'model.pkl'))
+
+FINAL_MODEL_EXPECTED_FEATURES = X_train.columns.tolist()
+print("All expected features from Final Model->\n")
+print(FINAL_MODEL_EXPECTED_FEATURES)
+
+#So, now, all necessary .pkl files created and saved in the current directory