gargaman07/Audio_Classification

1import json2import logging3import os4import shutil5import tempfile6from pathlib import Path7 8 9import numpy as np10import requests11import tensorflow as tf12import tensorflow_hub as hub13from acrcloud.recognizer import ACRCloudRecognizer14from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile15from fastapi.middleware.cors import CORSMiddleware16from fastapi.responses import HTMLResponse17from fastapi.templating import Jinja2Templates18from pydantic import BaseModel19from pydub import AudioSegment20from tensorflow.keras.models import load_model21import librosa22 23app = FastAPI()24 25# Add CORS middleware26app.add_middleware(27    CORSMiddleware,28    allow_origins=["*"],29    allow_credentials=True,30    allow_methods=["*"],31    allow_headers=["*"],32)33 34templates = Jinja2Templates(directory=".")35model = load_model('./models/neural_networks.h5')36# ACRCloud Configuration using SDK37ACRCLOUD_CONFIG = {38    'host': 'identify-ap-southeast-1.acrcloud.com',39    'access_key': 'c529996b7457352ca72e2ccb1fcbc4dd',40    'access_secret': 'MQitmw327GTfkoLhCzk90Uwcf2dL0DGhUvQvQwS0',41    'timeout': 1  # seconds42}43acr_recognizer = ACRCloudRecognizer(ACRCLOUD_CONFIG) 44 45# Load YAMNet model and labels46yamnet_model_handle = 'https://tfhub.dev/google/yamnet/1'47yamnet_model = hub.load(yamnet_model_handle)48 49with open("yamnet_class_map.csv", "r") as f:50    yamnet_classes = [line.strip().split(",")[2] for line in f.readlines()[1:]]51 52# # Set up ffmpeg path53# FFMPEG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ffmpeg-master-latest-win64-gpl", "bin")54# if os.path.exists(FFMPEG_PATH):55#     os.environ["PATH"] = FFMPEG_PATH + os.pathsep + os.environ["PATH"]56#     AudioSegment.converter = os.path.join(FFMPEG_PATH, "ffmpeg.exe")57#     AudioSegment.ffmpeg = os.path.join(FFMPEG_PATH, "ffmpeg.exe")58#     AudioSegment.ffprobe = os.path.join(FFMPEG_PATH, "ffprobe.exe")59 60# Comment out or remove the Windows-specific FFMPEG_PATH setup61# In Docker, ffmpeg will be installed via apt-get and should be in the PATH62# pydub should find it automatically.63# If issues arise, one might need to set AudioSegment.converter explicitly,64# but without the Windows-specific path.65# For example:66# AudioSegment.converter = "/usr/bin/ffmpeg" # or wherever ffmpeg is installed67# AudioSegment.ffmpeg = "/usr/bin/ffmpeg"68# AudioSegment.ffprobe = "/usr/bin/ffprobe"69# However, this is often not needed if ffmpeg is in the system PATH.70 71def extract_features(audio_path, max_length=100):72    y, sr = librosa.load(audio_path, sr=None)73    y_normalized = librosa.util.normalize(y)74    segments = librosa.effects.split(y_normalized, top_db=20)75 76    mfccs = []77    for start, end in segments:78        segment = y[start:end]79        mfcc = librosa.feature.mfcc(y=segment, sr=sr, n_mfcc=13)80        if mfcc.shape[1] > max_length:81            mfcc = mfcc[:, :max_length]82        else:83            pad_width = max_length - mfcc.shape[1]84            mfcc = np.pad(mfcc, pad_width=((0, 0), (0, pad_width)), mode='constant')85        mfccs.append(mfcc)86    87    return mfccs88 89def predict_vehicle_class(audio_path):90    features = extract_features(audio_path)91    92    # Normalize using training distribution (consider saving stats during training if accuracy matters)93    features = np.array(features)94    features = (features - np.mean(features)) / np.std(features)95 96    # Average predictions across all segments97    predictions = model.predict(features)98    averaged_prediction = np.mean(predictions, axis=0)99    predicted_class = int(np.argmax(averaged_prediction))  # Convert numpy.int64 to Python int100 101    return predicted_class102 103def convert_audio_to_wav(src_path: str, dst_path: str) -> bool:104    """Convert any audio file to WAV format using pydub."""105    try:106        # Get the file extension107        ext = os.path.splitext(src_path)[1].lower().lstrip('.')108        109        # Load the audio file with specific parameters110        audio = AudioSegment.from_file(111            src_path,112            format=ext,113            parameters=["-ar", "16000", "-ac", "1"]  # Set sample rate to 16kHz and mono114        )115        116        # Export as WAV with specific parameters117        audio.export(118            dst_path,119            format="wav",120            parameters=["-ar", "16000", "-ac", "1", "-acodec", "pcm_s16le"]121        )122        return True123    except Exception as e:124        logging.error(f"Error converting audio file: {str(e)}")125        return False126 127def classify_audio_with_yamnet(file_path):128    try:129        # Create a temporary WAV file130        with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as temp_wav:131            temp_wav_path = temp_wav.name132 133        # Convert the input file to WAV if needed134        if not convert_audio_to_wav(file_path, temp_wav_path):135            return {136                "success": False,137                "message": "Failed to convert audio file to WAV format"138            }139 140        try:141            # Load and process the audio142            waveform, sr = librosa.load(temp_wav_path, sr=16000)  # YAMNet expects 16kHz143            scores, embeddings, spectrogram = yamnet_model(waveform)144            scores_np = scores.numpy().mean(axis=0)  # average over time145 146            top5_i = np.argsort(scores_np)[::-1][:5]147            top_labels = [(yamnet_classes[i], float(scores_np[i])) for i in top5_i]  # Convert scores to Python float148 149            return {150                "success": True,151                "top_classes": top_labels152            }153        finally:154            # Clean up the temporary file155            if os.path.exists(temp_wav_path):156                os.unlink(temp_wav_path)157 158    except Exception as e:159        logging.exception("YAMNet classification failed:")160        return {161            "success": False,162            "message": f"Audio classification failed: {str(e)}"163        }164 165def is_vehicle_sound(yamnet_classes):166    """167    Check if any of the top YAMNet classifications are vehicle-related.168    Returns True if a vehicle sound is detected, along with the matched class and score.169    """170    vehicle_keywords = [171        # General vehicle terms172        'vehicle', 'automobile', 'motor vehicle',173        # Specific vehicle types174        'car', 'truck', 'bus', 'van', 'motorcycle', 'scooter',175        # Vehicle components176        'engine', 'motor', 'horn', 'siren', 'tire', 'wheel',177        # Vehicle sounds178        'revving', 'acceleration', 'braking', 'idling',179        # Transportation180        'transport', 'traffic', 'road'181    ]182    183    # Log the top classifications for debugging184    logging.info("Top YAMNet classifications:")185    for class_name, score in yamnet_classes:186        logging.info(f"- {class_name}: {score:.2f}")187    188    # Check each classification against vehicle keywords189    for class_name, score in yamnet_classes:190        class_name_lower = class_name.lower()191        for keyword in vehicle_keywords:192            if keyword in class_name_lower:193                logging.info(f"Vehicle sound detected: '{class_name}' (score: {score:.2f})")194                return True, class_name, score195    196    logging.info("No vehicle sounds detected in the audio")197    return False, None, 0.0198 199@app.post("/classify/")200async def classify_audio(file: UploadFile = File(...)):201    temp_filename = f"temp_classify_{file.filename}"202    file_content = await file.read()203 204    try:205        with open(temp_filename, "wb") as f:206            f.write(file_content)207 208        # First try music recognition209        result_json_str = acr_recognizer.recognize_by_file(temp_filename, 0)210        music_result = format_acrcloud_response(result_json_str)211 212        if music_result["success"]:213            # If music recognition was successful, return that result214            return {215                "success": True,216                "type": "music",217                "music_result": music_result218            }219        else:220            # If music recognition failed, try YAMNet classification221            yamnet_result = classify_audio_with_yamnet(temp_filename)222            if yamnet_result["success"]:223                # Check if the sound is vehicle-related224                is_vehicle, vehicle_class, vehicle_score = is_vehicle_sound(yamnet_result["top_classes"])225                if is_vehicle:226                    # If it's a vehicle sound, use the neural network for specific classification227                    vehicle_class = predict_vehicle_class(temp_filename)228                    vehicle_type = "Car" if vehicle_class == 0 else "Truck"229                    230                    return {231                        "success": True,232                        "type": "vehicle",233                        "vehicle_result": {234                            "vehicle_type": vehicle_type,235                            "detected_sound": vehicle_class,236                            "confidence": float(vehicle_score) * 100237                        }238                    }239                240                # If not a vehicle sound, return YAMNet classification241                return {242                    "success": True,243                    "type": "sound",244                    "sound_result": yamnet_result245                }246            else:247                return {248                    "success": False,249                    "message": "No music, vehicle, or sound patterns recognized."250                }251 252    except Exception as e:253        logging.exception("Error during classification:")254        return {"success": False, "message": str(e)}255 256    finally:257        if os.path.exists(temp_filename):258            os.remove(temp_filename)259 260@app.get("/", response_class=HTMLResponse)261async def read_root(request: Request):262    return templates.TemplateResponse("index.html", {"request": request})263 264@app.post("/recognize/")265async def recognize_song_acr(file: UploadFile = File(...)):266    temp_filename = f"temp_recognize_{file.filename}"267    file_content = await file.read()268 269    try:270        with open(temp_filename, "wb") as buffer:271            buffer.write(file_content)272        273        result_json_str = acr_recognizer.recognize_by_file(temp_filename, 0) 274        275        return format_acrcloud_response(result_json_str)276    except Exception as e:277        logging.exception("Error during SDK ACRCloud recognition:")278        return {"success": False, "message": f"Recognition failed: {str(e)}"}279    finally:280        # Changed: Ensure temp file is cleaned up281        if os.path.exists(temp_filename):282            os.remove(temp_filename)283 284@app.post("/upload/")285async def upload_song_acr(file: UploadFile = File(...), song_name: str = Form(None)): 286    temp_filename = f"temp_upload_{file.filename}"287    file_content = await file.read()288 289    try:290        with open(temp_filename, "wb") as buffer:291            buffer.write(file_content)292            293        result_json_str = acr_recognizer.recognize_by_file(temp_filename, 0)294        295        response_data = format_acrcloud_response(result_json_str)296        if song_name and response_data.get("success"):297            response_data["message_context"] = f"Recognition for (originally uploaded as '{song_name}')"298        elif song_name and not response_data.get("success"):299             response_data["message"] = f"Recognition for (originally uploaded as '{song_name}') failed: {response_data.get('message')}"300 301        return response_data302    except Exception as e:303        logging.exception("Error during SDK ACRCloud upload/recognition:")304        return {"success": False, "message": f"Upload/Recognition failed: {str(e)}"}305    finally:306        if os.path.exists(temp_filename):307            os.remove(temp_filename)308 309@app.post("/recognize-live-chunk/")310async def recognize_live_chunk(file: UploadFile = File(...)):311    file_content = await file.read()312 313    if not file_content:314        return {"success": False, "message": "Empty audio chunk received."}315 316    try:317        logging.info(f"Received live chunk, size: {len(file_content)} bytes, filename: {file.filename}")318        319        # First try music recognition320        result_json_str = acr_recognizer.recognize_by_filebuffer(file_content, 0)321        music_result = format_acrcloud_response(result_json_str)322 323        # Check if we got a valid music result324        if music_result["success"] and music_result.get("song_name"):325            # If we have a valid song name, return the music result326            return {327                "success": True,328                "type": "music",329                "music_result": music_result330            }331        332        # If no valid music result, try YAMNet classification333        with tempfile.NamedTemporaryFile(suffix='.webm', delete=False) as temp_file:334            temp_filename = temp_file.name335            temp_file.write(file_content)336 337        try:338            # Convert to WAV first339            wav_filename = temp_filename.replace('.webm', '.wav')340            if convert_audio_to_wav(temp_filename, wav_filename):341                yamnet_result = classify_audio_with_yamnet(wav_filename)342                343                if yamnet_result["success"]:344                    # Check if the sound is vehicle-related345                    is_vehicle, vehicle_class, vehicle_score = is_vehicle_sound(yamnet_result["top_classes"])346                    if is_vehicle:347                        # If it's a vehicle sound, use the neural network for specific classification348                        vehicle_class = predict_vehicle_class(wav_filename)349                        vehicle_type = "Car" if vehicle_class == 0 else "Truck"350                        351                        return {352                            "success": True,353                            "type": "vehicle",354                            "vehicle_result": {355                                "vehicle_type": vehicle_type,356                                "detected_sound": str(vehicle_class),  # Convert to string357                                "confidence": float(vehicle_score) * 100  # Convert to Python float358                            }359                        }360                    361                    # If not a vehicle sound, return YAMNet classification362                    return {363                        "success": True,364                        "type": "sound",365                        "sound_result": {366                            "top_classes": [(str(label), float(score)) for label, score in yamnet_result["top_classes"]]367                        }368                    }369            370            # If we get here, all recognition attempts failed371            return {372                "success": False,373                "message": "No music, vehicle, or sound patterns recognized."374            }375        finally:376            # Clean up temporary files377            if os.path.exists(temp_filename):378                os.remove(temp_filename)379            if os.path.exists(wav_filename):380                os.remove(wav_filename)381 382    except Exception as e:383        logging.exception("Error during audio processing:")384        return {"success": False, "message": f"Processing failed: {str(e)}"}385 386def format_acrcloud_response(result_json_str: str):387    """388    Parses the JSON string response from ACRCloud and formats it.389    """390    try:391        result = json.loads(result_json_str)392        logging.info(f"ACRCloud raw response: {result}")393        394        # Check if we have a valid music result395        if result.get("status", {}).get("code") == 0 and "metadata" in result and "music" in result["metadata"]:396            # Ensure 'music' list is not empty397            if not result["metadata"]["music"]:398                return {"success": False, "message": "No music metadata found in response."}399            400            music_info = result["metadata"]["music"][0]401            title = music_info.get("title")402            403            # If no title, it's not a valid music result404            if not title:405                return {"success": False, "message": "No song title found in response."}406            407            artists_list = music_info.get("artists", [])408            artists = ", ".join([artist["name"] for artist in artists_list if "name" in artist])409            album = music_info.get("album", {}).get("name")410            411            offset_seconds = music_info.get("play_offset_ms", 0) / 1000.0412            if offset_seconds == 0 and "sample_begin_time_offset_ms" in music_info:413                 offset_seconds = music_info.get("sample_begin_time_offset_ms", 0) / 1000.0414            415            confidence = music_info.get("score", 0)416            if confidence == 0 and "result_type" in result: 417                 confidence = result.get("result_type",0) * 25 418 419            return {420                "success": True,421                "song_name": title,422                "artists": artists,423                "album": album,424                "confidence": confidence,425                "offset_seconds": offset_seconds,426                "raw_acr_response": result427            }428        else:429            return {"success": False, "message": result.get("status", {}).get("msg", "Song not recognized or error in response.")}430    except json.JSONDecodeError:431        logging.error(f"Failed to decode ACRCloud JSON response: {result_json_str}")432        return {"success": False, "message": "Error parsing recognition server response."}433    except Exception as e:434        logging.error(f"Error processing ACRCloud response: {e} -- Response was: {result_json_str}")435        return {"success": False, "message": f"An unexpected error occurred: {str(e)}"}436 437@app.post("/predict/")438async def predict_audio(file: UploadFile = File(...)):439    # Save uploaded file temporarily440    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp:441        shutil.copyfileobj(file.file, tmp)442        tmp_path = tmp.name443 444    try:445        # Predict using the neural network446        predicted_class = predict_vehicle_class(tmp_path)447        return {"filename": file.filename, "predicted_class": int(predicted_class)}448    finally:449        os.remove(tmp_path)450 451# Mistral AI configuration452MISTRAL_API_KEY = "SDV5ynlJBEs0n15l2PDvO9eor1ki4dTI"453MISTRAL_API_URL = "https://api.mistral.ai/v1/chat/completions"454 455class ChatRequest(BaseModel):456    system_prompt: str457    user_message: str458 459@app.post("/chat-with-mistral/")460async def chat_with_mistral(request: ChatRequest):461    try:462        headers = {463            "Authorization": f"Bearer {MISTRAL_API_KEY}",464            "Content-Type": "application/json"465        }466 467        data = {468            "model": "mistral-small",469            "messages": [470                {471                    "role": "system",472                    "content": request.system_prompt473                },474                {475                    "role": "user",476                    "content": request.user_message477                }478            ]479        }480 481        response = requests.post(MISTRAL_API_URL, headers=headers, json=data)482 483        if response.status_code == 200:484            ai_response = response.json()["choices"][0]["message"]["content"]485            return {486                "success": True,487                "response": ai_response488            }489        else:490            raise HTTPException(491                status_code=500,492                detail=f"Error from Mistral API: {response.status_code} - {response.text}"493            )494 495    except Exception as e:496        raise HTTPException(497            status_code=500,498            detail=str(e)499        )500 501logging.basicConfig(level=logging.INFO)502