Spaces:
Sleeping
Sleeping
File size: 4,537 Bytes
46695b5 |
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 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
import pandas as pd
import gradio as gr
from sklearn.preprocessing import StandardScaler
from sklearn.svm import OneClassSVM
# Function to detect theft and provide insights
def detect_theft_from_file(uploaded_file):
# Determine the file type and read it
if uploaded_file.name.endswith('.csv'):
df = pd.read_csv(uploaded_file.name)
elif uploaded_file.name.endswith(('.xlsx', '.xls')):
df = pd.read_excel(uploaded_file.name)
else:
return "Unsupported file format. Please upload a CSV or Excel file."
# Ensure required format (Consumer ID + 12 months + optional Last Year)
if len(df.columns) < 13:
return "The file must have at least 13 columns: 'Consumer ID', 12 months (e.g., Jan-Dec), and optionally 'Last Year Avg'."
feature_columns = df.columns[1:13] # 12 months
data = df[feature_columns].values
# Normalize the data
scaler = StandardScaler()
data_scaled = scaler.fit_transform(data)
# One-Class SVM model
model = OneClassSVM(nu=0.1, kernel='rbf', gamma='scale')
model.fit(data_scaled)
# Predict anomalies
predictions = model.predict(data_scaled)
# Add Reasons and Tips
reasons = []
tips = []
for i, pred in enumerate(predictions):
if pred == -1: # Theft detected
current_pattern = data[i]
if 'Last Year Avg' in df.columns:
last_year_avg = df['Last Year Avg'].iloc[i]
if current_pattern.mean() < 0.5 * last_year_avg:
reasons.append("Unusual drop compared to last year's average.")
elif current_pattern.mean() > 1.5 * last_year_avg:
reasons.append("Unusual rise compared to last year's average.")
else:
reasons.append("Irregular monthly pattern detected.")
else:
reasons.append("Irregular monthly pattern detected.")
tips.append(
"1. Inspect the meter physically for tampering.\n"
"2. Compare with neighborhood usage for similar consumption patterns.\n"
"3. Check if there are bypass connections or rewiring."
)
else: # No theft detected
reasons.append("No irregularities detected.")
tips.append("No action required.")
# Results
df['Anomaly'] = ['Theft' if pred == -1 else 'No Theft' for pred in predictions]
df['Reason'] = reasons
df['Tips'] = tips
return df[['Consumer ID', 'Anomaly', 'Reason', 'Tips']]
# Function for manual input detection
def detect_theft_from_manual_input(consumer_id, *monthly_kwh):
# Create a DataFrame from manual input
df = pd.DataFrame(
{
'Consumer ID': [consumer_id],
**{f'Month {i+1}': [monthly_kwh[i]] for i in range(12)},
}
)
# Normalize the data
data = df.iloc[:, 1:].values # Select monthly data
scaler = StandardScaler()
data_scaled = scaler.fit_transform(data)
# One-Class SVM model
model = OneClassSVM(nu=0.1, kernel='rbf', gamma='scale')
model.fit(data_scaled)
# Predict anomaly
prediction = model.predict(data_scaled)
# Add Reason and Tips
if prediction[0] == -1: # Theft detected
reason = "Irregular monthly pattern detected."
tips = (
"1. Inspect the meter physically for tampering.\n"
"2. Compare with neighborhood usage for similar consumption patterns.\n"
"3. Check if there are bypass connections or rewiring."
)
else:
reason = "No irregularities detected."
tips = "No action required."
df['Anomaly'] = ['Theft' if prediction[0] == -1 else 'No Theft']
df['Reason'] = [reason]
df['Tips'] = [tips]
return df[['Consumer ID', 'Anomaly', 'Reason', 'Tips']]
# Gradio Interface
file_interface = gr.Interface(
fn=detect_theft_from_file,
inputs=gr.File(label="Upload Meter Data File (CSV, XLSX, XLS)"),
outputs=gr.Dataframe(label="Meter Theft Detection")
)
manual_input_interface = gr.Interface(
fn=detect_theft_from_manual_input,
inputs=[
gr.Textbox(label="Consumer ID", placeholder="Enter Consumer ID"),
*[gr.Number(label=f"Month {i+1} kWh") for i in range(12)],
],
outputs=gr.Dataframe(label="Meter Theft Detection")
)
# Combine interfaces
iface = gr.TabbedInterface(
interface_list=[file_interface, manual_input_interface],
tab_names=["Upload File", "Manual Input"]
)
# Launch the Gradio app
iface.launch()
|