import io import fnmatch import os from hashlib import sha1 from typing import List, Tuple import numpy as np from pydub import AudioSegment from pydub.utils import audioop from dejavu.third_party import wavio from s3_file import S3File def unique_hash(file_path: io.RawIOBase) -> str: """ Small function to generate a hash to uniquely generate a file. Inspired by MD5 version here: http://stackoverflow.com/a/1131255/712997 Works with large files. :param file_path: path to file. :return: a hash in an hexagesimal string form. """ s = sha1() buf = file_path.read() s.update(buf) return s.hexdigest().upper() def find_files(path: str, extensions: List[str]) -> List[Tuple[str, str]]: """ Get all files that meet the specified extensions. :param path: path to a directory with audio files. :param extensions: file extensions to look for. :return: a list of tuples with file name and its extension. """ # Allow both with ".mp3" and without "mp3" to be used for extensions extensions = [e.replace(".", "") for e in extensions] results = [] for dirpath, dirnames, files in os.walk(path): for extension in extensions: for f in fnmatch.filter(files, f"*.{extension}"): p = os.path.join(dirpath, f) results.append((p, extension)) return results def read_segments(file_name: S3File, limit: int = None) -> List[Tuple[List[List[int]], int, str]]: # pydub does not support 24-bit wav files, use wavio when this occurs pass def read(file_name: S3File, limit: int = None) -> Tuple[List[List[int]], int, str]: """ Reads any file supported by pydub (ffmpeg) and returns the data contained within. If file reading fails due to input being a 24-bit wav file, wavio is used as a backup. Can be optionally limited to a certain amount of seconds from the start of the file by specifying the `limit` parameter. This is the amount of seconds from the start of the file. :param file_name: S3 file to be read. :param limit: number of seconds to limit. :return: tuple list of (channels, sample_rate, content_file_hash). """ # pydub does not support 24-bit wav files, use wavio when this occurs try: audiofile = AudioSegment.from_file(file_name) # if limit: # offset = 120*1000 # audiofile = audiofile[offset:(limit * 2 * 1000)+offset] if limit: audiofile = audiofile[:limit * 1000] data = np.fromstring(audiofile.raw_data, np.int16) channels = [] for chn in range(audiofile.channels): channels.append(data[chn::audiofile.channels]) audiofile.frame_rate except audioop.error: audiofile = wavio.read(file_name) if limit: audiofile = audiofile[:limit * 1000] audiofile = audiofile.data.T audiofile = audiofile.astype(np.int16) channels = [] for chn in audiofile: channels.append(chn) return channels, audiofile.frame_rate, unique_hash(file_name) def get_audio_name_from_path(file_path: str) -> str: """ Extracts song name from a file path. :param file_path: path to an audio file. :return: file name """ return os.path.splitext(os.path.basename(file_path))