{ "cells": [ { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "import os\n", "os.environ[\"CUDA_VISIBLE_DEVICES\"] = \"3\"" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from io import BytesIO\n", "from urllib.request import urlopen\n", "import librosa\n", "from transformers import Qwen2AudioForConditionalGeneration, AutoProcessor\n", "\n", "processor = AutoProcessor.from_pretrained(\"Qwen/Qwen2-Audio-7B-Instruct\")\n", "model = Qwen2AudioForConditionalGeneration.from_pretrained(\"Qwen/Qwen2-Audio-7B-Instruct\", device_map=\"auto\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\n", "conversation = [\n", " {'role': 'system', 'content': 'You are a helpful assistant.'}, \n", " {\"role\": \"user\", \"content\": [\n", " {\"type\": \"audio\", \"audio_url\": \"https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen2-Audio/audio/glass-breaking-151256.mp3\"},\n", " {\"type\": \"text\", \"text\": \"What's that sound?\"},\n", " ]},\n", "]\n", "text = processor.apply_chat_template(conversation, add_generation_prompt=True, tokenize=False)\n", "audios = []\n", "for message in conversation:\n", " if isinstance(message[\"content\"], list):\n", " for ele in message[\"content\"]:\n", " if ele[\"type\"] == \"audio\":\n", " audios.append(\n", " librosa.load(\n", " BytesIO(urlopen(ele['audio_url']).read()), \n", " sr=processor.feature_extractor.sampling_rate)[0]\n", " )\n", "\n", "inputs = processor(text=text, audios=audios, return_tensors=\"pt\", padding=True)\n", "inputs.input_ids = inputs.input_ids.to(\"cuda\")\n", "\n", "generate_ids = model.generate(**inputs, max_length=256)\n", "generate_ids = generate_ids[:, inputs.input_ids.size(1):]\n", "\n", "response = processor.batch_decode(generate_ids, skip_special_tokens=True, clean_up_tokenization_spaces=False)[0]" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import os \n", "import torchaudio\n", "import torch\n", "import csv\n", "import numpy as np\n", "\n", "import sys\n", "sys.path.append(\"/home/christian/code/ast/src/models\")\n", "\n", "from ast_models import ASTModel \n", "# download pretrained model in this directory\n", "os.environ['TORCH_HOME'] = '/home/christian/code/ast/pretrained_models' \n", "# assume each input spectrogram has 100 time frames\n", "input_tdim = 100\n", "# assume the task has 527 classes\n", "label_dim = 527\n", "# create a pseudo input: a batch of 10 spectrogram, each with 100 time frames and 128 frequency bins \n", "test_input = torch.rand([10, input_tdim, 128]) \n", "# create an AST model\n" ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [], "source": [ "def make_features(audio_tensor, sr, mel_bins, target_length=1024):\n", "\n", " fbank = torchaudio.compliance.kaldi.fbank(\n", " audio_tensor, htk_compat=True, sample_frequency=sr, use_energy=False,\n", " window_type='hanning', num_mel_bins=mel_bins, dither=0.0,\n", " frame_shift=10)\n", "\n", " n_frames = fbank.shape[0]\n", "\n", " p = target_length - n_frames\n", " if p > 0:\n", " m = torch.nn.ZeroPad2d((0, 0, 0, p))\n", " fbank = m(fbank)\n", " elif p < 0:\n", " fbank = fbank[0:target_length, :]\n", "\n", " fbank = (fbank - (-4.2677393)) / (4.5689974 * 2)\n", " return fbank\n", "\n", "\n", "def load_label(label_csv):\n", " with open(label_csv, 'r') as f:\n", " reader = csv.reader(f, delimiter=',')\n", " lines = list(reader)\n", " labels = []\n", " ids = [] # Each label has a unique id such as \"/m/068hy\"\n", " for i1 in range(1, len(lines)):\n", " id = lines[i1][1]\n", " label = lines[i1][2]\n", " ids.append(id)\n", " labels.append(label)\n", " return labels" ] }, { "cell_type": "code", "execution_count": 16, "metadata": {}, "outputs": [], "source": [ "# 4. map the post-prob to label\n", "label_csv = '/home/christian/code/ast/egs/audioset/data/class_labels_indices.csv' # label and indices for audioset data\n", "\n", "labels = load_label(label_csv)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\n", "# load audio \n", "audio_path = \"/home/christian/audio/reference-audio-wav/02 Dreams.wav\"\n", "audio_path = \"/home/christian/audio/reference-audio-wav/02 Take Five.wav\"\n", "x, sr = torchaudio.load(audio_path)\n", "x = x.mean(dim=0, keepdim=True)[:, :sr*60]\n", "x = torchaudio.functional.resample(x, sr, 16000)\n", "\n", "# 1. make feature for predict\n", "feats = make_features(x, 16000, mel_bins=128) # shape(1024, 128)\n", "print(feats.shape)\n", "\n", "# assume each input spectrogram has 100 time frames\n", "input_tdim = feats.shape[0]\n", "\n", "# 3. feed the data feature to model\n", "feats_data = feats.expand(1, input_tdim, 128) # reshape the feature\n", "print(feats_data.shape)\n", "\n", "# load model\n", "ast_mdl = ASTModel(label_dim=label_dim, input_tdim=input_tdim, imagenet_pretrain=True, audioset_pretrain=True)\n", "ast_mdl.cpu()\n", "\n", "with torch.no_grad():\n", " output = ast_mdl.forward(feats_data)\n", " output = torch.sigmoid(output)\n", "result_output = output.data.cpu().numpy()[0]\n", "\n", "sorted_indexes = np.argsort(result_output)[::-1]\n", "\n", "# Print audio tagging top probabilities\n", "print('[*INFO] predice results:')\n", "for k in range(10):\n", " print('{}: {:.4f}'.format(np.array(labels)[sorted_indexes[k]],\n", " result_output[sorted_indexes[k]]))" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "suno_qwen", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10.16" } }, "nbformat": 4, "nbformat_minor": 2 }