diff --git a/README.md b/README.md index 45a883a..f485f24 100644 --- a/README.md +++ b/README.md @@ -1 +1,97 @@ -# MusBot \ No newline at end of file +# MusBot. Телеграм бот для распознавания музыки +## Задача +Разработать аналог приложения "Shazam" - Телеграм бота для распознавания музыки через голосовые или музыкальные файлы формата MP3. +## Анализ предметной области +Для того, чтобы представить музыку в удобном для Shazam или любого другого сервиса виде, мы выделили следующие шаги преобразования: +1. Анализ частот. Преобразование Фурье; +2. Создание «карты созвездий»; +3. Устранение времени. +### Анализ частот. Преобразование Фурье +Преобразование Фурье - математический метод, позволяющий представить музыку в её первоначальном виде – до того, как все частоты смешались в один поток. + +![image](https://github.com/miroutom/MusBot/assets/78901500/77eaa2fc-dc6d-461c-bad0-95892ac39333) + + +Использование данного метода позволяет нам получить график, именуемый спектрограммой, показывающий зависимость частот (ось Y) от времени (ось X). Ниже представлена реализация метода Фурье с помощью библиотеки `scipy`. + +![image](https://github.com/miroutom/MusBot/assets/78901500/9a675d4c-bb5a-4753-83e9-867a3ac7f747) + + +### Создание "карты созвездий" +Создание "карты созвездий" или "отпечатка" композиции - это процесс анализа спектрограммы и выделения на ней самых ярких областей в каждый момент времени. Использование данной карты позволяет сократить размер композиции, а также удалить все лишние звуки. + + +![image](https://github.com/miroutom/MusBot/assets/78901500/81934d68-afd0-425c-a8bf-1d1200152929) + + +### Устранение времени +На данном этапе возникла довольно серьёзная проблема - записанный пользователем фрагмент может оказаться в любом моменте песни: будь то начало, середина или конец. В итоге, алгоритм примет совершенно другую карту. Поэтому на данном шаге было принято решение найти какую-то опорную точку O и связать её с несколькими другими точками. Так, вместо конкретных частот с привязкой ко времени, мы храним информацию о том, как эти частоты связаны между собой. + +## Реализация алгоритма + +### Обработка аудио. Создание спектрограммы +Для создания спектрограммы сперва нужно было выбрать необходимую библиотеку. Мы предпочли `scipy`, так как в сравнении с её не менее известным аналогом `librosa`, она гораздо быстрее. +```Python +from scipy.io import wavfile +from scipy.signal import spectrogram +``` +Данная библиотека обрабатывает аудиофайлы в формате WAV. Но поскольку мы решили, что бот будет принимать форматы MP3 и OGG, необходимо было сделать предварительную обработку. +```Python + if input_file[-3:] == 'mp3': + sound = AudioSegment.from_mp3(input_file) + sound.export(output_file, format='wav') + elif input_file[-3:] == 'ogg': + sound = AudioSegment.from_ogg(input_file) + sound.export(output_file, format='wav') +``` + +После чего файл проверяется на количество каналов в нём (`scipy` обрабатывает только одноканальную музыку) и исходя из этого создаётся спектрограмма. +```Python + try: + f, t, Sxx = spectrogram(data[:, 0], samplerate, nperseg=FRAME_SIZE, + noverlap=HOP_SIZE, window='hamming') + except: + f, t, Sxx = spectrogram(data, samplerate, nperseg=FRAME_SIZE, + noverlap=HOP_SIZE, window='hamming') +``` + +Затем данная картинка отображается при помощи использования библиотеки `matplotlib`. + +![image](https://github.com/miroutom/MusBot/assets/78901500/89943812-6d59-4afa-a537-08c473e9eb5c) + + +### Создание "карты созвездий" +Данный этап был значительно упрощён в виду того, что библиотеки, реализующие `fingerprinting` оказались устаревшими или изменёнными, поэтому было принятно решение создать карту созвездий с использованием `np.array` и функции `flatten`, преобразовывающей многомерный массив в одномерный. +```Python + # read audio file + samplerate, data = wavfile.read(audio_file) + # generate spectrogram + try: + f, t, Sxx = spectrogram(data[:, 0], samplerate) + except: + f, t, Sxx = spectrogram(data, samplerate) + # replace zeros with epsilon to handle divide by zero errors + Sxx[Sxx == 0] = np.finfo(float).eps + # take logarithm of spectrogram + log_spectrogram = np.log(Sxx) + fingerprint = log_spectrogram.flatten() + return fingerprint +``` + +### Нахождение наибольшего совпадения +Для нахождения наибольшего совпадения мы написали функцию косинусного расстояния между векторами, работающей по следующему принципу: +1. Уравнять размеры векторов, заполнив меньший нулями +2. Вычислить скалярное произведение и норму векторов +3. Рассчитать косинусное расстояние как скалярное произведение, поделённое на произведение норм +```Python + # pad the smaller fingerprint with zeros to match the size of the larger fingerprint + if len(x) < len(y): + x = np.pad(x, (0, len(y) - len(x)), mode='constant') + elif len(y) < len(x): + y = np.pad(y, (0, len(x) - len(y)), mode='constant') + # calculate cosine distance + dot_product = np.dot(x, y) + norm_x = np.linalg.norm(x) + norm_y = np.linalg.norm(y) + return dot_product / (norm_x * norm_y) +``` diff --git a/bot.py b/bot.py new file mode 100644 index 0000000..6872af6 --- /dev/null +++ b/bot.py @@ -0,0 +1,164 @@ +import logging +import threading + +from aiogram import Bot, Dispatcher, types, executor +from dotenv import load_dotenv +from pathlib import Path +from pyglet.media import Player, load + +from spectr import create_spectrogram +from matching import finish, finish_debug, get_song +import os + +dotenv_path = Path('.env.txt') +load_dotenv(dotenv_path=dotenv_path) + +# logging so don't miss important messages +logging.basicConfig(level=logging.INFO) +bot_token = os.getenv("BOT_TOKEN") +# bot's object +bot = Bot(token=bot_token) +dp = Dispatcher(bot) + +command = '' + + +# handler for /start +@dp.message_handler(commands=['start']) +async def cmd_start(message: types.Message): + await message.answer("Hello! I'm music recognizing bot. Waiting for your commands...") + + +@dp.message_handler(commands=['run']) +async def run(message: types.Message): + global command + command = 'run' + await message.answer("Send me an audiofile or voice message. I'll show you its title") + + +@dp.message_handler(commands=['debug']) +async def debug(message: types.Message): + global command + command = 'debug' + await message.answer("Send me an audiofile or voice message. I'll show you its title " + "and some additional info") + + +player = None + + +def play_audio(audio_file): + global player + player = Player() + source = load(audio_file) + player.queue(source) + player.play() + + +def stop_audio(): + global player + if player is not None: + player.pause() + player.delete() + player = None + + +@dp.callback_query_handler(lambda c: c.data == 'play') +async def play_callback_handler(callback_query: types.CallbackQuery): + audio_file = get_song() + threading.Thread(target=play_audio, args=(audio_file,)).start() + + +@dp.callback_query_handler(lambda c: c.data == 'stop') +async def stop_callback_handler(callback_query: types.CallbackQuery): + stop_audio() + + +@dp.message_handler(commands=['play']) +async def play_music(message: types.Message): + audio_file = get_song() + await bot.send_audio(message.chat.id, audio=open(audio_file, 'rb')) + + +@dp.message_handler(commands=['help']) +async def help(message: types.Message): + await message.answer('All commands:\n' + '/start - run bot\n' + '/run - waiting for audio and shows its title\n' + '/debug - gives an additional info of your file\n' + '/play - sends an audio player') + + +user_home_dir = os.path.expanduser('~') +user = os.path.split(user_home_dir)[-1] +disk = os.path.split(user_home_dir)[0] +common_path = os.path.join(disk, user, 'Desktop') + +try: + if not os.path.isdir('Downloaded audio'): + os.chdir(common_path) + os.mkdir('Downloaded audio') +except: + print() + + +# handler for audio messages, voice messages and photos +@dp.message_handler(content_types=['audio', 'voice']) +async def show_spectrogram(message: types.Message): + global command + unique_id = path_to_file = file_info = '' + # getting a directory for saving a file + directory = os.path.join(disk, user, 'Desktop', 'Downloaded audio') + # checking a type of message + if message.audio is not None: + audio_id = message.audio.file_id + unique_id = message.audio.file_unique_id + file_info = await bot.get_file(audio_id) + path_to_file = os.path.join(directory, f'{unique_id}.mp3') + elif message.voice is not None: + audio_id = message.voice.file_id + unique_id = message.voice.file_unique_id + file_info = await bot.get_file(audio_id) + path_to_file = os.path.join(directory, f'{unique_id}.ogg') + + # downloading a file + try: + await bot.download_file(file_info.file_path, + destination=os.path.join(path_to_file)) + await message.answer("Your audio is successfully downloaded") + except: + await message.answer("Error downloading") + + print(command) + if command == 'run': + create_spectrogram(path_to_file, unique_id) + + path_to_photo = os.path.join(disk, user, 'Desktop', + 'Created spectrograms', f'{unique_id}.png') + try: + await message.answer(finish()) + except: + await message.answer("I'm Sorry. Something went wrong...") + elif command == 'debug': + create_spectrogram(path_to_file, unique_id) + + path_to_photo = os.path.join(disk, user, 'Desktop', + 'Created spectrograms', f'{unique_id}.png') + # showing spectrogram + try: + with open(path_to_photo, 'rb') as photo: + await message.answer_photo(photo, caption='Your spectrogram') + photo.close() + except: + await message.answer("Error showing") + + try: + await message.answer(finish_debug()) + except: + await message.answer("I'm Sorry. Something went wrong...") + else: + await message.answer('Try some existed commands, please') + + +if __name__ == '__main__': + executor.start_polling(dp, skip_updates=True) diff --git a/fingerprinting.py b/fingerprinting.py new file mode 100644 index 0000000..d7234a2 --- /dev/null +++ b/fingerprinting.py @@ -0,0 +1,46 @@ +from scipy.io import wavfile +from scipy.signal import spectrogram +import numpy as np +from scipy.ndimage.morphology import (generate_binary_structure, iterate_structure, + binary_erosion) +from scipy.ndimage.filters import maximum_filter + + +# function to generate a fingerprint of an audio file +def finger_print(audio_file): + # read audio file + samplerate, data = wavfile.read(audio_file) + # generate spectrogram + try: + f, t, Sxx = spectrogram(data[:, 0], samplerate) + except: + f, t, Sxx = spectrogram(data, samplerate) + # replace zeros with epsilon to handle divide by zero errors + Sxx[Sxx == 0] = np.finfo(float).eps + # take logarithm of spectrogram + log_spectrogram = np.log(Sxx) + fingerprint = log_spectrogram.flatten() + return fingerprint + + +def get_peaks(spectrogram, no_of_iteration=10, min_amplitude=10): + structure = generate_binary_structure(2, 2) + neighborhood = iterate_structure(structure, no_of_iteration) + + local_max = maximum_filter(spectrogram, footprint=neighborhood) == spectrogram + background = (spectrogram == 0) + eroded_background = binary_erosion(background, structure=neighborhood, border_value=1) + + # applying XOR between matrices to get boolean mask of spectrogram + detected_peaks = local_max ^ eroded_background + + # getting peaks, their freqs and times + peaks = spectrogram[detected_peaks].flatten() + peak_freqs, peak_times = np.where(detected_peaks) + + peak_idx = np.where(peaks > min_amplitude) + + freqs = peak_freqs[peak_idx] + times = peak_times[peak_idx] + + return list(zip(freqs, times)) diff --git a/matching.py b/matching.py new file mode 100644 index 0000000..0f8c1e6 --- /dev/null +++ b/matching.py @@ -0,0 +1,96 @@ +import numpy as np +import os +import spectr +from fingerprinting import finger_print +from pydub import AudioSegment +from mutagen.mp3 import MP3 + + +def cosine_distance(x, y): + # pad the smaller fingerprint with zeros to match the size of the larger fingerprint + if len(x) < len(y): + x = np.pad(x, (0, len(y) - len(x)), mode='constant') + elif len(y) < len(x): + y = np.pad(y, (0, len(x) - len(y)), mode='constant') + # calculate cosine distance + dot_product = np.dot(x, y) + norm_x = np.linalg.norm(x) + norm_y = np.linalg.norm(y) + return dot_product / (norm_x * norm_y) + + +# function to find the closest match for an audio file in a given set of fingerprints +def find_closest_match(audio_file, fingerprint_maps): + # generate fingerprint for query audio file + query_fp = finger_print(audio_file) + # initialize maximum distance and closest match + max_distance = 0 + closest_match = '' + # iterate through fingerprint maps and find closest match + for key in fingerprint_maps: + # calculate cosine distance between query fingerprint and current fingerprint + distance = cosine_distance(query_fp, fingerprint_maps[key]) + # update maximum distance and closest match if a closer match is found + if distance > max_distance: + max_distance = distance + closest_match = key + # return closest match and its distance from query audio file + return closest_match, max_distance + + +path = "D:\\PyCharm\\PyCharm Community Edition 2023.1\\pythonProject\\music" + +# generate fingerprints for all audio files in directory +fingerprint_maps = {} +for file_name in os.listdir(path): + file_path = os.path.join(path, file_name) + if file_path.endswith(".mp3"): + # convert mp3 file to wav format + output_file = os.path.join(path, f"{os.path.splitext(file_name)[0]}.wav") + sound = AudioSegment.from_mp3(file_path) + sound.export(output_file, format='wav') + try: + fp = finger_print(output_file) + fingerprint_maps[file_name] = fp + except Exception as e: + print(f"Error processing file {output_file}: {e}") + finally: + # delete the wav file + os.remove(output_file) + + +def get_duration(audio_file): + audio_ = MP3(audio_file) + duration_ = audio_.info.length + minutes = int(duration_ // 60) + seconds = int(duration_ % 60) + duration_format = f"{minutes:02d}:{seconds:02d}" + return duration_format + + +def finish(): + audio = os.path.join(spectr.output_file) + # find closest match for query audio file + closest_match_, max_distance_ = find_closest_match(audio, fingerprint_maps) + path_to_file = os.path.join(path, closest_match_) + duration = get_duration(path_to_file) + # print closest match + return f"Song: {closest_match_[:-4]}\nDuration: {duration}" + + +def finish_debug(): + audio = os.path.join(spectr.output_file) + # find closest match for query audio file + closest_match_, max_distance_ = find_closest_match(audio, fingerprint_maps) + path_to_file = os.path.join(path, closest_match_) + duration = get_duration(path_to_file) + # print closest match + return f"Song: {closest_match_[:-4]}\nDuration: {duration}\nDistance: {max_distance_}" + + +def get_song(): + audio = os.path.join(spectr.output_file) + # find closest match for query audio file + closest_match_, max_distance_ = find_closest_match(audio, fingerprint_maps) + path_to_file = os.path.join(path, closest_match_) + return path_to_file diff --git a/spectr.py b/spectr.py new file mode 100644 index 0000000..dbd9561 --- /dev/null +++ b/spectr.py @@ -0,0 +1,68 @@ +import matplotlib.pyplot as plt +from scipy.io import wavfile +from scipy.signal import spectrogram +import numpy as np +import os +from fingerprinting import get_peaks +from pydub import AudioSegment + +user_home_dir = os.path.expanduser('~') +user = os.path.split(user_home_dir)[-1] +disk = os.path.split(user_home_dir)[0] +common_path = os.path.join(disk, user, 'Desktop') + +FRAME_SIZE = 2 ** 12 +HOP_SIZE = int(FRAME_SIZE * 0.5) + +input_file = '' +output_file = '' + + +def create_spectrogram(path_to_file: str, unique_id: str): + global input_file, output_file + input_file = f'{path_to_file}' + output_file = f'{path_to_file}.wav' + + if input_file[-3:] == 'mp3': + sound = AudioSegment.from_mp3(input_file) + sound.export(output_file, format='wav') + elif input_file[-3:] == 'ogg': + sound = AudioSegment.from_ogg(input_file) + sound.export(output_file, format='wav') + + samplerate, data = wavfile.read(output_file) + + try: + f, t, Sxx = spectrogram(data[:, 0], samplerate, nperseg=FRAME_SIZE, + noverlap=HOP_SIZE, window='hamming') + except: + f, t, Sxx = spectrogram(data, samplerate, nperseg=FRAME_SIZE, + noverlap=HOP_SIZE, window='hamming') + + Sxx[Sxx == 0] = np.finfo(float).eps + + max_time = int(np.max(t)) + max_freq = int(np.max(f)) + + spec_peaks = get_peaks(Sxx[:max_freq, :max_time]) + plot_peaks = np.array(spec_peaks) + plot_peaks[:, 0] = (plot_peaks[:, 0] * max_freq) / len(f) + + plt.pcolormesh(t, f, 10 * np.log10(Sxx / Sxx.max()), vmin=-120, + vmax=0, cmap='inferno') + plt.colorbar() + plt.scatter(plot_peaks[:, 1], plot_peaks[:, 0], s=10, color='white') + plt.ylabel('Frequency [Hz]') + plt.xlabel('Time [s]') + + img = f'{unique_id}.png' + try: + if not os.path.isdir('Created spectrograms'): + os.chdir(common_path) + os.mkdir('Created spectrograms') + except: + print() + + path = os.path.join(disk, user, 'Desktop', 'Created spectrograms', img) + plt.savefig(path) + plt.clf() \ No newline at end of file