File size: 8,482 Bytes
d0cb32e e9b0e37 d0cb32e e9b0e37 d0cb32e 65c3f40 e9b0e37 65c3f40 e9b0e37 65c3f40 e9b0e37 65c3f40 e9b0e37 65c3f40 e9b0e37 65c3f40 d0cb32e 65c3f40 e9b0e37 d0cb32e 65c3f40 d0cb32e 07c6db0 65c3f40 e9b0e37 d0cb32e 65c3f40 d0cb32e e9b0e37 65c3f40 d0cb32e e9b0e37 d0cb32e 65c3f40 d0cb32e 65c3f40 d0cb32e 65c3f40 d0cb32e 65c3f40 d0cb32e 65c3f40 d0cb32e 65c3f40 d0cb32e 65c3f40 d0cb32e 65c3f40 d0cb32e 65c3f40 d0cb32e ee022c7 |
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 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 |
import gradio as gr
import numpy as np
import torch
import librosa
import soundfile as sf
from transformers import AutoFeatureExtractor, AutoModelForAudioClassification
import matplotlib.pyplot as plt
from matplotlib.colors import Normalize
import tempfile
import os
# Constants
SAMPLING_RATE = 16000
MODEL_NAME = "MIT/ast-finetuned-audioset-10-10-0.4593"
DEFAULT_THRESHOLD = 0.7
# Load model and feature extractor
feature_extractor = AutoFeatureExtractor.from_pretrained(MODEL_NAME)
model = AutoModelForAudioClassification.from_pretrained(MODEL_NAME)
# Equipment knowledge base
EQUIPMENT_RECOMMENDATIONS = {
"bearing": {
"high_frequency": "Recommend bearing replacement. High-frequency noise indicates wear or lubrication issues.",
"low_frequency": "Check for improper installation or contamination in bearings.",
"irregular": "Possible bearing cage damage. Schedule vibration analysis."
},
"pump": {
"cavitation": "Pump cavitation detected. Check suction conditions and NPSH.",
"impeller": "Impeller damage likely. Inspect and balance if needed.",
"misalignment": "Misalignment detected. Perform laser shaft alignment."
},
"motor": {
"electrical": "Electrical fault suspected. Check windings and connections.",
"mechanical": "Mechanical imbalance detected. Perform dynamic balancing.",
"bearing": "Motor bearing wear detected. Schedule replacement."
},
"compressor": {
"valve": "Compressor valve leakage suspected. Perform valve test.",
"pulsation": "Pulsation issues detected. Check dampeners and piping.",
"surge": "Compressor surge condition. Review control settings."
}
}
def analyze_frequency_patterns(audio, sr):
"""Analyze frequency patterns to identify potential issues"""
patterns = []
# Spectral analysis
spectral_centroid = librosa.feature.spectral_centroid(y=audio, sr=sr)[0]
spectral_rolloff = librosa.feature.spectral_rolloff(y=audio, sr=sr)[0]
mean_centroid = np.mean(spectral_centroid)
mean_rolloff = np.mean(spectral_rolloff)
if mean_centroid > 3000: # High frequency components
patterns.append("high_frequency")
elif mean_centroid < 1000: # Low frequency components
patterns.append("low_frequency")
if mean_rolloff > 8000: # Rich in harmonics
patterns.append("harmonic_rich")
return patterns
def generate_recommendation(prediction, confidence, audio, sr):
"""Generate maintenance recommendations based on analysis"""
if prediction == "Normal":
return "No immediate action required. Equipment operating within normal parameters."
patterns = analyze_frequency_patterns(audio, sr)
# Simple equipment type classifier based on frequency profile
spectral_flatness = librosa.feature.spectral_flatness(y=audio)[0]
mean_flatness = np.mean(spectral_flatness)
if mean_flatness < 0.2:
equipment_type = "bearing"
elif 0.2 <= mean_flatness < 0.6:
equipment_type = "pump"
else:
equipment_type = "motor" if np.mean(audio) < 0.1 else "compressor"
# Generate specific recommendations
recommendations = ["π§ Maintenance Recommendations:"]
recommendations.append(f"Detected issues in {equipment_type} with {confidence:.1%} confidence")
for pattern in patterns:
if pattern in EQUIPMENT_RECOMMENDATIONS.get(equipment_type, {}):
recommendations.append(f"β {EQUIPMENT_RECOMMENDATIONS[equipment_type][pattern]}")
# General recommendations
if prediction == "Anomaly":
recommendations.append("\nπ οΈ Suggested Actions:")
recommendations.append("1. Isolate equipment if possible")
recommendations.append("2. Perform visual inspection")
recommendations.append("3. Schedule detailed diagnostics")
recommendations.append(f"4. Review last maintenance records ({equipment_type})")
if confidence > 0.8:
recommendations.append("\nπ¨ Urgent: High confidence abnormality detected. Recommend immediate inspection!")
return "\n".join(recommendations)
def analyze_audio(audio_input, threshold=DEFAULT_THRESHOLD):
"""Process audio and detect anomalies"""
try:
# Handle file upload
if isinstance(audio_input, str):
audio, sr = sf.read(audio_input)
else: # Gradio file object
with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp:
tmp.write(audio_input.read())
tmp_path = tmp.name
audio, sr = sf.read(tmp_path)
os.unlink(tmp_path)
# Convert to mono and resample if needed
if len(audio.shape) > 1:
audio = np.mean(audio, axis=1)
if sr != SAMPLING_RATE:
audio = librosa.resample(audio, orig_sr=sr, target_sr=SAMPLING_RATE)
# Feature extraction and prediction
inputs = feature_extractor(audio, sampling_rate=SAMPLING_RATE, return_tensors="pt")
with torch.no_grad():
outputs = model(**inputs)
probs = torch.softmax(outputs.logits, dim=-1)
# Get results
predicted_class = "Normal" if probs[0][0] > threshold else "Anomaly"
confidence = probs[0][0].item() if predicted_class == "Normal" else 1 - probs[0][0].item()
# Generate spectrogram
spectrogram = librosa.feature.melspectrogram(y=audio, sr=SAMPLING_RATE, n_mels=64, fmax=8000)
db_spec = librosa.power_to_db(spectrogram, ref=np.max)
fig, ax = plt.subplots(figsize=(10, 4))
librosa.display.specshow(db_spec, x_axis='time', y_axis='mel', sr=SAMPLING_RATE, fmax=8000, ax=ax)
plt.colorbar(format='%+2.0f dB')
plt.title('Mel Spectrogram with Anomaly Detection')
# Mark anomalies on plot
if predicted_class == "Anomaly":
plt.text(0.5, 0.9, 'ANOMALY DETECTED', color='red',
ha='center', va='center', transform=ax.transAxes,
fontsize=14, bbox=dict(facecolor='white', alpha=0.8))
spec_path = os.path.join(tempfile.gettempdir(), 'spec.png')
plt.savefig(spec_path, bbox_inches='tight')
plt.close()
# Generate detailed recommendations
recommendations = generate_recommendation(predicted_class, confidence, audio, SAMPLING_RATE)
return (
predicted_class,
f"{confidence:.1%}",
spec_path,
recommendations
)
except Exception as e:
return f"Error: {str(e)}", "", None, ""
# Gradio Interface
with gr.Blocks(title="Industrial Diagnostic Assistant π¨βπ§", theme=gr.themes.Soft()) as demo:
gr.Markdown("""
# π Industrial Equipment Diagnostic Assistant
## Acoustic Anomaly Detection & Maintenance Recommendation System
""")
with gr.Row():
with gr.Column():
audio_input = gr.Audio(
label="Upload Equipment Recording (.wav)",
type="filepath",
source="upload"
)
threshold = gr.Slider(
minimum=0.5, maximum=0.95, step=0.05, value=DEFAULT_THRESHOLD,
label="Detection Sensitivity", interactive=True
)
analyze_btn = gr.Button("π Analyze & Diagnose", variant="primary")
with gr.Column():
result_label = gr.Label(label="Diagnosis Result")
confidence = gr.Textbox(label="Confidence Score")
spectrogram = gr.Image(label="Acoustic Analysis")
recommendations = gr.Textbox(
label="Maintenance Recommendations",
lines=10,
interactive=False
)
analyze_btn.click(
fn=analyze_audio,
inputs=[audio_input, threshold],
outputs=[result_label, confidence, spectrogram, recommendations]
)
gr.Markdown("""
### System Capabilities:
- Automatic anomaly detection in industrial equipment sounds
- Frequency pattern analysis to identify failure modes
- Equipment-specific maintenance recommendations
- Confidence-based urgency classification
**Tip:** For best results, use 5-10 second recordings of steady operation
""")
if __name__ == "__main__":
demo.launch()
|