bot update

README.md

@@ -1 +1,97 @@
-# MusBot
+# 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)
+```

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)

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))