resonance_engine.py

#!/usr/bin/env python3
"""
LYGO Resonance Engine v0.5.2
Image → Living Stereo Soundscape
Full standard synthesis + LDQ percussion mode.
"""

import cv2
import numpy as np
import soundfile as sf
import math
import argparse
import sys
from pathlib import Path
from typing import Optional, Dict, Any
import mido
from mido import MidiFile, MidiTrack, Message

# LDQ imports are LAZY: only loaded inside the LDQ branch.
# This keeps the "Standard Beat Tools / Factory Default" path completely independent
# of any LDQ code so the default engine always works even if LDQ modules have issues.

__version__ = "0.5.2"

# Artistic Presets
PRESETS = {
    "raw": {},
    "ambient": {
        "noise_vol": 0.03,
        "drone_vol": 0.04,
        "note_vol": 0.08,
        "glitch_vol": 0.005,
        "drone_attack": 6.0,
        "drone_decay": 6.0,
        "note_attack": 0.08,
        "note_decay": 0.45,
        "max_glitches": 5,
        "noise_lowpass_hz": 350,
        "root_freq_range": (40, 80),
        "theta_lock_range": (5, 9),
    },
    # "musical" / factory-strong defaults (for Standard Beat Tools to produce nice non-static output
    # matching historical working versions from LDQ protocol history + clean skill base).
    # These ensure the 4-layer (noise + drones + melody + glitch) has energy + filtered texture.
    "musical": {
        "noise_vol": 0.09,
        "drone_vol": 0.08,
        "note_vol": 0.13,
        "glitch_vol": 0.02,
        "noise_lowpass_hz": 650,
        "drone_attack": 4.0,
        "drone_decay": 5.0,
        "note_decay": 0.35,
    },
    "glitch": {
        "noise_vol": 0.08,
        "drone_vol": 0.04,
        "note_vol": 0.06,
        "glitch_vol": 0.03,
        "max_notes": 6,
        "max_glitches": 20,
        "note_decay": 0.20,
        "glitch_decay": 0.02,
        "noise_lowpass_hz": 1800,
        "root_freq_range": (30, 60),
        "theta_lock_range": (4, 8),
    },
    "ethereal": {
        "noise_vol": 0.02,
        "drone_vol": 0.04,
        "note_vol": 0.10,
        "glitch_vol": 0.008,
        "root_freq_range": (40, 95),
        "theta_lock_range": (6, 13),
        "note_attack": 0.10,
        "note_decay": 0.55,
        "noise_lowpass_hz": 300,
        "max_glitches": 4,
    },
    "cinematic": {
        "noise_vol": 0.04,
        "drone_vol": 0.06,
        "note_vol": 0.10,
        "glitch_vol": 0.015,
        "drone_attack": 4.5,
        "drone_decay": 4.5,
        "max_drones": 4,
        "noise_lowpass_hz": 700,
        "root_freq_range": (30, 70),
        "theta_lock_range": (4.5, 10),
    },
}

class ResonanceEngine:
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = {
            "sr": 44100,
            "duration": 15.0,
            "global_fade": 0.7,
            "soft_clip": True,
            "soft_clip_amount": 1.4,
            "max_drones": 4,
            "max_notes": 8,
            "max_glitches": 15,
            "noise_vol": 0.05,
            "drone_vol": 0.04,
            "note_vol": 0.10,
            "glitch_vol": 0.015,
            "root_freq_range": (30, 70),
            "theta_lock_range": (4.5, 11),
            "drone_attack": 3.5,
            "drone_decay": 3.5,
            "note_attack": 0.04,
            "note_decay": 0.30,
            "glitch_attack": 0.005,
            "glitch_decay": 0.015,
            "noise_lowpass_hz": 700,
            "random_seed": None,
            "verbose": True,
            "export_stems": False,
            "export_midi": False,
            "use_ldq": False,
            "genre_manifold": "None",
            "percussion_mode": "standard",
            "perceptual_polish": 0.5,
            "tempo_bpm": 140,
            "tempo_subdivision": 16,
            "swing": 0.0,
        }
        if config:
            self.config.update(config)
        self.image_path = None
        self.tempo_grid = None

    def _log(self, msg: str):
        if self.config.get("verbose", True):
            print(msg)

    def analyze_image(self, image_path: str) -> Dict[str, Any]:
        img = cv2.imread(str(image_path))
        if img is None:
            raise FileNotFoundError(f"Could not load image: {image_path}")
        if len(img.shape) == 2:
            img = cv2.cvtColor(img, cv2.COLOR_GRAY2BGR)
        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        h, w = gray.shape
        avg_blue, avg_green, avg_red, _ = cv2.mean(img)
        edges = cv2.Canny(gray, 50, 150)
        edge_density = np.sum(edges > 0) / (h * w)
        contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        lines = cv2.HoughLinesP(edges, 1, np.pi / 180, 50, minLineLength=28, maxLineGap=12)
        fast = cv2.FastFeatureDetector_create(threshold=38)
        keypoints = fast.detect(gray, None)
        return {
            "width": w, "height": h,
            "avg_red": avg_red, "avg_green": avg_green, "avg_blue": avg_blue,
            "edge_density": edge_density,
            "contours": contours,
            "lines": lines if lines is not None else [],
            "keypoints": keypoints,
        }

    def _generate_tone(self, freq: float, duration: float, wave_type: str = "sine") -> np.ndarray:
        sr = self.config["sr"]
        n = int(sr * duration)
        if n <= 0:
            return np.zeros(1, dtype=np.float32)
        t = np.linspace(0, duration, n, dtype=np.float32)
        if wave_type == "sine":
            return np.sin(freq * t * 2 * np.pi).astype(np.float32)
        elif wave_type == "sawtooth":
            return (2 * (t * freq - np.floor(0.5 + t * freq)) * 0.6).astype(np.float32)
        elif wave_type == "noise":
            return np.random.uniform(-0.3, 0.3, n).astype(np.float32)
        return np.zeros(n, dtype=np.float32)

    def _apply_envelope(self, audio: np.ndarray, attack: float, decay: float) -> np.ndarray:
        sr = self.config["sr"]
        n = len(audio)
        if n <= 0:
            return audio
        a = max(1, int(attack * sr))
        d = max(1, int(decay * sr))
        env = np.ones(n, dtype=np.float32)
        if n > a + d:
            env[:a] = np.linspace(0, 1, a, dtype=np.float32)
            env[-d:] = np.linspace(1, 0, d, dtype=np.float32)
        return (audio * env).astype(np.float32)

    def _stereo_pan(self, mono: np.ndarray, pan: float) -> np.ndarray:
        if mono.ndim == 1:
            mono = mono[:, np.newaxis]
        pan = max(-1.0, min(1.0, pan))
        left = math.cos((pan + 1) * math.pi / 4)
        right = math.sin((pan + 1) * math.pi / 4)
        return np.column_stack((mono * left, mono * right)).astype(np.float32)

    def _fft_lowpass(self, audio: np.ndarray, cutoff_hz: float) -> np.ndarray:
        if cutoff_hz <= 0 or len(audio) < 32:
            return audio
        sr = self.config["sr"]
        n = len(audio)
        fft = np.fft.rfft(audio)
        freqs = np.fft.rfftfreq(n, 1.0 / sr)
        fft[freqs > cutoff_hz] = 0
        return np.fft.irfft(fft, n=n).real.astype(np.float32)

    def _soft_limit(self, audio: np.ndarray) -> np.ndarray:
        if self.config["soft_clip"]:
            amt = self.config["soft_clip_amount"]
            return (np.tanh(audio * amt) / np.tanh(amt)).astype(np.float32)
        return audio

    def _freq_to_midi(self, freq: float) -> int:
        if freq <= 0:
            return 0
        return max(0, min(127, int(12 * math.log2(freq / 440) + 69)))

    def synthesize(self, features: Dict[str, Any], output_path: str):
        cfg = self.config
        if cfg["random_seed"] is not None:
            np.random.seed(cfg["random_seed"])

        sr = cfg["sr"]
        duration = cfg["duration"]
        n_total = int(sr * duration)
        if n_total <= 0:
            sf.write(output_path, np.zeros((1, 2)), sr)
            return

        audio = np.zeros((n_total, 2), dtype=np.float32)
        root = np.interp(features["avg_red"], [0, 255], cfg["root_freq_range"])
        theta = np.interp(features["avg_green"], [0, 255], cfg["theta_lock_range"])
        w, h = features["width"], features["height"]

        # ---- LDQ Protocol Percussion Mode (LAZY imports + guarded) ----
        if cfg.get("use_ldq") and cfg.get("percussion_mode") == "ldq":
            # Lazy load only when the user explicitly enables advanced LDQ mode.
            # Factory / Standard Beat Tools path never executes any of this.
            import ldq_fingerprint
            import ldq_genre_manifold
            import ldq_percussion
            import ldq_perceptual_layer
            import ldq_tempo_grid
            import ldq_music_production

            self.tempo_grid = ldq_tempo_grid.TempoGrid(
                bpm=cfg.get("tempo_bpm", 140),
                subdivision=cfg.get("tempo_subdivision", 16),
                swing=cfg.get("swing", 0.0)
            )

            self._log("🔬 LDQ Percussion Mode active")
            if self.image_path is not None:
                vgh = ldq_fingerprint.compute_vgh(self.image_path)
                self._log(f"🔑 VGH: {vgh[:16]}...")
            else:
                vgh = None

            if cfg.get("genre_manifold") and cfg["genre_manifold"] != "None":
                self._log(f"🎵 Genre: {cfg['genre_manifold']}")
                genre_params = ldq_genre_manifold.project_to_genre(features, cfg["genre_manifold"])
                theta = theta * (1 + genre_params.get("swing_amount", 0))
                if cfg["genre_manifold"] == "Dubstep":
                    cfg["tempo_bpm"] = 140
                    cfg["swing"] = 0.25
                elif cfg["genre_manifold"] == "Phonk":
                    cfg["tempo_bpm"] = 100
                    cfg["swing"] = 0.33
                elif cfg["genre_manifold"] == "Industrial":
                    cfg["tempo_bpm"] = 120
                    cfg["swing"] = 0.15
                self.tempo_grid = ldq_tempo_grid.TempoGrid(
                    bpm=cfg["tempo_bpm"],
                    subdivision=cfg["tempo_subdivision"],
                    swing=cfg["swing"]
                )

            # generate drums
            kick = ldq_percussion.generate_kick(features, sr, duration)
            snare = ldq_percussion.generate_snare(features, sr, duration)
            hihats = ldq_percussion.generate_hihats(features, sr, cfg["tempo_bpm"], duration)

            min_len = min(len(kick), len(snare), len(hihats), n_total)
            if min_len == 0:
                min_len = n_total

            audio[:min_len] += self._stereo_pan(kick[:min_len], 0.0) * 0.6
            audio[:min_len] += self._stereo_pan(snare[:min_len], 0.2) * 0.4
            audio[:min_len] += self._stereo_pan(hihats[:min_len], -0.2) * 0.3

            # music production layer (the sidechain that was crashing in old deploys)
            audio = ldq_music_production.apply_music_production(
                audio, features, sr, bpm=cfg["tempo_bpm"], sidechain=True
            )

            if cfg.get("perceptual_polish", 0.0) > 0:
                audio = ldq_perceptual_layer.apply_perceptual_mixing(audio, features, sr)

            if vgh is not None:
                audio = ldq_fingerprint.embed_fingerprint(audio, sr, vgh)
                self._log("🔐 Fingerprint embedded")

            audio = self._soft_limit(audio)
            fade = int(cfg["global_fade"] * sr)
            if fade > 0 and n_total > fade * 2:
                audio[:fade] *= np.linspace(0, 1, fade)[:, np.newaxis]
                audio[-fade:] *= np.linspace(1, 0, fade)[:, np.newaxis]

            sf.write(output_path, audio, sr)
            self._log(f"✓ Saved: {output_path}")
            return

        # ===== STANDARD SYNTHESIS (Original 4-layer system) =====
        # FOUNDATION LOG: This is the rock-solid working base ("buzzing + clear sound").
        # All advanced controls (BPM, swing, etc.) are mapped here lightly as STRICT modules
        # so they enhance without pulling full LDQ (per rebuild plan + user testing).
        self._log("🎵 Standard Synthesis active (FOUNDATION)")

        # === DETAILED FOUNDATION LOGGING (for methodical debugging) ===
        use_bpm_influence = True  # strict module: light tempo effect even in pure factory
        effective_bpm = cfg.get("tempo_bpm", 140)
        effective_swing = cfg.get("swing", 0.0)
        self._log(f"  LOG: PATH=STANDARD | bpm={effective_bpm} | swing={effective_swing} | seed={cfg.get('random_seed')} | noise_lowpass={cfg.get('noise_lowpass_hz')}")
        self._log(f"  LOG: PRESET_VOLS noise={cfg.get('noise_vol'):.3f} drone={cfg.get('drone_vol'):.3f} note={cfg.get('note_vol'):.3f} glitch={cfg.get('glitch_vol'):.3f}")

        # Collections for stems
        audio_noise = np.zeros((n_total, 2), dtype=np.float32)
        audio_drone = np.zeros((n_total, 2), dtype=np.float32)
        audio_melody = np.zeros((n_total, 2), dtype=np.float32)
        audio_glitch = np.zeros((n_total, 2), dtype=np.float32)
        melody_events = []

        # Layer 1: Texture Floor (Noise)
        if features["edge_density"] > 0.007:
            noise = self._generate_tone(0, duration, "noise")
            if cfg["noise_lowpass_hz"] > 0:
                noise = self._fft_lowpass(noise, cfg["noise_lowpass_hz"])
            noise = self._apply_envelope(noise, cfg["drone_attack"], cfg["drone_decay"])
            vol = min(features["edge_density"] * 1.0, cfg["noise_vol"])
            stereo_noise = self._stereo_pan(noise, 0.0) * vol
            audio += stereo_noise
            audio_noise += stereo_noise

        # Layer 2: Drones (Lines)
        for i, line in enumerate(features["lines"][:cfg["max_drones"]]):
            x1, _, x2, _ = line[0]
            length = math.hypot(x2 - x1, 0)
            detune = (i * 0.7) if cfg["random_seed"] is not None else 0
            freq = root + (max(1, int(length / 48)) * theta * 0.55) + detune
            tone = self._generate_tone(freq, duration, "sawtooth")
            tone = self._apply_envelope(tone, cfg["drone_attack"], cfg["drone_decay"])
            pan = (x1 / w) * 2 - 1
            stereo_drone = self._stereo_pan(tone, pan) * cfg["drone_vol"]
            audio += stereo_drone
            audio_drone += stereo_drone

        # Layer 3: Contours → Melody
        valid = [c for c in features["contours"] if 90 < cv2.contourArea(c) < (w * h * 0.6)]
        valid.sort(key=lambda c: cv2.boundingRect(c)[0])

        # STRICT BPM MODULE (light, foundation-safe): scale timing by tempo so BPM slider does audible work
        # without full LDQ TempoGrid. This is the "mapped perfectly" enhancement.
        tempo_factor = 120.0 / max(1, effective_bpm)   # >1 for slower BPM = longer spacing
        swing_amount = effective_swing

        for i, cnt in enumerate(valid[:cfg["max_notes"]]):
            area = cv2.contourArea(cnt)
            verts = len(cv2.approxPolyDP(cnt, 0.04 * cv2.arcLength(cnt, True), True))
            freq = (root * 3.7) + (verts * theta * 1.6)
            dur = min(2.6, 0.22 + (area / 13500))
            tone = self._generate_tone(freq, dur, "sine")
            tone = self._apply_envelope(tone, cfg["note_attack"], cfg["note_decay"])

            M = cv2.moments(cnt)
            cx = int(M["m10"] / M["m00"]) if M["m00"] != 0 else cv2.boundingRect(cnt)[0]
            start = (cx / w) * (duration - dur)

            # Apply light BPM + swing module (deterministic but controllable)
            if use_bpm_influence:
                start = start * tempo_factor
                if swing_amount > 0:
                    start += (i % 3 - 1) * swing_amount * 0.08   # micro humanizing swing

            idx = int(start * sr)
            end = min(idx + len(tone), n_total)
            pan = (cx / w) * 2 - 1
            stereo_note = self._stereo_pan(tone[:end-idx], pan) * cfg["note_vol"]
            audio[idx:end] += stereo_note
            audio_melody[idx:end] += stereo_note
            melody_events.append((freq, dur, start))

            if i == 0:
                self._log(f"  LOG: MELODY_MODULE first_event_start={start:.2f}s (bpm_factor={tempo_factor:.2f}, swing={swing_amount})")

        # Layer 4: Glitch / Micro events
        for i, kp in enumerate(features["keypoints"][:cfg["max_glitches"]]):
            x, y = kp.pt
            freq = root * 13.5 + (y % 85) * 1.4
            tone = self._generate_tone(freq, 0.05, "sine")
            tone = self._apply_envelope(tone, cfg["glitch_attack"], cfg["glitch_decay"])
            start = (y / h) * (duration - 0.05)
            idx = int(start * sr)
            end = min(idx + len(tone), n_total)
            pan = (x / w) * 2 - 1
            stereo_glitch = self._stereo_pan(tone[:end-idx], pan) * cfg["glitch_vol"]
            audio[idx:end] += stereo_glitch
            audio_glitch[idx:end] += stereo_glitch

        # Final polish
        audio = self._soft_limit(audio)
        fade = int(cfg["global_fade"] * sr)
        if fade > 0 and n_total > fade * 2:
            audio[:fade] *= np.linspace(0, 1, fade)[:, np.newaxis]
            audio[-fade:] *= np.linspace(1, 0, fade)[:, np.newaxis]

        peak = np.max(np.abs(audio))
        if peak > 0:
            audio = audio / peak * 0.97

        sf.write(output_path, audio, sr)
        self._log(f"✓ Saved: {output_path}  |  Peak: {peak:.3f}")
        self._log(f"  LOG_WORKING: STANDARD_FOUNDATION | effective_bpm={effective_bpm} | events={len(melody_events)} | peak={peak:.3f} | (bpm/swing now influence timing — test with extreme values)")

        # Export Stems
        if cfg.get("export_stems"):
            base = output_path.replace(".wav", "")
            for stem, name in [(audio_noise, "noise"), (audio_drone, "drone"),
                               (audio_melody, "melody"), (audio_glitch, "glitch")]:
                max_val = np.max(np.abs(stem))
                if max_val > 0:
                    stem = stem / max_val * 0.97
                sf.write(f"{base}_{name}.wav", stem, sr)
                self._log(f"✓ Stem saved: {base}_{name}.wav")

        # Export MIDI
        if cfg.get("export_midi") and melody_events:
            mid = MidiFile()
            track = MidiTrack()
            mid.tracks.append(track)
            ticks_per_beat = 480
            tempo = cfg.get("tempo_bpm", 120)
            tick_offset = 0
            for freq, dur, start in melody_events:
                midi_note = self._freq_to_midi(freq)
                duration_ticks = int(dur * ticks_per_beat * (tempo / 60))
                start_ticks = int(start * ticks_per_beat * (tempo / 60))
                track.append(Message('note_on', note=midi_note, velocity=64, time=start_ticks - tick_offset))
                track.append(Message('note_off', note=midi_note, velocity=64, time=duration_ticks))
                tick_offset = start_ticks + duration_ticks
            mid_path = output_path.replace(".wav", ".mid")
            mid.save(mid_path)
            self._log(f"✓ MIDI saved: {mid_path}")

    def process(self, image_path: str, output_path: str):
        self.image_path = image_path
        self._log(f"\n╔════════════════════════════════════════════╗")
        self._log(f"║           LYGO Resonance Engine v{__version__}      ║")
        self._log(f"║     Image → Living Stereo Soundscape       ║")
        self._log(f"╚════════════════════════════════════════════╝\n")
        self._log(f"Analyzing: {image_path}")
        features = self.analyze_image(image_path)
        self.synthesize(features, output_path)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("image", help="Input image path")
    parser.add_argument("-o", "--output", default=None)
    parser.add_argument("--duration", type=float, default=15.0)
    parser.add_argument("--style", choices=list(PRESETS.keys()), default="cinematic")
    parser.add_argument("--seed", type=int, default=None)
    parser.add_argument("--noise-filter", type=float, default=None)
    parser.add_argument("--stems", action="store_true")
    parser.add_argument("--midi", action="store_true")
    parser.add_argument("--batch", action="store_true")
    parser.add_argument("--quiet", action="store_true")
    parser.add_argument("--ldq", action="store_true")
    parser.add_argument("--genre", choices=["None", "Dubstep", "Phonk", "Industrial"], default="None")
    parser.add_argument("--percussion", choices=["standard", "ldq"], default="standard")
    parser.add_argument("--polish", type=float, default=0.5)
    parser.add_argument("--bpm", type=int, default=140)
    parser.add_argument("--swing", type=float, default=0.0)
    args = parser.parse_args()

    config = {
        "duration": args.duration,
        "random_seed": args.seed,
        "verbose": not args.quiet,
        "export_stems": args.stems,
        "export_midi": args.midi,
        "use_ldq": args.ldq,
        "genre_manifold": args.genre,
        "percussion_mode": args.percussion,
        "perceptual_polish": args.polish,
        "tempo_bpm": args.bpm,
        "swing": args.swing,
    }
    if args.noise_filter is not None:
        config["noise_lowpass_hz"] = args.noise_filter

    preset = PRESETS.get(args.style, {})
    config.update(preset)

    if args.batch:
        folder = Path(args.image)
        if not folder.is_dir():
            print("Error: --batch requires a folder path")
            return
        images = list(folder.glob("*.jpg")) + list(folder.glob("*.png")) + list(folder.glob("*.jpeg"))
        if not images:
            print("No images found")
            return
        for img in images:
            print(f"\nProcessing: {img.name}")
            out_path = f"resonance_{img.stem}.wav"
            engine = ResonanceEngine(config)
            engine.process(str(img), out_path)
        return

    out_path = args.output or f"resonance_{Path(args.image).stem}.wav"
    engine = ResonanceEngine(config)
    engine.process(args.image, out_path)


if __name__ == "__main__":
    main()