#!

/usr/bin/env python3
r"""
failure_analyzer_windows.py

Windows-ready preventive maintenance analysis tool tailored to CSVs with these

columns:
Created, Issue, Raised by, Entry Type, Resolved ?, Prod. Status, Attachments,
Image

Default folder set to:

C:\Users\fhussain\Downloads\Failure points

Features:
- Load all CSVs in the folder
- Parse 'Created' timestamp, extract issue text
- Compute per-component / per-reporter stats (here we treat 'Prod. Status' as
severity marker)
- TF-IDF + KMeans clustering on issue text to surface recurring failure groups
- IsolationForest anomaly detection for unusual events
- MTBF-like and downtime statistics (if downtime present)
- Tkinter UI for running analysis and exporting CSV/text reports
"""

import os
import glob
import traceback
from datetime import timedelta
import tkinter as tk
from tkinter import filedialog, messagebox, ttk
import pandas as pd
import numpy as np
import matplotlib
[Link]("TkAgg")
from [Link] import Figure
from [Link].backend_tkagg import FigureCanvasTkAgg

try:
from sklearn.feature_extraction.text import TfidfVectorizer
from [Link] import KMeans
from [Link] import IsolationForest
from [Link] import PCA
SKLEARN_AVAILABLE = True
except Exception:
# scikit-learn not installed / import failed — proceed but disable ML features
TfidfVectorizer = None
KMeans = None
IsolationForest = None
PCA = None
SKLEARN_AVAILABLE = False

# --------- CONFIG ----------

DEFAULT_FOLDER = r"C:\Users\fhussain\Downloads\Failure points"
OUTPUT_FOLDER_DEFAULT = [Link]([Link]("~"),
"failure_analyzer_output")
# --------------------------

COMMON_DATE_COLS = ['created', 'date', 'timestamp', 'time']

def list_csv_files(folder):
 pattern = [Link](folder, "*.csv")
return sorted([Link](pattern))

def load_and_concatenate(folder):
paths = list_csv_files(folder)
if not paths:
raise FileNotFoundError(f"No CSV files found in '{folder}'")
dfs = []
for p in paths:
try:
df = pd.read_csv(p, low_memory=False)
src = [Link](p)
df['_source_file'] = src
# machine name: filename without extension
df['_machine'] = [Link](src)[0]
[Link](df)
except Exception as e:
print(f"Warning: failed to read {p}: {e}")
if not dfs:
raise ValueError("No readable CSVs found.")
big = [Link](dfs, ignore_index=True, sort=False)
return big

def detect_column_name(df, candidates):

lower_to_col = {[Link]().strip(): c for c in [Link]}
for cand in candidates:
if cand in lower_to_col:
return lower_to_col[cand]
# try fuzzy: look if any column contains the candidate substring
for c_low, c_orig in lower_to_col.items():
for cand in candidates:
if cand in c_low:
return c_orig
return None

def preprocess(df):
df = [Link]()
# detect timestamp column - prefer exact 'Created'
ts_col = None
if 'Created' in [Link]:
ts_col = 'Created'
else:
ts_col = detect_column_name(df, [[Link]() for c in COMMON_DATE_COLS])

# detect text columns

issue_col = detect_column_name(df, ['issue','description','failure','issue
description','details']) or 'Issue'
raised_by_col = detect_column_name(df, ['raised by','reporter','reported_by'])
or 'Raised by'
status_col = detect_column_name(df, ['prod. status','prod
status','status','prod_status']) or 'Prod. Status'
resolved_col = detect_column_name(df,
['resolved ?','resolved','resolved?','resolved_flag']) or 'Resolved ?'
attachment_col = detect_column_name(df, ['attachments','attachment']) or
'Attachments'
image_col = detect_column_name(df, ['image','images']) or 'Image'

# parse timestamp
if ts_col in [Link]:
 df['_timestamp'] = pd.to_datetime(df[ts_col], errors='coerce',
infer_datetime_format=True)
else:
df['_timestamp'] = [Link]

# create safe text fields

df['_issue_text'] = df[issue_col].fillna('').astype(str) if issue_col in
[Link] else ''
df['_raised_by'] = df[raised_by_col].fillna('').astype(str) if raised_by_col in
[Link] else ''
df['_prod_status'] = df[status_col].fillna('').astype(str) if status_col in
[Link] else ''
df['_resolved'] = df[resolved_col].fillna('').astype(str) if resolved_col in
[Link] else ''
df['_attachments'] = df[attachment_col].fillna('').astype(str) if
attachment_col in [Link] else ''
df['_image'] = df[image_col].fillna('').astype(str) if image_col in [Link]
else ''

# derive weekday/hour
df['_weekday'] = df['_timestamp'].dt.day_name()
df['_hour'] = df['_timestamp'].[Link]

# normalized resolved boolean

df['_resolved_bool'] =
df['_resolved'].astype(str).[Link]().isin(['yes','y','true','resolved','1'])
# prepare follow-up resolution marker (if a later 'yes' exists for the same
machine+issue, the earlier event is considered resolved)
df['_resolved_followup'] = False
# ensure machine column exists (load_and_concatenate should set it)
if '_machine' not in [Link]:
df['_machine'] = [Link]('_source_file', '').astype(str).apply(lambda x:
[Link]([Link](x))[0] if x else '')

# detect follow-up resolution: for each machine+issue, if any later record is

resolved, mark earlier unresolved as resolved_by_followup
try:
# sort by timestamp to detect chronological order
df =
df.sort_values(['_machine','_issue_text','_timestamp']).reset_index(drop=True)
grp = [Link](['_machine','_issue_text'])
for (_, _), g in grp:
# indices in chronological order
idxs = [Link]()
resolved_flags = g['_resolved_bool'].tolist()
# if there's any True later, mark preceding False as followup resolved
for i, val in enumerate(resolved_flags):
if not val:
# if any later position has True
if any(resolved_flags[i+1:]):
[Link][idxs[i], '_resolved_followup'] = True
except Exception:
# if any error, leave followup flags False
pass

# severity tag from Prod. Status (UP -> low, DOWN -> high, PM -> medium, QUAL -
> medium)
def map_sev(s):
s_low = str(s).strip().upper()
 if s_low.startswith('DOWN'):
return 'HIGH'
if s_low.startswith('UP'):
return 'LOW'
if s_low.startswith('PM') or s_low.startswith('QUAL'):
return 'MEDIUM'
return 'UNKNOWN'
df['_severity'] = df['_prod_status'].apply(map_sev)

return df, {
'ts_col': ts_col,
'issue_col': issue_col,
'raised_by_col': raised_by_col,
'status_col': status_col,
'resolved_col': resolved_col
}

def compute_basic_metrics(df):
# counts overall and unresolved (respect follow-up resolution)
total_events = len(df)
# treat record as resolved if resolved_bool OR resolved_followup is True
resolved_mask = [Link]('_resolved_bool', [Link](False, index=[Link])) |
[Link]('_resolved_followup', [Link](False, index=[Link]))
unresolved = df[~resolved_mask]
unresolved_count = len(unresolved)
per_reporter =
[Link]('_raised_by').size().reset_index(name='count').sort_values('count',
ascending=False)
per_status =
[Link]('_prod_status').size().reset_index(name='count').sort_values('count',
ascending=False)
per_severity =
[Link]('_severity').size().reset_index(name='count').sort_values('count',
ascending=False)

# weekday/hour pivot — use the cleaned '_issue_text' (safer than relying on

original 'Issue' column)
ts_df = df[df['_timestamp'].notna()].copy()
if not ts_df.empty:
pivot = pd.pivot_table(ts_df, index='_weekday', columns='_hour',
values='_issue_text', aggfunc='count', fill_value=0)
# ensure consistent weekday order
weekdays =
['Monday','Tuesday','Wednesday','Thursday','Friday','Saturday','Sunday']
pivot = [Link](weekdays).fillna(0)
# ensure all hour columns 0..23 exist and are ordered
all_hours = list(range(24))
for h in all_hours:
if h not in [Link]:
pivot[h] = 0
pivot = pivot[all_hours]
else:
# empty pivot with expected columns
pivot = [Link](0,
index=['Monday','Tuesday','Wednesday','Thursday','Friday','Saturday','Sunday'],
columns=list(range(24)))

return {
'total_events': total_events,
 'unresolved_count': unresolved_count,
'per_reporter': per_reporter,
'per_status': per_status,
'per_severity': per_severity,
'weekday_hour': pivot
}

# --- new: analyze machines to find prone machines and simple upcoming failure
heuristic ---
def analyze_machines(df):
now = [Link]()
machines = []
for m, g in [Link]('_machine'):
cnt = len(g)
# unresolved respecting followup
resolved_mask = [Link]('_resolved_bool', [Link](False, index=[Link])) |
[Link]('_resolved_followup', [Link](False, index=[Link]))
unresolved_cnt = int((~resolved_mask).sum())
# top issues
top_issues = g['_issue_text'].value_counts().head(3).[Link]()
upcoming = []
# for each recurring issue compute median interval and days since last;
flag upcoming when last >= 0.8*median_interval
for issue, ig in [Link]('_issue_text'):
if len(ig) < 2:
continue
times = pd.to_datetime(ig['_timestamp'].dropna()).sort_values()
if len(times) < 2:
continue
diffs = [Link]().dt.total_seconds() / (3600*24)
med = float([Link]().median()) if not [Link]().empty else
None
if med and med > 0:
days_since_last = (now - [Link]()).total_seconds() / (3600*24)
# heuristic: if it's close to expected recurrence -> likely
upcoming
if days_since_last >= (0.8 * med):
[Link]({'issue': issue, 'median_days': med,
'days_since_last': days_since_last})
[Link]({
'machine': m,
'count': int(cnt),
'unresolved': int(unresolved_cnt),
'top_issues': top_issues,
'upcoming': upcoming
})
# sort machines by count desc
machines = sorted(machines, key=lambda x: x['count'], reverse=True)
return machines

def cluster_issues(df, n_clusters=6, max_features=1500):

# if sklearn is not available, return safe defaults (no clustering)
if not SKLEARN_AVAILABLE:
print("Warning: scikit-learn not available — clustering disabled. Install
with: pip install scikit-learn")
return [Link]([-1] * len(df), index=[Link]), None

texts = df['_issue_text'].fillna('').astype(str).values
non_empty = [t for t in texts if [Link]()]
 if len(non_empty) < 2:
return [Link]([-1] * len(df), index=[Link]), None

vect = TfidfVectorizer(max_features=max_features, stop_words='english',

ngram_range=(1,2))
X = vect.fit_transform(texts)
sample_count = [Link][0]
n_clusters = min(n_clusters, max(2, max(2, int(sample_count / 5))))
n_clusters = min(n_clusters, sample_count)
if n_clusters < 2:
return [Link]([-1] * len(df), index=[Link]), None

km = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)

labels = km.fit_predict(X)
# PCA for 2D plot (small data - safe to convert to array)
try:
pca = PCA(n_components=2, random_state=42)
Xred = pca.fit_transform([Link]())
except Exception:
Xred = None
return [Link](labels, index=[Link]), {'kmeans': km, 'vectorizer': vect,
'Xred': Xred}

def detect_anomalies(df):
# if sklearn is not available, skip anomaly detection
if not SKLEARN_AVAILABLE:
print("Warning: scikit-learn not available — anomaly detection disabled.
Install with: pip install scikit-learn")
return [Link]([False] * len(df), index=[Link]), None

feat = [Link]({
'hour': df['_hour'].fillna(-1),
'text_len': df['_issue_text'].[Link]().fillna(0),
# severity numeric
'sev': df['_severity'].map({'LOW':0,'MEDIUM':1,'HIGH':2}).fillna(0)
})
if len(feat) < 10:
return [Link]([False]*len(df), index=[Link]), None
iso = IsolationForest(contamination=0.02, random_state=42)
try:
pred = iso.fit_predict(feat)
anomalies = [Link](pred == -1, index=[Link])
return anomalies, iso
except Exception as e:
print("Anomaly detection failed:", e)
return [Link]([False]*len(df), index=[Link]), None

def generate_recommendations(df, metrics, clusters_labels, anomalies,

machine_analysis=None):
recs = []
# Machine-aware recommendations (prioritize by unresolved + upcoming)
if machine_analysis:
# sort by unresolved first then total count
sorted_m = sorted(machine_analysis, key=lambda x: (x['unresolved'],
x['count']), reverse=True)
top_machines = sorted_m[:5]
for m in top_machines:
name = m['machine'] or '<unknown>'
# highlight machines with unresolved events
 if m['unresolved'] > 0:
[Link](f"Inspect machine '{name}': {m['unresolved']}
unresolved events — review recent tickets and verify fixes were applied.")
# give concrete steps for top recurring issues
for issue in m['top_issues'][:3]:
[Link](f" - For {name}: check repeated symptom: '{issue}'. Run
diagnostics (logs, self-test), verify firmware/settings, and confirm operator
steps.")
# upcoming heuristics -> proactive actions
if m['upcoming']:
for up in m['upcoming'][:3]:
[Link](f" - Proactive: {name} likely to see
'{up['issue']}' again (median {int(up['median_days'])}d). Schedule inspection,
replace worn parts, and stage spares.")
# general quick actions
[Link](f"Action for {name}: gather recent logs, ensure spare parts
(specially for top issues) are available, and create an SOP for rapid triage.")
# separator
[Link]("-----")
# unresolved focus
if metrics['unresolved_count'] > 0:
[Link](f"There are {metrics['unresolved_count']} unresolved records.
Prioritize reviewing unresolved events and closing tickets with root-cause
documented.")

# frequent reporters
top_reporters = metrics['per_reporter'].head(5)
if not top_reporters.empty:
rp = ", ".join(top_reporters['Raised by'].astype(str).head(3)) if 'Raised
by' in top_reporters.columns else ",
".join(top_reporters['_raised_by'].astype(str).head(3))
[Link](f"Top reporters: {',
'.join(top_reporters['_raised_by'].head(3).astype(str))}. Consider interviewing
frequent reporters for hidden trends or recurring issues.")

# severity summary
[Link](f"Severity distribution: " + ", ".join(f"{r['_severity']}:
{int(r['count'])}" for _, r in metrics['per_severity'].iterrows()))

# clusters
if clusters_labels is not None:
counts = [Link](clusters_labels).value_counts().head(6)
for lab, cnt in [Link]():
[Link](f"Cluster {lab}: {int(cnt)} similar issue events. Create a
checklist or SOP for troubleshooting these symptoms and pre-stage common spare
parts.")

# anomalies
n_anom = int([Link]()) if anomalies is not None else 0
if n_anom > 0:
[Link](f"{n_anom} anomaly events detected — inspect these individually
(possible data-entry or unexpected failure modes).")

# time patterns
pivot = metrics['weekday_hour']
if pivot is not None and not [Link]:
day_sums = [Link](axis=1)
if not day_sums.empty:
peak = day_sums.idxmax()
 if day_sums.max() > day_sums.median() * 1.8:
[Link](f"Spike in events on {peak}. Review staffing, shift
handovers and processes on that day.")

# general advice
[Link]("General: maintain critical-spares list for frequent failure
symptoms, schedule PM checks on items in top clusters, and record RCA for repeat
failures to avoid recurrence.")
# deduplicate
dedup = []
for r in recs:
if r not in dedup:
[Link](r)
return dedup

# ------------- UI & orchestration ----------------

class App:
def __init__(self, root):
[Link] = root
[Link]("Failure Points - Preventive Maintenance Analyzer")
[Link]("1000x700")
frm = [Link](root, padding=8)
[Link](fill='both', expand=True)

control = [Link](frm)
[Link](side='top', fill='x', pady=6)
[Link](control, text="CSV folder:").pack(side='left')
self.path_var = [Link](value=DEFAULT_FOLDER)
self.path_entry = [Link](control, textvariable=self.path_var, width=60)
self.path_entry.pack(side='left', padx=6)
[Link](control, text="Browse", command=[Link]).pack(side='left')
[Link](control, text="Run Analysis",
command=[Link]).pack(side='left', padx=6)
[Link](control, text="Export Report",
command=self.export_report).pack(side='left', padx=6)

# -- new: error log CSV selector --

[Link](control, text=" Error log CSV:").pack(side='left', padx=(12,0))
self.error_var = [Link](value="")
self.error_entry = [Link](control, textvariable=self.error_var,
width=30, state='readonly')
self.error_entry.pack(side='left', padx=6)
[Link](control, text="Select Error Log CSV",
command=self.select_error_csv).pack(side='left')

[Link] = [Link](frm)
[Link](fill='both', expand=True, pady=6)
self.tab_summary = [Link]([Link]); [Link](self.tab_summary,
text='Summary')
self.tab_raw = [Link]([Link]); [Link](self.tab_raw, text='Raw Data
(first 200 rows)')
self.tab_plots = [Link]([Link]); [Link](self.tab_plots,
text='Charts')
self.tab_recs = [Link]([Link]); [Link](self.tab_recs,
text='Recommendations')

self.summary_text = [Link](self.tab_summary, wrap='word')

self.summary_text.pack(fill='both', expand=True)
 self.raw_text = [Link](self.tab_raw, wrap='none')
self.raw_text.pack(fill='both', expand=True)

[Link] = Figure(figsize=(8,5))
[Link] = FigureCanvasTkAgg([Link], master=self.tab_plots)
[Link].get_tk_widget().pack(fill='both', expand=True)

self.recs_text = [Link](self.tab_recs, wrap='word')

self.recs_text.pack(fill='both', expand=True)

[Link] = None
[Link] = None
[Link] = None
[Link] = None
self.output_folder = OUTPUT_FOLDER_DEFAULT
self.error_csv_path = None
[Link](self.output_folder, exist_ok=True)
# warn user early if scikit-learn is missing
if not SKLEARN_AVAILABLE:
try:
[Link]("Missing dependency",
"scikit-learn not installed — clustering and
anomaly detection are disabled.\n\nInstall with:\n pip install scikit-learn")
except Exception:
print("scikit-learn not installed — clustering and anomaly
detection are disabled. Install with: pip install scikit-learn")

def browse(self):
p = [Link](initialdir=self.path_var.get() or
[Link]("~"))
if p:
self.path_var.set(p)

def select_error_csv(self):
p = [Link](initialdir=self.path_var.get() or
[Link]("~"),
filetypes=[("CSV files", "*.csv"), ("All
files", "*.*")])
if p:
self.error_csv_path = p
self.error_var.set([Link](p))
# automatically run the analysis after selecting the file
[Link]()

def run(self):
folder = self.path_var.get().strip() or DEFAULT_FOLDER
if not [Link](folder):
[Link]("Error", f"Folder not found: {folder}")
return
try:
# load all CSVs from the folder
df = load_and_concatenate(folder)

# if an error-log CSV was selected, load it and append into the

dataframe
if getattr(self, 'error_csv_path', None):
try:
df_err = pd.read_csv(self.error_csv_path, low_memory=False)
 df_err['_source_file'] = [Link](self.error_csv_path)
df_err['_machine'] =
[Link]([Link](self.error_csv_path))[0]
df = [Link]([df, df_err], ignore_index=True, sort=False)
except Exception as e:
# warn but continue if the error log cannot be read
print(f"Warning: failed to read error log
{self.error_csv_path}: {e}")

df, detected = preprocess(df)

[Link] = df
metrics = compute_basic_metrics(df)
[Link] = metrics
labels, cluster_model = cluster_issues(df, n_clusters=6)
df['_cluster'] = labels
clusters_labels = [Link] if labels is not None else None
[Link] = cluster_model
anomalies, iso = detect_anomalies(df)
[Link] = anomalies

# machine analysis and store for UI

self.machine_analysis = analyze_machines(df)

recs = generate_recommendations(df, metrics, clusters_labels,

anomalies, getattr(self, 'machine_analysis', None))
self.populate_summary(df, detected, metrics)
self.populate_raw(df)
[Link](metrics, df, cluster_model)
self.populate_recs(recs)
[Link]("Done", "Analysis complete. Use Export Report to
save outputs.")
except Exception as e:
traceback.print_exc()
[Link]("Error", f"Analysis failed: {e}")

def populate_summary(self, df, detected, metrics):

self.summary_text.delete('1.0', [Link])
n_files = df['_source_file'].nunique() if '_source_file' in [Link] else
1
n_rows = len(df)
first = df['_timestamp'].min()
last = df['_timestamp'].max()
s = f"Loaded {n_rows} events from {n_files} CSV files.\nTime range: {first}
to {last}\n\nDetected mapping:\n"
for k,v in [Link]():
s += f" - {k}: {v}\n"
s += "\nMetrics:\n"
s += f" - Total events: {metrics['total_events']}\n"
s += f" - Unresolved count: {metrics['unresolved_count']}\n"
s += "\nTop statuses:\n"
s += metrics['per_status'].head(10).to_string(index=False)

# machines prone to failure

s += "\n\nTop machines by event count:\n"
if getattr(self, 'machine_analysis', None):
for m in self.machine_analysis[:10]:
s += f" - {m['machine']}: {m['count']} events, {m['unresolved']}
unresolved. Top issues: {', '.join(m['top_issues'][:3])}\n"
if m['upcoming']:
 for up in m['upcoming']:
s += f" * Upcoming (heuristic): '{up['issue']}' — median
every {int(up['median_days'])}d, last {int(up['days_since_last'])}d ago\n"
else:
s += " (no machine analysis available)\n"

self.summary_text.insert([Link], s)

def populate_raw(self, df):

self.raw_text.delete('1.0', [Link])
preview = [Link](200).to_string(index=False)
self.raw_text.insert([Link], preview)

def plot(self, metrics, df, cluster_model):

[Link]()
ax1 = [Link].add_subplot(221)
# severity counts
sev = metrics['per_severity']
if not [Link]:
[Link](sev['_severity'].astype(str), sev['count'])
ax1.set_title('Severity distribution')
ax2 = [Link].add_subplot(222)
st = metrics['per_status'].head(10)
if not [Link]:
[Link](st['_prod_status'].astype(str), st['count'][::-1])
ax2.set_title('Top statuses')
ax3 = [Link].add_subplot(223)
pivot = metrics['weekday_hour']
if pivot is not None and not [Link]:
im = [Link]([Link], aspect='auto')
ax3.set_yticks(range(len([Link])))
ax3.set_yticklabels([Link])
ax3.set_xlabel('Hour')
ax3.set_title('Failures by weekday/hour')
ax4 = [Link].add_subplot(224)
if cluster_model is not None and cluster_model.get('Xred') is not None:
Xred = cluster_model['Xred']
labs = cluster_model['kmeans'].labels_
[Link](Xred[:,0], Xred[:,1], c=labs, s=8)
ax4.set_title('Issue clusters (PCA)')
else:
[Link](0.1, 0.5, 'No cluster visualization available',
transform=[Link])
[Link]()

def populate_recs(self, recs):

self.recs_text.delete('1.0', [Link])
if not recs:
self.recs_text.insert([Link], "No recommendations generated.")
return
for i, r in enumerate(recs, start=1):
self.recs_text.insert([Link], f"{i}. {r}\n\n")

def export_report(self):
if [Link] is None or [Link] is None:
[Link]("Error", "No analysis to export. Run the analysis
first.")
return
out = self.output_folder
 try:
[Link]['per_reporter'].to_csv([Link](out,
'per_reporter.csv'), index=False)
[Link]['per_status'].to_csv([Link](out, 'per_status.csv'),
index=False)
[Link]['per_severity'].to_csv([Link](out,
'per_severity.csv'), index=False)
[Link]['weekday_hour'].to_csv([Link](out,
'weekday_hour.csv'))
# entire cleaned dataframe
[Link].to_csv([Link](out, 'cleaned_combined.csv'), index=False)
recs = self.recs_text.get('1.0', [Link]).strip()
with open([Link](out, '[Link]'), 'w', encoding='utf-
8') as f:
[Link](recs)
[Link]("Exported", f"Reports saved to: {out}")
except Exception as e:
traceback.print_exc()
[Link]("Export failed", str(e))

def main():
root = [Link]()
app = App(root)
[Link]()

if __name__ == "__main__":
main()