Update app.py

app.py

@@ -23,205 +23,265 @@ sample_rate = 44100
 amplitude_scaling_factor = 10.0
 
 
-# -----------------Filter----------------- #
-
-def butter_bandpass(sr, order=5):
-    """
-    This function designs a Butterworth bandpass filter.
-
-    Parameters:
-    sr (int): The sample rate of the audio.
-    order (int): The order of the filter.
+import numpy as np
+from scipy.io.wavfile import write
+from scipy.signal import find_peaks
+from scipy.fft import fft
+from tqdm import tqdm
+import time
+import matplotlib.pyplot as plt
+from scipy.io.wavfile import read
+from scipy import signal
+import gradio as gr
+import reedsolo
+import wavio
+from scipy.signal import butter, lfilter
 
-    Returns:
-    tuple: The filter coefficients `b` and `a`.
-    """
-    # Calculate the Nyquist frequency
-    nyquist = 0.5 * sr
+# ---------------Parameters--------------- #
 
-    # Normalize the cutoff frequencies
-    low = low_frequency / nyquist
-    high = high_frequency / nyquist
+low_frequency = 18000
+high_frequency = 19000
+bit_duration = 0.007
+sample_rate = 44100
+amplitude_scaling_factor = 10.0
 
-    # Design the Butterworth bandpass filter
-    coefficient = butter(order, [low, high], btype='band')
 
-    # Extract the filter coefficients
-    b = coefficient[0]
-    a = coefficient[1]
+# -----------------Record----------------- #
 
-    return b, a
+def record(audio):
+    try:
+        sr, data = audio
+        wavio.write("recorded.wav", data, sr)
+        main()
+        return f"Audio receive correctly"
+    except Exception as e:
+        return f"Error: {e}"
 
 
-def butter_bandpass_filter(data, sr, order=5):
-    """
-    This function applies the Butterworth bandpass filter to a given data.
+# -----------------Filter----------------- #
 
-    Parameters:
-    data (array): The audio data to be filtered.
-    sr (int): The sample rate of the audio.
-    order (int): The order of the filter.
+def butter_bandpass(lowcut, highcut, sr, order=5):
+    nyquist = 0.5 * sr
+    low = lowcut / nyquist
+    high = highcut / nyquist
+    coef = butter(order, [low, high], btype='band')
+    b = coef[0]
+    a = coef[1]
+    return b, a
 
-    Returns:
-    array: The filtered audio data.
-    """
-    # Get the filter coefficients
-    b, a = butter_bandpass(sr, order=order)
 
-    # Apply the filter to the data
+def butter_bandpass_filter(data, lowcut, highcut, sr, order=5):
+    b, a = butter_bandpass(lowcut, highcut, sr, order=order)
     y = lfilter(b, a, data)
-
     return y
 
 
-def filtered():
-    """
-    This function reads an audio file, applies the bandpass filter to the audio data,
-    and then writes the filtered data to an output file.
-
-    Returns:
-    str: A success message if the audio is filtered correctly, otherwise an error message.
-    """
-    try:
-        input_file = 'input_text.wav'
-        output_file = 'output_filtered_receiver.wav'
-
-        # Read the audio data from the input file
-        sr, data = read(input_file)
+def main():
+    input_file = 'recorded.wav'
+    output_file = 'output_filtered_receiver.wav'
+    lowcut = 17500
+    highcut = 19500
 
-        # Apply the bandpass filter to the audio data
-        filtered_data = butter_bandpass_filter(data, sr)
+    sr, data = read(input_file)
 
-        # Write the filtered data to the output file
-        write(output_file, sr, np.int16(filtered_data))
+    filtered_data = butter_bandpass_filter(data, lowcut, highcut, sr)
+    write(output_file, sr, np.int16(filtered_data))
+    return "Filtered Audio Generated"
+    
 
-        return "Filtered Audio Generated"
-    except Exception as e:
-        # If an error occurs, return an error message
-        return f"Error: {str(e)}"
-
-
-# -----------------Record----------------- #
+# -----------------Frame----------------- #
 
-def record(audio):
+def calculate_snr(data, start, end, target_frequency):
     """
-    This function records audio and writes it to a .wav file.
+    This function calculates the Signal-to-Noise Ratio (SNR) for a given frequency within a segment of data.
 
     Parameters:
-    audio (tuple): A tuple containing the sample rate and the audio data.
+    data (array): The audio data.
+    start (int): The start index of the segment.
+    end (int): The end index of the segment.
+    target_frequency (float): The frequency for which the SNR is to be calculated.
 
     Returns:
-    str: A success message if the audio is recorded correctly, otherwise an error message.
+    float: The calculated SNR.
     """
     try:
-        # Check if the audio tuple contains exactly two elements
-        if len(audio) != 2:
-            return f"Error: Expected a tuple with 2 elements, but got {len(audio)}"
+        # Extract the segment from the data
+        segment = data[start:end]
 
-        # Unpack the sample rate and data from the audio tuple
-        sr, data = audio
+        # Perform a Fast Fourier Transform on the segment
+        spectrum = np.fft.fft(segment)
 
-        # Write the audio data to a .wav file
-        wavio.write("recorded.wav", data, sr)
+        # Generate the frequencies corresponding to the FFT coefficients
+        frequencies = np.fft.fftfreq(len(spectrum), 1 / sample_rate)
 
-        # Call the filtered function to apply the bandpass filter to the audio data
-        filtered()
+        # Find the index of the target frequency
+        target_index = np.abs(frequencies - target_frequency).argmin()
 
-        # Return a success message
-        return f"Audio receive correctly"
-    except Exception as e:
-        # If an error occurs, return an error message
-        return f"Error: {str(e)}"
+        # Calculate the amplitude of the target frequency
+        amplitude = np.abs(spectrum[target_index])
 
+        # Define a noise segment
+        noise_segment = data[100:1000 + len(segment)]
 
-# -----------------Frame----------------- #
+        # Perform a Fast Fourier Transform on the noise segment
+        noise_spectrum = np.fft.fft(noise_segment)
 
-def calculate_snr(data, start, end, target_frequency):
-    segment = data[start:end]
-    spectrum = np.fft.fft(segment)
-    frequencies = np.fft.fftfreq(len(spectrum), 1 / sample_rate)
-    target_index = np.abs(frequencies - target_frequency).argmin()
-    amplitude = np.abs(spectrum[target_index])
+        # Calculate the amplitude of the noise at the target frequency
+        noise_amplitude = np.abs(noise_spectrum[target_index])
 
-    noise_segment = data[100:1000 + len(segment)]
-    noise_spectrum = np.fft.fft(noise_segment)
-    noise_amplitude = np.abs(noise_spectrum[target_index])
+        # Calculate the SNR
+        snr = 10 * np.log10(amplitude / noise_amplitude)
 
-    snr = 10 * np.log10(amplitude / noise_amplitude)
-    return snr
+        return snr
+    except Exception as e:
+        # If an error occurs, return an error message
+        return f"Error: {e}"
 
 
 def frame_analyse(filename):
-    sr, y = read(filename)
-
-    first_part_start = 0
-    first_part_end = len(y) // 2
-
-    second_part_start = len(y) // 2
-    second_part_end = len(y)
-
-    segment_length = 256
-    overlap_size = 128
-
-    f, t, sxx = signal.spectrogram(y, sr, nperseg=segment_length, noverlap=overlap_size)
-
-    plt.figure()
-    plt.pcolormesh(t, f, sxx, shading="gouraud")
-    plt.xlabel("Time [s]")
-    plt.ylabel("Frequency [Hz]")
-    plt.title("Spectrogram of the signal")
-    plt.show()
-
-    f0 = 18000
-
-    f_idx = np.argmin(np.abs(f - f0))
-
-    thresholds_start = calculate_snr(y, first_part_start, first_part_end, low_frequency)
-    thresholds_end = calculate_snr(y, second_part_start, second_part_end, high_frequency)
-
-    t_idx_start = np.argmax(sxx[f_idx] > thresholds_start)
-
-    t_start = t[t_idx_start]
-
-    t_idx_end = t_idx_start
-    while t_idx_end < len(t) and np.max(sxx[f_idx, t_idx_end:]) > thresholds_end:
-        t_idx_end += 1
+    """
+    This function analyses an audio file and returns the start and end times of the signal of interest.
 
-    t_end = t[t_idx_end]
+    Parameters:
+    filename (str): The path to the audio file.
 
-    return t_start, t_end
+    Returns:
+    tuple: The start and end times of the signal of interest.
+    """
+    try:
+        # Read the audio file
+        sr, y = read(filename)
+
+        # Define the start and end indices of the first and second parts of the audio data
+        first_part_start = 0
+        first_part_end = len(y) // 2
+        second_part_start = len(y) // 2
+        second_part_end = len(y)
+
+        # Define the segment length and overlap size for the spectrogram
+        segment_length = 256
+        overlap_size = 128
+
+        # Calculate the spectrogram of the audio data
+        f, t, sxx = signal.spectrogram(y, sr, nperseg=segment_length, noverlap=overlap_size)
+
+        # Plot the spectrogram
+        plt.figure()
+        plt.pcolormesh(t, f, sxx, shading="gouraud")
+        plt.xlabel("Time [s]")
+        plt.ylabel("Frequency [Hz]")
+        plt.title("Spectrogram of the signal")
+        plt.show()
+
+        # Define the target frequency
+        f0 = 18000
+
+        # Find the index of the target frequency
+        f_idx = np.argmin(np.abs(f - f0))
+
+        # Calculate the SNR thresholds for the start and end of the signal
+        thresholds_start = calculate_snr(y, first_part_start, first_part_end, low_frequency)
+        thresholds_end = calculate_snr(y, second_part_start, second_part_end, high_frequency)
+
+        # Find the start and end indices of the signal of interest
+        t_idx_start = np.argmax(sxx[f_idx] > thresholds_start)
+        t_idx_end = t_idx_start
+        while t_idx_end < len(t) and np.max(sxx[f_idx, t_idx_end:]) > thresholds_end:
+            t_idx_end += 1
+
+        # Convert the start and end indices to times
+        t_start = t[t_idx_start]
+        t_end = t[t_idx_end]
+
+        return t_start, t_end
+    except Exception as e:
+        # If an error occurs, return an error message
+        return f"Error: {e}"
 
 
 # -----------------Receiver----------------- #
 
 def dominant_frequency(signal_value):
+    """
+    This function calculates the dominant frequency in a given signal.
+
+    Parameters:
+    signal_value (array): The signal data.
+
+    Returns:
+    float: The dominant frequency.
+    """
+    # Perform a Fast Fourier Transform on the signal
     yf = fft(signal_value)
+
+    # Generate the frequencies corresponding to the FFT coefficients
     xf = np.linspace(0.0, sample_rate / 2.0, len(signal_value) // 2)
+
+    # Find the peaks in the absolute values of the FFT coefficients
     peaks, _ = find_peaks(np.abs(yf[0:len(signal_value) // 2]))
+
+    # Return the frequency corresponding to the peak with the highest amplitude
     return xf[peaks[np.argmax(np.abs(yf[0:len(signal_value) // 2][peaks]))]]
 
 
 def binary_to_text(binary):
+    """
+    This function converts a binary string to text.
+
+    Parameters:
+    binary (str): The binary string.
+
+    Returns:
+    str: The converted text.
+    """
     try:
+        # Convert each 8-bit binary number to a character and join them together
         return ''.join(chr(int(binary[i:i + 8], 2)) for i in range(0, len(binary), 8))
     except Exception as e:
-        return f"Except: {e}"
+        # If an error occurs, return an error message
+        return f"Error: {e}"
 
 
 def decode_rs(binary_string, ecc_bytes):
+    """
+    This function decodes a Reed-Solomon encoded binary string.
+
+    Parameters:
+    binary_string (str): The binary string.
+    ecc_bytes (int): The number of error correction bytes used in the encoding.
+
+    Returns:
+    str: The decoded binary string.
+    """
+    # Convert the binary string to a bytearray
     byte_data = bytearray(int(binary_string[i:i + 8], 2) for i in range(0, len(binary_string), 8))
+
+    # Initialize a Reed-Solomon codec
     rs = reedsolo.RSCodec(ecc_bytes)
+
+    # Decode the bytearray
     corrected_data_tuple = rs.decode(byte_data)
     corrected_data = corrected_data_tuple[0]
 
+    # Remove trailing null bytes
     corrected_data = corrected_data.rstrip(b'\x00')
 
+    # Convert the bytearray back to a binary string
     corrected_binary_string = ''.join(format(byte, '08b') for byte in corrected_data)
 
     return corrected_binary_string
 
 
 def manchester_decoding(binary_string):
+    """
+    This function decodes a Manchester encoded binary string.
+
+    Parameters:
+    binary_string (str): The binary string.
+
+    Returns:
+    str: The decoded binary string.
+    """
     decoded_string = ''
     for i in tqdm(range(0, len(binary_string), 2), desc="Decoding"):
         if i + 1 < len(binary_string):
@@ -236,16 +296,27 @@ def manchester_decoding(binary_string):
 
 
 def signal_to_binary_between_times(filename):
+    """
+    This function converts a signal to a binary string between specified times.
+
+    Parameters:
+    filename (str): The path to the audio file.
+
+    Returns:
+    str: The binary string.
+    """
+    # Get the start and end times of the signal of interest
     start_time, end_time = frame_analyse(filename)
 
+    # Read the audio file
     sr, data = read(filename)
 
+    # Calculate the start and end samples of the signal of interest
     start_sample = int((start_time - 0.007) * sr)
     end_sample = int((end_time - 0.007) * sr)
     binary_string = ''
 
-    start_analyse_time = time.time()
-
+    # Convert each sample to a binary digit
     for i in tqdm(range(start_sample, end_sample, int(sr * bit_duration))):
         signal_value = data[i:i + int(sr * bit_duration)]
         frequency = dominant_frequency(signal_value)
@@ -254,10 +325,10 @@ def signal_to_binary_between_times(filename):
         else:
             binary_string += '1'
 
+    # Find the start and end indices of the binary string
     index_start = binary_string.find("1000001")
     substrings = ["0111110", "011110"]
     index_end = -1
-
     for substring in substrings:
         index = binary_string.find(substring)
         if index != -1:
@@ -267,21 +338,33 @@ def signal_to_binary_between_times(filename):
     print("Binary String:", binary_string)
     binary_string_decoded = manchester_decoding(binary_string[index_start + 7:index_end])
 
+    # Decode the binary string
     decoded_binary_string = decode_rs(binary_string_decoded, 20)
 
     return decoded_binary_string
 
 
 def receive():
+    """
+    This function receives an audio signal, converts it to a binary string, and then converts the binary string to text.
+
+    Returns:
+    str: The received text.
+    """
     try:
+        # Convert the audio signal to a binary string
         audio_receive = signal_to_binary_between_times('output_filtered_receiver.wav')
+
+        # Convert the binary string to text
         return binary_to_text(audio_receive)
     except Exception as e:
+        # If an error occurs, return an error message
         return f"Error: {e}"
 
 
 # -----------------Interface----------------- #
 
+# Start a Gradio Blocks interface
 with gr.Blocks() as demo:
     input_audio = gr.Audio(sources=["upload"])
     output_text = gr.Textbox(label="Record Sound")