D3V1L1810/Primary_Sound_Classification

1import os2import tensorflow as tf3import tensorflow_hub as hub4import numpy as np5import csv6import requests7import json8import logging9import scipy10from scipy.io import wavfile11from pydub import AudioSegment12import io13from io import BytesIO14 15 16# Load the model17model = hub.load('Audio_Multiple_v1')18 19def class_names_from_csv(class_map_csv_text):20    """Returns list of class names corresponding to score vector."""21    class_names = []22    with tf.io.gfile.GFile(class_map_csv_text) as csvfile:23        reader = csv.DictReader(csvfile)24        for row in reader:25            class_names.append(row['display_name'])26    return class_names27 28class_map_path = model.class_map_path().numpy()29class_names = class_names_from_csv(class_map_path)30 31def ensure_sample_rate(original_sample_rate, waveform, desired_sample_rate=16000):32    if original_sample_rate != desired_sample_rate:33        desired_length = int(round(float(len(waveform)) / original_sample_rate * desired_sample_rate))34        waveform = np.array(scipy.signal.resample(waveform, desired_length), dtype=np.float32)35    return desired_sample_rate, waveform36 37def convert_mp3_to_wav(mp3_data):38    audio = AudioSegment.from_file(io.BytesIO(mp3_data), format="mp3")39    wav_buffer = io.BytesIO()40    audio.export(wav_buffer, format='wav')41    wav_buffer.seek(0)42    return wav_buffer.getvalue()43 44def process_audio_file(file_data, url, file_id):45    try:46        sample_rate, wav_data = wavfile.read(BytesIO(file_data))47        if wav_data.ndim > 1:48            wav_data = np.mean(wav_data, axis=1)49 50        sample_rate, wav_data = ensure_sample_rate(sample_rate, wav_data)51        waveform = wav_data / tf.int16.max52 53        scores, embeddings, spectrogram = model(waveform)54 55        scores_np = scores.numpy()56        spectrogram_np = spectrogram.numpy()57        mean_scores = np.mean(scores, axis=0)58 59        top_two_indices = np.argsort(mean_scores)[-2:][::-1]60        inferred_class = class_names[top_two_indices[0]]61 62        if inferred_class == "Silence" and len(top_two_indices) > 1:63            inferred_class = class_names[top_two_indices[1]]64 65        answer_dict = {'url': url, 'answer': [inferred_class], qcUser: None, "normalfileID": file_id}66        return answer_dict67    except Exception as e:68        logging.error(f"Error processing {url}: {e}")69        return None70 71def get_audio_data(url):72    response = requests.get(url)73    response.raise_for_status()74    return response.content75 76# def send_results_to_api(data, result_url):77#     headers = {"Content-Type": "application/json"}78#     try:79#         response = requests.post(result_url, json=data, headers=headers)80#         response.raise_for_status()  # Raise error for non-200 responses81#         return response.json()  # Return any JSON response from the API82#     except requests.exceptions.HTTPError as http_err:83#         logging.error(f"HTTP error occurred: {http_err}")84#         return {"error": f"HTTP error occurred: {http_err}"}85#     except requests.exceptions.RequestException as req_err:86#         logging.error(f"Request error occurred: {req_err}")87#         return {"error": f"Request error occurred: {req_err}"}88#     except ValueError as val_err:89#         logging.error(f"Error decoding JSON response: {val_err}")90#         return {"error": f"Error decoding JSON response: {val_err}"}91 92def process_audio(params):93    try:94        params = json.loads(params)95    except json.JSONDecodeError as e:96        return {"error": f"Invalid JSON input: {e.msg} at line {e.lineno} column {e.colno}"}97 98    audio_files = params.get("urls", [])99    if not params.get("normalfileID",[]):100        file_ids = [None]*len(audio_files)101    else:102        file_ids = params.get("normalfileID",[])103    # api = params.get("api", "")104    # job_id = params.get("job_id", "")105 106    solutions = []107    for audio_url,file_id in zip(audio_files, file_ids):108        audio_data = get_audio_data(audio_url)109 110        if audio_url.endswith(".mp3"):111            wav_data = convert_mp3_to_wav(audio_data)112            result = process_audio_file(wav_data, audio_url, file_id)113 114        elif audio_url.endswith(".wav"):115            result = process_audio_file(audio_data, audio_url, file_id)116 117        if result:118            solutions.append(result)119 120    # result_url = f"{api}/{job_id}"121    # send_results_to_api(solutions, result_url)122 123    return json.dumps({"solutions": solutions})124 125import gradio as gr126 127inputt = gr.Textbox(label="Parameters (JSON format) Eg. {'urls':['file1.mp3','file2.wav']}")128outputs = gr.JSON()129 130application = gr.Interface(fn=process_audio, inputs=inputt, outputs=outputs, title="Audio Classification with API Integration")131application.launch()