SonicForgeStudio/app/core/ai_dsp_engine.py

import numpy as np
import os

class AIDSPEngine:
    @staticmethod
    def find_exact_zero_crossing(y: np.ndarray, sr: int, target_time: float, window_ms: float = 50.0) -> float:
        """
        Locates the absolute nearest physical zero-crossing sample index to target_time (seconds).
        Returns the optimized timeline index position in seconds where amplitude hits 0 (x[i] * x[i+1] <= 0).
        """
        if len(y) == 0 or sr <= 0:
            return float(target_time)
            
        target_sample = int(target_time * sr)
        window_samples = max(2, int((window_ms / 1000.0) * sr))
        
        # Symmetrical boundary window centered around target_sample
        start_idx = max(0, target_sample - window_samples // 2)
        end_idx = min(len(y) - 1, target_sample + window_samples // 2)
        
        if end_idx <= start_idx:
            return float(target_time)
            
        y_segment = y[start_idx:end_idx]
        
        if len(y_segment) < 2:
            return float(target_time)

        # Handle multi-channel (2D) by reducing to 1D mono amplitude for zero-crossing analysis
        if y_segment.ndim > 1:
            y_analysis = np.mean(y_segment, axis=0)
        else:
            y_analysis = y_segment

        # Physical zero-crossing condition: y[i] * y[i+1] <= 0
        zero_crossings = np.where(y_analysis[:-1] * y_analysis[1:] <= 0)[0]
        
        if len(zero_crossings) == 0:
            # Fallback: if no sign change occurs, locate absolute minimum amplitude sample
            abs_min_idx = int(np.argmin(np.abs(y_analysis)))
            return float((abs_min_idx + start_idx) / sr)
            
        # Translate local segment indices back to absolute buffer coordinates
        absolute_crossings = zero_crossings + start_idx
        
        # Isolate the zero-crossing closest to raw target_sample
        distances = np.abs(absolute_crossings - target_sample)
        best_sample_idx = int(absolute_crossings[np.argmin(distances)])
        
        return float(best_sample_idx / sr)

    @classmethod
    def scan_best_loop_regions(cls, y: np.ndarray, sr: int, min_duration: float = 2.0, max_duration: float = 8.0) -> list:
        """
        Evaluates spectral Self-Similarity Matrices (Recurrence plots) to extract 
        the most musically periodic and cohesive loop segments within the track.
        """
        if len(y) == 0 or sr <= 0:
            return [{"start_time": 0.0, "end_time": min(4.0, max_duration), "score": 0.5}]

        # Ensure 1D mono audio array for spectral feature extraction
        if y.ndim > 1:
            y_mono = np.mean(y, axis=0)
        else:
            y_mono = y

        total_duration = len(y_mono) / sr
        if total_duration <= min_duration:
            t_start = cls.find_exact_zero_crossing(y_mono, sr, 0.0)
            t_end = cls.find_exact_zero_crossing(y_mono, sr, total_duration)
            return [{"start_time": t_start, "end_time": t_end, "score": 1.0}]

        best_score = 0.5
        t_start = 0.0
        t_end = min(total_duration, 4.0)

        try:
            from app.core.audio_features import chroma_stft as _chroma_stft
            # 1. Compute harmonic structural properties via Chroma (STFT-based,
            #    thay chroma_cqt de lo bo librosa/numba/llvmlite ~171MB)
            chroma = _chroma_stft(y=y_mono, sr=sr)
            
            # 2. Compile Self-Similarity Matrix (Cosine Recurrence Plot)
            #    thay sklearn.metrics.pairwise.cosine_similarity bang numpy
            c = chroma.T  # (n_frames, 12)
            norms = np.linalg.norm(c, axis=1, keepdims=True)
            ssm = (c @ c.T) / (norms @ norms.T + 1e-9)
            
            num_frames = ssm.shape[0]
            hop_length = 512
            frame_duration = hop_length / sr
            
            min_frames = int(min_duration / frame_duration)
            max_frames = int(max_duration / frame_duration)
            
            best_score = -1.0
            best_lag = min_frames
            
            for lag in range(min_frames, min(num_frames, max_frames + 1)):
                score = float(np.mean(np.diagonal(ssm, offset=lag)))
                if score > best_score:
                    best_score = score
                    best_lag = lag
                    
            start_frame = 0
            end_frame = min(num_frames - 1, start_frame + best_lag)
            t_start = start_frame * frame_duration
            t_end = end_frame * frame_duration
            
        except Exception:
            # Fallback DSP loop calculation if librosa/sklearn optional dependencies encounter edge cases
            energy = y_mono ** 2
            window = int(0.1 * sr)
            if len(energy) > window:
                smoothed_energy = np.convolve(energy, np.ones(window)/window, mode='valid')
                peak_idx = int(np.argmax(smoothed_energy))
                t_start = peak_idx / sr
                t_end = min(total_duration, t_start + min(4.0, max_duration))

        # 3. Lock boundaries to precise physical zero-crossings to prevent transient click noise
        t_start_zero = cls.find_exact_zero_crossing(y_mono, sr, t_start)
        t_end_zero = cls.find_exact_zero_crossing(y_mono, sr, t_end)
        
        return [{"start_time": t_start_zero, "end_time": t_end_zero, "score": float(best_score)}]

    @classmethod
    def slice_and_copy_with_zero_crossing(
        cls, 
        y: np.ndarray, 
        sr: int, 
        start_time: float, 
        end_time: float
    ) -> tuple:
        """
        Slices an audio data array from start_time to end_time using zero-crossing alignment. 
        Strictly bypasses linear or exponential fade configurations.
        """
        t_start_zero = cls.find_exact_zero_crossing(y, sr, start_time)
        t_end_zero = cls.find_exact_zero_crossing(y, sr, end_time)
        
        sample_start = int(t_start_zero * sr)
        sample_end = int(t_end_zero * sr)
        
        if sample_end <= sample_start:
            sample_end = min(len(y), sample_start + 100)
            
        if y.ndim > 1:
            y_sliced = np.copy(y[:, sample_start:sample_end])
        else:
            y_sliced = np.copy(y[sample_start:sample_end])
            
        return y_sliced, t_start_zero, t_end_zero