{ "cells": [ { "cell_type": "code", "execution_count": null, "id": "0", "metadata": {}, "outputs": [], "source": [ "from suno_utils.utils.text import read_jsonl\n", "from suno_utils.audio import Audio\n", "import numpy as np\n", "import matplotlib.pyplot as plt\n", "import pandas as pd\n", "import torch\n", "from suno_utils.tasks.dac_vae_fixed_25hz import encode_overlap, preload_models, decode, SPLIT_PREDICT_ARRAY_LEN" ] }, { "cell_type": "code", "execution_count": null, "id": "1", "metadata": {}, "outputs": [], "source": [ "model_filepath = \"s3://suno-data/minz/models/dac_vae_tuned_25hz.pth\"\n", "_ = preload_models(\n", " checkpoint_filepath=model_filepath,\n", ")" ] }, { "cell_type": "code", "execution_count": null, "id": "2", "metadata": {}, "outputs": [], "source": [ "sfx_data = read_jsonl(\"/app2/suno/data/diffusion/sfx/v2/combined_v3_w_extreme_metas_v0.jsonl\")" ] }, { "cell_type": "code", "execution_count": null, "id": "3", "metadata": {}, "outputs": [], "source": [ "def plot_audio(audio_array, sample_rate=44100, title=\"Audio Waveform\"):\n", " \"\"\"\n", " Plot an audio float array as a waveform.\n", " \n", " Args:\n", " audio_array (np.ndarray): Audio data as float array\n", " sample_rate (int): Sample rate in Hz (default: 44100)\n", " title (str): Plot title\n", " \"\"\"\n", " # Create time axis\n", " duration = len(audio_array) / sample_rate\n", " time = np.linspace(0, duration, len(audio_array))\n", " \n", " # Create the plot\n", " plt.figure(figsize=(16, 2))\n", " plt.plot(time, audio_array, linewidth=0.5)\n", " plt.xlabel('Time (seconds)')\n", " plt.ylim((-0.2, 0.2))\n", " plt.ylabel('Amplitude')\n", " plt.title(title)\n", " plt.grid(True, alpha=0.3)\n", " plt.tight_layout()\n", " plt.show()" ] }, { "cell_type": "code", "execution_count": null, "id": "4", "metadata": {}, "outputs": [], "source": [ "df = pd.DataFrame(sfx_data)\n", "df = df[df[\"dataset\"] == \"extreme\"]\n", "df = df[df[\"is_reliable\"]]\n", "df = df[df[\"duration_s\"] > 10]\n", "df.head()" ] }, { "cell_type": "code", "execution_count": null, "id": "5", "metadata": {}, "outputs": [], "source": [ "audio = Audio.from_s3(df.iloc[3][\"s3_filepath\"])\n", "#audio = audio.get_segment(from_s=0, to_s=3.0)\n", "plot_audio(audio.array_float, sample_rate=audio.sample_rate)\n", "print(audio)\n", "audio.play()" ] }, { "cell_type": "code", "execution_count": null, "id": "6", "metadata": {}, "outputs": [], "source": [ "\n", "audio = Audio.from_s3(df.iloc[1][\"s3_filepath\"], sample_rate = 48_000)\n", "#audio = audio.get_segment(from_s=0, to_s=3.0)\n", "audio_data = audio\n", "plot_audio(audio_data.array_float[:int(48_000 * 0.1)], sample_rate=48_000)\n", "print(audio_data, audio_data.array_float.shape)\n", "for i in range(2):\n", " output = encode_overlap([audio_data], normalize_volume=False)[0]\n", " output = decode(output)\n", " plot_audio(output.array_float[0][:int(48_000 * 0.1)], sample_rate=48_000)\n", " audio_data = output\n", " print(audio_data, audio_data.array_float.shape)" ] }, { "cell_type": "code", "execution_count": null, "id": "7", "metadata": {}, "outputs": [], "source": [ "import numpy as np\n", "import librosa\n", "from scipy import signal\n", "from scipy.stats import pearsonr\n", "import matplotlib.pyplot as plt\n", "\n", "class DrumLoopPhaseDetector:\n", " def __init__(self, audio_file, beats, expected_loop_beats=4):\n", " \"\"\"\n", " Initialize with focus on drum loop phase detection\n", " Note: beats may be incorrectly timed if first transient is missing\n", " \n", " Args:\n", " audio_file: path to audio file\n", " beats: numpy array with [time, beat_number] pairs (potentially inaccurate)\n", " expected_loop_beats: beats per loop (usually 4 for drums)\n", " \"\"\"\n", " self.audio, self.sr = librosa.load(audio_file)\n", " self.beats = beats\n", " self.beat_times = beats[:, 0]\n", " self.expected_loop_beats = expected_loop_beats\n", " \n", " # Don't trust the beat intervals if first beat might be wrong\n", " # Use a more robust estimate\n", " self.beat_interval = self._estimate_robust_beat_interval()\n", " self.loop_duration = self.beat_interval * expected_loop_beats\n", " \n", " def _estimate_robust_beat_interval(self):\n", " \"\"\"\n", " Estimate beat interval without relying on potentially incorrect first beat\n", " \"\"\"\n", " intervals = np.diff(self.beat_times)\n", " \n", " # Use median of intervals (ignores outliers from wrong first beat)\n", " median_interval = np.median(intervals)\n", " return median_interval\n", "\n", " def _test_interval_quality(self, interval1, interval2):\n", " \"\"\"\n", " Test which interval produces better loop alignment\n", " \"\"\"\n", " scores = []\n", " for interval in [interval1, interval2]:\n", " loop_samples = int(interval * self.expected_loop_beats * self.sr)\n", " if loop_samples * 2 <= len(self.audio):\n", " loop1 = self.audio[:loop_samples]\n", " loop2 = self.audio[loop_samples:loop_samples*2]\n", " if len(loop1) == len(loop2):\n", " score = np.corrcoef(loop1, loop2)[0, 1]\n", " scores.append(score if not np.isnan(score) else 0)\n", " else:\n", " scores.append(0)\n", " else:\n", " scores.append(0)\n", " \n", " return interval1 if scores[0] > scores[1] else interval2\n", " \n", " def extract_loop_segments(self, num_loops=3):\n", " \"\"\"Extract multiple loop segments for comparison\"\"\"\n", " loop_samples = int(self.loop_duration * self.sr)\n", " segments = []\n", " \n", " for i in range(num_loops):\n", " start_sample = i * loop_samples\n", " end_sample = start_sample + loop_samples\n", " \n", " if end_sample <= len(self.audio):\n", " segment = self.audio[start_sample:end_sample]\n", " segments.append(segment)\n", " \n", " return segments\n", " \n", " def method1_envelope_correlation(self):\n", " \"\"\"\n", " Compare amplitude envelopes of loop segments to find optimal alignment\n", " \"\"\"\n", " segments = self.extract_loop_segments(3)\n", " if len(segments) < 2:\n", " return {'optimal_offset': 0, 'confidence': 0}\n", " \n", " # Calculate envelope for each segment\n", " envelopes = []\n", " for segment in segments:\n", " envelope = np.abs(signal.hilbert(segment))\n", " # Smooth the envelope\n", " envelope = signal.savgol_filter(envelope, window_length=101, polyorder=3)\n", " envelopes.append(envelope)\n", " \n", " # Test different phase offsets\n", " max_offset_samples = int(0.15 * self.sr) # Test up to 150ms\n", " best_offset = 0\n", " best_correlation = -1\n", " \n", " correlations = []\n", " offsets = range(0, max_offset_samples, int(0.001 * self.sr)) # 1ms steps\n", " \n", " for offset in offsets:\n", " if offset < len(envelopes[0]) and offset < len(envelopes[1]):\n", " # Compare shifted versions\n", " env1_shifted = envelopes[0][offset:]\n", " env2 = envelopes[1][:len(env1_shifted)]\n", " \n", " if len(env1_shifted) > 0 and len(env2) > 0:\n", " corr = np.corrcoef(env1_shifted, env2)[0, 1]\n", " if not np.isnan(corr):\n", " correlations.append(corr)\n", " if corr > best_correlation:\n", " best_correlation = corr\n", " best_offset = offset\n", " else:\n", " correlations.append(0)\n", " else:\n", " correlations.append(0)\n", " else:\n", " correlations.append(0)\n", " \n", " best_offset_seconds = best_offset / self.sr\n", " confidence = max(0, (best_correlation - 0.7) / 0.3) if best_correlation > 0.7 else 0\n", " \n", " return {\n", " 'optimal_offset': best_offset_seconds,\n", " 'best_correlation': best_correlation,\n", " 'confidence': confidence,\n", " 'correlations': correlations,\n", " 'offset_samples': list(offsets)\n", " }\n", " \n", " def method2_spectral_template_matching(self):\n", " \"\"\"\n", " Use spectral features and template matching\n", " \"\"\"\n", " # Extract spectrograms for each loop\n", " hop_length = 256\n", " segments = self.extract_loop_segments(4)\n", " if len(segments) < 2:\n", " return {'optimal_offset': 0, 'confidence': 0}\n", " \n", " spectrograms = []\n", " for segment in segments:\n", " stft = librosa.stft(segment, hop_length=hop_length)\n", " spectrogram = np.abs(stft)\n", " # Focus on percussive frequencies (reduce to key frequency bands)\n", " spectrogram = spectrogram[:spectrogram.shape[0]//2] # Lower half of spectrum\n", " spectrograms.append(spectrogram)\n", " \n", " # Compare spectrograms with different phase shifts\n", " max_offset_frames = int(0.15 * self.sr / hop_length)\n", " best_offset = 0\n", " best_correlation = -1\n", " \n", " for offset in range(0, max_offset_frames):\n", " if offset < spectrograms[0].shape[1]:\n", " # Shift first spectrogram\n", " spec1_shifted = spectrograms[0][:, offset:]\n", " spec2 = spectrograms[1]\n", " \n", " # Make same size for comparison\n", " min_frames = min(spec1_shifted.shape[1], spec2.shape[1])\n", " if min_frames > 10: # Need enough frames for meaningful comparison\n", " spec1_cut = spec1_shifted[:, :min_frames]\n", " spec2_cut = spec2[:, :min_frames]\n", " \n", " # Flatten and correlate\n", " corr = np.corrcoef(spec1_cut.flatten(), spec2_cut.flatten())[0, 1]\n", " \n", " if not np.isnan(corr) and corr > best_correlation:\n", " best_correlation = corr\n", " best_offset = offset\n", " \n", " best_offset_seconds = best_offset * hop_length / self.sr\n", " confidence = max(0, (best_correlation - 0.5) / 0.5) if best_correlation > 0.5 else 0\n", " \n", " return {\n", " 'optimal_offset': best_offset_seconds,\n", " 'best_correlation': best_correlation,\n", " 'confidence': confidence\n", " }\n", " \n", " def method3_onset_density_alignment(self):\n", " \"\"\"\n", " Look at onset density patterns to find proper alignment\n", " \"\"\"\n", " # Detect all onsets with high precision\n", " onset_frames = librosa.onset.onset_detect(\n", " y=self.audio,\n", " sr=self.sr,\n", " hop_length=128, # Higher precision\n", " delta=0.02,\n", " units='time'\n", " )\n", " \n", " # Create onset density function\n", " time_resolution = 0.005 # 5ms resolution\n", " max_time = min(self.loop_duration * 3, len(self.audio) / self.sr)\n", " time_bins = np.arange(0, max_time, time_resolution)\n", " onset_density = np.histogram(onset_frames[onset_frames < max_time], bins=time_bins)[0]\n", " \n", " # Extract density patterns for each loop\n", " loop_length_bins = int(self.loop_duration / time_resolution)\n", " patterns = []\n", " \n", " for i in range(3): # First 3 loops\n", " start_bin = i * loop_length_bins\n", " end_bin = start_bin + loop_length_bins\n", " if end_bin <= len(onset_density):\n", " pattern = onset_density[start_bin:end_bin]\n", " patterns.append(pattern)\n", " \n", " if len(patterns) < 2:\n", " return {'optimal_offset': 0, 'confidence': 0}\n", " \n", " # Find best alignment between patterns\n", " max_offset_bins = int(0.15 / time_resolution) # 150ms max offset\n", " best_offset = 0\n", " best_correlation = -1\n", " \n", " for offset in range(max_offset_bins):\n", " if offset < len(patterns[0]):\n", " pattern1_shifted = patterns[0][offset:]\n", " pattern2 = patterns[1][:len(pattern1_shifted)]\n", " \n", " if len(pattern1_shifted) > 10 and len(pattern2) > 10:\n", " corr = np.corrcoef(pattern1_shifted, pattern2)[0, 1]\n", " if not np.isnan(corr) and corr > best_correlation:\n", " best_correlation = corr\n", " best_offset = offset\n", " \n", " best_offset_seconds = best_offset * time_resolution\n", " confidence = max(0, (best_correlation - 0.3) / 0.7) if best_correlation > 0.3 else 0\n", " \n", " return {\n", " 'optimal_offset': best_offset_seconds,\n", " 'best_correlation': best_correlation,\n", " 'confidence': confidence,\n", " 'onset_patterns': patterns\n", " }\n", "\n", " def method4_beat_phase_alignment(self):\n", " \"\"\"\n", " Determine if beats are correctly aligned with musical structure\n", " Assumes beat timing intervals are correct, but phase may be wrong\n", " \"\"\"\n", " # Extract audio segments around each detected beat\n", " beat_window = int(0.1 * self.sr) # 100ms window around each beat\n", " beat_segments = []\n", " \n", " for beat_time in self.beat_times[:16]: # First 16 beats\n", " center_sample = int(beat_time * self.sr)\n", " start_sample = max(0, center_sample - beat_window//2)\n", " end_sample = min(len(self.audio), center_sample + beat_window//2)\n", " \n", " if end_sample - start_sample == beat_window:\n", " segment = self.audio[start_sample:end_sample]\n", " beat_segments.append(segment)\n", " \n", " if len(beat_segments) < 8:\n", " return {'optimal_phase_shift': 0, 'confidence': 0}\n", " \n", " # Test different phase shifts of the beat positions\n", " max_shift = int(0.15 * self.sr) # Test shifts up to 150ms\n", " shift_step = int(0.002 * self.sr) # 2ms steps\n", " \n", " best_shift = 0\n", " best_consistency = -1\n", " \n", " for shift in range(-max_shift, max_shift + 1, shift_step):\n", " consistency_scores = []\n", " \n", " # For each beat position, extract audio with the phase shift\n", " shifted_segments = []\n", " for beat_time in self.beat_times[:16]:\n", " shifted_center = int((beat_time * self.sr) + shift)\n", " start_sample = max(0, shifted_center - beat_window//2)\n", " end_sample = min(len(self.audio), shifted_center + beat_window//2)\n", " \n", " if end_sample - start_sample == beat_window:\n", " segment = self.audio[start_sample:end_sample]\n", " shifted_segments.append(segment)\n", " \n", " if len(shifted_segments) < 8:\n", " continue\n", " \n", " # Measure consistency of beat patterns\n", " # Group by beat position in measure (1, 2, 3, 4)\n", " beat_groups = [[], [], [], []]\n", " for i, segment in enumerate(shifted_segments):\n", " beat_in_measure = i % 4\n", " beat_groups[beat_in_measure].append(segment)\n", " \n", " # Calculate consistency within each beat position\n", " for group in beat_groups:\n", " if len(group) >= 2:\n", " # Compare segments at same beat position\n", " correlations = []\n", " for i in range(len(group)):\n", " for j in range(i+1, len(group)):\n", " corr = np.corrcoef(group[i], group[j])[0, 1]\n", " if not np.isnan(corr):\n", " correlations.append(corr)\n", " \n", " if correlations:\n", " consistency_scores.append(np.mean(correlations))\n", " \n", " # Overall consistency is mean of all beat position consistencies\n", " if consistency_scores:\n", " overall_consistency = np.mean(consistency_scores)\n", " if overall_consistency > best_consistency:\n", " best_consistency = overall_consistency\n", " best_shift = shift\n", " \n", " best_shift_seconds = best_shift / self.sr\n", " confidence = max(0, (best_consistency - 0.3) / 0.7) if best_consistency > 0.3 else 0\n", " \n", " return {\n", " 'optimal_phase_shift': best_shift_seconds,\n", " 'best_consistency': best_consistency,\n", " 'confidence': confidence,\n", " 'beat_segments': beat_segments[:4] # Keep first 4 for visualization\n", " }\n", "\n", " def method5_measure_boundary_detection(self):\n", " \"\"\"\n", " Find where measure boundaries should actually be by analyzing the loop structure\n", " Uses the fact that drum loops typically have strong patterns at measure boundaries\n", " \"\"\"\n", " # Look for the strongest repeating pattern (measure length)\n", " measure_samples = int(self.loop_duration * self.sr)\n", " \n", " # Test different starting points for the measure\n", " max_offset_samples = int(0.2 * self.sr) # 200ms max\n", " offset_step = int(0.005 * self.sr) # 5ms steps\n", " \n", " best_offset = 0\n", " best_correlation = -1\n", " correlations = []\n", " \n", " for offset in range(0, max_offset_samples, offset_step):\n", " # Extract measures with this offset\n", " measure1_start = offset\n", " measure1_end = offset + measure_samples\n", " measure2_start = offset + measure_samples\n", " measure2_end = offset + 2 * measure_samples\n", " \n", " if measure2_end <= len(self.audio):\n", " measure1 = self.audio[measure1_start:measure1_end]\n", " measure2 = self.audio[measure2_start:measure2_end]\n", " \n", " # Calculate correlation between measures\n", " corr = np.corrcoef(measure1, measure2)[0, 1]\n", " if not np.isnan(corr):\n", " correlations.append(corr)\n", " if corr > best_correlation:\n", " best_correlation = corr\n", " best_offset = offset\n", " else:\n", " correlations.append(0)\n", " else:\n", " correlations.append(0)\n", " \n", " best_offset_seconds = best_offset / self.sr\n", " \n", " # Calculate where the first beat should be relative to this measure boundary\n", " # The first beat should be at the measure boundary\n", " current_first_beat_time = self.beat_times[0]\n", " optimal_first_beat_time = best_offset_seconds\n", " \n", " beat_phase_correction = optimal_first_beat_time - current_first_beat_time\n", " \n", " confidence = max(0, (best_correlation - 0.5) / 0.5) if best_correlation > 0.5 else 0\n", " \n", " return {\n", " 'measure_boundary_offset': best_offset_seconds,\n", " 'beat_phase_correction': beat_phase_correction,\n", " 'measure_correlation': best_correlation,\n", " 'confidence': confidence,\n", " 'correlation_curve': correlations\n", " }\n", "\n", " def detect_phase_offset(self):\n", " \"\"\"\n", " Detect phase offset assuming beat intervals are correct but phase may be wrong\n", " \"\"\"\n", " results = {\n", " 'envelope': self.method1_envelope_correlation(),\n", " 'spectral': self.method2_spectral_template_matching(), \n", " 'onset_density': self.method3_onset_density_alignment(),\n", " 'beat_phase': self.method4_beat_phase_alignment(),\n", " 'measure_boundary': self.method5_measure_boundary_detection()\n", " }\n", " \n", " # Prioritize methods that work with correct beat intervals but wrong phase\n", " phase_methods = ['beat_phase', 'measure_boundary']\n", " pattern_methods = ['envelope', 'spectral', 'onset_density']\n", " \n", " # Get estimates from phase-specific methods\n", " phase_estimates = []\n", " phase_confidences = []\n", " \n", " for method in phase_methods:\n", " result = results[method]\n", " confidence = result['confidence']\n", " \n", " if method == 'beat_phase':\n", " offset = result['optimal_phase_shift']\n", " elif method == 'measure_boundary':\n", " offset = result['beat_phase_correction']\n", " \n", " if confidence > 0.2:\n", " phase_estimates.append(offset * confidence)\n", " phase_confidences.append(confidence)\n", " \n", " # Get estimates from pattern methods\n", " pattern_estimates = []\n", " pattern_confidences = []\n", " \n", " for method in pattern_methods:\n", " result = results[method]\n", " confidence = result['confidence']\n", " \n", " if confidence > 0.2:\n", " offset = result['optimal_offset']\n", " pattern_estimates.append(offset * confidence)\n", " pattern_confidences.append(confidence)\n", " \n", " # Combine estimates - prefer phase methods since they're designed for this problem\n", " if phase_confidences:\n", " phase_weighted_avg = sum(phase_estimates) / sum(phase_confidences)\n", " phase_confidence = np.mean(phase_confidences)\n", " else:\n", " phase_weighted_avg = 0\n", " phase_confidence = 0\n", " \n", " if pattern_confidences:\n", " pattern_weighted_avg = sum(pattern_estimates) / sum(pattern_confidences)\n", " pattern_confidence = np.mean(pattern_confidences)\n", " else:\n", " pattern_weighted_avg = 0\n", " pattern_confidence = 0\n", " \n", " # Choose best estimate\n", " if phase_confidence > pattern_confidence:\n", " estimated_offset = phase_weighted_avg\n", " overall_confidence = phase_confidence\n", " primary_method = 'beat_phase_alignment'\n", " else:\n", " estimated_offset = pattern_weighted_avg\n", " overall_confidence = pattern_confidence\n", " primary_method = 'pattern_matching'\n", " \n", " # If both methods agree reasonably well, increase confidence\n", " if (phase_confidence > 0.3 and pattern_confidence > 0.3 and \n", " abs(phase_weighted_avg - pattern_weighted_avg) < 0.02):\n", " overall_confidence = min(1.0, overall_confidence * 1.3)\n", " estimated_offset = (phase_weighted_avg + pattern_weighted_avg) / 2\n", " \n", " # Detection threshold\n", " is_offset_detected = abs(estimated_offset) > 0.012 and overall_confidence > 0.25 # 12ms threshold\n", " \n", " return {\n", " 'is_phase_offset_detected': is_offset_detected,\n", " 'estimated_offset_seconds': estimated_offset,\n", " 'confidence': overall_confidence,\n", " 'primary_detection_method': primary_method,\n", " #'detailed_results': results,\n", " #'phase_method_estimate': phase_weighted_avg,\n", " #'pattern_method_estimate': pattern_weighted_avg,\n", " 'methods_agreement': abs(phase_weighted_avg - pattern_weighted_avg) < 0.02\n", " }\n", " \n", " def visualize_detection(self):\n", " \"\"\"\n", " Create visualization to help verify detection and diagnose beat timing issues\n", " \"\"\"\n", " result = self.detect_phase_offset()\n", " \n", " fig, axes = plt.subplots(4, 1, figsize=(15, 12))\n", " \n", " # Plot 1: Raw audio with provided beats vs detected onsets\n", " time_axis = np.linspace(0, len(self.audio)/self.sr, len(self.audio))\n", " axes[0].plot(time_axis, self.audio, alpha=0.7, color='blue', label='Audio')\n", " \n", " # Show provided beat positions\n", " for i, beat_time in enumerate(self.beat_times[:12]): # First 12 beats\n", " axes[0].axvline(beat_time, color='red', alpha=0.5, linestyle='--', \n", " label='Provided Beats' if i == 0 else '')\n", " \n", " axes[0].set_title('Audio with Beat Timing Comparison')\n", " axes[0].set_xlim(0, min(6, len(self.audio)/self.sr)) # First 6 seconds\n", " axes[0].legend()\n", " \n", " # Plot 2: Loop alignment comparison\n", " loop_samples = int(self.loop_duration * self.sr)\n", " time_axis_loop = np.linspace(0, self.loop_duration, loop_samples)\n", " \n", " # Original alignment\n", " colors = ['blue', 'orange', 'green']\n", " for i in range(3):\n", " start_sample = i * loop_samples\n", " end_sample = start_sample + loop_samples\n", " if end_sample <= len(self.audio):\n", " loop_audio = self.audio[start_sample:end_sample]\n", " axes[1].plot(time_axis_loop, loop_audio, alpha=0.6, \n", " color=colors[i], label=f'Loop {i+1} (Original)')\n", " axes[1].set_title('Original Loop Alignment')\n", " axes[1].legend()\n", " \n", " # Plot 3: Corrected alignment (if offset detected)\n", " if result['is_phase_offset_detected']:\n", " offset_samples = int(result['estimated_offset_seconds'] * self.sr)\n", " for i in range(3):\n", " start_sample = i * loop_samples + offset_samples\n", " end_sample = start_sample + loop_samples\n", " if end_sample <= len(self.audio):\n", " loop_audio = self.audio[start_sample:end_sample]\n", " axes[2].plot(time_axis_loop, loop_audio, alpha=0.6,\n", " color=colors[i], label=f'Loop {i+1} (Corrected)')\n", " axes[2].set_title(f'Corrected Alignment (offset: {result[\"estimated_offset_seconds\"]:.3f}s)')\n", " axes[2].legend()\n", " else:\n", " axes[2].text(0.5, 0.5, 'No offset detected', transform=axes[2].transAxes, \n", " ha='center', va='center', fontsize=14)\n", " axes[2].set_title('No Correction Applied')\n", " \n", " # Plot 4: Diagnostic information\n", " axes[3].axis('off')\n", " \n", " # Create diagnostic text\n", " diag_text = f\"\"\"DIAGNOSTIC INFORMATION:\n", " \n", "Phase Offset Detected: {result['is_phase_offset_detected']}\n", "Estimated Offset: {result['estimated_offset_seconds']:.3f}s ({result['estimated_offset_seconds']*1000:.1f}ms)\n", "Confidence: {result['confidence']:.3f}\n", "Primary Method: {result['primary_detection_method']}\n", "\n", "Method Confidence Scores:\"\"\"\n", " \n", " for method, details in result['detailed_results'].items():\n", " confidence = details.get('confidence', 0)\n", " diag_text += f\"\\n- {method}: {confidence:.3f}\"\n", " \n", " axes[3].text(0.02, 0.98, diag_text, transform=axes[3].transAxes, \n", " fontsize=10, verticalalignment='top', fontfamily='monospace')\n", " \n", " plt.tight_layout()\n", " return fig\n", "\n", "\n", "def detect_drum_loop_offset(audio_file, beats):\n", " \"\"\"\n", " Detect phase offset in drum loop\n", " \n", " Returns:\n", " tuple: (result_dict, detector_object)\n", " \"\"\"\n", " detector = DrumLoopPhaseDetector(audio_file, beats)\n", " result = detector.detect_phase_offset()\n", " return result, detector" ] }, { "cell_type": "code", "execution_count": null, "id": "8", "metadata": {}, "outputs": [], "source": [ "import suno_utils.tasks.audio_features.beat_this_downbeat\n", "from suno_utils.audio import Audio\n", "import numpy as np\n", "import matplotlib.pyplot as plt\n", "\n", "import logging\n", "logging.basicConfig(level=logging.INFO)\n", "\n", "extractor = suno_utils.tasks.audio_features.beat_this_downbeat.BeatThisDownbeatExtractor(device=\"cuda\", model_path=\"s3://suno-data/m4burns/beat_this_rc_12l.pt\")" ] }, { "cell_type": "code", "execution_count": null, "id": "9", "metadata": {}, "outputs": [], "source": [ "fp = \"/home/sara/sara/Untitled (11).mp3\"\n", "audio = Audio.from_file(fp)\n", "audio_mono = Audio.convert(audio, n_channels=1, sample_rate=audio.sample_rate, byte_width=audio.byte_width)\n", "out = extractor.extract(audio_mono)\n", "beats_refined = np.array(out[\"downbeats\"])\n", "result, detector = detect_drum_loop_offset(fp, beats_refined)\n", "print(result.keys())\n", "print(result)" ] }, { "cell_type": "code", "execution_count": null, "id": "10", "metadata": {}, "outputs": [], "source": [ "all = []\n", "for idx in range(100):\n", " fp = df.iloc[idx][\"s3_filepath\"]\n", " audio = Audio.from_s3(fp)\n", " audio.to_wav(\"temp.wav\")\n", " audio_mono = Audio.convert(audio, n_channels=1, sample_rate=audio.sample_rate, byte_width=audio.byte_width)\n", " out = extractor.extract(audio_mono)\n", " beats_refined = np.array(out[\"downbeats\"])\n", " result, detector = detect_drum_loop_offset(\"temp.wav\", beats_refined)\n", " all.append(result)" ] }, { "cell_type": "code", "execution_count": null, "id": "11", "metadata": {}, "outputs": [], "source": [ "for a in all:\n", " if a['is_phase_offset_detected'] and a['estimated_offset_seconds'] < 0.05:\n", " print(a)" ] }, { "cell_type": "code", "execution_count": null, "id": "12", "metadata": {}, "outputs": [], "source": [ "fp = \"/home/sara/sara/Untitled (6).wav\"\n", "audio = Audio.from_file(fp)\n", "audio_mono = Audio.convert(audio, n_channels=1, sample_rate=audio.sample_rate, byte_width=audio.byte_width)\n", "out = extractor.extract(audio_mono)\n", "beats_refined = np.array(out[\"downbeats\"])[:,0]" ] }, { "cell_type": "code", "execution_count": null, "id": "13", "metadata": {}, "outputs": [], "source": [ "import numpy as np\n", "import matplotlib.pyplot as plt\n", "from scipy.io import wavfile\n", "import librosa\n", "\n", "def plot_beats_on_audio(audio_data, beat_times, sample_rate=22050, title=\"Audio Waveform with Beat Times\"):\n", " \"\"\"\n", " Plot beat times as vertical lines overlaid on audio waveform\n", " \n", " Parameters:\n", " audio_data: numpy array of audio samples\n", " beat_times: numpy array of beat times in seconds\n", " sample_rate: sample rate of audio (default 22050 Hz)\n", " title: plot title\n", " \"\"\"\n", " \n", " # Create time axis for audio\n", " duration = len(audio_data) / sample_rate\n", " time_axis = np.linspace(0, duration, len(audio_data))\n", " \n", " # Create the plot\n", " plt.figure(figsize=(15, 6))\n", " \n", " # Plot the audio waveform\n", " plt.plot(time_axis, audio_data, color='lightblue', alpha=0.7, linewidth=0.5, label='Audio')\n", " \n", " # Plot beat times as vertical lines\n", " for beat_time in beat_times:\n", " if beat_time <= duration: # Only plot beats within audio duration\n", " plt.axvline(x=beat_time, color='red', linestyle='-', alpha=0.8, linewidth=1.5)\n", " \n", " # Customize the plot\n", " plt.xlabel('Time (seconds)')\n", " plt.ylabel('Amplitude')\n", " plt.title(title)\n", " plt.grid(True, alpha=0.3)\n", " plt.legend(['Audio Waveform', 'Beat Times'])\n", " \n", " # Set reasonable limits\n", " plt.xlim(0, duration)\n", " \n", " plt.tight_layout()\n", " plt.show()\n", "\n", "def plot_beats_on_audio_zoomed(audio_data, beat_times, sample_rate=48000, \n", " start_time=0, end_time=10, title=\"Audio Waveform with Beat Times (Zoomed)\"):\n", " \"\"\"\n", " Plot a zoomed section of audio with beat times\n", " \"\"\"\n", " # Calculate sample indices for zoom window\n", " start_sample = int(start_time * sample_rate)\n", " end_sample = int(end_time * sample_rate)\n", " \n", " # Extract audio segment\n", " audio_segment = audio_data[start_sample:end_sample]\n", " time_axis = np.linspace(start_time, end_time, len(audio_segment))\n", " \n", " # Filter beat times to zoom window\n", " beats_in_window = beat_times[(beat_times >= start_time) & (beat_times <= end_time)]\n", " \n", " plt.figure(figsize=(15, 6))\n", " \n", " # Plot audio segment\n", " plt.plot(time_axis, audio_segment, color='lightblue', alpha=0.7, linewidth=0.8, label='Audio')\n", " \n", " # Plot beats in window\n", " for beat_time in beats_in_window:\n", " plt.axvline(x=beat_time, color='red', linestyle='-', alpha=0.8, linewidth=2)\n", " \n", " plt.xlabel('Time (seconds)')\n", " plt.ylabel('Amplitude')\n", " plt.title(title)\n", " plt.grid(True, alpha=0.3)\n", " plt.legend(['Audio Waveform', 'Beat Times'])\n", " plt.xlim(start_time, end_time)\n", " \n", " plt.tight_layout()\n", " plt.show()\n", "\n", "def plot_beats_with_spectrogram(audio_data, beat_times, sample_rate=22050):\n", " \"\"\"\n", " Create a subplot with both waveform and spectrogram, with beats overlaid on both\n", " \"\"\"\n", " fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(15, 10), sharex=True)\n", " \n", " # Time axis\n", " duration = len(audio_data) / sample_rate\n", " time_axis = np.linspace(0, duration, len(audio_data))\n", " \n", " # Top plot: Waveform with beats\n", " ax1.plot(time_axis, audio_data, color='lightblue', alpha=0.7, linewidth=0.5)\n", " for beat_time in beat_times:\n", " if beat_time <= duration:\n", " ax1.axvline(x=beat_time, color='red', linestyle='-', alpha=0.8, linewidth=1.5)\n", " ax1.set_ylabel('Amplitude')\n", " ax1.set_title('Audio Waveform with Beat Times')\n", " ax1.grid(True, alpha=0.3)\n", " \n", " # Bottom plot: Spectrogram with beats\n", " # Compute spectrogram\n", " f, t, Sxx = plt.specgram(audio_data, Fs=sample_rate, NFFT=1024, noverlap=512, \n", " cmap='viridis', ax=ax2)\n", " \n", " # Overlay beats on spectrogram\n", " for beat_time in beat_times:\n", " if beat_time <= duration:\n", " ax2.axvline(x=beat_time, color='red', linestyle='-', alpha=0.8, linewidth=1.5)\n", " \n", " ax2.set_xlabel('Time (seconds)')\n", " ax2.set_ylabel('Frequency (Hz)')\n", " ax2.set_title('Spectrogram with Beat Times')\n", " \n", " plt.tight_layout()\n", " plt.show()\n", "\n", "# Example usage:\n", "# Assuming you have your data loaded as:\n", "# audio_data = your_numpy_audio_array\n", "# beat_times = your_numpy_beat_times_array\n", "# sample_rate = your_sample_rate (e.g., 22050, 44100)\n", "\n", "# Basic plot\n", "# plot_beats_on_audio(audio_data, beat_times, sample_rate)\n", "\n", "# Zoomed plot (first 10 seconds)\n", "# plot_beats_on_audio_zoomed(audio_data, beat_times, sample_rate, 0, 10)\n", "\n", "# Combined waveform + spectrogram\n", "# plot_beats_with_spectrogram(audio_data, beat_times, sample_rate)\n", "\n", "# If you need to load audio from file:\n", "def load_audio_file(filename):\n", " \"\"\"\n", " Load audio file and return audio data and sample rate\n", " Supports various formats via librosa\n", " \"\"\"\n", " try:\n", " # Using librosa (handles many formats)\n", " audio_data, sample_rate = librosa.load(filename, sr=None)\n", " return audio_data, sample_rate\n", " except:\n", " try:\n", " # Fallback to scipy for WAV files\n", " sample_rate, audio_data = wavfile.read(filename)\n", " # Convert to float and normalize if needed\n", " if audio_data.dtype == np.int16:\n", " audio_data = audio_data.astype(np.float32) / 32768.0\n", " elif audio_data.dtype == np.int32:\n", " audio_data = audio_data.astype(np.float32) / 2147483648.0\n", " return audio_data, sample_rate\n", " except Exception as e:\n", " print(f\"Error loading audio file: {e}\")\n", " return None, None\n", "\n", "def plot_beat_interval_distribution(beat_times, title=\"Distribution of Beat Intervals\"):\n", " \"\"\"\n", " Plot distribution of time differences between consecutive beats\n", " \n", " Parameters:\n", " beat_times: numpy array of beat times in seconds\n", " title: plot title\n", " \"\"\"\n", " # Calculate time differences between consecutive beats\n", " beat_intervals = np.diff(beat_times)\n", " \n", " # Create figure with subplots\n", " fig, (ax1, ax2, ax3) = plt.subplots(1, 3, figsize=(18, 6))\n", " \n", " # 1. Histogram of beat intervals\n", " ax1.hist(beat_intervals, bins=50, alpha=0.7, color='skyblue', edgecolor='black')\n", " ax1.set_xlabel('Beat Interval (seconds)')\n", " ax1.set_ylabel('Frequency')\n", " ax1.set_title('Histogram of Beat Intervals')\n", " ax1.grid(True, alpha=0.3)\n", " \n", " # Add statistics to the plot\n", " mean_interval = np.mean(beat_intervals)\n", " std_interval = np.std(beat_intervals)\n", " ax1.axvline(mean_interval, color='red', linestyle='--', linewidth=2, \n", " label=f'Mean: {mean_interval:.3f}s')\n", " ax1.axvline(mean_interval + std_interval, color='orange', linestyle='--', alpha=0.7,\n", " label=f'+1 STD: {mean_interval + std_interval:.3f}s')\n", " ax1.axvline(mean_interval - std_interval, color='orange', linestyle='--', alpha=0.7,\n", " label=f'-1 STD: {mean_interval - std_interval:.3f}s')\n", " ax1.legend()\n", " \n", " # 2. Box plot of beat intervals\n", " ax2.boxplot(beat_intervals, vert=True, patch_artist=True, \n", " boxprops=dict(facecolor='lightblue', alpha=0.7))\n", " ax2.set_ylabel('Beat Interval (seconds)')\n", " ax2.set_title('Box Plot of Beat Intervals')\n", " ax2.grid(True, alpha=0.3)\n", " \n", " # 3. Beat intervals over time (line plot)\n", " ax3.plot(beat_times[1:], beat_intervals, 'o-', alpha=0.7, markersize=3, linewidth=1)\n", " ax3.set_xlabel('Time (seconds)')\n", " ax3.set_ylabel('Beat Interval (seconds)')\n", " ax3.set_title('Beat Intervals Over Time')\n", " ax3.grid(True, alpha=0.3)\n", " \n", " plt.tight_layout()\n", " plt.show()\n", " \n", " # Print summary statistics\n", " print(f\"\\n=== Beat Interval Statistics ===\")\n", " print(f\"Number of beats: {len(beat_times)}\")\n", " print(f\"Number of intervals: {len(beat_intervals)}\")\n", " print(f\"Mean interval: {mean_interval:.4f} seconds ({60/mean_interval:.1f} BPM)\")\n", " print(f\"Median interval: {np.median(beat_intervals):.4f} seconds ({60/np.median(beat_intervals):.1f} BPM)\")\n", " print(f\"Standard deviation: {std_interval:.4f} seconds\")\n", " print(f\"Min interval: {np.min(beat_intervals):.4f} seconds ({60/np.min(beat_intervals):.1f} BPM)\")\n", " print(f\"Max interval: {np.max(beat_intervals):.4f} seconds ({60/np.max(beat_intervals):.1f} BPM)\")\n", " print(f\"25th percentile: {np.percentile(beat_intervals, 25):.4f} seconds\")\n", " print(f\"75th percentile: {np.percentile(beat_intervals, 75):.4f} seconds\")\n", "\n", "def plot_tempo_analysis(beat_times, window_size=10, title=\"Tempo Analysis\"):\n", " \"\"\"\n", " Analyze tempo variations using a sliding window approach\n", " \n", " Parameters:\n", " beat_times: numpy array of beat times in seconds\n", " window_size: number of beats to include in each tempo calculation\n", " title: plot title\n", " \"\"\"\n", " if len(beat_times) < window_size + 1:\n", " print(f\"Not enough beats for tempo analysis. Need at least {window_size + 1} beats.\")\n", " return\n", " \n", " # Calculate instantaneous BPM using sliding window\n", " bpm_values = []\n", " time_points = []\n", " \n", " for i in range(len(beat_times) - window_size):\n", " # Calculate average interval over window\n", " window_beats = beat_times[i:i + window_size + 1]\n", " avg_interval = np.mean(np.diff(window_beats))\n", " bpm = 60.0 / avg_interval\n", " bpm_values.append(bpm)\n", " time_points.append(beat_times[i + window_size // 2]) # Center of window\n", " \n", " bpm_values = np.array(bpm_values)\n", " time_points = np.array(time_points)\n", " \n", " # Create plots\n", " fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(15, 8), sharex=True)\n", " \n", " # Plot BPM over time\n", " ax1.plot(time_points, bpm_values, 'b-', alpha=0.7, linewidth=1.5)\n", " ax1.set_ylabel('BPM')\n", " ax1.set_title(f'Tempo Variation Over Time (Window Size: {window_size} beats)')\n", " ax1.grid(True, alpha=0.3)\n", " \n", " # Add horizontal lines for mean and std\n", " mean_bpm = np.mean(bpm_values)\n", " std_bpm = np.std(bpm_values)\n", " ax1.axhline(mean_bpm, color='red', linestyle='--', alpha=0.8, \n", " label=f'Mean: {mean_bpm:.1f} BPM')\n", " ax1.axhline(mean_bpm + std_bpm, color='orange', linestyle='--', alpha=0.6)\n", " ax1.axhline(mean_bpm - std_bpm, color='orange', linestyle='--', alpha=0.6)\n", " ax1.legend()\n", " \n", " # Plot BPM distribution\n", " ax2.hist(bpm_values, bins=30, alpha=0.7, color='lightgreen', edgecolor='black')\n", " ax2.axvline(mean_bpm, color='red', linestyle='--', linewidth=2, \n", " label=f'Mean: {mean_bpm:.1f} BPM')\n", " ax2.set_xlabel('BPM')\n", " ax2.set_ylabel('Frequency')\n", " ax2.set_title('BPM Distribution')\n", " ax2.grid(True, alpha=0.3)\n", " ax2.legend()\n", " \n", " plt.tight_layout()\n", " plt.show()\n", " \n", " print(f\"\\n=== Tempo Analysis (Window Size: {window_size}) ===\")\n", " print(f\"Mean BPM: {mean_bpm:.2f}\")\n", " print(f\"BPM Standard Deviation: {std_bpm:.2f}\")\n", " print(f\"BPM Range: {np.min(bpm_values):.1f} - {np.max(bpm_values):.1f}\")\n", " print(f\"Tempo Stability: {((std_bpm/mean_bpm)*100):.1f}% coefficient of variation\")\n", "\n", "# Example usage:\n", "# Assuming you have your beat times loaded:\n", "# beat_times = your_numpy_beat_times_array\n", "\n", "# Plot beat interval distribution\n", "# plot_beat_interval_distribution(beat_times)\n", "\n", "# Plot tempo analysis with different window sizes\n", "# plot_tempo_analysis(beat_times, window_size=10) # Analysis over 10 beats\n", "# plot_tempo_analysis(beat_times, window_size=20) # Analysis over 20 beats\n", "\n", "# Combined analysis - plot everything\n", "def analyze_beats_comprehensive(audio_data, beat_times, sample_rate=22050):\n", " \"\"\"\n", " Comprehensive beat analysis including waveform, intervals, and tempo\n", " \"\"\"\n", " print(\"=== Comprehensive Beat Analysis ===\")\n", " \n", " # 1. Plot audio with beats\n", " plot_beats_on_audio(audio_data, beat_times, sample_rate, \n", " \"Audio Waveform with Detected Beats\")\n", " \n", " # 2. Analyze beat intervals\n", " plot_beat_interval_distribution(beat_times, \n", " \"Beat Interval Distribution Analysis\")\n", " \n", " # 3. Tempo analysis\n", " plot_tempo_analysis(beat_times, window_size=10, \n", " title=\"Tempo Variation Analysis\")\n", "\n", "# Example with file loading:\n", "# audio_data, sample_rate = load_audio_file('your_audio_file.wav')\n", "# if audio_data is not None:\n", "# analyze_beats_comprehensive(audio_data, beat_times, sample_rate)" ] }, { "cell_type": "code", "execution_count": null, "id": "14", "metadata": {}, "outputs": [], "source": [ "plot_beats_on_audio_zoomed(audio.array_float, beats_refined, end_time=4)" ] }, { "cell_type": "code", "execution_count": null, "id": "15", "metadata": {}, "outputs": [], "source": [ "plot_beat_interval_distribution(beats_refined)" ] }, { "cell_type": "code", "execution_count": null, "id": "16", "metadata": {}, "outputs": [], "source": [ "diffs = []\n", "\n", "for i in range(1, len(beats_refined)):\n", " diff = beats_refined[i] - beats_refined[i - 1]\n", " diffs.append(diff)" ] }, { "cell_type": "code", "execution_count": null, "id": "17", "metadata": {}, "outputs": [], "source": [ "import numpy as np\n", "\n", "def detect_first_beat_cutoff(beat_times, threshold_std=2.0, min_beats=5):\n", " \"\"\"\n", " Detect if the first beat interval suggests the beginning of the audio is cut off\n", " \n", " Parameters:\n", " beat_times: numpy array of beat times in seconds\n", " threshold_std: number of standard deviations from mean to consider \"far\"\n", " min_beats: minimum number of beats required for analysis\n", " \n", " Returns:\n", " dict with detection results and statistics\n", " \"\"\"\n", " if len(beat_times) < min_beats:\n", " return {\n", " 'cutoff_detected': False,\n", " 'reason': f'Insufficient beats for analysis (need >= {min_beats})',\n", " 'first_interval': None,\n", " 'mean_interval': None,\n", " 'std_interval': None,\n", " 'z_score': None,\n", " 'threshold_used': threshold_std\n", " }\n", " \n", " # Calculate beat intervals\n", " beat_intervals = np.diff(beat_times)\n", " first_interval = beat_intervals[0]\n", " \n", " # Use intervals 2-end to calculate \"normal\" rhythm (excluding first)\n", " normal_intervals = beat_intervals[1:] # Skip first interval\n", " mean_normal = np.mean(normal_intervals)\n", " std_normal = np.std(normal_intervals)\n", " \n", " # Calculate z-score of first interval relative to \"normal\" rhythm\n", " if std_normal == 0:\n", " z_score = 0 # All intervals are identical\n", " else:\n", " z_score = abs(first_interval - mean_normal) / std_normal\n", " \n", " # Detect cutoff\n", " cutoff_detected = z_score > threshold_std\n", " \n", " # Determine if it's shorter (cut off) or longer (late start)\n", " if cutoff_detected:\n", " if first_interval < mean_normal:\n", " cutoff_type = \"First beat interval too short - likely audio cut-off\"\n", " else:\n", " cutoff_type = \"First beat interval too long - possible late start\"\n", " else:\n", " cutoff_type = \"Normal first beat timing\"\n", " \n", " return {\n", " 'cutoff_detected': cutoff_detected,\n", " 'cutoff_type': cutoff_type,\n", " 'first_interval': first_interval,\n", " 'mean_normal_interval': mean_normal,\n", " 'std_normal_interval': std_normal,\n", " 'z_score': z_score,\n", " 'threshold_used': threshold_std,\n", " 'confidence': min(z_score / threshold_std, 2.0) if cutoff_detected else 0.0,\n", " 'first_bpm': 60.0 / first_interval,\n", " 'normal_bpm': 60.0 / mean_normal\n", " }\n", "\n", "def print_cutoff_analysis(beat_times, threshold_std=2.0):\n", " \"\"\"\n", " Print a formatted analysis of first beat cutoff detection\n", " \"\"\"\n", " result = detect_first_beat_cutoff(beat_times, threshold_std)\n", " \n", " print(\"=== First Beat Cut-off Analysis ===\")\n", " print(f\"Cut-off detected: {'YES' if result['cutoff_detected'] else 'NO'}\")\n", " \n", " if result['first_interval'] is not None:\n", " print(f\"First interval: {result['first_interval']:.4f}s ({result['first_bpm']:.1f} BPM)\")\n", " print(f\"Normal mean: {result['mean_normal_interval']:.4f}s ({result['normal_bpm']:.1f} BPM)\")\n", " print(f\"Standard deviation: {result['std_normal_interval']:.4f}s\")\n", " print(f\"Z-score: {result['z_score']:.2f} (threshold: {result['threshold_used']})\")\n", " \n", " if result['cutoff_detected']:\n", " print(f\"Analysis: {result['cutoff_type']}\")\n", " print(f\"Confidence: {result['confidence']:.1f}x threshold\")\n", " else:\n", " print(f\"Reason: {result['reason']}\")\n", "\n", "# Quick batch analysis function\n", "def analyze_multiple_files(beat_times_list, file_names=None, threshold_std=2.0):\n", " \"\"\"\n", " Analyze multiple files for first beat cutoffs\n", " \n", " Parameters:\n", " beat_times_list: list of numpy arrays, each containing beat times\n", " file_names: optional list of file names for labeling\n", " threshold_std: detection threshold\n", " \n", " Returns:\n", " list of detection results\n", " \"\"\"\n", " results = []\n", " \n", " for i, beat_times in enumerate(beat_times_list):\n", " file_name = file_names[i] if file_names else f\"File_{i+1}\"\n", " result = detect_first_beat_cutoff(beat_times, threshold_std)\n", " result['file_name'] = file_name\n", " results.append(result)\n", " \n", " # Print summary\n", " print(\"=== Batch Analysis Summary ===\")\n", " cutoff_count = sum(1 for r in results if r['cutoff_detected'])\n", " print(f\"Files with cutoff detected: {cutoff_count}/{len(results)}\")\n", " \n", " for result in results:\n", " status = \"CUT-OFF\" if result['cutoff_detected'] else \"OK\"\n", " z_score = result['z_score'] if result['z_score'] is not None else 0\n", " print(f\"{result['file_name']}: {status} (z-score: {z_score:.2f})\")\n", " \n", " return results\n", "\n", "# Example usage:\n", "\"\"\"\n", "# Single file analysis\n", "beat_times = np.array([0.1, 0.8, 1.4, 2.0, 2.6, 3.2, 3.8]) # Example with short first interval\n", "result = detect_first_beat_cutoff(beat_times)\n", "print_cutoff_analysis(beat_times)\n", "\n", "# Batch analysis\n", "beat_times_list = [beat_times1, beat_times2, beat_times3]\n", "file_names = ['song1.wav', 'song2.wav', 'song3.wav']\n", "results = analyze_multiple_files(beat_times_list, file_names, threshold_std=1.5)\n", "\"\"\"" ] }, { "cell_type": "code", "execution_count": null, "id": "18", "metadata": {}, "outputs": [], "source": [ "result = detect_first_beat_cutoff(beats_refined)\n", "print(result)" ] }, { "cell_type": "code", "execution_count": null, "id": "19", "metadata": {}, "outputs": [], "source": [ "all = []\n", "\n", "for idx in range(100):\n", " fp = df.iloc[idx][\"s3_filepath\"]\n", " audio = Audio.from_s3(fp)\n", " audio_mono = Audio.convert(audio, n_channels=1, sample_rate=audio.sample_rate, byte_width=audio.byte_width)\n", " out = extractor.extract(audio_mono)\n", " beats_refined = np.array(out[\"downbeats\"])\n", " result = detect_first_beat_cutoff(beats_refined)\n", " all.append(result)" ] }, { "cell_type": "code", "execution_count": null, "id": "20", "metadata": {}, "outputs": [], "source": [ "for a in all:\n", " if a['cutoff_detected'].item():\n", " print(a)" ] }, { "cell_type": "code", "execution_count": null, "id": "21", "metadata": {}, "outputs": [], "source": [ "import modal\n", "import json\n", "from uuid import uuid4\n", "from tqdm import tqdm" ] }, { "cell_type": "code", "execution_count": null, "id": "22", "metadata": {}, "outputs": [], "source": [ "modal_sfx = modal.Cls.from_name(\"upsample-diff_seeds_sfx_v1-dev\", \"UpsampleStub\")" ] }, { "cell_type": "code", "execution_count": null, "id": "23", "metadata": {}, "outputs": [], "source": [ "id = str(uuid4())\n", "input = {\n", " \"id\": id,\n", " \"prompt_audio\": \"\",\n", " \"prompt_text\": \"\",\n", " \"model_name\": \"diff_seeds_sfx_v1\",\n", " \"metadata\": {\n", " \"prompt\": \"\",\n", " \"tags\": \"reggae drum loop, acoustic drums\",\n", " },\n", "}" ] }, { "cell_type": "code", "execution_count": null, "id": "24", "metadata": {}, "outputs": [], "source": [ "all = []\n", "\n", "for idx in tqdm(range(50)):\n", " id = str(uuid4())\n", " input = {\n", " \"id\": id,\n", " \"prompt_audio\": \"\",\n", " \"prompt_text\": \"\",\n", " \"model_name\": \"diff_seeds_sfx_v1\",\n", " \"metadata\": {\n", " \"prompt\": \"\",\n", " \"tags\": \"reggae drum loop, acoustic drums, tempo: 110\",\n", " },\n", " }\n", " output = modal_sfx.upsample.remote(json.dumps(input))\n", "\n", " audio = Audio.from_s3(id)\n", " audio_mono = Audio.convert(audio, n_channels=1, sample_rate=audio.sample_rate, byte_width=audio.byte_width)\n", " out = extractor.extract(audio_mono)\n", " beats_refined = np.array(out[\"downbeats\"])\n", " result = detect_first_beat_cutoff(beats_refined)\n", " all.append(result)" ] }, { "cell_type": "code", "execution_count": null, "id": "25", "metadata": {}, "outputs": [], "source": [ "for a in all:\n", " print(a)" ] }, { "cell_type": "code", "execution_count": null, "id": "26", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "suno_clean", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10.15" } }, "nbformat": 4, "nbformat_minor": 5 }