add src

src/features/__init__.py

@@ -0,0 +1,12 @@
+from .extraction import (    
+    LowLevelFeatureExtractor, 
+    HighLevelFeatureExtractor
+)
+from .visualization import (
+    plot_low_level_feature_dist,
+    plot_high_level_feature_dist,
+    perform_pca_and_plot,
+    perform_kernel_pca_and_plot,
+    perform_pca_and_plot_3d
+)
+

src/features/extraction/README.md

@@ -0,0 +1,72 @@
+# Features Extraction Module
+
+This module is an essential part of a system designed to identify deepfake audio recordings. It contains all the necessary tools for extracting a wide range of audio features that help differentiate real audio from synthetic ones.
+
+## Purpose
+
+The main purpose of this module is to provide a reliable framework for pulling various audio features from raw sound data. These features include spectral, prosodic, and voice quality aspects, each offering valuable insights into the characteristics of audio signals. By examining these features, the system can identify patterns and irregularities that are commonly found in deepfake audio, improving the effectiveness of detection methods.
+
+## Modular Design
+
+The module's design is highly modular, making it easy to integrate and scale. Researchers and developers can conveniently expand existing feature categories or add new ones as new developments in audio analysis become available. This flexibility ensures that the system can adapt and evolve without needing major changes, supporting ongoing improvements in deepfake detection technology.
+
+## Module Overview
+
+### Directory Structure
+
+- `__init__.py`: Initializes the `features.extraction` package, allowing classes and functions to be imported.
+- `features_list.py`: Manages lists of feature names used throughout the various extraction modules.
+- `high_level_features_extractor.py`: Extracts complex statistical features from detailed low-level features.
+- `low_level_features_extractor.py`: Pulls detailed low-level features from raw audio data.
+- `prosodic_features.py`: Focuses on features related to the rhythm and intonation of speech.
+- `spectral_features.py`: Extracts features that describe the audio spectrum.
+- `voice_quality_features.py`: Collects features that show the quality and traits of the voice.
+- `stat_measures.py`: Provides statistical analysis tools to compute measures such as mean, variance, and skewness for extracted detailed low-level features.
+
+## Workflow
+
+### 1. Input
+
+The workflow starts with raw audio data which is preprocessed to get it ready for feature extraction.
+
+### 2. Detailed Low-Level Feature Extraction
+
+This phase involves pulling out various audio features:
+
+- **Spectral Features**: Such as MFCCs, FFT, and spectral centroids.
+- **Temporal Features**: Like zero-crossing rate and peak amplitude.
+- **Prosodic Features**: Including measures like speaking rate and pitch.
+- **Voice Quality Features**: Evaluating quality through metrics like jitter and shimmer.
+
+### 3. Transformation and Aggregation
+
+After extracting features, the data is summarized statistically and segmented (sometimes using rolling windows). This helps focus on the most informative parts of the features and reduces the amount of data.
+
+### 4. High-Level Feature Extraction
+
+The summarized data is then used to compute higher-order statistical features such as mean, standard deviation, skewness, and kurtosis, providing a summary suitable for machine learning models.
+
+### 5. Output
+
+The end result is a structured array of high-level features for each audio sample, ready for further analysis or direct use in machine learning algorithms.
+
+## Usage
+
+```python
+from features.extraction.low_level_features_extractor import LowLevelFeatureExtractor
+from features.extraction.high_level_features_extractor import HighLevelFeatureExtractor
+
+# Initialize extractors
+low_level_extractor = LowLevelFeatureExtractor()
+high_level_extractor = HighLevelFeatureExtractor()
+
+# Process audio data
+audio_data = {'audio_arr': your_audio_array, 'srate': your_sampling_rate}
+low_level_features = low_level_extractor.extract_features(audio_data)
+high_level_features = high_level_extractor.compute high_level_features(low_level_features)
+
+print(high_level_features)
+```
+
+## Future Integration
+The module is designed for easy integration with data preprocessing pipelines and machine learning frameworks. It allows for simple updates, such as adding new feature categories or improving existing ones, ensuring the system stays current with the latest in audio analysis. This flexibility is particularly important for building a reliable audio deepfake detection system.

src/features/extraction/__init__.py

@@ -0,0 +1,2 @@
+from .low_level_features_extractor import LowLevelFeatureExtractor
+from .high_level_features_extractor import HighLevelFeatureExtractor

src/features/extraction/features_list.py

@@ -0,0 +1,69 @@
+
+# Spectral features
+DEFAULT_SPECTRAL_FEATURES = [
+    'spectral_centroid', 
+    'spectral_bandwidth', 
+    'spectral_contrast', 
+    'spectral_flatness', 
+    'spectral_rolloff', 
+    'zero_crossing_rate', 
+    'mfccs', 
+    'chroma_stft', 
+    'spectral_flux'
+]
+ALL_SPECTRAL_FEATURES = [
+    'spectral_centroid', 
+    'spectral_bandwidth', 
+    'spectral_contrast', 
+    'spectral_flatness', 
+    'spectral_rolloff', 
+    'zero_crossing_rate', 
+    'mfccs', 
+    'chroma_stft', 
+    'spectral_flux'
+]
+
+# Prosodic features
+DEFAULT_PROSODIC_FEATURES = ['f0', 'energy', 'speaking_rate', 'pauses', 'formants']
+ALL_PROSODIC_FEATURES = ['f0', 'energy', 'speaking_rate', 'pauses',  'formants']
+
+# Voice Quality Features
+DEFAULT_VOICE_QUALITY_FEATURES = [
+    'jitter',
+    'shimmer',
+    'hnr',
+    'speech_rate'
+]
+ALL_VOICE_QUALITY_FEATURES = [
+    'jitter_local',	
+    'jitter_rap', 
+    'jitter_ppq5', 
+    'shimmer_local', 
+    'shimmer_apq3', 
+    'shimmer_apq5', 
+    'shimmer_dda', 
+    'hnr',
+    'voicedcount',
+    'npause',
+    'intensity_duration',
+    'speakingrate',
+    'articulationrate',
+    'asd',
+    'totalpauseduration'
+]
+
+
+# Default features to extract
+DEFAULT_FEATURES = {
+    'spectral': DEFAULT_SPECTRAL_FEATURES,
+    'prosodic': DEFAULT_PROSODIC_FEATURES,
+    'voice_quality': DEFAULT_VOICE_QUALITY_FEATURES
+}
+
+# All features to extract
+ALL_FEATURES = {
+    'spectral': ALL_SPECTRAL_FEATURES,
+    'prosodic': ALL_PROSODIC_FEATURES,
+    'voice_quality': ALL_VOICE_QUALITY_FEATURES
+}
+

src/features/extraction/high_level_features_extractor.py

@@ -0,0 +1,159 @@
+import numpy as np
+from .stat_measures import StatisticalMeasures
+from .features_list import ALL_FEATURES
+
+
+class HighLevelFeatureExtractor:
+    """
+    A class to extract high-level statistical measures from low-level audio features, which include
+    spectral, prosodic, voice quality characteristics and more.
+
+    Attributes:
+        measures (list of str): List of statistical measures to compute for each feature.
+
+    Methods:
+        compute_high_level_features(feature_dict): Computes high-level features for a given set of audio features.
+        high_level_feature_generator(low_level_gen): Generates high-level features from a generator of low-level features.
+    """
+    def __init__(self, measures=None):
+        """
+        Initializes the HighLevelFeatureExtractor with a list of statistical measures.
+
+        Args:
+            measures (list of str, optional): A list of statistical measures to apply. Default measures include
+                                              mean, standard deviation, variance, min, max, range, and percentiles.
+        """
+        self.measures = measures if measures is not None else ['mean', 'std', 'var', 'min', 'max', 'range', '25th_percentile', '50th_percentile', '75th_percentile', 'skew', 'kurtosis']
+
+    def compute_high_level_features(self, feature_dict):
+        """
+        Computes high-level features for a dictionary of extracted low-level features.
+
+        Args:
+            feature_dict (dict): Dictionary containing low-level audio feature arrays.
+
+        Returns:
+            dict: A dictionary containing high-level statistical features.
+        """
+        features = {
+            'audio_id': feature_dict['audio_id'],
+            'real_or_fake': feature_dict['real_or_fake']
+        }
+
+        # Compute high-level spectral, prosodic, and voice quality features
+        features.update(self._compute_spectral_features(feature_dict, ALL_FEATURES['spectral']))
+        features.update(self._compute_prosodic_features(feature_dict, ALL_FEATURES['prosodic']))
+        features.update(self._compute_voice_quality_features(feature_dict, ALL_FEATURES['voice_quality']))
+
+        return features
+    
+    def high_level_feature_generator(self, low_level_gen):
+        """
+        Generator to process each set of low-level features and compute high-level features.
+
+        Args:
+            low_level_gen (generator): Generator yielding dictionaries of low-level features.
+
+        Yields:
+            dict: High-level features computed from each low-level feature set.
+        """
+        for low_level_features in low_level_gen:
+            yield self.compute_high_level_features(feature_dict=low_level_features)
+    
+    
+    ##################################################################
+    ### Additional private methods to compute each type of feature ###
+    ##################################################################
+    def _compute_spectral_features(self, feature_dict, spectral_features):
+        """
+        Computes high-level statistical features for spectral features.
+        """
+        spectral_features_dict = {}
+        for feature_name in spectral_features:
+            if feature_name in ['mfccs', 'chroma_stft']:
+                continue
+            feature_array = feature_dict.get(feature_name)
+            if feature_array is not None:
+                stats = StatisticalMeasures.compute_statistical_measures(feature_array, self.measures)
+                spectral_features_dict.update({f"{feature_name}_{key}": value for key, value in stats.items()})
+                
+        # Compute MFCC Features
+        if 'mfccs' in ALL_FEATURES['spectral'] and 'mfccs' in feature_dict:
+            spectral_features_dict.update(self._compute_mfcc_features(feature_dict['mfccs']))
+            
+        # Compute Chroma Features
+        if 'chroma_stft' in ALL_FEATURES['spectral'] and 'chroma_stft' in feature_dict:
+            spectral_features_dict.update(self._compute_chroma_features(feature_dict['chroma_stft']))
+            
+        return spectral_features_dict
+    
+    def _compute_mfcc_features(self, mfccs_flat):
+        """
+        MFCC features are computed from a flattened array of MFCC coefficients.
+        """
+        mfccs = mfccs_flat.reshape(-1, 13)  # Reshape the flattened array to its original form
+        mfcc_features_dict = {}
+        for i in range(mfccs.shape[1]):
+            feature_array = mfccs[:, i]
+            stats = StatisticalMeasures.compute_statistical_measures(feature_array, self.measures)
+            mfcc_features_dict.update({f"mfcc_{i+1}_{key}": value for key, value in stats.items()})
+        return mfcc_features_dict
+    
+    def _compute_chroma_features(self, chroma_flat):
+        """
+        Chroma features are computed from a flattened array of chroma coefficients.
+        """
+        chroma = chroma_flat.reshape(-1, 12)  # Reshape the flattened array to its original form
+        chroma_features_dict = {}
+        for i in range(chroma.shape[1]):
+            feature_array = chroma[:, i]
+            stats = StatisticalMeasures.compute_statistical_measures(feature_array, self.measures)
+            chroma_features_dict.update({f"chroma_{i+1}_{key}": value for key, value in stats.items()})
+        return chroma_features_dict
+    
+    def _compute_prosodic_features(self, feature_dict, prosodic_features):
+        """
+        Computes high-level statistical features for prosodic features.
+        """
+        prosodic_features_dict = {}
+        for feature_name in prosodic_features:
+            if feature_name in ['speaking_rate', 'pauses', 'formants']:
+                continue
+            feature_array = feature_dict.get(feature_name)
+            if isinstance(feature_array, dict):
+                raise TypeError(f"Expected array for {feature_name}, but got {type(feature_array).__name__}")
+            if feature_array is not None:
+                stats = StatisticalMeasures.compute_statistical_measures(feature_array, self.measures)
+                prosodic_features_dict.update({f"{feature_name}_{key}": value for key, value in stats.items()})
+
+        if 'speaking_rate' in prosodic_features and 'speaking_rate' in feature_dict:
+            prosodic_features_dict['speaking_rate'] = feature_dict['speaking_rate']
+
+        if 'pauses' in prosodic_features and 'pauses' in feature_dict:
+            pauses = feature_dict['pauses']
+            if pauses:
+                pause_durations = np.array([end - start for start, end in pauses])
+                pause_stats = StatisticalMeasures.compute_statistical_measures(pause_durations, self.measures)
+                prosodic_features_dict.update({f"pause_{key}": value for key, value in pause_stats.items()})
+            else:
+                for measure in self.measures:
+                    prosodic_features_dict[f'pause_{measure}'] = np.nan
+                    
+        if 'formants' in prosodic_features and 'formants' in feature_dict:
+            formant_values = feature_dict['formants']
+            if formant_values:
+                for key, value in formant_values.items():
+                    prosodic_features_dict[key] = value
+                
+        return prosodic_features_dict
+    
+    def _compute_voice_quality_features(self, feature_dict, voice_quality_features):
+        """
+        Computes high-level statistical features for voice quality features.
+        """
+        voice_quality_features_dict = {}
+        for feature_name in voice_quality_features:
+            feature_value = feature_dict.get(feature_name)
+            if feature_value is not None:
+                voice_quality_features_dict[feature_name] = feature_value
+        return voice_quality_features_dict

src/features/extraction/low_level_features_extractor.py

@@ -0,0 +1,112 @@
+import librosa
+import parselmouth
+from parselmouth.praat import call
+
+import numpy as np
+from tqdm import tqdm
+
+from .spectral_features import SpectralFeatureExtractor
+from .prosodic_features import ProsodicFeatureExtractor
+from .voice_quality_features import VoiceQualityFeatureExtractor
+from .features_list import DEFAULT_FEATURES
+
+
+class LowLevelFeatureExtractor:
+    """
+    A class to orchestrate the extraction of low-level audio features across spectral, prosodic,
+    and voice quality domains.
+
+    Attributes:
+        target_sr (int): The target sampling rate for audio resampling.
+        include_spectral (bool): Flag to include spectral feature extraction.
+        include_prosodic (bool): Flag to include prosodic feature extraction.
+        include_voice_quality (bool): Flag to include voice quality feature extraction.
+        spectral_features (list): List of spectral features to extract.
+        prosodic_features (list): List of prosodic features to extract.
+        voice_quality_features (list): List of voice quality features to extract.
+
+    Methods:
+        resample_audio(audio_arr, orig_sr): Resamples the audio to the target sampling rate.
+        extract_features(row): Extracts all configured features for a single audio example.
+        low_level_feature_generator(df): Generator that processes a DataFrame of audio examples.
+    """
+    def __init__(self, target_sr=16000, include_only=None, spectral_features=None, prosodic_features=None, voice_quality_features=None):
+        """
+        Initializes the LowLevelFeatureExtractor with optional feature lists and inclusion flags.
+
+        Args:
+            target_sr (int): Desired sampling rate for the analysis.
+            include_only (list of str): Optional, specify which feature groups to include.
+            spectral_features (list of str): Specific spectral features to extract.
+            prosodic_features (list of str): Specific prosodic features to extract.
+            voice_quality_features (list of str): Specific voice quality features to extract.
+        """
+        self.target_sr = target_sr
+        self.include_spectral = True
+        self.include_prosodic = True
+        self.include_voice_quality = True
+        self.spectral_features = spectral_features if spectral_features is not None else DEFAULT_FEATURES['spectral']
+        self.prosodic_features = prosodic_features if prosodic_features is not None else DEFAULT_FEATURES['prosodic']
+        self.voice_quality_features = voice_quality_features if voice_quality_features is not None else DEFAULT_FEATURES['voice_quality']
+
+        if include_only is not None:
+            self.include_spectral      = 'spectral' in include_only
+            self.include_prosodic      = 'prosodic' in include_only
+            self.include_voice_quality = 'voice_quality' in include_only
+            
+            if not self.include_spectral:
+                self.spectral_features = []
+            
+            if not self.include_prosodic:
+                self.prosodic_features = []
+            
+            if not self.include_voice_quality:
+                self.voice_quality_features = []
+
+    def resample_audio(self, audio_arr, orig_sr):
+        """
+        Resamples the given audio array from its original sampling rate to the target rate.
+        """
+        return librosa.resample(audio_arr, orig_sr=orig_sr, target_sr=self.target_sr)
+
+    def extract_features(self, row):
+        """
+        Extracts features from a single row of audio data, which includes audio id, array, and other metadata.
+        """
+        audio_id = row['audio_id']
+        audio_arr = row['audio_arr']
+        orig_sr = row['srate']
+        real_or_fake = row['real_or_fake']
+
+        y = self.resample_audio(audio_arr, orig_sr)
+        
+        features = {}
+        
+        if self.include_spectral:
+                spectral_extractor = SpectralFeatureExtractor(y, self.target_sr)
+                features.update(spectral_extractor.extract(self.spectral_features))
+
+        if self.include_prosodic:
+            prosodic_extractor = ProsodicFeatureExtractor(y, self.target_sr, audio_arr, orig_sr)
+            features.update(prosodic_extractor.extract(self.prosodic_features))
+            
+        if self.include_voice_quality:
+            voice_quality_extractor = VoiceQualityFeatureExtractor(audio_arr, orig_sr)
+            features.update(voice_quality_extractor.extract(self.voice_quality_features))
+
+        features = {**{'audio_id': audio_id, 'real_or_fake': real_or_fake}, **features}
+
+        return features
+
+    def low_level_feature_generator(self, df):
+        """
+        A generator that processes a DataFrame of audio examples to extract features.
+
+        Args:
+            df (pandas.DataFrame): DataFrame containing columns with audio data and metadata.
+
+        Yields:
+            dict: A dictionary of extracted features for each audio file.
+        """
+        for _, row in tqdm(df.iterrows(), total=len(df), desc="Processing Audios"):
+            yield self.extract_features(row)

src/features/extraction/prosodic_features.py

@@ -0,0 +1,127 @@
+import librosa
+import parselmouth
+from parselmouth.praat import call
+
+import numpy as np
+
+
+class ProsodicFeatureExtractor:
+    """
+    A class for extracting various prosodic features from audio data.
+
+    Attributes:
+        y (numpy.array): Audio time series.
+        sr (int): Sampling rate of the audio time series.
+        audio_arr (numpy.array): Original audio array for parselmouth processing.
+        orig_sr (int): Original sampling rate of the audio array 
+
+    Methods:
+        extract(features_to_extract=None): Extracts specified prosodic features from audio.
+        extract_f0(): Extracts fundamental frequency (F0) from audio.
+        extract_energy(): Extracts energy from audio.
+        extract_speaking_rate(): Estimates the speaking rate from audio.
+        extract_pauses(): Detects pauses from audio.
+        extract_formants(): Extracts formant frequencies from audio.
+    """
+    def __init__(self, y, sr, audio_arr, orig_sr):
+        """
+        Initializes the ProsodicFeatureExtractor with audio data.
+        """
+        self.y = y
+        self.sr = sr
+        self.audio_arr = audio_arr
+        self.orig_sr = orig_sr
+
+    def extract(self, features_to_extract=None):
+        """
+        Extracts the specified prosodic features.
+
+        Args:
+            features_to_extract (list, optional): List of feature names to extract.
+                Defaults to all available features if None.
+
+        Returns:
+            dict: A dictionary containing the extracted features.
+        """
+        feature_funcs = {
+            'f0': self.extract_f0,
+            'energy': self.extract_energy,
+            'speaking_rate': self.extract_speaking_rate,
+            'pauses': self.extract_pauses,
+            'formants': self.extract_formants
+        }
+
+        if features_to_extract is None:
+            features_to_extract = feature_funcs.keys()
+
+        features = {}
+        for feature in features_to_extract:
+            if feature in feature_funcs:
+                result = feature_funcs[feature]()
+                if isinstance(result, tuple):
+                    features.update(result)
+                else:
+                    features[feature] = result
+                    
+        return features
+
+    def extract_f0(self):
+        """
+        Extracts the fundamental frequency (F0) using PYIN algorithm.
+        """
+        f0, voiced_flag, voiced_probs = librosa.pyin(self.y, fmin=librosa.note_to_hz('C2'), fmax=librosa.note_to_hz('C7'))
+        f0 = np.nan_to_num(f0)
+        return f0
+
+    def extract_energy(self):
+        """
+        Extracts the root-mean-square (RMS) energy from the audio.
+        """
+        return librosa.feature.rms(y=self.y)[0]
+
+
+    def extract_speaking_rate(self):
+        """
+        Estimates the speaking rate by calculating the number of syllables per second.
+        """
+        try:
+            snd = parselmouth.Sound(self.audio_arr, sampling_frequency=self.orig_sr)
+            total_duration = snd.get_total_duration()
+            intensity = snd.to_intensity()
+            intensity_values = intensity.values.T
+            threshold = 0.3 * max(intensity_values)
+            syllable_count = len([1 for i in intensity_values if i > threshold])
+            speaking_rate = syllable_count / total_duration
+            return speaking_rate
+        except Exception as e:
+            print(f'Error extracting speaking rate: {e}')
+            return None
+
+    def extract_pauses(self):
+        """
+        Identifies and timestamps pauses in the audio.
+        """
+        try:
+            snd = parselmouth.Sound(self.audio_arr, sampling_frequency=self.orig_sr)
+            silences = call(snd, "To TextGrid (silences)", 100, 0, -25, 0.1, 0.1, "silent", "sounding")
+            pauses = [(call(silences, "Get start time of interval", 1, i), call(silences, "Get end time of interval", 1, i)) for i in range(1, call(silences, "Get number of intervals", 1) + 1) if call(silences, "Get label of interval", 1, i) == "silent"]
+            return pauses
+        except Exception as e:
+            print(f'Error extracting pauses: {e}')
+            return None
+        
+    def extract_formants(self):
+        """
+        Extracts the first three formant frequencies using the Burg method.
+        """
+        try:
+            snd = parselmouth.Sound(self.audio_arr, sampling_frequency=self.orig_sr)
+            formant = call(snd, "To Formant (burg)", 0.025, 5, 5500, 0.025, 50)
+            formant_values = {}
+            for i in range(1, 4):  # Extracting the first three formants
+                formant_values[f'F{i}_mean'] = call(formant, "Get mean", i, 0, 0, "Hertz")
+                formant_values[f'F{i}_stdev'] = call(formant, "Get standard deviation", i, 0, 0, "Hertz")
+            return formant_values
+        except Exception as e:
+            print(f'Error extracting formants: {e}')
+            return {}

src/features/extraction/spectral_features.py

@@ -0,0 +1,117 @@
+import librosa
+import numpy as np
+
+
+class SpectralFeatureExtractor:
+    """
+    A class to extract various spectral features from audio data using the librosa library.
+
+    Attributes:
+        y (numpy.array): Audio time series.
+        sr (int): Sampling rate of the audio time series.
+
+    Methods:
+        extract(features_to_extract=None): Extracts specified spectral features from audio.
+        spectral_centroid(): Computes the spectral centroid of the audio.
+        spectral_bandwidth(): Computes the spectral bandwidth of the audio.
+        spectral_contrast(): Computes the spectral contrast of the audio.
+        spectral_flatness(): Computes the spectral flatness of the audio.
+        spectral_rolloff(): Computes the spectral rolloff of the audio.
+        zero_crossing_rate(): Computes the zero crossing rate of the audio.
+        mfccs(): Computes the Mel-frequency cepstral coefficients (MFCCs) of the audio.
+        chroma_stft(): Computes the chromagram from a waveform or power spectrogram.
+        spectral_flux(): Computes the spectral flux of the audio.
+    """
+    def __init__(self, y, sr):
+        """
+        Initializes the SpectralFeatureExtractor with audio data.
+        """
+        self.y = y
+        self.sr = sr
+
+    def extract(self, features_to_extract=None):
+        """
+        Extracts the specified spectral features.
+        
+        Args:
+            features_to_extract (list of str, optional): A list of feature names to extract.
+                Defaults to extracting all available features if None.
+
+        Returns:
+            dict: A dictionary containing the extracted features.
+        """
+        feature_funcs = {
+            'spectral_centroid': self.spectral_centroid,
+            'spectral_bandwidth': self.spectral_bandwidth,
+            'spectral_contrast': self.spectral_contrast,
+            'spectral_flatness': self.spectral_flatness,
+            'spectral_rolloff': self.spectral_rolloff,
+            'zero_crossing_rate': self.zero_crossing_rate,
+            'mfccs': self.mfccs,
+            'chroma_stft': self.chroma_stft,
+            'spectral_flux': self.spectral_flux
+        }
+
+        if features_to_extract is None:
+            features_to_extract = feature_funcs.keys()
+
+        features = {}
+        for feature in features_to_extract:
+            if feature in feature_funcs:
+                features[feature] = feature_funcs[feature]()
+        return features
+
+    def spectral_centroid(self):
+        """
+        Computes the spectral centroid of the audio.
+        """
+        return librosa.feature.spectral_centroid(y=self.y, sr=self.sr).flatten()
+
+    def spectral_bandwidth(self):
+        """
+        Computes the spectral bandwidth of the audio.
+        """
+        return librosa.feature.spectral_bandwidth(y=self.y, sr=self.sr).flatten()
+
+    def spectral_contrast(self):
+        """
+        Computes the spectral contrast of the audio.
+        """
+        return librosa.feature.spectral_contrast(y=self.y, sr=self.sr).flatten()
+
+    def spectral_flatness(self):
+        """
+        Computes the spectral flatness of the audio.
+        """
+        return librosa.feature.spectral_flatness(y=self.y).flatten()
+
+    def spectral_rolloff(self):
+        """
+        Computes the spectral rolloff point of the audio.
+        """
+        return librosa.feature.spectral_rolloff(y=self.y, sr=self.sr).flatten()
+
+    def zero_crossing_rate(self):
+        """
+        Computes the zero crossing rate of the audio.
+        """
+        return librosa.feature.zero_crossing_rate(self.y).flatten()
+
+    def mfccs(self):
+        """
+        Computes the Mel-frequency cepstral coefficients (MFCCs) of the audio.
+        """
+        return librosa.feature.mfcc(y=self.y, sr=self.sr, n_mfcc=13).flatten()
+
+    def chroma_stft(self):
+        """
+        Computes the chromagram from a waveform or power spectrogram.
+        """
+        return librosa.feature.chroma_stft(y=self.y, sr=self.sr).flatten()
+
+    def spectral_flux(self):
+        """
+        Computes the spectral flux of the audio, indicating the rate of change in the power spectrum.
+        """
+        S = np.abs(librosa.stft(self.y))
+        return np.sqrt(np.sum(np.diff(S, axis=1)**2, axis=0))

src/features/extraction/stat_measures.py

@@ -0,0 +1,49 @@
+import numpy as np
+from scipy.stats import skew, kurtosis
+
+
+class StatisticalMeasures:
+    """
+    A class dedicated to computing various statistical measures on a given feature array.
+    This utility class provides a method to compute a predefined set of statistical measures
+    which are commonly used in data analysis and feature extraction.
+
+    Methods:
+    --------
+    compute_statistical_measures(feature_array, measures=None):
+        Computes selected statistical measures from a provided numerical array.
+    """
+    @staticmethod
+    def compute_statistical_measures(feature_array, measures=None):
+        
+        if measures is None:
+            measures = ['mean', 'std', 'var', 'min', 'max', 'range', '25th_percentile', '50th_percentile', '75th_percentile', 'skew', 'kurtosis']
+        
+        stats = {}
+        if 'mean' in measures:
+            stats['mean'] = np.mean(feature_array)
+        if 'std' in measures:
+            stats['std'] = np.std(feature_array)
+        if 'var' in measures:
+            stats['var'] = np.var(feature_array)
+        if 'min' in measures:
+            stats['min'] = np.min(feature_array)
+        if 'max' in measures:
+            stats['max'] = np.max(feature_array)
+        if 'range' in measures:
+            stats['range'] = np.ptp(feature_array)
+        if '25th_percentile' in measures:
+            stats['25th_percentile'] = np.percentile(feature_array, 25)
+        if '50th_percentile' in measures:
+            stats['50th_percentile'] = np.percentile(feature_array, 50)
+        if '75th_percentile' in measures:
+            stats['75th_percentile'] = np.percentile(feature_array, 75)
+        if 'skew' in measures and len(np.unique(feature_array)) > 1:
+            stats['skew'] = skew(feature_array)
+        else:
+            stats['skew'] = np.nan
+        if 'kurtosis' in measures and len(np.unique(feature_array)) > 1:
+            stats['kurtosis'] = kurtosis(feature_array)
+        else:
+            stats['kurtosis'] = np.nan
+        return stats

src/features/extraction/voice_quality_features.py

@@ -0,0 +1,272 @@
+import parselmouth
+from parselmouth.praat import call
+import numpy as np
+import math
+
+
+class VoiceQualityFeatureExtractor:
+    """
+    A class to extract various voice quality features from audio data.
+
+    Attributes:
+        audio_arr (numpy.array): The audio array used for processing.
+        orig_sr (int): The original sampling rate of the audio.
+
+    Methods:
+        extract(features_to_extract=None): Main method to extract specified voice quality features.
+        extract_jitter(): Extracts measures of frequency variation (jitter).
+        extract_shimmer(): Extracts measures of amplitude variation (shimmer).
+        extract_hnr(): Extracts the Harmonics-to-Noise Ratio (HNR).
+        extract_speech_rate(): Calculates various speech rate metrics.
+        measure_speech_rate(voiceID): Helper method to perform detailed speech rate analysis.
+    """
+    def __init__(self, audio_arr, orig_sr):
+        """
+        Initializes the VoiceQualityFeatureExtractor with audio data.
+        """
+        self.audio_arr = audio_arr
+        self.orig_sr = orig_sr
+
+    def extract(self, features_to_extract=None):
+        """
+        Extracts specified voice quality features from the audio data.
+        
+        Args:
+            features_to_extract (list of str, optional): A list of feature names to extract.
+                Defaults to extracting all available features if None.
+
+        Returns:
+            dict: A dictionary containing the extracted features.
+        """
+        feature_funcs = {
+            'jitter': self.extract_jitter,
+            'shimmer': self.extract_shimmer,
+            'hnr': self.extract_hnr,
+            'speech_rate': self.extract_speech_rate
+        }
+
+        if features_to_extract is None:
+            features_to_extract = feature_funcs.keys()
+
+        features = {}
+        for feature in features_to_extract:
+            if feature in feature_funcs:
+                feature_values = feature_funcs[feature]()
+                if isinstance(feature_values, dict):
+                    features.update(feature_values)
+                else:
+                    features[feature] = feature_values
+        return features
+
+    def extract_jitter(self):
+        """
+        Extracts jitter measures from the audio data.
+        """
+        try:
+            snd = parselmouth.Sound(self.audio_arr, sampling_frequency=self.orig_sr)
+            point_process = call(snd, "To PointProcess (periodic, cc)", 75, 500)
+            jitter_local = call(point_process, "Get jitter (local)", 0, 0, 0.0001, 0.02, 1.3)
+            jitter_rap = call(point_process, "Get jitter (rap)", 0, 0, 0.0001, 0.02, 1.3)
+            jitter_ppq5 = call(point_process, "Get jitter (ppq5)", 0, 0, 0.0001, 0.02, 1.3)
+            return {
+                'jitter_local': jitter_local,
+                'jitter_rap': jitter_rap,
+                'jitter_ppq5': jitter_ppq5
+            }
+        except Exception as e:
+            print(f'Error extracting jitter: {e}')
+            return {
+                'jitter_local': np.nan,
+                'jitter_rap': np.nan,
+                'jitter_ppq5': np.nan
+            }
+
+    def extract_shimmer(self):
+        """
+        Extracts shimmer measures from the audio data.
+        """
+        try:
+            snd = parselmouth.Sound(self.audio_arr, sampling_frequency=self.orig_sr)
+            point_process = call(snd, "To PointProcess (periodic, cc)", 75, 500)
+            shimmer_local = call([snd, point_process], "Get shimmer (local)", 0, 0, 0.0001, 0.02, 1.3, 1.6)
+            shimmer_apq3 = call([snd, point_process], "Get shimmer (apq3)", 0, 0, 0.0001, 0.02, 1.3, 1.6)
+            shimmer_apq5 = call([snd, point_process], "Get shimmer (apq5)", 0, 0, 0.0001, 0.02, 1.3, 1.6)
+            shimmer_dda = call([snd, point_process], "Get shimmer (dda)", 0, 0, 0.0001, 0.02, 1.3, 1.6)
+            return {
+                'shimmer_local': shimmer_local,
+                'shimmer_apq3': shimmer_apq3,
+                'shimmer_apq5': shimmer_apq5,
+                'shimmer_dda': shimmer_dda
+            }
+        except Exception as e:
+            print(f'Error extracting shimmer: {e}')
+            return {
+                'shimmer_local': np.nan,
+                'shimmer_apq3': np.nan,
+                'shimmer_apq5': np.nan,
+                'shimmer_dda': np.nan
+            }
+
+    def extract_hnr(self):
+        """
+        Extracts the Harmonics-to-Noise Ratio (HNR) from the audio data.
+        """
+        try:
+            snd = parselmouth.Sound(self.audio_arr, sampling_frequency=self.orig_sr)
+            harmonicity = call(snd, "To Harmonicity (cc)", 0.01, 75, 0.1, 1.0)
+            hnr = call(harmonicity, "Get mean", 0, 0)
+            return {'hnr': hnr}
+        except Exception as e:
+            print(f'Error extracting HNR: {e}')
+            return {'hnr': np.nan}
+        
+        
+    def extract_speech_rate(self):
+        """
+        Calculates and extracts various metrics related to speech rate.
+        """
+        try:
+            sound = parselmouth.Sound(self.audio_arr, sampling_frequency=self.orig_sr)
+            (voicedcount, npause, originaldur, intensity_duration, speakingrate, articulationrate, asd, totalpauseduration) = self.measure_speech_rate(sound)
+            return {
+                'voicedcount': voicedcount,
+                'npause': npause,
+                'originaldur': originaldur,
+                'intensity_duration': intensity_duration,
+                'speakingrate': speakingrate,
+                'articulationrate': articulationrate,
+                'asd': asd,
+                'totalpauseduration': totalpauseduration
+            }
+        except Exception as e:
+            print(f'Error extracting speech rate: {e}')
+            return {
+                'voicedcount': np.nan,
+                'npause': np.nan,
+                'originaldur': np.nan,
+                'intensity_duration': np.nan,
+                'speakingrate': np.nan,
+                'articulationrate': np.nan,
+                'asd': np.nan,
+                'totalpauseduration': np.nan
+            }
+    
+    
+    def measure_speech_rate(self, voiceID):
+        """
+        Performs a detailed analysis to measure various speech rate metrics from the given audio.
+
+        This method calculates metrics like the number of voiced segments, number of pauses,
+        the total original duration of the audio, the duration of voiced segments, speaking rate,
+        articulation rate, average syllable duration, and the total duration of pauses.
+        """
+        silencedb = -25
+        mindip = 2
+        minpause = 0.3
+        
+        sound = parselmouth.Sound(voiceID)
+        originaldur = sound.get_total_duration()
+        intensity = sound.to_intensity(50)
+        start = call(intensity, "Get time from frame number", 1)
+        nframes = call(intensity, "Get number of frames")
+        end = call(intensity, "Get time from frame number", nframes)
+        min_intensity = call(intensity, "Get minimum", 0, 0, "Parabolic")
+        max_intensity = call(intensity, "Get maximum", 0, 0, "Parabolic")
+
+        # get .99 quantile to get maximum (without influence of non-speech sound bursts)
+        max_99_intensity = call(intensity, "Get quantile", 0, 0, 0.99)
+
+        # estimate Intensity threshold
+        threshold = max_99_intensity + silencedb
+        threshold2 = max_intensity - max_99_intensity
+        threshold3 = silencedb - threshold2
+        if threshold < min_intensity:
+            threshold = min_intensity
+
+        # get pauses (silences) and speakingtime
+        textgrid = call(intensity, "To TextGrid (silences)", threshold3, minpause, 0.1, "silent", "sounding")
+        silencetier = call(textgrid, "Extract tier", 1)
+
+        silencetable = call(silencetier, "Down to TableOfReal", "sounding")
+        npauses = call(silencetable, "Get number of rows")
+
+        speakingtot = 0
+        for ipause in range(npauses):
+            pause = ipause + 1
+            beginsound = call(silencetable, "Get value", pause, 1)
+            endsound = call(silencetable, "Get value", pause, 2)
+            speakingdur = endsound - beginsound
+            speakingtot += speakingdur
+        total_pause_duration = originaldur - speakingtot
+
+        intensity_matrix = call(intensity, "Down to Matrix")
+        sound_from_intensity_matrix = call(intensity_matrix, "To Sound (slice)", 1)
+        intensity_duration = call(sound_from_intensity_matrix, "Get total duration")
+        intensity_max = call(sound_from_intensity_matrix, "Get maximum", 0, 0, "Parabolic")
+        point_process = call(sound_from_intensity_matrix, "To PointProcess (extrema)", "Left", "yes", "no", "Sinc70")
+        numpeaks = call(point_process, "Get number of points")
+        t = [call(point_process, "Get time from index", i + 1) for i in range(numpeaks)]
+
+        timepeaks = []
+        peakcount = 0
+        intensities = []
+        for i in range(numpeaks):
+            value = call(sound_from_intensity_matrix, "Get value at time", t[i], "Cubic")
+            if value > threshold:
+                peakcount += 1
+                intensities.append(value)
+                timepeaks.append(t[i])
+
+        validpeakcount = 0
+        currenttime = timepeaks[0]
+        currentint = intensities[0]
+        validtime = []
+
+        for p in range(peakcount - 1):
+            following = p + 1
+            followingtime = timepeaks[p + 1]
+            dip = call(intensity, "Get minimum", currenttime, timepeaks[p + 1], "None")
+            diffint = abs(currentint - dip)
+            if diffint > mindip:
+                validpeakcount += 1
+                validtime.append(timepeaks[p])
+            currenttime = timepeaks[following]
+            currentint = call(intensity, "Get value at time", timepeaks[following], "Cubic")
+
+        pitch = sound.to_pitch_ac(0.02, 30, 4, False, 0.03, 0.25, 0.01, 0.35, 0.25, 450)
+        voicedcount = 0
+        voicedpeak = []
+
+        for time in range(validpeakcount):
+            querytime = validtime[time]
+            whichinterval = call(textgrid, "Get interval at time", 1, querytime)
+            whichlabel = call(textgrid, "Get label of interval", 1, whichinterval)
+            value = pitch.get_value_at_time(querytime) 
+            if not math.isnan(value):
+                if whichlabel == "sounding":
+                    voicedcount += 1
+                    voicedpeak.append(validtime[time])
+
+        timecorrection = originaldur / intensity_duration
+        call(textgrid, "Insert point tier", 1, "syllables")
+        for i in range(len(voicedpeak)):
+            position = (voicedpeak[i] * timecorrection)
+            call(textgrid, "Insert point", 1, position, "")
+
+        speakingrate = voicedcount / originaldur
+        
+        # Handling division by zero for articulationrate
+        if speakingtot != 0:
+            articulationrate = voicedcount / speakingtot
+        else:
+            articulationrate = float('nan')
+
+        # Handling division by zero for asd
+        if voicedcount != 0:
+            asd = speakingtot / voicedcount
+        else:
+            asd = float('nan')  
+            
+        npause = npauses - 1
+
+        return voicedcount, npause, originaldur, intensity_duration, speakingrate, articulationrate, asd, total_pause_duration

src/models/predict.py

@@ -0,0 +1,36 @@
+import joblib
+import pandas as pd
+from sklearn.impute import SimpleImputer
+from sklearn.preprocessing import StandardScaler
+
+def select_features(features_df: pd.DataFrame):
+    selected_features = ['spectral_contrast_var', 'spectral_contrast_range', 'spectral_contrast_mean', 'F3_mean', 'F2_stdev', 'F3_stdev', 'F1_stdev', 'mfcc_13_std', 'F2_mean', 'mfcc_6_75th_percentile', 'mfcc_12_75th_percentile', 'mfcc_9_75th_percentile', 'mfcc_3_75th_percentile', 'mfcc_12_50th_percentile', 'mfcc_9_50th_percentile', 'mfcc_2_50th_percentile', 'mfcc_5_50th_percentile', 'mfcc_7_50th_percentile', 'f0_skew', 'pause_std', 'asd', 'pause_75th_percentile', 'chroma_11_50th_percentile', 'chroma_3_50th_percentile', 'chroma_6_50th_percentile', 'spectral_flux_skew', 'mfcc_12_25th_percentile', 'mfcc_6_25th_percentile', 'mfcc_2_25th_percentile', 'spectral_bandwidth_min', 'zero_crossing_rate_skew', 'chroma_1_range', 'speaking_rate', 'chroma_12_range', 'chroma_2_range', 'chroma_3_range', 'chroma_5_range', 'chroma_10_range', 'spectral_flatness_skew', 'chroma_6_range', 'chroma_8_range', 'chroma_7_range', 'chroma_9_range', 'f0_kurtosis', 'chroma_11_range', 'spectral_bandwidth_kurtosis', 'chroma_6_max', 'chroma_10_max', 'chroma_2_max', 'chroma_12_max', 'chroma_5_max', 'chroma_7_max', 'chroma_4_max', 'chroma_1_max', 'chroma_11_max', 'chroma_4_std', 'chroma_6_std', 'chroma_7_std', 'chroma_3_max', 'chroma_12_std', 'chroma_11_std', 'chroma_2_std', 'chroma_10_std', 'chroma_3_std', 'chroma_9_std', 'chroma_8_std', 'chroma_5_std', 'chroma_1_std', 'zero_crossing_rate_range', 'mfcc_1_skew', 'spectral_rolloff_range', 'f0_25th_percentile', 'pause_skew', 'chroma_9_min', 'mfcc_13_mean', 'mfcc_11_mean', 'zero_crossing_rate_min', 'spectral_bandwidth_max', 'mfcc_10_max', 'f0_75th_percentile', 'mfcc_5_max', 'mfcc_6_mean', 'mfcc_3_max', 'jitter_local', 'spectral_flux_25th_percentile', 'spectral_flatness_min', 'energy_min', 'shimmer_local', 'spectral_flatness_range']
+    print(f"Number of features {len(selected_features)}")
+    features_df = features_df[selected_features]
+    return features_df
+
+def impute_missing_values(features_df: pd.DataFrame):
+    """Impute missing values in feature set."""
+    imputer = SimpleImputer(strategy='mean')
+    features_imputed = imputer.fit_transform(features_df)
+    scaler = StandardScaler()
+    features_scaled = scaler.fit_transform(features_imputed)
+    return features_scaled
+
+def preprocess(features_df: pd.DataFrame):
+    features_df = select_features(features_df)
+    features_df = impute_missing_values(features_df)
+    return features_df
+
+def load_model():
+    model_path = './models/logistic_regression_model.pkl'
+    model = joblib.load(model_path)
+    return model
+
+def predict(features_df: pd.DataFrame):
+    features_df = preprocess(features_df)
+    model = load_model()
+    predictions = model.predict(features_df)
+    # Prediction Probabilities
+    prediction_probabilities = model.predict_proba(features_df)
+    return predictions, prediction_probabilities