TrixyZ/trainer/utils/file_utils.py

from cmath import e
from genericpath import isdir
import os
from re import A
import wave
import logging
from pydub import AudioSegment
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import librosa.display
import collections

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
trainer_classes = {}

def ensure_directory_exists(directory):
    if not os.path.exists(directory):
        os.makedirs(directory)

def list_files_in_directory(directory, extension=None, deep=0):
    if not os.path.exists(directory):
        raise FileNotFoundError(f"Directory {directory} does not exist.")
    #isarray = type(extension).__name__ in ('list', 'tuple', 'dict')
    isarray = isinstance(extension, (list, tuple, dict))
    files = []
    for file in os.listdir(directory):
        if os.path.isdir(directory+file+"/") and deep>0 and file != "processed":
            dfs = list_files_in_directory(directory+file+"/", extension=extension, deep=deep-1)
            for df in dfs:
                files.append(df)
            continue
        if extension:
            if isarray:
                found=False
                for e in extension:
                    if file.endswith(e):
                        found=True
                if found==False:
                    continue
            elif not file.endswith(extension):
                continue
        files.append(directory+file)
    return files

def list_directories_in_directory(directory):
    if not os.path.exists(directory):
        raise FileNotFoundError(f"Directory {directory} does not exist.")
    
    directories = [d for d in os.listdir(directory) if os.path.isdir(os.path.join(directory, d))]
    return directories

def save_file(directory, filename, data):
    ensure_directory_exists(directory)
    file_path = os.path.join(directory, filename)
    with open(file_path, 'wb') as file:
        file.write(data)

def delete_file(file_path):
    if os.path.exists(file_path):
        os.remove(file_path)

def get_file_metadata(file_path):
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"File {file_path} does not exist.")
    
    try:
        with wave.open(file_path, 'rb') as wf:
            frames = wf.getnframes()
            rate = wf.getframerate()
            length = frames / float(rate)
        size = os.path.getsize(file_path) / 1024  # size in KB
        return round(length, 2), round(size, 2)
    except wave.Error as e:
        logger.error(f"Error processing WAV file {file_path}: {e}")
        raise ValueError(f"Error processing WAV file: {e}")

def is_valid_wav(file_path):
    try:
        with wave.open(file_path, 'rb') as wf:
            return True
    except wave.Error:
        return False

def get_file_size(file_path):
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"File {file_path} does not exist.")
    size = os.path.getsize(file_path)
    return get_Size_String(size)

def get_Size_String(value):
    ext = ["Byte","KB","MB","GB","TB"]
    i = 0
    while value > 1024:
        value /= 1024
        i += 1
    return f"{value:.2f} {ext[i]}"

def update_training_time(model_name, additional_time):
    training_time_path = f"./model/hotword/{model_name}/training_time.txt"
    ensure_directory_exists(os.path.dirname(training_time_path))
    if os.path.exists(training_time_path):
        with open(training_time_path, "r") as f:
            total_time = int(f.read().strip())
    else:
        total_time = 0
    total_time += additional_time
    with open(training_time_path, "w") as f:
        f.write(str(total_time))

def get_training_time(model_name):
    training_time_path = f"./model/hotword/{model_name}/training_time.txt"
    if os.path.exists(training_time_path):
        with open(training_time_path, "r") as f:
            return int(f.read().strip())
    return 0

def resample_training_data(training_directory, background_directory=None):
    print("  training_directory: "+training_directory)
    print("  background_directory: "+background_directory)
    input_dir = os.path.join(training_directory)
    output_dir = os.path.join(input_dir, 'processed/')
    print("  input_dir: "+input_dir)
    print("  output_dir: "+output_dir)
    ensure_directory_exists(output_dir)

    hotword_files = list_files_in_directory(input_dir, extension='.wav')
    if background_directory:
        background_files = list_files_in_directory(background_directory, extension='.wav')
    else:
        background_files = []

    for hotword_file in hotword_files:
        hotword_audio = AudioSegment.from_wav(hotword_file)
        fname1=os.path.splitext(os.path.basename(hotword_file))[0]+".wav"
        hotword_audio.export(os.path.join(output_dir, fname1), format='wav')  # Save raw hotword

        for background_file in background_files:
            fname1=os.path.splitext(os.path.basename(hotword_file))[0]
            fname2=os.path.splitext(os.path.basename(background_file))[0]
            mixed_filename = f"{fname1}_{fname2}.wav"
            ofile = os.path.join(output_dir, mixed_filename)
            if os.path.exists(ofile)==False:
                background_audio = AudioSegment.from_wav(background_file)
                mixed_audio = hotword_audio.overlay(background_audio, position=0)
                print(f"  Mixed Filename: {mixed_filename}")
                mixed_audio.export(ofile, format='wav')

def preprocessing_training_data(model_dir,data_dir, sc_data_dir, fail_data_dir):
    import librosa
    
    data_path_dict = {
        0: list_files_in_directory(fail_data_dir, extension=['.wav','.mp3','.m4a'], deep=3),
        1: list_files_in_directory(data_dir, extension=['.wav','.mp3','.m4a'], deep=3),
        2: list_files_in_directory(sc_data_dir, extension=['.wav','.mp3','.m4a'], deep=3)
    }

    #walley_sample = "training/background/rec04.wav"
    #data, sample_rate = librosa.load(walley_sample)
    #plt.title("Wave Form")
    #librosa.display.waveplot(data, sr=sample_rate)
    #plt.show()
    #mfccs = librosa.feature.mfcc(y=data, sr=sample_rate, n_mfcc=40)
    #print("Shape of mfcc:", mfccs.shape)
    #plt.title("MFCC")
    #librosa.display.specshow(mfccs, sr=sample_rate, x_axis='time')
    #plt.show()

    all_data = []
    print("  Load training data")
    for class_label, list_of_files in data_path_dict.items():
        for single_file in list_of_files:
            try:
                if single_file[0:2]=="./":
                    single_file = single_file[2:]
                print(f"    {class_label}: {single_file}")
                audio, sample_rate = librosa.load(single_file)
                print(f"      Sample-Rate: {sample_rate}")
                mfcc = librosa.feature.mfcc(y=audio, sr=sample_rate, n_mfcc=40)
                print(f"      mfcc: {mfcc}")
                mfcc_processed = np.mean(mfcc.T, axis=0)
                print(f"      mfcc_processed: mfcc_processed")
                all_data.append([mfcc_processed, class_label])
                print("      Next")
            except Exception as e:
                print(f"      Exception: {str(e)}")
        print(f"  Info: Succesfully Preprocessed Class Label {class_label}")

    df = pd.DataFrame(all_data, columns=["feature", "class_label"])
    df.to_pickle(model_dir+"/audio_data.csv")