Spaces:
Running
Running
| """Module containing common function. | |
| """ | |
| import os | |
| import traceback | |
| import numpy as np | |
| from pathlib import Path | |
| import config as cfg | |
| def collect_audio_files(path: str): | |
| """Collects all audio files in the given directory. | |
| Args: | |
| path: The directory to be searched. | |
| Returns: | |
| A sorted list of all audio files in the directory. | |
| """ | |
| # Get all files in directory with os.walk | |
| files = [] | |
| for root, _, flist in os.walk(path): | |
| for f in flist: | |
| if not f.startswith(".") and f.rsplit(".", 1)[-1].lower() in cfg.ALLOWED_FILETYPES: | |
| files.append(os.path.join(root, f)) | |
| return sorted(files) | |
| def readLines(path: str): | |
| """Reads the lines into a list. | |
| Opens the file and reads its contents into a list. | |
| It is expected to have one line for each species or label. | |
| Args: | |
| path: Absolute path to the species file. | |
| Returns: | |
| A list of all species inside the file. | |
| """ | |
| return Path(path).read_text(encoding="utf-8").splitlines() if path else [] | |
| def list_subdirectories(path: str): | |
| """Lists all directories inside a path. | |
| Retrieves all the subdirectories in a given path without recursion. | |
| Args: | |
| path: Directory to be searched. | |
| Returns: | |
| A filter sequence containing the absolute paths to all directories. | |
| """ | |
| return filter(lambda el: os.path.isdir(os.path.join(path, el)), os.listdir(path)) | |
| def random_split(x, y, val_ratio=0.2): | |
| """Splits the data into training and validation data. | |
| Makes sure that each class is represented in both sets. | |
| Args: | |
| x: Samples. | |
| y: One-hot labels. | |
| val_ratio: The ratio of validation data. | |
| Returns: | |
| A tuple of (x_train, y_train, x_val, y_val). | |
| """ | |
| # Set numpy random seed | |
| np.random.seed(cfg.RANDOM_SEED) | |
| # Get number of classes | |
| num_classes = y.shape[1] | |
| # Initialize training and validation data | |
| x_train, y_train, x_val, y_val = [], [], [], [] | |
| # Split data | |
| for i in range(num_classes): | |
| # Get indices of current class | |
| indices = np.where(y[:, i] == 1)[0] | |
| # Get number of samples for each set | |
| num_samples = len(indices) | |
| num_samples_train = max(1, int(num_samples * (1 - val_ratio))) | |
| num_samples_val = max(0, num_samples - num_samples_train) | |
| # Randomly choose samples for training and validation | |
| np.random.shuffle(indices) | |
| train_indices = indices[:num_samples_train] | |
| val_indices = indices[num_samples_train:num_samples_train + num_samples_val] | |
| # Append samples to training and validation data | |
| x_train.append(x[train_indices]) | |
| y_train.append(y[train_indices]) | |
| x_val.append(x[val_indices]) | |
| y_val.append(y[val_indices]) | |
| # Concatenate data | |
| x_train = np.concatenate(x_train) | |
| y_train = np.concatenate(y_train) | |
| x_val = np.concatenate(x_val) | |
| y_val = np.concatenate(y_val) | |
| # Shuffle data | |
| indices = np.arange(len(x_train)) | |
| np.random.shuffle(indices) | |
| x_train = x_train[indices] | |
| y_train = y_train[indices] | |
| indices = np.arange(len(x_val)) | |
| np.random.shuffle(indices) | |
| x_val = x_val[indices] | |
| y_val = y_val[indices] | |
| return x_train, y_train, x_val, y_val | |
| def mixup(x, y, augmentation_ratio=0.25, alpha=0.2): | |
| """Apply mixup to the given data. | |
| Mixup is a data augmentation technique that generates new samples by | |
| mixing two samples and their labels. | |
| Args: | |
| x: Samples. | |
| y: One-hot labels. | |
| augmentation_ratio: The ratio of augmented samples. | |
| alpha: The beta distribution parameter. | |
| Returns: | |
| Augmented data. | |
| """ | |
| # Calculate the number of samples to augment based on the ratio | |
| num_samples_to_augment = int(len(x) * augmentation_ratio) | |
| for _ in range(num_samples_to_augment): | |
| # Randomly choose one instance from the dataset | |
| index = np.random.choice(len(x)) | |
| x1, y1 = x[index], y[index] | |
| # Randomly choose a different instance from the dataset | |
| second_index = np.random.choice(len(x)) | |
| while second_index == index: | |
| second_index = np.random.choice(len(x)) | |
| x2, y2 = x[second_index], y[second_index] | |
| # Generate a random mixing coefficient (lambda) | |
| lambda_ = np.random.beta(alpha, alpha) | |
| # Mix the embeddings and labels | |
| mixed_x = lambda_ * x1 + (1 - lambda_) * x2 | |
| mixed_y = lambda_ * y1 + (1 - lambda_) * y2 | |
| # Replace one of the original samples and labels with the augmented sample and labels | |
| x[index] = mixed_x | |
| y[index] = mixed_y | |
| return x, y | |
| def label_smoothing(y, alpha=0.1): | |
| # Subtract alpha from correct label when it is >0 | |
| y[y > 0] -= alpha | |
| # Assigned alpha to all other labels | |
| y[y == 0] = alpha / y.shape[0] | |
| return y | |
| def upsampling(x, y, ratio=0.5, mode="repeat"): | |
| """Balance data through upsampling. | |
| We upsample minority classes to have at least 10% (ratio=0.1) of the samples of the majority class. | |
| Args: | |
| x: Samples. | |
| y: One-hot labels. | |
| ratio: The minimum ratio of minority to majority samples. | |
| mode: The upsampling mode. Either 'repeat', 'mean' or 'smote'. | |
| Returns: | |
| Upsampled data. | |
| """ | |
| # Set numpy random seed | |
| np.random.seed(cfg.RANDOM_SEED) | |
| # Determin min number of samples | |
| min_samples = int(np.max(y.sum(axis=0)) * ratio) | |
| x_temp = [] | |
| y_temp = [] | |
| if mode == 'repeat': | |
| # For each class with less than min_samples ranomdly repeat samples | |
| for i in range(y.shape[1]): | |
| while y[:, i].sum() + len(y_temp) < min_samples: | |
| # Randomly choose a sample from the minority class | |
| random_index = np.random.choice(np.where(y[:, i] == 1)[0]) | |
| # Append the sample and label to a temp list | |
| x_temp.append(x[random_index]) | |
| y_temp.append(y[random_index]) | |
| elif mode == 'mean': | |
| # For each class with less than min_samples | |
| # select two random samples and calculate the mean | |
| for i in range(y.shape[1]): | |
| x_temp = [] | |
| y_temp = [] | |
| while y[:, i].sum() + len(y_temp) < min_samples: | |
| # Randomly choose two samples from the minority class | |
| random_indices = np.random.choice(np.where(y[:, i] == 1)[0], 2) | |
| # Calculate the mean of the two samples | |
| mean = np.mean(x[random_indices], axis=0) | |
| # Append the mean and label to a temp list | |
| x_temp.append(mean) | |
| y_temp.append(y[random_indices[0]]) | |
| elif mode == 'smote': | |
| # For each class with less than min_samples apply SMOTE | |
| for i in range(y.shape[1]): | |
| x_temp = [] | |
| y_temp = [] | |
| while y[:, i].sum() + len(y_temp) < min_samples: | |
| # Randomly choose a sample from the minority class | |
| random_index = np.random.choice(np.where(y[:, i] == 1)[0]) | |
| # Get the k nearest neighbors | |
| k = 5 | |
| distances = np.sqrt(np.sum((x - x[random_index])**2, axis=1)) | |
| indices = np.argsort(distances)[1:k+1] | |
| # Randomly choose one of the neighbors | |
| random_neighbor = np.random.choice(indices) | |
| # Calculate the difference vector | |
| diff = x[random_neighbor] - x[random_index] | |
| # Randomly choose a weight between 0 and 1 | |
| weight = np.random.uniform(0, 1) | |
| # Calculate the new sample | |
| new_sample = x[random_index] + weight * diff | |
| # Append the new sample and label to a temp list | |
| x_temp.append(new_sample) | |
| y_temp.append(y[random_index]) | |
| # Append the temp list to the original data | |
| if len(x_temp) > 0: | |
| x = np.vstack((x, np.array(x_temp))) | |
| y = np.vstack((y, np.array(y_temp))) | |
| # Shuffle data | |
| indices = np.arange(len(x)) | |
| np.random.shuffle(indices) | |
| x = x[indices] | |
| y = y[indices] | |
| return x, y | |
| def saveToCache(cache_file: str, x_train: np.ndarray, y_train: np.ndarray, labels: list[str]): | |
| """Saves the training data to a cache file. | |
| Args: | |
| cache_file: The path to the cache file. | |
| x_train: The training samples. | |
| y_train: The training labels. | |
| labels: The list of labels. | |
| """ | |
| # Create cache directory | |
| os.makedirs(os.path.dirname(cache_file), exist_ok=True) | |
| # Save to cache | |
| np.savez_compressed(cache_file, x_train=x_train, y_train=y_train, labels=labels) | |
| def loadFromCache(cache_file: str): | |
| """Loads the training data from a cache file. | |
| Args: | |
| cache_file: The path to the cache file. | |
| Returns: | |
| A tuple of (x_train, y_train, labels). | |
| """ | |
| # Load from cache | |
| cache = np.load(cache_file, allow_pickle=True) | |
| # Get data | |
| x_train = cache["x_train"] | |
| y_train = cache["y_train"] | |
| labels = cache["labels"] | |
| return x_train, y_train, labels | |
| def clearErrorLog(): | |
| """Clears the error log file. | |
| For debugging purposes. | |
| """ | |
| if os.path.isfile(cfg.ERROR_LOG_FILE): | |
| os.remove(cfg.ERROR_LOG_FILE) | |
| def writeErrorLog(ex: Exception): | |
| """Writes an exception to the error log. | |
| Formats the stacktrace and writes it in the error log file configured in the config. | |
| Args: | |
| ex: An exception that occurred. | |
| """ | |
| with open(cfg.ERROR_LOG_FILE, "a") as elog: | |
| elog.write("".join(traceback.TracebackException.from_exception(ex).format()) + "\n") | |