Upload edsp_project.py

edsp_project.py

@@ -0,0 +1,202 @@
+import streamlit as st
+import pandas as pd
+import numpy as np
+from scipy.signal import butter, lfilter, cheby1
+import plotly.graph_objects as go
+import matplotlib.pyplot as plt
+
+# Function to apply FFT
+def apply_fft(data, sampling_rate):
+    n = len(data)
+    fft_result = np.fft.fft(data)
+    freq = np.fft.fftfreq(n, 1 / sampling_rate)
+    
+
+    return freq, np.abs(fft_result)
+# Function to apply Wiener filter
+def apply_wiener_filter(noisy_signal, noise_level):
+    signal_power = np.mean(noisy_signal ** 2)
+    noise_power = noise_level ** 2
+    transfer_function = 1 - noise_power / signal_power
+    filtered_signal = noisy_signal * transfer_function
+    return filtered_signal
+
+# Function to apply IIR filter
+def apply_iir_filter(data, cutoff_freq, sampling_rate, filter_type='butter'):
+    nyquist = 0.5 * sampling_rate
+    normal_cutoff = cutoff_freq / nyquist
+
+    if filter_type == 'butter':
+        b, a = butter(4, normal_cutoff, btype='low', analog=False)
+    elif filter_type == 'chebyshev':
+        b, a = cheby1(4, 0.5, normal_cutoff, btype='low', analog=False)
+    else:
+        raise ValueError("Invalid filter type")
+
+    filtered_data = lfilter(b, a, data)
+    return filtered_data
+
+# Function to classify motor signal
+def classify_motor_signal(data, sampling_rate, noise_level):
+    _, fft_result = apply_fft(data, sampling_rate)
+    num_components = 5
+    threshold_multiplier = 1.5
+    
+    # Calculate mean amplitude of the first few frequency components
+    first_few_amplitudes = np.abs(fft_result[1:num_components + 1])
+    mean_first_few_amplitude = np.mean(first_few_amplitudes)
+    
+    # Calculate mean amplitude of the entire FFT spectrum
+    mean_total_amplitude = np.mean(np.abs(fft_result))
+    
+    # Calculate the ratio of mean_first_few_amplitude to mean_total_amplitude
+    ratio = mean_first_few_amplitude / mean_total_amplitude
+    
+    # Set the threshold based on the ratio
+    threshold = threshold_multiplier
+    
+    if ratio > threshold:
+        return "Faulty Motor: High-frequency components detected."
+    else:
+        return "Healthy Motor: No significant fault detected."
+
+# Function to generate power spectrum
+def generate_power_spectrum(signal, sampling_rate):
+    n = len(signal)
+    frequencies = np.fft.fftfreq(n, d=1/sampling_rate)
+    spectrum = np.abs(np.fft.fft(signal))
+    return frequencies, spectrum
+
+# Function to identify peak in the power spectrum
+def identify_peak(frequencies, spectrum):
+    peak_frequency = frequencies[np.argmax(spectrum)]
+    peak_amplitude = np.max(spectrum)
+    return peak_frequency, peak_amplitude
+def calculate_mean_variance(data):
+    mean_value = np.mean(data)
+    variance_value = np.var(data)
+    return mean_value, variance_value
+
+def apply_tukey_test(data, threshold=1.5):
+    q75, q25 = np.percentile(data, [75, 25])
+    iqr = q75 - q25
+    lower_bound = q25 - threshold * iqr
+    upper_bound = q75 + threshold * iqr
+    outliers = (data < lower_bound) | (data > upper_bound)
+    return outliers
+
+# Function to classify motor based on mean, variance, and Tukey test
+def classify_motor_stats(data):
+    mean_value, variance_value = calculate_mean_variance(data)
+    outliers = apply_tukey_test(data)
+
+    if np.any(outliers):
+        return "Faulty Motor: Outliers detected."
+    else:
+        return "Healthy Motor: No significant outliers detected."
+
+
+# Streamlit app
+def main():
+    st.title("Motor Signal Analysis App")
+
+    # Define tabs for different scenarios
+    tab_names = ["No Load - Healthy", "No Load - Faulty", "Full Load - Healthy", "Full Load - Faulty"]
+    selected_tab = st.radio("Select Scenario", tab_names)
+
+    # Set default values
+    default_noise_level = 0.1
+    default_cutoff_freq = 50
+
+    # Upload CSV file
+    uploaded_file = st.file_uploader(f"Upload CSV for {selected_tab}", type=["csv"])
+
+    if uploaded_file is not None:
+        # Load data
+        df = pd.read_csv(uploaded_file)
+        data_column = df.columns[0]  # Assume data is in the first column
+        data = df[data_column].values
+
+        # Display original signal graph
+        st.subheader("Original Signal")
+        original_trace = go.Scatter(x=df.index, y=df[data_column], mode='lines', name='Original Signal')
+        st.plotly_chart([original_trace])
+
+        # Sampling rate
+        sampling_rate = st.number_input("Enter the sampling rate", min_value=1, value=1000)
+
+        # Apply FFT
+        freq, fft_result = apply_fft(data, sampling_rate)
+
+        # Plot FFT
+        st.subheader("FFT Analysis")
+        fft_trace = go.Scatter(x=np.abs(freq), y=np.abs(fft_result), mode='lines', name='FFT Result')
+
+        # Set layout parameters
+        layout_fft = go.Layout(
+    title='FFT Result Analysis',
+    xaxis=dict(title='Frequency (Hz)', range=[0, 200]),  # Set the range up to 200 Hz
+    yaxis=dict(title='Amplitude',range=[0,20000]),  # Remove or comment out 'autorange' setting
+)
+
+        st.plotly_chart(go.Figure(data=[fft_trace], layout=layout_fft))
+
+        # Apply selected filter
+        apply_filter_option = st.checkbox("Apply Filter")
+        if apply_filter_option:
+            # Cutoff frequency for IIR filter
+            cutoff_freq = st.number_input("Enter the cutoff frequency", min_value=1, value=default_cutoff_freq)
+
+            # Filter type for IIR filter
+            filter_type = st.selectbox("Select Filter Type", ['butter', 'chebyshev', 'wiener'], index=0)
+
+            # Apply selected filter
+            if filter_type == 'butter':
+                filtered_data = apply_iir_filter(fft_result, cutoff_freq, sampling_rate, filter_type='butter')
+            elif filter_type == 'chebyshev':
+                filtered_data = apply_iir_filter(fft_result, cutoff_freq, sampling_rate, filter_type='chebyshev')
+            elif filter_type == 'wiener':
+                filtered_data = apply_wiener_filter(fft_result, default_noise_level)
+
+            # Generate power spectrum after applying filter
+            frequencies_spectrum, spectrum = generate_power_spectrum(filtered_data, sampling_rate)
+
+            # Identify peak in power spectrum
+            peak_frequency, peak_amplitude = identify_peak(frequencies_spectrum, spectrum)
+
+            # Plot the results using Plotly
+            st.subheader(f"{filter_type.capitalize()} Filtered Signal")
+            filtered_trace = go.Scatter(x=np.abs(freq), y=np.abs(filtered_data), mode='lines',
+                                    name=f'{filter_type.capitalize()} Filtered Signal')
+            layout_filtered = go.Layout(
+            title='Filtered signal',
+            xaxis=dict(title='Frequency (Hz)',range=[0,200]),
+            yaxis=dict(title='Amplitude',range=[0,20000]),  # Remove or comment out 'autorange' setting
+        )
+
+            st.plotly_chart(go.Figure(data=[filtered_trace], layout=layout_filtered))
+        
+
+            st.subheader(f"Power Spectrum after {filter_type.capitalize()} Filter")
+            power_spectrum_trace = go.Scatter(x=frequencies_spectrum, y=np.abs(spectrum), mode='lines',
+                                            name=f'Power Spectrum ({filter_type.capitalize()} Filter)')
+
+            # Set layout parameters for power spectrum plot
+            layout_power_spectrum = go.Layout(
+                title=f'Power Spectrum after {filter_type.capitalize()} Filter',
+                xaxis=dict(title='Frequency (Hz)'),
+                yaxis=dict(title='Amplitude'),
+            )
+
+            st.plotly_chart(go.Figure(data=[power_spectrum_trace], layout=layout_power_spectrum))
+
+            # Display peak frequency and amplitude
+            st.subheader("Peak in Power Spectrum")
+            st.write(f"Frequency: {peak_frequency} Hz, Amplitude: {peak_amplitude}")
+
+            motor_status_stats = classify_motor_stats(data=data)
+            st.info(f"Motor Status (Mean, Variance, Tukey Test): {motor_status_stats}")
+    
+# Run the app
+if __name__ == "__main__":
+    main()