Add child-voice processing and disordered voice verification

Author: martinakaduc

finetune/constants.py

@@ -56,6 +56,7 @@
 SYSTEM_PROMPT = "You are Qwen, a virtual human developed by the Qwen Team, Alibaba Group, capable of perceiving auditory and visual inputs, as well as generating text and speech."
 NORMAL_SPEECH_GENERATION_PROMPT = "Repeat exactly the following sentence and do not alter it. Return only the sentence and nothing else. The sentence is: {transcript}"
 DISORDERED_SPEECH_GENERATION_PROMPT = """Assuming you are a child with speech disordered. {disorder_description} Generate a corresponding transcript with the `{disorder_type} disorder` when the child say: "{transcript}". Return only the transcript text and nothing else."""
+VERIFICATION_PROMPT = """You are a highly experienced Speech-Language Pathologist (SLP). An audio recording will be provided, typically consisting of a speech prompt from a pathologist followed by a child's repetition. The prompt the child is trying to repeat is as follows: {words}. Based on your professional expertise: 1. Assess the child's speech in the recording for signs of typical development or potential speech-language disorder. 2. Conclude your analysis with one of the following labels only: 'typically_developing' or 'speech_disorder'. 3. Provide your response as the correct label without any additional explanation, commentary, or unnecessary text."""  # noqa: E501
 
 DISORDER_DESCRIPTION = {
     "addition": "The children with `addition disorder` will insert of an extra sound into a word.",
@@ -66,3 +67,4 @@
 
 # set use audio in video
 USE_AUDIO_IN_VIDEO = True
+SAMPLE_RATE = 24000

finetune/utils.py

@@ -4,6 +4,8 @@
     DISORDERED_SPEECH_GENERATION_PROMPT,
     DISORDER_DESCRIPTION,
     USE_AUDIO_IN_VIDEO,
+    SAMPLE_RATE,
+    VERIFICATION_PROMPT,
 )
 from transformers import (
     Qwen2_5OmniForConditionalGeneration,
@@ -21,6 +23,18 @@
 import re
 import os
 import json
+import pyworld as pw
+import numpy as np
+import librosa
+import tempfile
+from scipy.signal import butter, lfilter, sosfilt
+
+
+def get_temp_file_path(suffix=".wav"):
+    temp_dir = tempfile.gettempdir()
+    filename = next(tempfile._get_candidate_names()) + suffix
+    return os.path.join(temp_dir, filename)
+
 
 def load_model(model_name, attn_implementation=None):
     if "qwen2.5-omni" in model_name.lower():
@@ -37,7 +51,9 @@ def load_model(model_name, attn_implementation=None):
     return generation_config, model, processor
 
 
-def inference(model, processor, prompts, speaker="Ethan", return_audio=False, **kwargs):
+def inference(
+    model, processor, prompts, audios=[], speaker="Ethan", return_audio=False, **kwargs
+):
     """
     Perform inference using the model and processor.
 
@@ -52,7 +68,7 @@ def inference(model, processor, prompts, speaker="Ethan", return_audio=False, **
     """
 
     conversations = []
-    for prompt in prompts:
+    for idx, prompt in enumerate(prompts):
         conversation = [
             {
                 "role": "system",
@@ -67,6 +83,13 @@ def inference(model, processor, prompts, speaker="Ethan", return_audio=False, **
                 ],
             },
         ]
+        try:
+            if len(audios) > idx and audios[idx] is not None:
+                conversation[1]["content"].insert(
+                    0, {"type": "audio", "audio": audios[idx]}
+                )
+        except Exception as e:
+            print(f"Error processing audio for prompt {idx}: {e}")
         conversations.append(conversation)
 
     # Preparation for inference
@@ -126,7 +149,7 @@ def save_sound_file(audio, file_path):
         audio (np.ndarray): The audio data to save.
         file_path (str): The path where the audio file will be saved.
     """
-    sf.write(file_path, audio, samplerate=24000)
+    sf.write(file_path, audio, samplerate=SAMPLE_RATE)
 
 
 def verify_transcript(transcripts):
@@ -153,6 +176,37 @@ def verify_transcript(transcripts):
     return verified
 
 
+def verify_speech(transcript, audio, label, model, processor, generation_config):
+    """
+    Verify the speech by checking if the transcript matches the audio.
+
+    Args:
+        transcript (str): The transcript to verify.
+        audio_path (str): The path to the audio file.
+        label (str): The expected label for the transcript.
+
+    Returns:
+        bool: True if the transcript matches the audio, False otherwise.
+    """
+    audio_path = get_temp_file_path(suffix=".wav")
+    save_sound_file(audio, audio_path)
+    output = inference(
+        model,
+        processor,
+        prompts=[VERIFICATION_PROMPT.format(words=transcript)],
+        audios=[audio_path],
+        num_return_sequences=1,
+        do_sample=False,
+        max_new_tokens=8,
+        return_audio=False,
+        generation_config=generation_config,
+    )
+    if label.lower() in output[0].strip().lower():
+        return True
+
+    return False
+
+
 def post_process_transcripts(transcripts):
     """
     Post-process the transcripts to remove unnecessary characters and whitespace.
@@ -444,7 +498,8 @@ def create_dataset(model_name, num_samples, data_dir="data", iteration=0, seed=4
 
     # Define prompts and disorder types
     list_of_disorder_types = list(DISORDER_DESCRIPTION.keys())
-    list_speakers = ["Chelsie", "Ethan"]
+    list_speakers = ["Chelsea", "Ethan"]
+    genders = {"Chelsea": "female", "Ethan": "male"}
     trial = 1
 
     # Generate normal speech transcripts
@@ -477,6 +532,7 @@ def create_dataset(model_name, num_samples, data_dir="data", iteration=0, seed=4
             )
             dc_transcript = post_process_transcripts(dc_transcript)
             if dc_transcript[0].strip().lower() == transcript.strip().lower():
+                audio = adult_to_child_voice(audio, gender=genders[speaker])
                 normal_audios.append(audio)
                 has_correct_audio = True
                 break
@@ -504,13 +560,18 @@ def create_dataset(model_name, num_samples, data_dir="data", iteration=0, seed=4
     )
 
     # Generate disordered speech transcripts
-    disordered_transcripts, disordered_original_transcripts, disordered_audios, disordered_types = [], [], [], []
+    (
+        disordered_transcripts,
+        disordered_original_transcripts,
+        disordered_audios,
+        disordered_types,
+    ) = ([], [], [], [])
     for transcript in tqdm(
         normal_transcripts, desc="Generating disordered speech audios"
     ):
         speaker = random.choice(list_speakers)
         disorder_type = random.choice(list_of_disorder_types)
-        dc_transcripts, audios = inference(
+        dc_transcripts, audio = inference(
             model,
             processor,
             prompts=[
@@ -527,11 +588,22 @@ def create_dataset(model_name, num_samples, data_dir="data", iteration=0, seed=4
             return_audio=True,
         )
         dc_transcripts = post_process_transcripts(dc_transcripts)
-        diff_count = editdistance.eval(dc_transcripts[0].strip().lower(), transcript.strip().lower())
-        if diff_count > 0 and diff_count <= 10:
+        diff_count = editdistance.eval(
+            dc_transcripts[0].strip().lower(), transcript.strip().lower()
+        )
+        audio = adult_to_child_voice(audio, gender=genders[speaker])
+        audio_verified = verify_speech(
+            transcript.strip().lower(),
+            audio,
+            "speech_disorder",
+            model,
+            processor,
+            generation_config,
+        )
+        if diff_count > 0 and diff_count <= 10 and audio_verified:
             disordered_transcripts.append(dc_transcripts[0])
             disordered_original_transcripts.append(transcript)
-            disordered_audios.append(audios)
+            disordered_audios.append(audio)
             disordered_types.append(disorder_type)
         else:
             # If the disordered transcript is the same as the normal one, retry
@@ -565,7 +637,22 @@ def create_dataset(model_name, num_samples, data_dir="data", iteration=0, seed=4
                     generation_config=generation_config,
                 )
                 dc_transcript = post_process_transcripts(dc_transcript)
-                if dc_transcript[0].strip().lower() == disordered_transcript.strip().lower():
+                if (
+                    dc_transcript[0].strip().lower()
+                    == disordered_transcript.strip().lower()
+                ):
+                    audio = adult_to_child_voice(audio, gender=genders[speaker])
+                    audio_verified = verify_speech(
+                        transcript.strip().lower(),
+                        audio,
+                        "speech_disorder",
+                        model,
+                        processor,
+                        generation_config,
+                    )
+                    if not audio_verified:
+                        continue
+
                     disordered_audios.append(audio)
                     has_correct_audio = True
                     break
@@ -614,3 +701,140 @@ def create_dataset(model_name, num_samples, data_dir="data", iteration=0, seed=4
     add_dataset_config(dataset_name, f"{dataset_name}/{dataset_name}.json")
 
     return dataset_name
+
+
+def butter_bandpass(lowcut, highcut, fs, order=4):
+    nyquist = fs / 2
+    low = lowcut / nyquist
+    high = highcut / nyquist
+    b, a = butter(order, [low, high], btype="band")
+    return b, a
+
+
+def butter_highpass(cutoff, sr, order=4):
+    nyq = 0.5 * sr
+    normal_cutoff = cutoff / nyq
+    b, a = butter(order, normal_cutoff, btype="high", analog=False)
+    return b, a
+
+
+def apply_eq(y, sr, gender):
+    """Apply high-pass filter at 200 Hz."""
+    # High-pass filter at 200 Hz
+    b, a = butter_highpass(200, sr, order=4)
+    filtered = lfilter(b, a, y)
+    return filtered
+
+
+def add_breathiness(y, sr, level=0.02):
+    """Add light breathiness by mixing filtered white noise."""
+    noise = np.random.randn(len(y))
+    b, a = butter_bandpass(3000, 8000, sr, order=4)
+    breath = lfilter(b, a, noise)
+    return y + (breath * level)
+
+
+def dynamic_pitch_modulation(y, sr, semitone_range=0.5, segment_length=0.2):
+    """Small random pitch changes every few hundred ms."""
+    n_samples = len(y)
+    seg_size = int(segment_length * sr)
+    output = np.zeros_like(y)
+    for start in range(0, n_samples, seg_size):
+        end = min(start + seg_size, n_samples)
+        pitch_shift_amt = np.random.uniform(-semitone_range, semitone_range)
+        shifted = librosa.effects.pitch_shift(
+            y[start:end], sr=sr, n_steps=pitch_shift_amt
+        )
+        output[start:end] = shifted[: end - start]
+    return output
+
+
+def design_peaking_eq(f0, Q, gain_db, sr):
+    """Design a peaking EQ filter (second-order section)."""
+    # Convert gain in dB to linear
+    A = 10 ** (gain_db / 40)
+    w0 = 2 * np.pi * f0 / sr
+    alpha = np.sin(w0) / (2 * Q)
+
+    b0 = 1 + alpha * A
+    b1 = -2 * np.cos(w0)
+    b2 = 1 - alpha * A
+    a0 = 1 + alpha / A
+    a1 = -2 * np.cos(w0)
+    a2 = 1 - alpha / A
+
+    # Normalize
+    b = np.array([b0, b1, b2]) / a0
+    a = np.array([1, a1 / a0, a2 / a0])
+
+    # Convert to SOS format for stability
+    from scipy.signal import tf2sos
+
+    sos = tf2sos(b, a)
+    return sos
+
+
+def apply_3band_eq(y, sr):
+    # 100 Hz: +1 dB, Q=1
+    sos1 = design_peaking_eq(100, Q=1, gain_db=1, sr=sr)
+    # 900 Hz: -7 dB, Q=1
+    sos2 = design_peaking_eq(900, Q=1, gain_db=-7, sr=sr)
+    # 2500 Hz: -10 dB, Q=1
+    sos3 = design_peaking_eq(2500, Q=1, gain_db=-10, sr=sr)
+
+    # Apply sequentially
+    y_eq = sosfilt(sos1, y)
+    y_eq = sosfilt(sos2, y_eq)
+    y_eq = sosfilt(sos3, y_eq)
+    return y_eq
+
+
+def adult_to_child_voice(input_sound, gender):
+    """
+    Convert adult voice to child-like voice using pitch + formant shift.
+
+    Parameters:
+        input_sound (str): Path to input sound file.
+        output_wav (str): Path to save output WAV file.
+        gender (str): 'male' or 'female'
+    """
+    # Load audio
+    x = apply_3band_eq(input_sound, sr=SAMPLE_RATE)
+    if x.ndim > 1:  # Convert stereo to mono
+        x = np.mean(x, axis=1)
+
+    # WORLD analysis
+    f0, timeaxis = pw.harvest(x, SAMPLE_RATE)  # Fundamental frequency
+    sp = pw.cheaptrick(x, f0, timeaxis, SAMPLE_RATE)  # Spectral envelope
+    ap = pw.d4c(x, f0, timeaxis, SAMPLE_RATE)  # Aperiodicity
+
+    # Define pitch & formant scaling
+    if gender.lower() == "male":
+        pitch_scale = random.uniform(1.7, 1.9)  # Shift pitch higher
+    elif gender.lower() == "female":
+        pitch_scale = random.uniform(1.2, 1.4)
+    else:
+        raise ValueError("gender must be 'male' or 'female'")
+
+    # Apply pitch shift
+    f0 *= pitch_scale
+
+    # WORLD synthesis
+    y = pw.synthesize(f0, sp, ap, SAMPLE_RATE)
+
+    # Step 1: Dynamic pitch modulation
+    # y = dynamic_pitch_modulation(y, SAMPLE_RATE)
+
+    # Step 2: Spectral EQ (reduce lows, boost highs)
+    y = apply_eq(y, SAMPLE_RATE, gender)
+
+    # Step 3: Add breathiness
+    # y = add_breathiness(y, SAMPLE_RATE, level=random.uniform(0, 5e-4))
+
+    # Step 4: Speed up tempo (slightly)
+    y = librosa.effects.time_stretch(y, rate=random.uniform(1.1, 1.2))
+
+    # Normalize to avoid clipping
+    # y = y / np.max(np.abs(y) + 1e-6)
+
+    return y