Create app.py

app.py

@@ -0,0 +1,125 @@
+try:
+    import os
+    import tensorflow as tf
+    import tensorflow_hub as hub
+    import numpy as np
+    import csv
+    import requests
+    import json
+    import logging
+    import scipy
+    from scipy.io import wavfile
+    from pydub import AudioSegment
+    import io
+    from io import BytesIO
+
+    model = hub.load('Audio_Multiple_v1')     
+
+    def class_names_from_csv(class_map_csv_text):          
+        """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):
+        if original_sample_rate != desired_sample_rate:                  
+            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_data):                
+        audio = AudioSegment.from_file(io.BytesIO(mp3_data), format="mp3")
+        wav_buffer = io.BytesIO()
+        audio.export(wav_buffer, format='wav')
+        wav_buffer.seek(0)
+        return wav_buffer
+
+
+    def process_audio_file(file_data, url):
+        try:
+            sample_rate, wav_data = wavfile.read(BytesIO(file_data))
+            if wav_data.ndim > 1:                       
+                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
+
+            scores, embeddings, spectrogram = model(waveform)       
+
+            scores_np = scores.numpy()
+            spectrogram_np = spectrogram.numpy()
+            mean_scores = np.mean(scores, axis=0)
+
+            top_two_indices = np.argsort(mean_scores)[-2:][::-1]        
+            inferred_class = class_names[top_two_indices[0]]
+        
+            if inferred_class == "Silence" and len(top_two_indices) > 1:     
+                inferred_class = class_names[top_two_indices[1]]
+
+            answer_dict = {'file_name': url, 'class_name': inferred_class}
+            return answer_dict
+        except Exception as e:
+            logging.error(f"Error processing {url}: {e}")
+            return None
+
+
+    def get_audio_data(url):
+        response = requests.get(url)
+        response.raise_for_status()
+        return response.content
+    
+    def send_results_to_api(data, result_url):
+        # Example function to send results to an API
+        headers = {"Content-Type": "application/json"}
+        response = requests.post(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}"}
+    
+    def process_audio(params):
+        try:
+            params = json.loads(params)
+        except json.JSONDecodeError as e:
+            return {"error": f"Invalid JSON input: {e.msg} at line {e.lineno} column {e.colno}"}
+
+        audio_files = params.get("audio_files", [])
+        api = params.get("api", "")
+        job_id = params.get("job_id", "")
+
+        solutions = []
+        for audio_url in audio_files:
+            audio_data = get_audio_data(audio_url)
+
+            if audio_url.endswith(".mp3"):            
+                wav_data = convert_mp3_to_wav(audio_data)
+                result = process_audio_file(wav_data, audio_url)
+
+            elif audio_url.endswith(".wav"):           
+                result = process_audio_file(audio_data, audio_url)
+            
+            if result:
+                solutions.append(result)
+
+        result_url = f"{api}/{job_id}"
+        send_results_to_api(solutions, result_url)
+
+        return json.dumps({"solutions": solutions}, indent=4)
+
+    import gradio as gr
+    inputt = gr.Textbox(label="Parameters (JSON format) Eg. {'audio_files':['file1.mp3','file2.wav'], 'api':'https://api.example.com', 'job_id':'1001'}")
+    outputs = gr.JSON()
+
+    application = gr.Interface(fn=process_audio, inputs=inputt, outputs=outputs, title="Audio Classification with API Integration")
+    application.launch()
+
+except Exception as e:
+    print(e)