[+] Calculate freq statistics

src/formant.py

@@ -1,11 +1,16 @@
 from __future__ import annotations
 
 import csv
+import json
 import os
+from dataclasses import dataclass
 from multiprocessing import Pool
 from os import PathLike
 from pathlib import Path
+from typing import Iterable
 
+import jsonpickle as jsonpickle
+import numpy as np
 import tqdm
 import matplotlib.pyplot as plt
 import numpy
@@ -49,6 +54,50 @@ def load_vox_celeb_asab_dict(path: PathLike) -> dict[str, str]:
         return {row[0]: row[2] for row in csv.reader(f, delimiter='\t') if row[0].startswith('id')}
 
 
+def loop_id_dirs() -> Iterable[Path]:
+    # Loop through all ids
+    for id in agab:
+        id_dir = vox_celeb_dir.joinpath(id)
+
+        # Check if directory exists
+        if not id_dir.is_dir():
+            continue
+
+        yield id, id_dir
+
+
+def get_audio_paths(id_dir: Path, audio_suffix: str = 'wav') -> list[str]:
+    """
+    Get all audio paths under one person
+
+    :param id_dir: Person ID directory
+    :param audio_suffix: Select only files with this suffix
+    :return: audio paths
+    """
+    audios = []
+
+    # Loop through all videos
+    for vid in os.listdir(id_dir):
+        vid_dir = id_dir.joinpath(vid)
+
+        # Check if it's a video directory
+        if not vid_dir.is_dir():
+            continue
+
+        # Loop through all audios
+        for aud in os.listdir(vid_dir):
+            aud_dir = vid_dir.joinpath(aud)
+
+            # Check if end with suffix
+            if not aud.endswith(audio_suffix):
+                continue
+
+            # Add
+            audios.append(str(aud_dir))
+
+    return audios
+
+
 def compute_vox_celeb_helper(aud_dir: str):
     """
     Compute one audio file
@@ -61,40 +110,12 @@ def compute_vox_celeb_helper(aud_dir: str):
 
 
 def compute_vox_celeb():
-    vox_celeb_dir = Path('C:/Workspace/EECS 6414/Datasets/VoxCeleb1/wav')
-    audio_suffix = 'wav'
-
     print('Finding audio files...')
-
-    asab = load_vox_celeb_asab_dict(vox_celeb_dir.joinpath('../vox1_meta.csv'))
     queue: list[str] = []
 
     # Loop through all ids
-    for id in asab:
-        id_dir = vox_celeb_dir.joinpath(id)
-
-        # Check if directory exists
-        if not id_dir.is_dir():
-            continue
-
-        # Loop through all videos
-        for vid in os.listdir(id_dir):
-            vid_dir = id_dir.joinpath(vid)
-
-            # Check if it's a video directory
-            if not vid_dir.is_dir():
-                continue
-
-            # Loop through all audios
-            for aud in os.listdir(vid_dir):
-                aud_dir = vid_dir.joinpath(aud)
-
-                # Check if end with suffix
-                if not aud.endswith(audio_suffix):
-                    continue
-
-                # Add to queue
-                queue.append(str(aud_dir))
+    for id, id_dir in loop_id_dirs():
+        queue += get_audio_paths(id_dir)
 
     print(f'There are {len(queue)} audio files to process.')
     print('Starting processing...')
@@ -105,8 +126,82 @@ def compute_vox_celeb():
             pass
 
 
+@dataclass
+class FrequencyStats:
+    pitch: Statistics
+    f1: Statistics
+    f2: Statistics
+    f3: Statistics
+    f1ratio: Statistics
+    f2ratio: Statistics
+    f3ratio: Statistics
+
+
+@dataclass
+class Statistics:
+    mean: float
+    median: float
+    q1: float
+    q3: float
+    iqr: float
+    min: float
+    max: float
+    n: int
+
+
+def calculate_statistics(arr: np.ndarray) -> FrequencyStats:
+    """
+    Calculate frequency data array statistics
+
+    :param arr: n-by-4 Array from calculate_freq_info
+    :return: Statistics
+    """
+    result: list[Statistics] = []
+
+    def calc_col_stats(col: np.ndarray) -> Statistics:
+        q1 = np.quantile(col, 0.25)
+        q3 = np.quantile(col, 0.75)
+        return Statistics(
+            float(np.mean(col)),
+            float(np.median(col)),
+            float(q1),
+            float(q3),
+            float(q3 - q1),
+            float(np.min(col)),
+            float(np.max(col)),
+            len(arr)
+        )
+
+    for i in range(0, 4):
+        result.append(calc_col_stats(arr[:, i]))
+
+    for i in range(1, 4):
+        result.append(calc_col_stats(np.divide(arr[:, i], arr[:, 0])))
+
+    return FrequencyStats(*result)
+
+
+def celeb_statistics():
+    # Loop through all ids
+    for id, id_dir in loop_id_dirs():
+
+        # Load all files
+        cumulative: np.ndarray = np.concatenate([np.load(f) for f in get_audio_paths(id_dir, 'npy')])
+
+        # Remove out NaN values
+        cumulative = cumulative[~np.isnan(cumulative).any(axis=1), :]
+        result = calculate_statistics(cumulative)
+
+        # Write results
+        with open(f'{id_dir}/stats.json', 'w') as jsonfile:
+            jsonfile.write(jsonpickle.encode(result, jsonfile, indent=1))
+
+
 if __name__ == '__main__':
-    compute_vox_celeb()
+    vox_celeb_dir = Path('C:/Workspace/EECS 6414/Datasets/VoxCeleb1/wav')
+    agab = load_vox_celeb_asab_dict(vox_celeb_dir.joinpath('../vox1_meta.csv'))
+
+    # print(calculate_freq_info(parselmouth.Sound('../00001.wav')))
+    celeb_statistics()
     # print(calculate_freq_info(parselmouth.Sound('D:/Downloads/Vowels-Extract-Z-44kHz.flac')))
     # print(calculate_freq_info(parselmouth.Sound('D:/Downloads/Vowels-Azalea.flac')))
-