D3V1L1810/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# Load the model16model = hub.load('Audio_Multiple_v1')17 18def class_names_from_csv(class_map_csv_text):19    """Returns list of class names corresponding to score vector."""20    class_names = []21    with tf.io.gfile.GFile(class_map_csv_text) as csvfile:22        reader = csv.DictReader(csvfile)23        for row in reader:24            class_names.append(row['display_name'])25    return class_names26 27class_map_path = model.class_map_path().numpy()28class_names = class_names_from_csv(class_map_path)29 30def ensure_sample_rate(original_sample_rate, waveform, desired_sample_rate=16000):31    if original_sample_rate != desired_sample_rate:  # Resample waveform if required32        desired_length = int(round(float(len(waveform)) / original_sample_rate * desired_sample_rate))33        waveform = np.array(scipy.signal.resample(waveform, desired_length), dtype=np.float32)34    return desired_sample_rate, waveform35 36def convert_mp3_to_wav(mp3_data):37    audio = AudioSegment.from_file(io.BytesIO(mp3_data), format="mp3")38    wav_buffer = io.BytesIO()39    audio.export(wav_buffer, format='wav')40    wav_buffer.seek(0)41    return wav_buffer.getvalue()42 43def process_audio_file(file_data, url,file_id):44    try:45        sample_rate, wav_data = wavfile.read(BytesIO(file_data))      46        47        if wav_data.ndim > 1:                        48            wav_data = np.mean(wav_data, axis=1)49        sample_rate, wav_data = ensure_sample_rate(sample_rate, wav_data)50 51        waveform = wav_data / tf.int16.max52 53        scores, embeddings, spectrogram = model(waveform)      54 55        scores_np = scores.numpy()56        mean_scores = np.mean(scores, axis=0)57 58        inferred_class = class_names[mean_scores.argmax()]    59        60        confidence_threshold = 0.60                 61        confident_classes = set()62 63        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']64        for frame_scores in scores_np:65            for i, score in enumerate(frame_scores):66                if score > confidence_threshold:67                    class_name = class_names[i]68 69                    if class_name =='Child speech, kid speaking':70                        class_name='Child speech'71                    elif class_name =='Vehicle horn, car horn, honking':72                        class_name='Vehicle horn'73                    elif class_name =='Railroad car, train wagon':74                        class_name='Train/wagon'75                    elif class_name=='Rail transport':76                        class_name='Train/wagon'77 78                    if class_name not in exclusion_list:79                        confident_classes.add(class_name)80 81        confident_classes = sorted(confident_classes)82        confident_classes_list = list(confident_classes)83        84        answer_dict = {'url': url, 'answer': confident_classes_list, "qcUser" : None, "normalfileID": file_id}85        return answer_dict86    87    except Exception as e:88        logging.error(f"Error processing {url}: {e}")89        return None90 91def get_audio_data(url):92    response = requests.get(url)93    response.raise_for_status()94    return response.content95 96# def send_results_to_api(data, result_url):97#     headers = {"Content-Type": "application/json"}98#     try:99#         response = requests.post(result_url, json=data, headers=headers)100#         response.raise_for_status()  # Raise error for non-200 responses101#         return response.json()  # Return any JSON response from the API102#     except requests.exceptions.HTTPError as http_err:103#         logging.error(f"HTTP error occurred: {http_err}")104#         return {"error": f"HTTP error occurred: {http_err}"}105#     except requests.exceptions.RequestException as req_err:106#         logging.error(f"Request error occurred: {req_err}")107#         return {"error": f"Request error occurred: {req_err}"}108#     except ValueError as val_err:109#         logging.error(f"Error decoding JSON response: {val_err}")110#         return {"error": f"Error decoding JSON response: {val_err}"}111 112def process_audio(params):113    try:114        params = json.loads(params)115    except json.JSONDecodeError as e:116        return {"error": f"Invalid JSON input: {e.msg} at line {e.lineno} column {e.colno}"}117 118    audio_files = params.get("urls", [])119    if not params.get("normalfileID",[]):120        file_ids = [None]*len(audio_files)121    else:122        file_ids = params.get("normalfileID",[])123    # api = params.get("api", "")124    # job_id = params.get("job_id", "")125 126    solutions = []127    for audio_url, file_id in zip(audio_files, file_ids):128        audio_data = get_audio_data(audio_url)129 130        if audio_url.endswith(".mp3"):            131            wav_data = convert_mp3_to_wav(audio_data)132            result = process_audio_file(wav_data, audio_url, file_id)133 134        elif audio_url.endswith(".wav"):           135            result = process_audio_file(audio_data, audio_url, file_id)136        137        if result:138            solutions.append(result)139 140    # result_url = f"{api}/{job_id}"141    # send_results_to_api(solutions, result_url)142 143    return json.dumps({"solutions": solutions})144 145import gradio as gr146 147inputt = gr.Textbox(label="Parameters (JSON format) Eg. {'urls':['file1.mp3','file2.wav']}")148outputs = gr.JSON()149 150application = gr.Interface(fn=process_audio, inputs=inputt, outputs=outputs, title="Audio Classification with API Integration")151application.launch()152