Update app.py

app.py

@@ -5,7 +5,7 @@ import zipfile
 import librosa
 import numpy as np
 from pydub import AudioSegment
-from moviepy.editor import AudioFileClip, ImageClip
+from moviepy.editor import AudioFileClip, ImageClip, CompositeVideoClip
 import subprocess
 from pathlib import Path
 import sys
@@ -14,145 +14,199 @@ import sys
 OUTPUT_DIR = Path("nightpulse_output")
 TEMP_DIR = Path("temp_processing")
 
-def process_track(audio_file, cover_art_image):
-    zip_path = None
-    video_path = None
+# --- Core Logic Functions ---
 
+def analyze_and_separate(audio_file):
+    """Phase 1: Separate Stems and return file paths for preview"""
     try:
-        # --- 0. Input Validation ---
         if not audio_file:
             raise ValueError("No audio file provided.")
             
-        # --- 1. Setup Directories ---
-        if OUTPUT_DIR.exists():
-            shutil.rmtree(OUTPUT_DIR)
-        if TEMP_DIR.exists():
-            shutil.rmtree(TEMP_DIR)
-        
-        # Create subfolders for organization
+        # Cleanup
+        if OUTPUT_DIR.exists(): shutil.rmtree(OUTPUT_DIR)
+        if TEMP_DIR.exists(): shutil.rmtree(TEMP_DIR)
         (OUTPUT_DIR / "Stems").mkdir(parents=True, exist_ok=True)
         (OUTPUT_DIR / "Loops").mkdir(parents=True, exist_ok=True)
         TEMP_DIR.mkdir(parents=True, exist_ok=True)
-        
+
         filename = Path(audio_file).stem
         
-        # --- 2. Analyze BPM (Librosa) ---
+        # 1. BPM Detection
         print(f"Analyzing {filename}...")
-        try:
-            y, sr = librosa.load(audio_file, duration=60, mono=True)
-            tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
-            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:
-            print(f"BPM Warning: {e}")
-            detected_bpm = 120 
-        
-        # --- 3. AI Stem Separation (Demucs) ---
+        y, sr = librosa.load(audio_file, duration=60, mono=True)
+        tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
+        bpm = int(round(tempo[0])) if np.ndim(tempo) > 0 else int(round(tempo))
+        print(f"Detected BPM: {bpm}")
+
+        # 2. Demucs Separation
         print("Separating stems...")
-        try:
-            subprocess.run([
-                sys.executable, "-m", "demucs", 
-                "-n", "htdemucs", 
-                "--out", str(TEMP_DIR), 
-                audio_file
-            ], check=True, capture_output=True)
-        except subprocess.CalledProcessError as e:
-            raise RuntimeError(f"Demucs failed: {e.stderr.decode()}")
-        
+        subprocess.run([
+            sys.executable, "-m", "demucs", "-n", "htdemucs", "--out", str(TEMP_DIR), audio_file
+        ], check=True, capture_output=True)
+
+        # 3. Locate Stems
         demucs_out = TEMP_DIR / "htdemucs"
         track_folder = next(demucs_out.iterdir(), None)
+        if not track_folder: raise FileNotFoundError("Demucs failed to output files.")
         
-        if not track_folder: raise FileNotFoundError("Demucs output folder missing.")
-
-        # Map the raw stems
-        drums_path = track_folder / "drums.wav"
-        melody_path = track_folder / "other.wav"
-        bass_path = track_folder / "bass.wav"
-        vocals_path = track_folder / "vocals.wav"
-
-        # --- 4. EXPORT FULL STEMS (New Step) ---
-        # Copy the full length files to the Stems folder
-        if drums_path.exists(): shutil.copy(drums_path, OUTPUT_DIR / "Stems" / f"{detected_bpm}BPM_Full_Drums.wav")
-        if melody_path.exists(): shutil.copy(melody_path, OUTPUT_DIR / "Stems" / f"{detected_bpm}BPM_Full_Melody.wav")
-        if bass_path.exists(): shutil.copy(bass_path, OUTPUT_DIR / "Stems" / f"{detected_bpm}BPM_Full_Bass.wav")
-        if vocals_path.exists(): shutil.copy(vocals_path, OUTPUT_DIR / "Stems" / f"{detected_bpm}BPM_Full_Vocals.wav")
-
-        # --- 5. Loop Logic (The Chop) ---
-        if detected_bpm <= 0: detected_bpm = 120
-        ms_per_beat = (60 / detected_bpm) * 1000
+        drums = track_folder / "drums.wav"
+        bass = track_folder / "bass.wav"
+        melody = track_folder / "other.wav"
+        vocals = track_folder / "vocals.wav"
+        
+        # Return paths to the UI and the BPM
+        return str(drums), str(bass), str(melody), str(vocals), bpm, str(track_folder)
+
+    except Exception as e:
+        raise gr.Error(f"Separation Failed: {str(e)}")
+
+def package_and_export(track_folder_str, bpm, start_offset_sec, cover_art):
+    """Phase 2: Chop Loops, Generate Video (Zoom Effect), Zip"""
+    try:
+        track_folder = Path(track_folder_str)
+        drums = track_folder / "drums.wav"
+        bass = track_folder / "bass.wav"
+        melody = track_folder / "other.wav"
+        vocals = track_folder / "vocals.wav"
+
+        # 1. Save Full Stems
+        for stem, name in [(drums, "Drums"), (bass, "Bass"), (melody, "Melody"), (vocals, "Vocals")]:
+            if stem.exists():
+                shutil.copy(stem, OUTPUT_DIR / "Stems" / f"{bpm}BPM_Full_{name}.wav")
+
+        # 2. Create Loops (With User Offset)
+        ms_per_beat = (60 / bpm) * 1000
         eight_bars_ms = ms_per_beat * 4 * 8
+        start_ms = start_offset_sec * 1000
         
-        def create_loop(source_path, output_name):
-            if not source_path.exists(): return None, None
-            
-            audio = AudioSegment.from_wav(str(source_path))
-            start = len(audio) // 3
-            end = start + eight_bars_ms
+        created_loops = {} # Store paths for video generation
+
+        def make_loop(src, name):
+            if not src.exists(): return None
+            audio = AudioSegment.from_wav(str(src))
             
-            if len(audio) < end:
-                start = 0
-                end = min(len(audio), eight_bars_ms)
+            # Use User Offset
+            end_ms = start_ms + eight_bars_ms
+            if len(audio) < end_ms: # Fallback if offset is too late
+                s = 0
+                e = min(len(audio), eight_bars_ms)
+            else:
+                s, e = start_ms, end_ms
                 
-            loop = audio[start:int(end)].fade_in(15).fade_out(15).normalize()
-            # Save to Loops folder
-            out_name = OUTPUT_DIR / "Loops" / f"{detected_bpm}BPM_{output_name}.wav"
-            loop.export(out_name, format="wav")
-            return out_name, loop
-
-        loop_drums, _ = create_loop(drums_path, "DrumLoop")
-        loop_melody, _ = create_loop(melody_path, "MelodyLoop")
-        create_loop(bass_path, "BassLoop")
+            loop = audio[s:int(e)].fade_in(15).fade_out(15).normalize()
+            out_path = OUTPUT_DIR / "Loops" / f"{bpm}BPM_{name}.wav"
+            loop.export(out_path, format="wav")
+            return out_path
+
+        created_loops['drums'] = make_loop(drums, "DrumLoop")
+        created_loops['bass'] = make_loop(bass, "BassLoop")
+        created_loops['melody'] = make_loop(melody, "MelodyLoop")
         
-        # --- 6. Video Generation ---
-        if cover_art_image and loop_melody:
-            print("Rendering Video...")
-            try:
-                vid_out = OUTPUT_DIR / "Promo_Video.mp4"
-                audio_clip = AudioFileClip(str(loop_melody))
-                img_clip = ImageClip(cover_art_image)
-                img_clip = img_clip.resize(width=1080)
-                img_clip = img_clip.set_duration(audio_clip.duration)
-                img_clip = img_clip.set_audio(audio_clip)
-                img_clip.fps = 24
+        # 3. Generate Video (Slow Zoom Effect)
+        video_path = None
+        if cover_art and created_loops['melody']:
+            print("Rendering Dynamic Video...")
+            vid_out = OUTPUT_DIR / "Promo_Video.mp4"
+            audio_clip = AudioFileClip(str(created_loops['melody']))
+            duration = audio_clip.duration
+            
+            # Load Image
+            img = ImageClip(cover_art).resize(width=1080)
+            
+            # THE ZOOM EFFECT: Resize image from 1.0x to 1.1x over time
+            # We crop the center to handle the zoom so it doesn't change output size
+            w, h = img.size
+            
+            def zoom_effect(t):
+                # Zoom factor goes from 1.0 to 1.15
+                scale = 1 + 0.15 * (t / duration)
+                return scale
                 
-                img_clip.write_videofile(str(vid_out), codec="libx264", audio_codec="aac", logger=None)
-                video_path = str(vid_out)
-            except Exception as e:
-                print(f"Video skipped: {e}")
+            # Apply zoom (this is a bit heavy, simple resize is safer for CPU)
+            # Alternative: Simple Pan or just Static if Pillow is tricky.
+            # Let's do a simple resize animation
+            img = img.resize(lambda t : 1 + 0.04*t) # Slow 4% zoom
+            img = img.set_position(('center', 'center'))
+            img = img.set_duration(duration)
+            img = img.set_audio(audio_clip)
+            
+            # Composite to ensure frame size stays constant (1080 width)
+            # Note: resizing makes it grow, we need to crop or center. 
+            # For simplicity in v1, we let it grow slightly or stick to static if 'resize' fails.
+            # We will use the simple robust method:
+            
+            final_clip = CompositeVideoClip([img], size=(1080, 1920))
+            final_clip.duration = duration
+            final_clip.audio = audio_clip
+            final_clip.fps = 24
+            
+            final_clip.write_videofile(str(vid_out), codec="libx264", audio_codec="aac", logger=None)
+            video_path = str(vid_out)
 
-        # --- 7. Zip Export ---
+        # 4. Zip
         zip_file = "NightPulse_Pack.zip"
         with zipfile.ZipFile(zip_file, 'w') as zf:
             for root, dirs, files in os.walk(OUTPUT_DIR):
                 for file in files:
                     file_path = Path(root) / file
-                    # Create structure inside zip (Stems/file.wav, Loops/file.wav)
                     arcname = file_path.relative_to(OUTPUT_DIR)
                     zf.write(file_path, arcname)
-        zip_path = zip_file
-        
-        return zip_path, video_path
+                    
+        return zip_file, video_path
 
     except Exception as e:
-        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"),
-        gr.Image(type="filepath", label="Upload Cover Art")
-    ],
-    outputs=[
-        gr.File(label="Download Full Pack (ZIP)"),
-        gr.Video(label="Preview Video")
-    ],
-    title="Night Pulse Audio | Automator",
-    description="Night Pulse Pipeline v1.3 (Full Stems Included)"
-)
+        raise gr.Error(f"Packaging Failed: {str(e)}")
+
+
+# --- GUI (Blocks) ---
+
+with gr.Blocks(title="Night Pulse | Command Center", theme=gr.themes.Base()) as app:
+    gr.Markdown("# 🎛️ Night Pulse Audio | Command Center")
+    gr.Markdown("Transform Suno tracks into commercial sample packs.")
+    
+    # State storage
+    stored_folder = gr.State()
+    stored_bpm = gr.State()
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            input_audio = gr.Audio(type="filepath", label="1. Upload Master Track")
+            input_art = gr.Image(type="filepath", label="Cover Art (9:16)")
+            btn_analyze = gr.Button("🔍 Phase 1: Deconstruct & Analyze", variant="primary")
+            
+        with gr.Column(scale=1):
+            gr.Markdown("### 2. Stem Preview")
+            # Audio Players
+            p_drums = gr.Audio(label="Drums Stem")
+            p_bass = gr.Audio(label="Bass Stem")
+            p_melody = gr.Audio(label="Melody Stem")
+            p_vocals = gr.Audio(label="Vocals Stem")
+    
+    gr.Markdown("---")
+    
+    with gr.Row():
+        with gr.Column():
+            gr.Markdown("### 3. Loop Settings")
+            slider_start = gr.Slider(minimum=0, maximum=120, value=15, label="Loop Start Time (Seconds)", info="Where should the 8-bar loop start?")
+            btn_package = gr.Button("📦 Phase 2: Package & Export", variant="primary")
+        
+        with gr.Column():
+            gr.Markdown("### 4. Final Output")
+            out_zip = gr.File(label="Download Pack (ZIP)")
+            out_video = gr.Video(label="Promo Video (Dynamic)")
+
+    # Events
+    btn_analyze.click(
+        fn=analyze_and_separate,
+        inputs=[input_audio],
+        outputs=[p_drums, p_bass, p_melody, p_vocals, stored_bpm, stored_folder]
+    )
+    
+    btn_package.click(
+        fn=package_and_export,
+        inputs=[stored_folder, stored_bpm, slider_start, input_art],
+        outputs=[out_zip, out_video]
+    )
 
 if __name__ == "__main__":
-    iface.launch()
+    app.launch()