scripts/process_data.py

import os
import argparse
import numpy as np
import pandas as pd
from multiprocessing import Pool, cpu_count
from tqdm import tqdm
import warnings

# Import our custom modules
from src.utils.config import load_config
from src.utils.labels import get_label
from src.preprocessing.preprocess import preprocess_audio
from src.feature_extraction.extract import extract_mel_spectrograms

warnings.filterwarnings('ignore')

# Global config placeholder for the workers
worker_config = None

def init_worker(config):
    """This function runs once when each worker process starts."""
    global worker_config
    worker_config = config

def process_file(task):
    """Worker function for multiprocessing."""
    global worker_config
    
    file_path       = task["file_path"]
    label           = task["label"]
    machine         = task["machine"]
    label_name      = task["label_name"]
    base_name       = task["base_name"]
    machine_out_dir = task["machine_out_dir"]

    save_name = f"{machine}_{label_name}_{base_name}.npz"
    save_path = os.path.join(machine_out_dir, save_name)

    if os.path.exists(save_path):
        try:
            with np.load(save_path) as d:
                return (save_name, label, machine, d["features"].shape[0], save_path)
        except Exception:
            os.remove(save_path)

    try:
        # Pass config values to preprocessing
        chunks = preprocess_audio(
            file_path=file_path,
            target_sr=worker_config['audio']['target_sr'],
            chunk_duration=worker_config['audio']['chunk_duration'],
            step_duration=worker_config['audio']['step_duration'],
            trim_top_db=worker_config['audio']['trim_top_db']
        )

        # Pass config values to feature extraction
        features = extract_mel_spectrograms(
            chunks=chunks,
            sr=worker_config['audio']['target_sr'],
            n_fft=worker_config['features']['n_fft'],
            hop_length=worker_config['features']['hop_length'],
            n_mels=worker_config['features']['n_mels']
        )

        np.savez(save_path, features=features)
        return (save_name, label, machine, features.shape[0], save_path)

    except Exception as e:
        # It's helpful to see the actual error during debugging
        return f"Error processing {file_path}: {str(e)}"

def main():
    # Allow overriding config path from command line
    parser = argparse.ArgumentParser(description="Process Audio Data")
    parser.add_argument('--config', type=str, default='config/default.yaml', help='Path to config file')
    args = parser.parse_args()

    # Load config in the main process
    main_config = load_config(args.config)
    
    input_dir = main_config['paths']['input_dir']
    output_dir = main_config['paths']['output_dir']
    
    os.makedirs(output_dir, exist_ok=True)
    print(f"Input  : {input_dir}")
    print(f"Output : {output_dir}")

    # Gather tasks
    tasks = []
    # Check if directory exists to avoid crashes
    if not os.path.exists(input_dir):
        print(f"❌ Error: Input directory '{input_dir}' not found.")
        return

    for machine in sorted(os.listdir(input_dir)):
        machine_path = os.path.join(input_dir, machine)
        if not os.path.isdir(machine_path): continue

        machine_out_dir = os.path.join(output_dir, machine)
        os.makedirs(machine_out_dir, exist_ok=True)

        # Your structure: Machine 1 -> machine_data -> Normal
        data_path = os.path.join(machine_path, "machine_data")
        if not os.path.exists(data_path): continue

        for label_name in ["Normal", "Abnormal"]:
            class_path = os.path.join(data_path, label_name)
            if not os.path.exists(class_path): continue

            label = get_label(machine, label_name)

            for file in sorted(os.listdir(class_path)):
                if file.endswith(".wav"):
                    tasks.append({
                        "file_path": os.path.join(class_path, file),
                        "label": label,
                        "machine": machine,
                        "label_name": label_name,
                        "base_name": os.path.splitext(file)[0],
                        "machine_out_dir": machine_out_dir,
                    })

    print(f"Total audio files to process: {len(tasks)}")

    # Execute Multiprocessing with Initializer
    all_results = []
    # Use initializer to pass config to workers
    with Pool(processes=cpu_count(), initializer=init_worker, initargs=(main_config,)) as pool:
        for res in tqdm(pool.imap_unordered(process_file, tasks), total=len(tasks)):
            if isinstance(res, tuple):
                all_results.append(res)
            else:
                print(res) # Print the error string if it failed

    # Save Metadata
    if all_results:
        df = pd.DataFrame(all_results, columns=["file", "label", "machine", "num_chunks", "file_path"])
        csv_path = os.path.join(output_dir, main_config['paths']['csv_name'])
        df.to_csv(csv_path, index=False)
        print(f"✅ Metadata saved to {csv_path}")
    else:
        print("❌ No files were processed successfully.")

if __name__ == "__main__":
    main()