My code:

import os
import torch
import torchaudio
import logging
import numpy as np
import pandas as pd
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta
from concurrent.futures import ThreadPoolExecutor, as_completed
import csv
import gc  # Import garbage collector
from threading import Lock
import warnings

from transformers import (
    AutoModelForSpeechSeq2Seq, 
    AutoProcessor, 
    pipeline,
    WhisperProcessor, 
    WhisperForConditionalGeneration,Wav2Vec2PreTrainedModel,Wav2Vec2Processor,Wav2Vec2Model
)
from pyannote.audio import Pipeline  # Ensure you have Pyannote installed

# =======================
# Suppress Non-Critical Warnings
# =======================
# Suppress specific warnings from transformers and pyannote.audio
warnings.filterwarnings("ignore", category=UserWarning, module="speechbrain.inference")
warnings.filterwarnings("ignore", category=UserWarning, module="transformers")
warnings.filterwarnings("ignore", category=FutureWarning, module="transformers")

import transformers
transformers.logging.set_verbosity_error()  # Suppress transformers library warnings

# =======================
# Setup and Constants
# =======================

# Directories
RECORDINGS_DIR = r"X:\Bidvest_ASC_recordings\9247a83d-680b-4ecc-aa0e-40d03d60ae2a.AscVolume"  # Directory with recordings
OUTPUT_DIR = r"X:\Bidvest_ASC_recordings\segments"  # Directory to save audio segments
network_base_path = r"\\172.21.18.180\Bidvest_ASC_recordings"

# Sample Rate and Chunk Length
SAMPLE_RATE = 16000  # Desired sample rate
CHUNK_LENGTH = 30    # seconds
BATCH_SIZE = 1       # Set to 1 for simplicity

# Hugging Face Token (Replace with your actual token)
hf_token = "hf_liKIskKwsaWRovlsemOvecQolMLjZMktV"  # Replace with your actual token

# Paths for CSV output
FULL_TRANSCRIPTION_CSV = r"C:\Users\FourieH\full_transcription_results_with_analysis.csv"
DIARIZED_SEGMENTS_CSV = r"C:\Users\FourieH\diarized_segments_results_with_analysis.csv"

# Initialize logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Device configuration
device = "cpu"  # Change to 'cuda' if GPU is available
torch_dtype = torch.float32

# =======================
# Model Loading
# =======================

# Load Whisper model and processor for transcription
logging.info("Loading Whisper model and processor...")
model_id = "openai/whisper-large-v3-turbo"
try:
    processor = WhisperProcessor.from_pretrained(model_id)
    model = WhisperForConditionalGeneration.from_pretrained(
        model_id,
        torch_dtype=torch.float32,
        low_cpu_mem_usage=True,
        attn_implementation="eager"  # Addressing WhisperModel attention warning
    ).to(device)
    
    transcription_pipe = pipeline(
        "automatic-speech-recognition",
        model=model,
        tokenizer=processor.tokenizer,
        feature_extractor=processor.feature_extractor,
        chunk_length_s=CHUNK_LENGTH,
        batch_size=BATCH_SIZE,
        device=device,
    )
    logging.info("Whisper model and processor loaded successfully.")
except Exception as e:
    logging.error(f"Failed to load Whisper model and processor: {e}")
    exit(1)

generate_kwargs = {
    "max_new_tokens": 300,
    "num_beams": 5,
    "return_timestamps": "word",
    "language": "english"
}

# Load Pyannote Speaker Diarization Pipeline
logging.info("Loading Pyannote speaker diarization pipeline...")
try:
    diarization_pipeline = Pipeline.from_pretrained(
        "pyannote/speaker-diarization-3.1",
        use_auth_token=hf_token
    )
    logging.info("Speaker diarization pipeline loaded successfully.")
except Exception as e:
    logging.error(f"Failed to load diarization pipeline: {e}")
    exit(1)

# Load Sentiment Analysis Pipelines
logging.info("Loading Sentiment Analysis pipelines...")
sentiment_models = {
    "twitter_roberta_latest": "cardiffnlp/twitter-roberta-base-sentiment-latest",
    "distilbert_multilingual": "lxyuan/distilbert-base-multilingual-cased-sentiments-student",
    "amazon_review": "LiYuan/amazon-review-sentiment-analysis",
    "twitter_xlm_fine": "cardiffnlp/twitter-xlm-roberta-base-sentiment-multilingual"
}

sentiment_pipelines = {}
for model_name, model_path in sentiment_models.items():
    try:
        sentiment_pipelines[model_name] = pipeline(
            "text-classification",
            model=model_path,
            top_k=None  # Replaces return_all_scores=True
        )
        logging.info(f"Loaded sentiment analysis model: {model_name}")
    except Exception as e:
        logging.error(f"Failed to load sentiment model {model_name}: {e}")

# Load Audio Classification Model
logging.info("Loading audio classification model...")
try:
    audio_classifier = pipeline(
        "audio-classification",
        model="somosnlp-hackathon-2022/wav2vec2-base-finetuned-sentiment-classification-MESD"
    )
    logging.info("Audio classification model loaded successfully.")
except Exception as e:
    logging.error(f"Failed to load audio classification model: {e}")
    exit(1)

# Load Emotion Prediction Model
class RegressionHead(torch.nn.Module):
    """Regression head for emotions."""
    def __init__(self, config):
        super().__init__()
        self.dense = torch.nn.Linear(config.hidden_size, config.hidden_size)
        self.dropout = torch.nn.Dropout(config.final_dropout)
        self.out_proj = torch.nn.Linear(config.hidden_size, config.num_labels)

    def forward(self, features):
        x = self.dropout(features)
        x = self.dense(x)
        x = torch.tanh(x)
        x = self.dropout(x)
        return self.out_proj(x)

class EmotionModel(Wav2Vec2PreTrainedModel):
    """Speech emotion classifier."""
    def __init__(self, config):
        super().__init__(config)
        self.config = config
        self.wav2vec2 = Wav2Vec2Model(config)
        self.classifier = RegressionHead(config)
        self.init_weights()

    def forward(self, input_values):
        outputs = self.wav2vec2(input_values)
        hidden_states = outputs.last_hidden_state
        hidden_states = torch.mean(hidden_states, dim=1)
        logits = self.classifier(hidden_states)
        return logits

# Load Emotion Prediction Model from Hugging Face
logging.info("Loading Emotion Prediction model...")
try:
    emotion_model_name = 'audeering/wav2vec2-large-robust-12-ft-emotion-msp-dim'
    emotion_processor = Wav2Vec2Processor.from_pretrained(emotion_model_name)
    emotion_model = EmotionModel.from_pretrained(emotion_model_name).to(device)
    emotion_model.eval()
    emotion_labels = ['Arousal', 'Dominance', 'Valence']
    logging.info("Emotion Prediction model loaded successfully.")
except Exception as e:
    logging.error(f"Failed to load Emotion Prediction model: {e}")
    exit(1)

# =======================
# CSV Initialization
# =======================

# Define sentiment analysis column prefixes
sentiment_prefixes = [
    "Positive-twitter_roberta_latest",
    "Negative-twitter_roberta_latest",
    "Neutral-twitter_roberta_latest",
    "Positive-distilbert_multilingual",
    "Negative-distilbert_multilingual",
    "Neutral-distilbert_multilingual",
    "Positive-amazon_review",
    "Negative-amazon_review",
    "Neutral-amazon_review",
    "1 Star-amazon_review",
    "2 Star-amazon_review",
    "3 Star-amazon_review",
    "4 Star-amazon_review",
    "5 Star-amazon_review",
    "Positive-twitter_xlm_fine",
    "Negative-twitter_xlm_fine",
    "Neutral-twitter_xlm_fine"
]

# Define audio classification labels
audio_class_labels = ['Happiness', 'Anger', 'Neutral']

# Define emotion score columns
emotion_score_columns = [f'Emotion Score - {label}' for label in emotion_labels]

# Define full transcription CSV headers
full_transcription_headers = [
    'RecKey',
    'Full Transcription',
    'Emotion Classification',
    'Emotion Score',
    'Network Hyperlink'
] + sentiment_prefixes + audio_class_labels

# Define diarized segments CSV headers
diarized_segments_headers = [
    'RecKey',
    'Start Time',
    'End Time',
    'Speaker',
    'Transcribed Text',
    'Order Number',
    'Emotional Class',
    'Emotional Class Score',
    'Segment Audio Hyperlink'
] + sentiment_prefixes + audio_class_labels + emotion_score_columns

# Initialize CSV files with headers
def initialize_csv(file_path, headers):
    if not os.path.exists(file_path):
        with open(file_path, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=headers)
            writer.writeheader()
        logging.info(f"Initialized CSV file with headers: {file_path}")

# Initialize both CSVs
initialize_csv(FULL_TRANSCRIPTION_CSV, full_transcription_headers)
initialize_csv(DIARIZED_SEGMENTS_CSV, diarized_segments_headers)

# Create locks for thread-safe CSV writing
full_csv_lock = Lock()
diarized_csv_lock = Lock()

# =======================
# Helper Functions
# =======================

def get_rec_key(audio_file_path):
    """
    Extracts a unique recording key from the corresponding XML file.
    """
    xml_file_path = audio_file_path.replace('.wav', '.xml')
    if not os.path.exists(xml_file_path):
        logging.warning(f"XML file does not exist for {audio_file_path}")
        return "UnknownRecKey"
    try:
        tree = ET.parse(xml_file_path)
        root = tree.getroot()
        caller_number = root.findtext('.//callerNumber').strip()
        end_time_str = root.findtext('.//endtime').strip()
        end_time = datetime.strptime(end_time_str[:16], '%Y-%m-%dT%H:%M')
        adjusted_end_time = end_time + timedelta(hours=2)
        return f"{caller_number}{adjusted_end_time.strftime('%Y%m%d%H%M')}"
    except Exception as e:
        logging.error(f"Error parsing XML for {audio_file_path}: {e}")
        return "UnknownRecKey"

def get_recording_id(file_path):
    """
    Extracts the recording ID from the filename.
    """
    filename = os.path.basename(file_path)
    return filename.split('.')[0]

def load_and_resample_audio(file_path, sample_rate=SAMPLE_RATE):
    """
    Loads an audio file and resamples it to the desired sample rate.
    """
    waveform, original_sample_rate = torchaudio.load(file_path)
    if waveform.size(0) > 1:
        waveform = torch.mean(waveform, dim=0, keepdim=True)
    if original_sample_rate != sample_rate:
        try:
            resampler = torchaudio.transforms.Resample(orig_freq=original_sample_rate, new_freq=sample_rate)
            waveform = resampler(waveform)
        except Exception as e:
            logging.error(f"Resampling failed for {file_path}: {e}")
            return None
    return waveform.squeeze(0)

def classify_emotion_prediction(preprocessed_audio):
    """
    Uses the custom Wav2Vec2 model for emotion prediction.
    """
    inputs = emotion_processor(preprocessed_audio, sampling_rate=SAMPLE_RATE, return_tensors="pt")
    input_values = inputs['input_values'].to(device)
    with torch.no_grad():
        logits = emotion_model(input_values)
    return logits.detach().cpu().numpy()

def classify_sentiment(text, model_pipelines):
    """
    Performs sentiment analysis using the loaded sentiment models.
    Returns a dictionary of sentiment scores.
    """
    sentiment_scores = {}
    for model_name, analyzer in model_pipelines.items():
        try:
            result = analyzer(text, top_k=None)  # Updated to use top_k=None
            for score_info in result:
                label = score_info['label'].lower()
                score = score_info['score']
                if model_name == "twitter_roberta_latest" and label in ['positive', 'negative', 'neutral']:
                    sentiment_scores[f'{label.capitalize()}-{model_name}'] = score
                elif 'positive' in label:
                    sentiment_scores[f'Positive-{model_name}'] = score
                elif 'negative' in label:
                    sentiment_scores[f'Negative-{model_name}'] = score
                elif 'neutral' in label:
                    sentiment_scores[f'Neutral-{model_name}'] = score
                elif 'star' in label:
                    star_rating = label.split()[0]
                    sentiment_scores[f'{star_rating} Star-{model_name}'] = score
        except Exception as e:
            logging.error(f"Error during sentiment analysis with {model_name}: {e}")
    return sentiment_scores

def align_words_with_diarization(diarization_result, word_timestamps, recording_id, rec_key, waveform, file_path):
    """
    Aligns transcribed words with speaker diarization results and segments the audio accordingly.
    Returns a list of dictionaries representing each diarized segment with analyses.
    """
    aligned_script = []
    segment_number = 1
    diarized_segments = [
        (segment.start, segment.end, label)
        for segment, _, label in diarization_result.itertracks(yield_label=True)
    ]
    current_speaker = None
    current_segment_start = None
    current_segment_end = None
    current_words = []
    last_segment_end_time = 0

    for word_info in word_timestamps:
        word = word_info['text']
        start_time = word_info['timestamp'][0]
        end_time = word_info['timestamp'][1]

        if start_time is None or end_time is None:
            logging.warning(f"Invalid timestamps for word '{word}': {word_info['timestamp']}")
            continue

        # Ensure start_time and end_time are positive and in correct order
        if start_time < 0 or end_time <= start_time:
            logging.warning(f"Invalid word timestamps: start_time={start_time}, end_time={end_time}")
            continue

        for (seg_start, seg_end, speaker_label) in diarized_segments:
            if seg_start <= start_time <= seg_end:
                if current_speaker != speaker_label:
                    if current_speaker is not None:
                        # Calculate segment duration
                        segment_duration = current_segment_end - current_segment_start
                        if segment_duration < 2.0:  # Minimum 2 seconds
                            logging.warning(f"Segment {segment_number} is too short ({segment_duration:.2f}s). Skipping.")
                            break

                        # Ensure slice indices are within waveform bounds
                        total_duration = len(waveform) / SAMPLE_RATE
                        padded_start = max(current_segment_start, 0)
                        padded_end = min(current_segment_end, total_duration)

                        start_idx = int(padded_start * SAMPLE_RATE)
                        end_idx = int(padded_end * SAMPLE_RATE)

                        if end_idx <= start_idx:
                            logging.warning(f"Segment {segment_number} has invalid slice indices: start_idx={start_idx}, end_idx={end_idx}. Skipping.")
                            break

                        segment_waveform = waveform[start_idx:end_idx]
                        if segment_waveform.size(0) < SAMPLE_RATE * 2:  # At least 2 seconds
                            logging.warning(f"Segment {segment_number} has insufficient samples ({segment_waveform.size(0)}). Skipping.")
                            break

                        # Save the current segment and perform analyses
                        segment_filename = os.path.join(
                            OUTPUT_DIR, f"{recording_id}_speaker_{current_speaker}_segment_{segment_number}.wav"
                        )
                        try:
                            torchaudio.save(segment_filename, segment_waveform.unsqueeze(0), SAMPLE_RATE)
                        except Exception as e:
                            logging.error(f"Failed to save segment {segment_filename}: {e}")
                            break

                        network_segment_path = segment_filename.replace(r"X:\Bidvest_ASC_recordings", network_base_path)

                        # Sentiment Analysis
                        sentiment_scores = classify_sentiment(' '.join(current_words), sentiment_pipelines)

                        # Emotion Prediction
                        emotion_prediction_scores = classify_emotion_prediction(segment_waveform.numpy())
                        # Handle potential NaNs and ensure valid scores
                        emotion_prediction_dict = {f'Emotion Score - {label}': (score if not np.isnan(score) else 'N/A') 
                                                   for label, score in zip(emotion_labels, emotion_prediction_scores[0])}

                        # Audio Classification
                        try:
                            classification_results = audio_classifier(segment_filename, sample_rate=SAMPLE_RATE)
                            audio_classification = {result['label']: (result['score'] if not np.isnan(result['score']) else 'N/A') 
                                                    for result in classification_results}
                        except Exception as e:
                            logging.error(f"Error during audio classification for {segment_filename}: {e}")
                            audio_classification = {}

                        # Combine all data into one dictionary
                        segment_data = {
                            'RecKey': rec_key,
                            'Start Time': current_segment_start,
                            'End Time': current_segment_end,
                            'Speaker': current_speaker,
                            'Transcribed Text': ' '.join(current_words),
                            'Order Number': segment_number,
                            'Emotional Class': 'N/A',  # Removed SpeechBrain Emotion Class
                            'Emotional Class Score': 'N/A',  # Removed SpeechBrain Emotion Score
                            'Segment Audio Hyperlink': network_segment_path
                        }
                        segment_data.update(sentiment_scores)
                        segment_data.update(emotion_prediction_dict)
                        segment_data.update(audio_classification)

                        aligned_script.append(segment_data)
                        segment_number += 1

                    # Reset for the new speaker
                    current_speaker = speaker_label
                    current_segment_start = max(last_segment_end_time, start_time - 2)
                    current_segment_end = end_time + 2
                    current_words = [word]
                else:
                    # Extend end time and add the word to the current segment for continuity
                    current_segment_end = max(current_segment_end, end_time + 2)
                    current_words.append(word)

                last_segment_end_time = current_segment_end
                break

    # Final check for the last segment
    if current_speaker is not None and current_words:
        # Calculate segment duration
        segment_duration = current_segment_end - current_segment_start
        if segment_duration >= 2.0:
            # Ensure slice indices are within waveform bounds
            total_duration = len(waveform) / SAMPLE_RATE
            padded_start = max(current_segment_start, 0)
            padded_end = min(current_segment_end, total_duration)

            start_idx = int(padded_start * SAMPLE_RATE)
            end_idx = int(padded_end * SAMPLE_RATE)

            if end_idx > start_idx:
                segment_waveform = waveform[start_idx:end_idx]
                if segment_waveform.size(0) >= SAMPLE_RATE * 2:
                    # Save the current segment and perform analyses
                    segment_filename = os.path.join(
                        OUTPUT_DIR, f"{recording_id}_speaker_{current_speaker}_segment_{segment_number}.wav"
                    )
                    try:
                        torchaudio.save(segment_filename, segment_waveform.unsqueeze(0), SAMPLE_RATE)
                    except Exception as e:
                        logging.error(f"Failed to save segment {segment_filename}: {e}")
                        return aligned_script  # Return what has been appended so far

                    network_segment_path = segment_filename.replace(r"X:\Bidvest_ASC_recordings", network_base_path)

                    # Sentiment Analysis
                    sentiment_scores = classify_sentiment(' '.join(current_words), sentiment_pipelines)

                    # Emotion Prediction
                    emotion_prediction_scores = classify_emotion_prediction(segment_waveform.numpy())
                    # Handle potential NaNs and ensure valid scores
                    emotion_prediction_dict = {f'Emotion Score - {label}': (score if not np.isnan(score) else 'N/A') 
                                               for label, score in zip(emotion_labels, emotion_prediction_scores[0])}

                    # Audio Classification
                    try:
                        classification_results = audio_classifier(segment_filename, sample_rate=SAMPLE_RATE)
                        audio_classification = {result['label']: (result['score'] if not np.isnan(result['score']) else 'N/A') 
                                                for result in classification_results}
                    except Exception as e:
                        logging.error(f"Error during audio classification for {segment_filename}: {e}")
                        audio_classification = {}

                    # Combine all data into one dictionary
                    segment_data = {
                        'RecKey': rec_key,
                        'Start Time': current_segment_start,
                        'End Time': current_segment_end,
                        'Speaker': current_speaker,
                        'Transcribed Text': ' '.join(current_words),
                        'Order Number': segment_number,
                        'Emotional Class': 'N/A',  # Removed SpeechBrain Emotion Class
                        'Emotional Class Score': 'N/A',  # Removed SpeechBrain Emotion Score
                        'Segment Audio Hyperlink': network_segment_path
                    }
                    segment_data.update(sentiment_scores)
                    segment_data.update(emotion_prediction_dict)
                    segment_data.update(audio_classification)

                    aligned_script.append(segment_data)
            else:
                logging.warning(f"Final segment has invalid slice indices and is too short. Skipping.")
        else:
            logging.warning(f"Final segment is too short ({segment_duration:.2f}s). Skipping.")

    return aligned_script

def write_to_csv(file_path, headers, data_dict, lock):
    """
    Writes a single row dictionary to a CSV file in a thread-safe manner.
    """
    try:
        with lock:
            with open(file_path, 'a', newline='', encoding='utf-8') as f:
                writer = csv.DictWriter(f, fieldnames=headers)
                # Ensure all keys in data_dict are present in headers
                row = {key: data_dict.get(key, 'N/A') for key in headers}
                writer.writerow(row)
    except Exception as e:
        logging.error(f"Failed to write to CSV {file_path}: {e}")

def process_audio_file(file):
    """
    Processes a single audio file: transcription, diarization, sentiment analyses, emotion predictions, and audio classification.
    Writes results directly to CSV files to minimize memory usage.
    """
    if not file.endswith(".wav"):
        return
    audio_file_path = os.path.join(RECORDINGS_DIR, file)
    rec_key = get_rec_key(audio_file_path)
    recording_id = get_recording_id(audio_file_path)
    
    try:
        waveform = load_and_resample_audio(audio_file_path, SAMPLE_RATE)
        if waveform is None:
            logging.error(f"Failed to load waveform for {audio_file_path}. Skipping.")
            return
    except Exception as e:
        logging.error(f"Error loading audio file {audio_file_path}: {e}")
        return

    if waveform.size(0) < SAMPLE_RATE * CHUNK_LENGTH:
        logging.warning(f"Audio is too short for processing: {file}")
        return

    try:
        # Transcription
        result = transcription_pipe(waveform.numpy(), return_timestamps="word", generate_kwargs=generate_kwargs)
        transcription_text = result['text']
        word_timestamps = result.get('chunks', [])
        if not word_timestamps:
            logging.warning(f"No transcription chunks found for {audio_file_path}. Skipping.")
            return
        # Validate each chunk
        total_duration = len(waveform) / SAMPLE_RATE
        valid_chunks = []
        for chunk in word_timestamps:
            start, end = chunk['timestamp']
            if start < 0 or end > total_duration or end <= start:
                logging.warning(f"Invalid chunk timestamps: start_time={start}, end_time={end}. Skipping this chunk.")
                continue
            valid_chunks.append(chunk)
        if not valid_chunks:
            logging.warning(f"All transcription chunks for {audio_file_path} are invalid. Skipping.")
            return
    except Exception as e:
        logging.error(f"Error during transcription for {audio_file_path}: {e}")
        return

    try:
        # Speaker Diarization
        diarization_result = diarization_pipeline(audio_file_path)
    except Exception as e:
        logging.error(f"Error during diarization for {audio_file_path}: {e}")
        return

    try:
        # Sentiment Analysis on Full Transcript
        sentiment_scores_full = classify_sentiment(transcription_text, sentiment_pipelines)
    except Exception as e:
        logging.error(f"Error during sentiment analysis for {audio_file_path}: {e}")
        sentiment_scores_full = {}

    try:
        # Emotion Prediction for Full Transcript
        emotion_prediction_scores_full = classify_emotion_prediction(waveform.numpy())
        if emotion_prediction_scores_full.size > 0:
            emotion_prediction_dict_full = {f'Emotion Score - {label}': (score if not np.isnan(score) else 'N/A') 
                                           for label, score in zip(emotion_labels, emotion_prediction_scores_full[0])}
        else:
            emotion_prediction_dict_full = {f'Emotion Score - {label}': 'N/A' for label in emotion_labels}
    except Exception as e:
        logging.error(f"Error during emotion prediction for {audio_file_path}: {e}")
        emotion_prediction_dict_full = {f'Emotion Score - {label}': 'N/A' for label in emotion_labels}

    try:
        # Audio Classification for Full Transcript
        classification_results_full = audio_classifier(audio_file_path, sample_rate=SAMPLE_RATE)
        audio_classification_full = {result['label']: (result['score'] if not np.isnan(result['score']) else 'N/A') 
                                    for result in classification_results_full}
    except Exception as e:
        logging.error(f"Error during audio classification for {audio_file_path}: {e}")
        audio_classification_full = {}
    
    # Network Path
    network_path = audio_file_path.replace(r"X:\Bidvest_ASC_recordings", network_base_path)

    # Prepare full transcription data
    full_transcription_data = {
        'RecKey': rec_key,
        'Full Transcription': transcription_text,
        'Emotion Classification': 'N/A',  # Removed SpeechBrain Emotion Class
        'Emotion Score': 'N/A',            # Removed SpeechBrain Emotion Score
        'Network Hyperlink': network_path
    }
    full_transcription_data.update(sentiment_scores_full)
    full_transcription_data.update(emotion_prediction_dict_full)
    full_transcription_data.update(audio_classification_full)

    # Write full transcription data to CSV
    write_to_csv(FULL_TRANSCRIPTION_CSV, full_transcription_headers, full_transcription_data, full_csv_lock)

    try:
        # Aligned Segments
        aligned_segments = align_words_with_diarization(
            diarization_result,
            valid_chunks,
            recording_id,
            rec_key,
            waveform,
            audio_file_path
        )
    except Exception as e:
        logging.error(f"Error during alignment for {audio_file_path}: {e}")
        aligned_segments = []

    # Write each segment to CSV
    for segment in aligned_segments:
        write_to_csv(DIARIZED_SEGMENTS_CSV, diarized_segments_headers, segment, diarized_csv_lock)

    # Cleanup
    del waveform
    del aligned_segments
    gc.collect()

    logging.info(f"Processed file: {file}")

def write_to_csv(file_path, headers, data_dict, lock):
    """
    Writes a single row dictionary to a CSV file in a thread-safe manner.
    """
    try:
        with lock:
            with open(file_path, 'a', newline='', encoding='utf-8') as f:
                writer = csv.DictWriter(f, fieldnames=headers)
                # Ensure all keys in data_dict are present in headers
                row = {key: data_dict.get(key, 'N/A') for key in headers}
                writer.writerow(row)
    except Exception as e:
        logging.error(f"Failed to write to CSV {file_path}: {e}")

def process_audio_file(file):
    """
    Processes a single audio file: transcription, diarization, sentiment analyses, emotion predictions, and audio classification.
    Writes results directly to CSV files to minimize memory usage.
    """
    if not file.endswith(".wav"):
        return
    audio_file_path = os.path.join(RECORDINGS_DIR, file)
    rec_key = get_rec_key(audio_file_path)
    recording_id = get_recording_id(audio_file_path)
    
    try:
        waveform = load_and_resample_audio(audio_file_path, SAMPLE_RATE)
        if waveform is None:
            logging.error(f"Failed to load waveform for {audio_file_path}. Skipping.")
            return
    except Exception as e:
        logging.error(f"Error loading audio file {audio_file_path}: {e}")
        return

    if waveform.size(0) < SAMPLE_RATE * CHUNK_LENGTH:
        logging.warning(f"Audio is too short for processing: {file}")
        return

    try:
        # Transcription
        result = transcription_pipe(waveform.numpy(), return_timestamps="word", generate_kwargs=generate_kwargs)
        transcription_text = result['text']
        word_timestamps = result.get('chunks', [])
        if not word_timestamps:
            logging.warning(f"No transcription chunks found for {audio_file_path}. Skipping.")
            return
        # Validate each chunk
        total_duration = len(waveform) / SAMPLE_RATE
        valid_chunks = []
        for chunk in word_timestamps:
            start, end = chunk['timestamp']
            if start < 0 or end > total_duration or end <= start:
                logging.warning(f"Invalid chunk timestamps: start_time={start}, end_time={end}. Skipping this chunk.")
                continue
            valid_chunks.append(chunk)
        if not valid_chunks:
            logging.warning(f"All transcription chunks for {audio_file_path} are invalid. Skipping.")
            return
    except Exception as e:
        logging.error(f"Error during transcription for {audio_file_path}: {e}")
        return

    try:
        # Speaker Diarization
        diarization_result = diarization_pipeline(audio_file_path)
    except Exception as e:
        logging.error(f"Error during diarization for {audio_file_path}: {e}")
        return

    try:
        # Sentiment Analysis on Full Transcript
        sentiment_scores_full = classify_sentiment(transcription_text, sentiment_pipelines)
    except Exception as e:
        logging.error(f"Error during sentiment analysis for {audio_file_path}: {e}")
        sentiment_scores_full = {}

    try:
        # Emotion Prediction for Full Transcript
        emotion_prediction_scores_full = classify_emotion_prediction(waveform.numpy())
        if emotion_prediction_scores_full.size > 0:
            emotion_prediction_dict_full = {f'Emotion Score - {label}': (score if not np.isnan(score) else 'N/A') 
                                           for label, score in zip(emotion_labels, emotion_prediction_scores_full[0])}
        else:
            emotion_prediction_dict_full = {f'Emotion Score - {label}': 'N/A' for label in emotion_labels}
    except Exception as e:
        logging.error(f"Error during emotion prediction for {audio_file_path}: {e}")
        emotion_prediction_dict_full = {f'Emotion Score - {label}': 'N/A' for label in emotion_labels}

    try:
        # Audio Classification for Full Transcript
        classification_results_full = audio_classifier(audio_file_path, sample_rate=SAMPLE_RATE)
        audio_classification_full = {result['label']: (result['score'] if not np.isnan(result['score']) else 'N/A') 
                                    for result in classification_results_full}
    except Exception as e:
        logging.error(f"Error during audio classification for {audio_file_path}: {e}")
        audio_classification_full = {}
    
    # Network Path
    network_path = audio_file_path.replace(r"X:\Bidvest_ASC_recordings", network_base_path)

    # Prepare full transcription data
    full_transcription_data = {
        'RecKey': rec_key,
        'Full Transcription': transcription_text,
        'Emotion Classification': 'N/A',  # Removed SpeechBrain Emotion Class
        'Emotion Score': 'N/A',            # Removed SpeechBrain Emotion Score
        'Network Hyperlink': network_path
    }
    full_transcription_data.update(sentiment_scores_full)
    full_transcription_data.update(emotion_prediction_dict_full)
    full_transcription_data.update(audio_classification_full)

    # Write full transcription data to CSV
    write_to_csv(FULL_TRANSCRIPTION_CSV, full_transcription_headers, full_transcription_data, full_csv_lock)

    try:
        # Aligned Segments
        aligned_segments = align_words_with_diarization(
            diarization_result,
            valid_chunks,
            recording_id,
            rec_key,
            waveform,
            audio_file_path
        )
    except Exception as e:
        logging.error(f"Error during alignment for {audio_file_path}: {e}")
        aligned_segments = []

    # Write each segment to CSV
    for segment in aligned_segments:
        write_to_csv(DIARIZED_SEGMENTS_CSV, diarized_segments_headers, segment, diarized_csv_lock)

    # Cleanup
    del waveform
    del aligned_segments
    gc.collect()

    logging.info(f"Processed file: {file}")

def write_to_csv(file_path, headers, data_dict, lock):
    """
    Writes a single row dictionary to a CSV file in a thread-safe manner.
    """
    try:
        with lock:
            with open(file_path, 'a', newline='', encoding='utf-8') as f:
                writer = csv.DictWriter(f, fieldnames=headers)
                # Ensure all keys in data_dict are present in headers
                row = {key: data_dict.get(key, 'N/A') for key in headers}
                writer.writerow(row)
    except Exception as e:
        logging.error(f"Failed to write to CSV {file_path}: {e}")

def process_audio_file(file):
    """
    Processes a single audio file: transcription, diarization, sentiment analyses, emotion predictions, and audio classification.
    Writes results directly to CSV files to minimize memory usage.
    """
    if not file.endswith(".wav"):
        return
    audio_file_path = os.path.join(RECORDINGS_DIR, file)
    rec_key = get_rec_key(audio_file_path)
    recording_id = get_recording_id(audio_file_path)
    
    try:
        waveform = load_and_resample_audio(audio_file_path, SAMPLE_RATE)
        if waveform is None:
            logging.error(f"Failed to load waveform for {audio_file_path}. Skipping.")
            return
    except Exception as e:
        logging.error(f"Error loading audio file {audio_file_path}: {e}")
        return

    if waveform.size(0) < SAMPLE_RATE * CHUNK_LENGTH:
        logging.warning(f"Audio is too short for processing: {file}")
        return

    try:
        # Transcription
        result = transcription_pipe(waveform.numpy(), return_timestamps="word", generate_kwargs=generate_kwargs)
        transcription_text = result['text']
        word_timestamps = result.get('chunks', [])
        if not word_timestamps:
            logging.warning(f"No transcription chunks found for {audio_file_path}. Skipping.")
            return
        # Validate each chunk
        total_duration = len(waveform) / SAMPLE_RATE
        valid_chunks = []
        for chunk in word_timestamps:
            start, end = chunk['timestamp']
            if start < 0 or end > total_duration or end <= start:
                logging.warning(f"Invalid chunk timestamps: start_time={start}, end_time={end}. Skipping this chunk.")
                continue
            valid_chunks.append(chunk)
        if not valid_chunks:
            logging.warning(f"All transcription chunks for {audio_file_path} are invalid. Skipping.")
            return
    except Exception as e:
        logging.error(f"Error during transcription for {audio_file_path}: {e}")
        return

    try:
        # Speaker Diarization
        diarization_result = diarization_pipeline(audio_file_path)
    except Exception as e:
        logging.error(f"Error during diarization for {audio_file_path}: {e}")
        return

    try:
        # Sentiment Analysis on Full Transcript
        sentiment_scores_full = classify_sentiment(transcription_text, sentiment_pipelines)
    except Exception as e:
        logging.error(f"Error during sentiment analysis for {audio_file_path}: {e}")
        sentiment_scores_full = {}

    try:
        # Emotion Prediction for Full Transcript
        emotion_prediction_scores_full = classify_emotion_prediction(waveform.numpy())
        if emotion_prediction_scores_full.size > 0:
            emotion_prediction_dict_full = {f'Emotion Score - {label}': (score if not np.isnan(score) else 'N/A') 
                                           for label, score in zip(emotion_labels, emotion_prediction_scores_full[0])}
        else:
            emotion_prediction_dict_full = {f'Emotion Score - {label}': 'N/A' for label in emotion_labels}
    except Exception as e:
        logging.error(f"Error during emotion prediction for {audio_file_path}: {e}")
        emotion_prediction_dict_full = {f'Emotion Score - {label}': 'N/A' for label in emotion_labels}

    try:
        # Audio Classification for Full Transcript
        classification_results_full = audio_classifier(audio_file_path, sample_rate=SAMPLE_RATE)
        audio_classification_full = {result['label']: (result['score'] if not np.isnan(result['score']) else 'N/A') 
                                    for result in classification_results_full}
    except Exception as e:
        logging.error(f"Error during audio classification for {audio_file_path}: {e}")
        audio_classification_full = {}
    
    # Network Path
    network_path = audio_file_path.replace(r"X:\Bidvest_ASC_recordings", network_base_path)

    # Prepare full transcription data
    full_transcription_data = {
        'RecKey': rec_key,
        'Full Transcription': transcription_text,
        'Emotion Classification': 'N/A',  # Removed SpeechBrain Emotion Class
        'Emotion Score': 'N/A',            # Removed SpeechBrain Emotion Score
        'Network Hyperlink': network_path
    }
    full_transcription_data.update(sentiment_scores_full)
    full_transcription_data.update(emotion_prediction_dict_full)
    full_transcription_data.update(audio_classification_full)

    # Write full transcription data to CSV
    write_to_csv(FULL_TRANSCRIPTION_CSV, full_transcription_headers, full_transcription_data, full_csv_lock)

    try:
        # Aligned Segments
        aligned_segments = align_words_with_diarization(
            diarization_result,
            valid_chunks,
            recording_id,
            rec_key,
            waveform,
            audio_file_path
        )
    except Exception as e:
        logging.error(f"Error during alignment for {audio_file_path}: {e}")
        aligned_segments = []

    # Write each segment to CSV
    for segment in aligned_segments:
        write_to_csv(DIARIZED_SEGMENTS_CSV, diarized_segments_headers, segment, diarized_csv_lock)

    # Cleanup
    del waveform
    del aligned_segments
    gc.collect()

    logging.info(f"Processed file: {file}")


def main():
    processed_files = 0
    max_files = 50000000  # Adjust as needed

    # Ensure output directory exists
    os.makedirs(OUTPUT_DIR, exist_ok=True)

    # Define the number of worker threads based on system capability
    MAX_WORKERS = 2  # Start with 2 to prevent system overload

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        wav_files = [file for file in os.listdir(RECORDINGS_DIR) if file.endswith(".wav")]
        futures = {executor.submit(process_audio_file, file): file for file in wav_files}

        for future in as_completed(futures):
            file = futures[future]
            try:
                future.result()
                processed_files += 1
                if processed_files >= max_files:
                    logging.info(f"Reached maximum file processing limit: {max_files}")
                    break
                if processed_files % 100 == 0:
                    logging.info(f"Processed {processed_files} files...")
            except Exception as e:
                logging.error(f"Exception occurred while processing file '{file}': {e}")

    logging.info(f"Processing complete. Total files processed: {processed_files}")

if __name__ == "__main__":
    main()

Output:

2024-11-12 11:40:43,659 - INFO - Loading Whisper model and processor...
2024-11-12 11:40:50,652 - INFO - Whisper model and processor loaded successfully.
2024-11-12 11:40:50,654 - INFO - Loading Pyannote speaker diarization pipeline...
2024-11-12 11:40:53,024 - INFO - Speaker diarization pipeline loaded successfully.
2024-11-12 11:40:53,024 - INFO - Loading Sentiment Analysis pipelines...
2024-11-12 11:40:54,376 - INFO - Loaded sentiment analysis model: twitter_roberta_latest
2024-11-12 11:40:55,124 - INFO - Loaded sentiment analysis model: distilbert_multilingual
2024-11-12 11:40:56,087 - INFO - Loaded sentiment analysis model: amazon_review
2024-11-12 11:40:58,441 - INFO - Loaded sentiment analysis model: twitter_xlm_fine
2024-11-12 11:40:58,441 - INFO - Loading audio classification model...
2024-11-12 11:40:59,207 - INFO - Audio classification model loaded successfully.
2024-11-12 11:40:59,223 - INFO - Loading Emotion Prediction model...
2024-11-12 11:41:02,118 - INFO - Emotion Prediction model loaded successfully.
2024-11-12 11:41:30,297 - WARNING - Audio is too short for processing: 223f6b02-1a2c-4cb9-a5ab-409acaaa3d2a.wav
2024-11-12 11:43:58,950 - WARNING - Invalid chunk timestamps: start_time=9.8, end_time=9.8. Skipping this chunk.
2024-11-12 11:43:58,965 - WARNING - Invalid chunk timestamps: start_time=23.64, end_time=23.64. Skipping this chunk.
2024-11-12 11:43:58,965 - WARNING - Invalid chunk timestamps: start_time=23.64, end_time=23.64. Skipping this chunk.
2024-11-12 11:43:58,971 - WARNING - Invalid chunk timestamps: start_time=23.64, end_time=23.64. Skipping this chunk.
2024-11-12 11:43:58,973 - WARNING - Invalid chunk timestamps: start_time=32.18, end_time=32.18. Skipping this chunk.
2024-11-12 11:43:58,974 - WARNING - Invalid chunk timestamps: start_time=32.88, end_time=32.88. Skipping this chunk.
2024-11-12 11:43:58,975 - WARNING - Invalid chunk timestamps: start_time=59.48, end_time=59.48. Skipping this chunk.
2024-11-12 11:43:58,976 - WARNING - Invalid chunk timestamps: start_time=59.48, end_time=59.48. Skipping this chunk.
2024-11-12 11:43:58,977 - WARNING - Invalid chunk timestamps: start_time=59.48, end_time=59.48. Skipping this chunk.
2024-11-12 11:43:58,978 - WARNING - Invalid chunk timestamps: start_time=60.18, end_time=60.18. Skipping this chunk.
2024-11-12 11:43:58,979 - WARNING - Invalid chunk timestamps: start_time=61.54, end_time=61.54. Skipping this chunk.
2024-11-12 11:43:58,979 - WARNING - Invalid chunk timestamps: start_time=61.54, end_time=61.54. Skipping this chunk.
2024-11-12 11:43:58,980 - WARNING - Invalid chunk timestamps: start_time=69.32, end_time=69.32. Skipping this chunk.
C:\Users\FourieH\AppData\Local\anaconda3\Lib\site-packages\pyannote\audio\models\blocks\pooling.py:104: UserWarning: std(): degrees of freedom is <= 0. Correction should be strictly less than the reduction factor (input numel divided by output numel). (Triggered internally at C:\actions-runner\_work\pytorch\pytorch\builder\windows\pytorch\aten\src\ATen\native\ReduceOps.cpp:1823.)
  std = sequences.std(dim=-1, correction=1)
2024-11-12 11:46:05,538 - WARNING - Segment 7 is too short (1.40s). Skipping.
2024-11-12 11:46:05,538 - WARNING - Segment 7 is too short (1.40s). Skipping.
2024-11-12 11:46:05,538 - WARNING - Segment 7 is too short (1.96s). Skipping.
2024-11-12 11:46:13,454 - ERROR - Error during transcription for X:\Bidvest_ASC_recordings\9247a83d-680b-4ecc-aa0e-40d03d60ae2a.AscVolume\4f225a17-eb3a-4916-8019-d27e8daea0d3.wav: index 302 is out of bounds for dimension 2 with size 302
2024-11-12 11:46:13,646 - WARNING - Audio is too short for processing: 529dfb24-793e-4295-a388-5552f9191ae4.wav
2024-11-12 11:46:13,744 - WARNING - Audio is too short for processing: 64f2b892-3be7-4922-a045-590b166b6fc4.wav
2024-11-12 11:46:28,057 - WARNING - Segment 12 is too short (0.96s). Skipping.
2024-11-12 11:46:28,057 - WARNING - Segment 12 is too short (0.96s). Skipping.
2024-11-12 11:46:28,057 - WARNING - Segment 12 is too short (0.96s). Skipping.
2024-11-12 11:46:28,057 - WARNING - Segment 12 is too short (0.96s). Skipping.
2024-11-12 11:46:28,070 - WARNING - Final segment is too short (0.96s). Skipping.
2024-11-12 11:46:29,079 - INFO - Processed file: 2f82ef2d-620a-431e-af4d-4c5cf287a443.wav