# -*- coding: utf-8 -*-
"""MiniProjectBDA_Review1.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1KQQu8_rON6eGoBJaX9ufEDEjuX_KJzN4

# **Title: Anomaly Detection for Energy Usage Optimization**

#**Problem Statement:**


The project aims to develop an anomaly detection model to predict whether the energy usage in a building is anomalous or not. The significance of this project lies in the fact that anomalous energy usage implies energy wastage, which can have both environmental and economic implications. By identifying and addressing such instances, we can significantly contribute to energy conservation and cost reduction.

# **Dataset Description**

**train.csv**

*building_id* - Unique building id code.

*timestamp* - When the measurement was taken

*meter_reading*- Electricity consumption in kWh.

*anomaly* - Whether this reading is anomalous (1) or not (0).
"""

!pip install pyspark

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, split, substring, when
from pyspark.sql.types import IntegerType
from datetime import datetime

spark = SparkSession.builder.appName("EnergyAnomalyDetection").getOrCreate()

train = spark.read.csv("train.csv", header=True, inferSchema=True)

train.show()

print("Shape:", (train.count(), len(train.columns)))

train.select([col(c).alias(c) for c in train.columns]).na.fill(0).show()

train = train.withColumn("new", split(train["timestamp"], " "))
train = train.withColumn("date", col("new")[0])
train = train.withColumn("time", substring(col("new")[1], 0, 2).cast(IntegerType()))
train = train.drop("new", "timestamp")
train.show()

from pyspark.sql.functions import mean
numeric_columns = [col_name for col_name, data_type in train.dtypes if data_type in ['double', 'float', 'int']]
mean_values = train.select([mean(col(column)).alias(column) for column in numeric_columns]).collect()[0].asDict()

for column in numeric_columns:
    train = train.withColumn(column, when(col(column).isNull(), mean_values[column]).otherwise(col(column)))

train.show()

train = train.withColumn("month", substring(col("date"), 6, 2).cast(IntegerType()))
train = train.withColumn("day", substring(col("date"), -2, 2).cast(IntegerType()))
train = train.drop("date")

train.show()

from pyspark.sql.functions import udf
import pyspark.sql.functions as F
@udf(IntegerType())
def weekend_or_weekday_udf(year, month, day):
    try:
        d = datetime(year, month, day)
        if d.weekday() > 4:
            return 1
        else:
            return 0
    except ValueError:
        return None

train = train.withColumn("weekend", weekend_or_weekday_udf(F.lit(2016), col("month"), col("day")))


train = train.withColumn("weekend", when(col("weekend").isNull(), 0).otherwise(col("weekend")))

train.show()

import matplotlib.pyplot as plt
from pyspark.sql import SparkSession
from pyspark.sql.functions import mean
import numpy as np

data = train.groupBy('weekend').agg(mean('meter_reading').alias('mean_meter_reading')).collect()

weekend_mean = data[1]['mean_meter_reading']
weekday_mean = data[0]['mean_meter_reading']

labels = ['Weekday Mean Usage', 'Weekend Mean Usage']
sizes = [weekday_mean, weekend_mean]
colors = plt.cm.Paired(np.arange(len(labels)))

plt.pie(sizes, labels=labels, colors=colors, autopct='%1.1f%%')
plt.axis('equal')

plt.show()

data = train.groupBy('month').agg(mean('meter_reading').alias('mean_meter_reading')).collect()

months = [row['month'] for row in data]
mean_readings = [row['mean_meter_reading'] for row in data]

plt.figure(figsize=(15, 5))
plt.bar(months, mean_readings)
plt.title('Mean usage monthly.', fontsize=20)
plt.xlabel('Month', fontsize=15)
plt.ylabel('Mean Meter Reading', fontsize=15)
plt.xticks(months)
plt.show()

data = train.groupBy('day').agg(mean('meter_reading').alias('mean_meter_reading')).collect()

days = [row['day'] for row in data]
mean_readings = [row['mean_meter_reading'] for row in data]

plt.figure(figsize=(15, 5))
plt.bar(days, mean_readings)
plt.title('Mean usage daily.', fontsize=20)
plt.xlabel('Day', fontsize=15)
plt.ylabel('Mean Meter Reading', fontsize=15)
plt.xticks(days)
plt.show()

neg = train.filter(train['anomaly'] == 0)
pos = train.filter(train['anomaly'] == 1)

neg_count = neg.count()
pos_count = pos.count()

print("Negative Shape:", neg_count)
print("Positive Shape:", pos_count)

train.show(5)

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.sql.types import IntegerType

inputColumns = ['building_id', 'meter_reading', 'time', 'month', 'day', 'weekend']
outputColumn = "anomaly"


for col_name in inputColumns:
    train = train.withColumn(col_name, train[col_name].cast(IntegerType()))

vector_assembler = VectorAssembler(inputCols=inputColumns, outputCol="features")

lr = LogisticRegression(labelCol=outputColumn, featuresCol="features")

stages = [vector_assembler, lr]
pipeline = Pipeline(stages=stages)


(train_df, test_df) = train.randomSplit([0.8, 0.2], seed=10)


pipeline_model = pipeline.fit(train_df)

predictions = pipeline_model.transform(test_df)

predictions.show(5)

from pyspark.ml.evaluation import BinaryClassificationEvaluator

evaluator = BinaryClassificationEvaluator(labelCol="anomaly")
accuracy = evaluator.evaluate(predictions)
print("Accuracy: ",accuracy)

pipeline_model.write().overwrite().save("/content/trained_model")

pip install gradio

import gradio as gr
from pyspark.ml import PipelineModel
from pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler


loaded_model = PipelineModel.load("/content/trained_model")

def predict(building_id, meter_reading, time, month, day, weekend):

    user_input = spark.createDataFrame([(building_id, meter_reading, time, month, day, weekend)],
                                       ["building_id", "meter_reading", "time", "month", "day", "weekend"])


    input_columns = ["building_id", "meter_reading", "time", "month", "day", "weekend"]
    assembler = VectorAssembler(inputCols=input_columns, outputCol="features")
    user_input = assembler.transform(user_input)
    try:
      prediction_result = predictions.select("prediction").first()[0]
    except Exception as e:
      print("Error occurred during prediction:", e)
      prediction_result = None
    return prediction_result


iface = gr.Interface(fn=predict,
                     inputs=["number", "number", "number", "number", "number", "number"],
                     outputs="label")


iface.launch()