Update app.py

app.py

@@ -1,395 +1,180 @@
 import gradio as gr
-import librosa
-import numpy as np
 import os
 import shutil
 import zipfile
-import tempfile
-import soundfile as sf
-import traceback
+import librosa
+import numpy as np
+from pydub import AudioSegment
+from moviepy.editor import AudioFileClip, ImageClip
 import subprocess
-from typing import Tuple, List
+from pathlib import Path
+import sys
 
 # --- Configuration ---
-OUTPUT_FOLDER_NAME = "PRO_LOOP_PACK"
-
-# Mapping of model selection to Spleeter config and resulting stem types
-STEM_MODELS = {
-    '2-Stems (Vocals/Inst)': {
-        'spleeter_config': '2stems',
-        'stems': ['vocals', 'accompaniment'], # Spleeter output names
-        'display_stems': ['Vocals', 'Instrumental'] # User-facing names
-    },
-    '4-Stems (Drums, Bass, Vocals, Other)': {
-        'spleeter_config': '4stems',
-        'stems': ['vocals', 'drums', 'bass', 'other'],
-        'display_stems': ['Vocals', 'Drums', 'Bass', 'Other']
-    },
-    '5-Stems (Drums, Bass, Vocals, Piano, Other)': {
-        'spleeter_config': '5stems',
-        'stems': ['vocals', 'drums', 'bass', 'piano', 'other'],
-        'display_stems': ['Vocals', 'Drums', 'Bass', 'Piano', 'Other']
-    },
-}
-LOOP_BAR_LENGTHS = [4, 6, 8]
-
-# Key Detection Templates (as defined previously)
-KEY_TEMPLATES = {
-    'major': [6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.16, 3.61, 3.28, 2.91],
-    'minor': [6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.91, 3.03, 3.34]
-}
-NOTES = ['C', 'C#', 'D', 'D#', 'E', 'F', 'F#', 'G', 'G#', 'A', 'A#', 'B']
-
-# --- Utility Functions ---
-
-def save_segment(filepath: str, audio_data: np.ndarray, sr: int):
-    """Utility function to save a NumPy audio array as a WAV file."""
-    # Spleeter outputs 44100Hz audio, so we explicitly set the sample rate
-    sf.write(filepath, audio_data, sr, format='WAV', subtype='PCM_16')
-
-def detect_key_and_mode(y: np.ndarray, sr: int) -> str:
-    """Estimates the musical key (e.g., 'C Major' or 'A Minor')."""
-    try:
-        chroma = librosa.feature.chroma_cqt(y=y, sr=sr)
-        chroma_mean = np.mean(chroma, axis=1)
-        chroma_mean /= chroma_mean.sum()
-        
-        best_key = "Unknown"
-        max_correlation = -1.0
-
-        for i, note in enumerate(NOTES):
-            # Check major keys
-            major_template = np.roll(KEY_TEMPLATES['major'], i)
-            corr_major = np.dot(chroma_mean, major_template)
-            
-            if corr_major > max_correlation:
-                max_correlation = corr_major
-                best_key = f"{note} Major"
-            
-            # Check minor keys
-            minor_template = np.roll(KEY_TEMPLATES['minor'], i)
-            corr_minor = np.dot(chroma_mean, minor_template)
-
-            if corr_minor > max_correlation:
-                max_correlation = corr_minor
-                best_key = f"{note} Minor"
-                
-        if max_correlation < 0.2:
-            return "KeyDetectionAmbiguous"
-
-        return best_key.replace(' ', '')
-        
-    except Exception as e:
-        print(f"Key Detection Failed: {e}")
-        return "KeyDetectionFailed"
+# We use Path objects for robust cross-platform compatibility
+OUTPUT_DIR = Path("nightpulse_output")
+TEMP_DIR = Path("temp_processing")
 
-def separate_stems(audio_path: str, model_name: str, output_dir: str) -> str:
+def process_track(audio_file, cover_art_image):
     """
-    Executes Spleeter separation via subprocess.
-    Requires 'spleeter' package to be installed in the environment.
+    Main pipeline function.
+    Returns: (zip_path, video_path)
     """
-    spleeter_config = STEM_MODELS[model_name]['spleeter_config']
-    
-    # Spleeter output folder will be a subfolder named after the input file (without extension)
-    # We clean this up later.
-    
-    # Spleeter command: spleeter separate -o {output_dir} -p {config} {input_file}
-    command = [
-        "spleeter", "separate", 
-        "-o", output_dir, 
-        "-p", f"spleeter:{spleeter_config}", 
-        audio_path
-    ]
-    
-    try:
-        # Run Spleeter command
-        result = subprocess.run(command, check=True, capture_output=True, text=True)
-        print("Spleeter Output:", result.stdout)
-        print("Spleeter Errors:", result.stderr)
-        
-        # Spleeter creates a sub-directory based on the input filename.
-        # We need to find that subdirectory.
-        base_filename = os.path.splitext(os.path.basename(audio_path))[0]
-        spleeter_output_path = os.path.join(output_dir, base_filename)
-        
-        if not os.path.isdir(spleeter_output_path):
-            raise FileNotFoundError(f"Spleeter output directory not found at: {spleeter_output_path}")
-
-        return spleeter_output_path
-    
-    except subprocess.CalledProcessError as e:
-        raise RuntimeError(f"Spleeter command failed. Check if 'spleeter' is installed. Output: {e.stdout}, Error: {e.stderr}")
-    except Exception as e:
-        raise RuntimeError(f"Error during Spleeter execution: {e}")
-
-# --- Main Processing Function ---
-
-def create_market_ready_pack(
-    audio_file_path: str,
-    one_shot_sensitivity: float,
-    stem_model_selection: str,
-    progress=gr.Progress()
-) -> Tuple[str | None, str]:
-    """
-    Processes the input audio file, generates loops and one-shots,
-    and packages them into a market-ready ZIP file.
-    """
-    temp_dir = None
-    
-    if not audio_file_path:
-        return None, "Error: Please upload an audio file before proceeding."
+    # Initialize return variables to None to prevent 'UnboundLocalError'
+    zip_path = None
+    video_path = None
 
     try:
-        # 1. Setup Temporary Directories
-        temp_dir = tempfile.mkdtemp()
-        output_root = os.path.join(temp_dir, OUTPUT_FOLDER_NAME)
-        os.makedirs(output_root, exist_ok=True)
-        
-        progress(0.05, desc="Loading and Verifying Audio...")
-
-        # Robust Audio Loading (Load full mix for analysis)
-        y_full, sr = librosa.load(audio_file_path, sr=None, mono=True)
-        if y_full.size == 0:
-             raise ValueError("Loaded audio is empty.")
-
-        # 2. Advanced Audio Analysis (Tempo and Key)
-        progress(0.15, desc="Analyzing Tempo and Musical Key...")
-        
-        tempo = 120.0 
-        start_sample = 0
-        key_mode_name = "120BPM_UnknownKey"
-        
-        try:
-            tempo, beat_frames = librosa.beat.beat_track(y=y_full, sr=sr, trim=True)
-            key_mode_name = detect_key_and_mode(y_full, sr)
+        # --- 0. Input Validation (Robustness Check) ---
+        if not audio_file:
+            raise ValueError("No audio file provided. Please upload a track.")
             
-            samples_per_beat = int((60 / tempo) * sr)
-            start_sample = librosa.frames_to_samples(beat_frames[0]) if beat_frames.size > 0 else 0
-            
-            gr.Info(f"Analysis Complete: {int(tempo)} BPM, {key_mode_name}.")
-            key_mode_name = f"{int(tempo)}BPM_{key_mode_name}"
+        # --- 1. Setup Directories ---
+        # Clean previous runs to prevent file mixing
+        if OUTPUT_DIR.exists():
+            shutil.rmtree(OUTPUT_DIR)
+        if TEMP_DIR.exists():
+            shutil.rmtree(TEMP_DIR)
+        OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+        TEMP_DIR.mkdir(parents=True, exist_ok=True)
+        
+        filename = Path(audio_file).stem
+        
+        # --- 2. Analyze BPM & Key (Librosa) ---
+        print(f"Analyzing {filename}...")
+        try:
+            # Load 60s for better context, mono=True for BPM analysis
+            y, sr = librosa.load(audio_file, duration=60, mono=True)
+            tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
             
+            # Robustness: Handle different librosa versions returning array vs float
+            if np.ndim(tempo) > 0:
+                detected_bpm = int(round(tempo[0]))
+            else:
+                detected_bpm = int(round(tempo))
+                
+            print(f"Detected BPM: {detected_bpm}")
         except Exception as e:
-            gr.Warning(f"Warning: Tempo or Key detection failed ({e}). Using default 120 BPM and 'Unknown Key'.")
-            samples_per_beat = int((60 / 120.0) * sr) # Fallback beat timing
-
-        # 3. REAL STEM SEPARATION using Spleeter
-        progress(0.25, desc=f"Separating Stems using {stem_model_selection} model...")
+            print(f"BPM Detection Warning: {e}")
+            detected_bpm = 120 # Safe Fallback
         
-        spleeter_output_path = separate_stems(audio_file_path, stem_model_selection, output_root)
-        spleeter_stems = STEM_MODELS[stem_model_selection]['stems']
-        display_stems = STEM_MODELS[stem_model_selection]['display_stems']
-        
-        # Dictionary to hold the audio data for each stem from Spleeter's output
-        stem_audio_data = {}
-        for spleeter_name, display_name in zip(spleeter_stems, display_stems):
-            stem_filepath = os.path.join(spleeter_output_path, f"{spleeter_name}.wav")
-            if not os.path.exists(stem_filepath):
-                gr.Warning(f"Stem file not found for {display_name}. Skipping this stem.")
-                continue
-
-            # Load the separated stem audio (it will be aligned and resampled by Spleeter)
-            # We enforce mono loading for consistent processing later
-            y_stem, sr_stem = librosa.load(stem_filepath, sr=sr, mono=True) 
-            
-            # Align the start of the stem using the previously detected global beat
-            y_stem_aligned = y_stem[start_sample:]
-            stem_audio_data[display_name] = y_stem_aligned
-        
-        # Clean up Spleeter's intermediate directory
-        shutil.rmtree(spleeter_output_path)
-        
-        if not stem_audio_data:
-            raise RuntimeError("No separated stems were successfully processed. Check Spleeter output.")
-
-        # 4. Generate Loops (4, 6, 8 Bars)
-        progress(0.45, desc="Generating Time-Aligned Loops...")
-        
-        for stem_name, y_stem in stem_audio_data.items():
-            loops_dir = os.path.join(output_root, 'LOOPS', stem_name)
-            os.makedirs(loops_dir, exist_ok=True)
+        # --- 3. AI Stem Separation (Demucs) ---
+        print("Separating stems with Demucs...")
+        try:
+            # We call demucs as a module using sys.executable to ensure we use the correct python environment
+            subprocess.run([
+                sys.executable, "-m", "demucs", 
+                "-n", "htdemucs", 
+                "--out", str(TEMP_DIR), 
+                audio_file
+            ], check=True, capture_output=True)
+        except subprocess.CalledProcessError as e:
+            # Capture the specific error from the subprocess
+            raise RuntimeError(f"Demucs processing failed. Error: {e.stderr.decode()}")
+        
+        # Locate separated stems (Robust Path Finding)
+        demucs_out = TEMP_DIR / "htdemucs"
+        # Demucs might normalize filenames (spaces -> underscores), so we just find the first folder
+        track_folder = next(demucs_out.iterdir(), None)
+        
+        if not track_folder:
+             raise FileNotFoundError("Demucs output folder could not be found.")
+
+        drums_path = track_folder / "drums.wav"
+        melody_path = track_folder / "other.wav"
+        bass_path = track_folder / "bass.wav"
+        
+        if not drums_path.exists():
+            raise FileNotFoundError(f"Stems were not generated in {track_folder}")
+
+        # --- 4. Loop Logic (Pydub) ---
+        # Calculate duration of 8 bars in milliseconds
+        if detected_bpm <= 0: detected_bpm = 120
+        ms_per_beat = (60 / detected_bpm) * 1000
+        eight_bars_ms = ms_per_beat * 4 * 8
+        
+        def create_loop(source_path, output_name):
+            if not source_path.exists():
+                return None, None
+                
+            audio = AudioSegment.from_wav(str(source_path))
             
-            samples_per_bar = samples_per_beat * 4 # Assuming 4/4 time signature
+            # Smart Chop: Grab the "middle" 8 bars to avoid intro/outro silence
+            start_time = len(audio) // 3 
+            end_time = start_time + eight_bars_ms
             
-            for num_bars in LOOP_BAR_LENGTHS:
-                samples_per_loop = samples_per_bar * num_bars
+            # Safety check if audio is shorter than 8 bars
+            if len(audio) < end_time:
+                start_time = 0
+                end_time = min(len(audio), eight_bars_ms)
                 
-                for i in range(0, len(y_stem) - samples_per_loop + 1, samples_per_loop):
-                    try:
-                        loop_segment = y_stem[i:i + samples_per_loop]
-                        
-                        if len(loop_segment) < samples_per_loop * 0.9: 
-                            continue
-                            
-                        index = i // samples_per_loop + 1
-                        # Naming convention: {BPM_Key}_{Stem}_{Bars}Bar_{Index}.wav
-                        filename = f"{key_mode_name}_{stem_name}_{num_bars}Bar_{index:02d}.wav"
-                        save_segment(os.path.join(loops_dir, filename), loop_segment, sr)
-                    except Exception as e:
-                        gr.Warning(f"Error slicing {num_bars}-bar loop for {stem_name}: {e}")
-                        continue
-
-        # 5. Generate One-Shots (Transient Detection)
-        progress(0.70, desc="Generating One-Shots (Transient Detection)...")
-        
-        # Sensitivity mapping: 1=Few/Loud (large pre_max), 10=Many/Quiet (small pre_max)
-        pre_max_frames = int(12 - one_shot_sensitivity) 
-        if pre_max_frames < 2: pre_max_frames = 2
-        
-        pre_slice_samples = int(sr * 0.05)
-        post_slice_samples = int(sr * 0.25) 
-
-        for stem_name, y_stem in stem_audio_data.items():
-            shots_dir = os.path.join(output_root, 'ONESHOTS', stem_name)
-            os.makedirs(shots_dir, exist_ok=True)
+            loop = audio[start_time:end_time]
+            # 15ms fade to prevent clicks
+            loop = loop.fade_in(15).fade_out(15).normalize() 
             
+            out_filename = f"{detected_bpm}BPM_{output_name}.wav"
+            out_file = OUTPUT_DIR / out_filename
+            loop.export(out_file, format="wav")
+            return out_file, loop
+
+        # Generate Loops
+        loop_drums_path, _ = create_loop(drums_path, "DrumLoop")
+        loop_melody_path, melody_audio = create_loop(melody_path, "MelodyLoop")
+        create_loop(bass_path, "BassLoop")
+        
+        # --- 5. Video Generation (MoviePy) ---
+        # Logic: Only generate video if User uploaded Art AND we successfully made a melody loop
+        if cover_art_image is not None and loop_melody_path is not None:
+            print("Rendering Promo Video...")
             try:
-                o_env = librosa.onset.onset_strength(y=y_stem, sr=sr, aggregate=np.median)
-                onset_frames = librosa.onset.onset_detect(
-                    onset_envelope=o_env, 
-                    sr=sr, 
-                    units='frames',
-                    pre_max=pre_max_frames, 
-                    post_max=pre_max_frames // 2,
-                    wait=10
-                )
-                onset_samples = librosa.frames_to_samples(onset_frames)
+                video_out_path = OUTPUT_DIR / "Promo_Video_Reel.mp4"
+                
+                audio_clip = AudioFileClip(str(loop_melody_path))
+                image_clip = ImageClip(cover_art_image)
+                
+                # Resize logic: Fit to width 1080 (standard), maintain aspect ratio
+                image_clip = image_clip.resize(width=1080)
                 
-                for i, sample_index in enumerate(onset_samples):
-                    start = max(0, sample_index - pre_slice_samples)
-                    end = min(len(y_stem), sample_index + post_slice_samples)
-                    
-                    shot_segment = y_stem[start:end]
-                    
-                    if len(shot_segment) > int(sr * 0.05):
-                        filename = f"{key_mode_name}_{stem_name}_OneShot_{i+1:03d}.wav"
-                        save_segment(os.path.join(shots_dir, filename), shot_segment, sr)
+                # Set duration to match audio loop
+                image_clip = image_clip.set_duration(audio_clip.duration)
+                image_clip = image_clip.set_audio(audio_clip)
+                image_clip.fps = 24
+                
+                image_clip.write_videofile(str(video_out_path), codec="libx264", audio_codec="aac", logger=None)
+                video_path = str(video_out_path)
             except Exception as e:
-                gr.Warning(f"Error during One-Shot detection for {stem_name}. Skipping. Details: {e}")
-                continue
-
-
-        # 6. Packaging (License and ZIP)
-        progress(0.90, desc="Creating License and Packaging Files...")
-        
-        # Create the License.txt file
-        license_content = f"""
-        -- PROFESSIONAL LOOP PACK LICENSE AGREEMENT --
+                print(f"Video generation skipped due to error: {e}")
+                # We don't fail the whole pipeline here, we just skip the video part
+                video_path = None
         
-        Product: {OUTPUT_FOLDER_NAME}
-        BPM/Key Reference: {key_mode_name}
-        Separation Model Used: {stem_model_selection}
+        # --- 6. Zip It Up ---
+        zip_file_path = "NightPulse_Pack.zip"
+        with zipfile.ZipFile(zip_file_path, 'w') as zipf:
+            for file in OUTPUT_DIR.iterdir():
+                zipf.write(file, file.name)
+        zip_path = zip_file_path
         
-        1. Royalty-Free Use: All sounds, loops, and one-shots within this pack are 
-           100% royalty-free for commercial use in musical compositions, sound design, 
-           and public performances. You may use them in your own tracks and sell those 
-           tracks without owing any additional royalties to the creator.
-        
-        2. Restrictions: Redistribution, repackaging, or re-selling of the individual 
-           sounds or loops as part of another sound library or sample pack is strictly 
-           prohibited.
-           
-        3. Generated: {os.uname().nodename}
-        """
-        
-        license_filepath = os.path.join(output_root, 'License.txt')
-        with open(license_filepath, 'w') as f:
-            f.write(license_content.strip())
-            
-        # Create the final ZIP file
-        zip_filename = os.path.join(temp_dir, f"{OUTPUT_FOLDER_NAME}_{key_mode_name}.zip")
-        
-        with zipfile.ZipFile(zip_filename, 'w', zipfile.ZIP_DEFLATED) as zf:
-            for root, dirs, files in os.walk(output_root):
-                for file in files:
-                    full_path = os.path.join(root, file)
-                    relative_path = os.path.relpath(full_path, temp_dir)
-                    zf.write(full_path, relative_path)
-                    
-        progress(1.0, desc="Packaging Complete!")
-        
-        shutil.rmtree(output_root)
-
-        return zip_filename, f"✅ Success! Your market-ready '{os.path.basename(zip_filename)}' is ready. Key/BPM: {key_mode_name}. Stems Processed: {', '.join(display_stems)}. Download below."
+        return zip_path, video_path
 
     except Exception as e:
-        error_message = f"Critical Error: {e}"
-        print(f"Full Traceback: {traceback.format_exc()}")
-        
-        if temp_dir and os.path.exists(temp_dir):
-            shutil.rmtree(temp_dir)
-            gr.Warning("Cleaned up temporary files after failure.")
-            
-        return None, f"❌ Processing failed. {error_message}. If Spleeter failed, ensure it is installed correctly."
-
-
-# --- Gradio Interface Definition ---
-
-with gr.Blocks(title="Market-Ready Loop Pack Generator") as demo:
-    gr.Markdown(
-        """
-        # 🎧 Professional Loop Pack Automation Tool
-        
-        Upload a full music track, select your stem separation model, and generate a 
-        complete, royalty-free sample pack including time-aligned loops and transient-detected one-shots.
-        """
-    )
-
-    with gr.Row():
-        audio_input = gr.Audio(
-            type="filepath", 
-            sources=["upload"], 
-            label="1. Upload Full Mix Audio File (WAV/MP3/FLAC)",
-        )
-        
-        stem_model_input = gr.Dropdown(
-            label="2. Select Stem Separation Model",
-            choices=list(STEM_MODELS.keys()),
-            value='4-Stems (Drums, Bass, Vocals, Other)',
-            allow_custom_value=False,
-            info="Choose the number and type of stems to split the audio into (requires Spleeter installation)."
-        )
-
-    with gr.Row():
-        sensitivity_slider = gr.Slider(
-            minimum=1, 
-            maximum=10, 
-            step=1, 
-            value=6, 
-            label="3. One-Shot Sensitivity (1=Few/Loud, 10=Many/Quiet)",
-            info="Controls the transient detection threshold for one-shot slicing."
-        )
-
-        generate_button = gr.Button("🚀 Generate Loop Pack", variant="primary")
-
-    with gr.Column(scale=1):
-        status_output = gr.Textbox(label="Status / Feedback", interactive=False)
-        zip_output = gr.File(label="4. Download Final Loop Pack ZIP")
-
-    # Define the core process action
-    generate_button.click(
-        fn=create_market_ready_pack,
-        inputs=[audio_input, sensitivity_slider, stem_model_input],
-        outputs=[zip_output, status_output]
-    )
-    
-    gr.Markdown(
-        """
-        ---
-        **Final Pack Structure (Example):**
-        - `PRO_LOOP_PACK_128BPM_CMinor.zip`
-          - `License.txt`
-          - `LOOPS/`
-            - `Drums/` (e.g., `128BPM_CMinor_Drums_4Bar_01.wav`)
-            - `Bass/` 
-            - `Vocals/` 
-            - ... (based on model selected)
-          - `ONESHOTS/`
-            - `Drums/` (e.g., `128BPM_CMinor_Drums_OneShot_001.wav`)
-            - `Bass/`
-            - ...
-        """
-    )
+        # This catches ANY crash and shows it in the UI as a red Error box
+        raise gr.Error(f"System Error: {str(e)}")
+
+# --- UI Definition ---
+iface = gr.Interface(
+    fn=process_track,
+    inputs=[
+        gr.Audio(type="filepath", label="Upload Suno Track (MP3/WAV)"),
+        gr.Image(type="filepath", label="Upload Cover Art (Optional)")
+    ],
+    outputs=[
+        gr.File(label="Download Completed Pack (ZIP)"),
+        gr.Video(label="Preview Promo Video")
+    ],
+    title="Night Pulse Audio | Automator",
+    description="<b>Night Pulse Pipeline v1.0</b><br>Upload a Suno track to automatically separate stems, normalize, chop loops, and generate a promo video.",
+    theme="default"
+)
 
 if __name__ == "__main__":
-    demo.launch(enable_queue=True)
+    iface.launch()