Create app.py

app.py

@@ -0,0 +1,119 @@
+import os
+import tensorflow as tf
+import tensorflow_hub as hub
+import numpy as np
+import csv
+import requests
+import json
+from scipy.io import wavfile
+from pydub import AudioSegment
+import gradio as gr
+
+model = hub.load('Audio_Multiple_v1')       #loading model
+
+def class_names_from_csv(class_map_csv_text):            # Function to load class names from CSV.
+    """Returns list of class names corresponding to score vector."""
+    class_names = []
+    with tf.io.gfile.GFile(class_map_csv_text) as csvfile:
+        reader = csv.DictReader(csvfile)
+        for row in reader:
+            class_names.append(row['display_name'])
+    return class_names
+
+class_map_path = model.class_map_path().numpy()
+class_names = class_names_from_csv(class_map_path)
+
+def ensure_sample_rate(original_sample_rate, waveform, desired_sample_rate=16000):   #to ensure/make a standard sample rate for audio
+    if original_sample_rate != desired_sample_rate:  # Resample waveform if required
+        desired_length = int(round(float(len(waveform)) / original_sample_rate * desired_sample_rate))
+        waveform = scipy.signal.resample(waveform, desired_length)
+    return desired_sample_rate, waveform
+
+def convert_mp3_to_wav(mp3_file_path):        #if audio file is mp3 then convert it to wav
+    audio = AudioSegment.from_mp3(mp3_file_path)
+    wav_file_path = mp3_file_path.replace('.mp3', '.wav')
+    audio.export(wav_file_path, format='wav')
+    return wav_file_path
+
+def process_audio_file(file_path):          #reading audio file and making stereo to mono
+    sample_rate, wav_data = wavfile.read(file_path, 'rb')
+    if wav_data.ndim > 1:  # Convert stereo to mono if needed
+        wav_data = np.mean(wav_data, axis=1)
+    sample_rate, wav_data = ensure_sample_rate(sample_rate, wav_data)
+
+    waveform = wav_data / tf.int16.max
+    waveform = tf.convert_to_tensor(waveform, dtype=tf.float32)
+
+    scores, embeddings, spectrogram = model(waveform)
+
+    scores_np = scores.numpy()
+    mean_scores = np.mean(scores, axis=0)
+
+    inferred_class = class_names[mean_scores.argmax()]
+
+    confidence_threshold = 0.60
+    confident_classes = set()
+
+    #excluding some unwanted classes
+    exclusion_list = ['Mechanisms','Domestic animals, pets', 'Animal', 'Silence', 'Alarm', 'Wind chime', 'Water', 'Livestock, farm animals, working animals', 'Wild animals', 'Bleat', 'Siren', 'Computer keyboard', 'Toot', 'Shatter', 'Bird','Caw', 'Independent music', 'Tender music', 'Ocean', 'House music', 'Middle Eastern music', 'Swing music', 'Soul music', 'Shofar', 'Motor vehicle (road)', 'White noise','Pink noise', 'Cacophony', 'Sidetone', 'Static', 'Outside, rural or natural', 'Outside, urban or manmade', 'Inside, public space', 'Inside, large room or hall', 'Inside, small room', 'Sound effect'  ]
+    for frame_scores in scores_np:
+        for i, score in enumerate(frame_scores):
+            if score > confidence_threshold:
+                class_name = class_names[i]
+
+                if class_name == 'Child speech, kid speaking':
+                    class_name = 'Child speech'
+                elif class_name == 'Vehicle horn, car horn, honking':
+                    class_name = 'Vehicle horn'
+                elif class_name == 'Railroad car, train wagon':
+                    class_name = 'Train/wagon'
+                elif class_name == 'Rail transport':
+                    class_name = 'Train/wagon'
+
+                if class_name not in exclusion_list:
+                    confident_classes.add(class_name)
+
+    confident_classes = sorted(confident_classes)
+    return confident_classes
+
+#main function to get the input
+def process_audio(params):
+    try:
+        params = json.loads(params)
+    except json.JSONDecodeError:
+        return {"error": "Invalid JSON input"}
+
+    audio_files = params.get("audio_files", [])
+    api = params.get("api", "")
+    job_id = params.get("job_id", "")
+
+    solutions = []
+    for audio in audio_files:
+        if audio.endswith(".mp3"):
+            wav_file_path = convert_mp3_to_wav(audio)
+            class_names = process_audio_file(wav_file_path)
+        elif audio.endswith(".wav"):
+            class_names = process_audio_file(audio)
+        
+        answer_dict = {'file_name': audio, 'class_names': class_names}
+        solutions.append(answer_dict)
+
+    result_url = f"{api}/{job_id}"
+    
+    # send_results_to_api(solutions, result_url)
+
+    return json.dumps({"solutions": solutions}, indent=4)
+
+def send_results_to_api(data, result_url):
+    headers = {"Content-Type": "application/json"}
+    response = requests.patch(result_url, json=data, headers=headers)
+    if response.status_code == 200:
+        return response.json()  # Return any response from the API if needed
+    else:
+        return {"error": f"Failed to send results to API: {response.status_code}"}
+
+inputt = gr.Textbox(label="Parameters (JSON format) Eg. audio_files:['',''], api:'', job_id:''")
+outputts = gr.JSON()
+
+application = gr.Interface(fn=process_audio, inputs=inputt, outputs=outputts, title="Audio Classification with API Integration")
+application.launch()