SonicForgeStudio/app/core/dsp_utils.py

import numpy as np
from pydub import AudioSegment
from app.core.audio_features import load as _load

def find_zero_crossing(y: np.ndarray, sr: int, target_time: float, window_seconds: float = 0.04) -> float:
    """
    Tìm điểm zero-crossing gần nhất với mốc thời gian đích (giây) để tránh click/pop.
    
    Args:
        y: Mảng biên độ âm thanh (1D numpy array, Mono)
        sr: Tần số lấy mẫu (Sample Rate)
        target_time: Vị trí mong muốn cắt (giây)
        window_seconds: Cửa sổ quét (mặc định 40ms)
        
    Returns:
        float: Thời gian của điểm zero-crossing gần nhất (giây)
    """
    if target_time is None or target_time < 0:
        return target_time
        
    target_sample = int(target_time * sr)
    window_samples = int(window_seconds * sr)
    
    # Xác định giới hạn vùng quét an toàn
    start_idx = max(0, target_sample - window_samples)
    end_idx = min(len(y) - 2, target_sample + window_samples)
    
    if start_idx >= end_idx:
        return target_time
    
    # Lấy phân khúc sóng âm trong cửa sổ quét
    y_window = y[start_idx:end_idx]
    
    # Tìm các điểm đổi dấu: y[i] * y[i+1] <= 0
    # Sử dụng np.sign và np.diff để tìm điểm đổi dấu nhanh chóng
    signs = np.sign(y_window)
    # Bất kỳ vị trí nào diff != 0 nghĩa là có sự đổi dấu (đi qua điểm 0)
    zero_crossings = np.where(np.diff(signs) != 0)[0]
    
    if len(zero_crossings) == 0:
        return target_time  # Không tìm thấy, trả về vị trí gốc
        
    # Chuyển chỉ số vùng quét về chỉ số mảng tuyệt đối
    absolute_crossings = zero_crossings + start_idx
    
    # Tìm điểm gần với target_sample nhất
    distances = np.abs(absolute_crossings - target_sample)
    closest_sample_idx = absolute_crossings[np.argmin(distances)]
    
    # Trả về thời gian tương ứng (giây)
    return float(closest_sample_idx / sr)

def find_nearest_zero_crossing_file(file_path: str, target_time_sec: float, search_window_sec: float = 0.04) -> float:
    """
    Tìm điểm zero-crossing từ file âm thanh.
    Wrapper cho hàm find_zero_crossing để tương thích với code cũ.
    """
    try:
        # Load mono audio for zero crossing analysis
        y, sr = _load(file_path, sr=None, mono=True)
        return find_zero_crossing(y, sr, target_time_sec, search_window_sec)
    except Exception as e:
        print(f"Error finding zero crossing: {e}")
        return target_time_sec

def find_nearest_zero_crossing(y: np.ndarray, sr: int, target_time_sec: float, search_window_sec: float = 0.04) -> float:
    """
    Tương thích với code cũ. Gọi đến hàm find_zero_crossing mới.
    """
    return find_zero_crossing(y, sr, target_time_sec, search_window_sec)

def apply_micro_fade(segment: AudioSegment, fade_duration_ms: int = 50) -> AudioSegment:
    """
    Áp dụng micro-fades (fade-in và fade-out) để triệt tiêu click/pop.
    """
    if len(segment) > fade_duration_ms * 2:
        return segment.fade_in(fade_duration_ms).fade_out(fade_duration_ms)
    elif len(segment) > fade_duration_ms:
        return segment.fade_in(fade_duration_ms // 2).fade_out(fade_duration_ms // 2)
    return segment


def apply_micro_crossfade(original: np.ndarray, edited: np.ndarray, start_sample: int, fade_len_ms: int = 10, sr: int = 44100) -> np.ndarray:
    """
    Áp dụng bộ lọc mờ biên Micro-crossfade (10ms) tại hai đầu điểm ráp nối
    để triệt tiêu tiếng click/pop khi Apply & Merge Back (22_CLIENT_DESK.md §2.2).
    
    Output(t) = (1 - alpha(t)) * Original(t) + alpha(t) * Edited(t - T_start)
    """
    fade_samples = int((fade_len_ms / 1000.0) * sr)
    if fade_samples <= 0 or len(original) == 0:
        return edited

    output = np.copy(original)
    edited_len = len(edited)
    end_sample = min(len(original), start_sample + edited_len)
    actual_len = end_sample - start_sample

    if actual_len <= 0:
        return output

    fade_in_len = min(fade_samples, actual_len)
    fade_out_len = min(fade_samples, actual_len)

    alpha_in = np.linspace(0.0, 1.0, fade_in_len)
    alpha_out = np.linspace(1.0, 0.0, fade_out_len)

    output[start_sample:end_sample] = edited[:actual_len]

    # Fade in at start splice point
    for i in range(fade_in_len):
        idx = start_sample + i
        if idx < len(original):
            output[idx] = (1.0 - alpha_in[i]) * original[idx] + alpha_in[i] * edited[i]

    # Fade out at end splice point
    for i in range(fade_out_len):
        idx = end_sample - fade_out_len + i
        edit_idx = actual_len - fade_out_len + i
        if idx < len(original) and edit_idx < len(edited):
            output[idx] = alpha_out[i] * edited[edit_idx] + (1.0 - alpha_out[i]) * original[idx]

    return output


def generate_peak_waveform(file_path: str, num_peaks: int = 800) -> dict:
    """
    Tạo dữ liệu peak waveform cho hiển thị đồ thị sóng âm trên Frontend.
    Dùng để vẽ waveform đồng bộ với Client (thay thế Web Audio API decodeAudioData).
    
    Args:
        file_path: Đường dẫn tệp âm thanh
        num_peaks: Số lượng điểm peak trả về (tương ứng pixel width trên UI)
        
    Returns:
        dict: {"peaks": [...], "duration": float, "sample_rate": int}
    """
    # Load mono audio
    y, sr = _load(file_path, sr=None, mono=True)
    
    total_samples = len(y)
    duration = float(total_samples) / sr
    
    if total_samples == 0:
        return {
            "peaks": [],
            "duration": 0.0,
            "sample_rate": sr
        }
    
    # Tính kích thước mỗi chunk
    samples_per_peak = max(1, total_samples // num_peaks)
    
    peaks = []
    for i in range(0, total_samples, samples_per_peak):
        chunk = y[i:i + samples_per_peak]
        if len(chunk) > 0:
            # Peak = giá trị tuyệt đối lớn nhất trong chunk
            peak_val = float(np.max(np.abs(chunk)))
            peaks.append(round(peak_val, 6))
    
    # Giới hạn đúng số lượng peaks yêu cầu
    if len(peaks) > num_peaks:
        peaks = peaks[:num_peaks]
    
    return {
        "peaks": peaks,
        "duration": round(duration, 4),
        "sample_rate": sr
    }


def generate_rms_waveform(file_path: str, num_points: int = 800) -> dict:
    """
    Tạo dữ liệu RMS waveform (mượt hơn peak waveform).
    
    Args:
        file_path: Đường dẫn tệp âm thanh
        num_points: Số lượng điểm RMS trả về
        
    Returns:
        dict: {"rms": [...], "duration": float, "sample_rate": int}
    """
    y, sr = _load(file_path, sr=None, mono=True)
    
    total_samples = len(y)
    duration = float(total_samples) / sr
    
    if total_samples == 0:
        return {
            "rms": [],
            "duration": 0.0,
            "sample_rate": sr
        }
    
    samples_per_point = max(1, total_samples // num_points)
    
    rms_values = []
    for i in range(0, total_samples, samples_per_point):
        chunk = y[i:i + samples_per_point]
        if len(chunk) > 0:
            rms_val = float(np.sqrt(np.mean(chunk ** 2)))
            rms_values.append(round(rms_val, 6))
    
    if len(rms_values) > num_points:
        rms_values = rms_values[:num_points]
    
    return {
        "rms": rms_values,
        "duration": round(duration, 4),
        "sample_rate": sr
    }