shared/utils/notification_sound.py

"""Add commentMore actions
Notification sounds for Wan2GP video generation application
Pure Python audio notification system with multiple backend support
"""

import os
import sys
import threading
import time
import numpy as np

os.environ["PYGAME_HIDE_SUPPORT_PROMPT"] = "1"

_cached_waveforms = {}
_sample_rate = 44100
_mixer_initialized = False
_mixer_lock = threading.Lock()

def _generate_notification_beep(volume=50, sample_rate=_sample_rate):
    """Generate pleasant C major chord notification sound"""
    if volume == 0:
        return np.array([])

    volume = max(0, min(100, volume))

    # Volume curve mapping
    if volume <= 25:
        volume_mapped = (volume / 25.0) * 0.5
    elif volume <= 50:
        volume_mapped = 0.5 + ((volume - 25) / 25.0) * 0.25
    elif volume <= 75:
        volume_mapped = 0.75 + ((volume - 50) / 25.0) * 0.25
    else:
        volume_mapped = 1.0 + ((volume - 75) / 25.0) * 0.05

    volume = volume_mapped

    # C major chord frequencies
    freq_c, freq_e, freq_g = 261.63, 329.63, 392.00
    duration = 0.8
    t = np.linspace(0, duration, int(sample_rate * duration), False)

    # Generate chord components
    wave = (
        np.sin(freq_c * 2 * np.pi * t) * 0.4
        + np.sin(freq_e * 2 * np.pi * t) * 0.3
        + np.sin(freq_g * 2 * np.pi * t) * 0.2
    )

    # Normalize
    max_amplitude = np.max(np.abs(wave))
    if max_amplitude > 0:
        wave = wave / max_amplitude * 0.8

    # ADSR envelope
    def apply_adsr_envelope(wave_data):
        length = len(wave_data)
        attack_time = int(0.2 * length)
        decay_time = int(0.1 * length)
        release_time = int(0.5 * length)

        envelope = np.ones(length)

        if attack_time > 0:
            envelope[:attack_time] = np.power(np.linspace(0, 1, attack_time), 3)

        if decay_time > 0:
            start_idx, end_idx = attack_time, attack_time + decay_time
            envelope[start_idx:end_idx] = np.linspace(1, 0.85, decay_time)

        if release_time > 0:
            start_idx = length - release_time
            envelope[start_idx:] = 0.85 * np.exp(-4 * np.linspace(0, 1, release_time))

        return wave_data * envelope

    wave = apply_adsr_envelope(wave)

    # Simple low-pass filter
    def simple_lowpass_filter(signal, cutoff_ratio=0.8):
        window_size = max(3, int(len(signal) * 0.001))
        if window_size % 2 == 0:
            window_size += 1

        kernel = np.ones(window_size) / window_size
        padded = np.pad(signal, window_size // 2, mode="edge")
        filtered = np.convolve(padded, kernel, mode="same")
        return filtered[window_size // 2 : -window_size // 2]

    wave = simple_lowpass_filter(wave)

    # Add reverb
    if len(wave) > sample_rate // 4:
        delay_samples = int(0.12 * sample_rate)
        reverb = np.zeros_like(wave)
        reverb[delay_samples:] = wave[:-delay_samples] * 0.08
        wave = wave + reverb

    # Apply volume & final normalize
    wave = wave * volume * 0.5
    max_amplitude = np.max(np.abs(wave))
    if max_amplitude > 0.85:
        wave = wave / max_amplitude * 0.85

    return wave

def _get_cached_waveform(volume):
    """Return cached waveform for volume"""
    if volume not in _cached_waveforms:
        _cached_waveforms[volume] = _generate_notification_beep(volume)
    return _cached_waveforms[volume]


def play_audio_with_pygame(audio_data, sample_rate=_sample_rate):
    """Play audio with pygame backend"""
    global _mixer_initialized
    try:
        import pygame

        with _mixer_lock:
            if not _mixer_initialized:
                pygame.mixer.pre_init(frequency=sample_rate, size=-16, channels=2, buffer=512)
                pygame.mixer.init()
                _mixer_initialized = True

            mixer_info = pygame.mixer.get_init()
            if mixer_info is None or mixer_info[2] != 2:
                return False

            audio_int16 = (audio_data * 32767).astype(np.int16)
            if len(audio_int16.shape) > 1:
                audio_int16 = audio_int16.flatten()

            stereo_data = np.zeros((len(audio_int16), 2), dtype=np.int16)
            stereo_data[:, 0] = audio_int16
            stereo_data[:, 1] = audio_int16

            sound = pygame.sndarray.make_sound(stereo_data)
            pygame.mixer.stop()
            sound.play()

            duration_ms = int(len(audio_data) / sample_rate * 1000) + 50
            pygame.time.wait(duration_ms)

            return True

    except ImportError:
        return False
    except Exception as e:
        print(f"Pygame error: {e}")
        return False

def play_audio_with_sounddevice(audio_data, sample_rate=_sample_rate):
    """Play audio using sounddevice backend"""
    try:
        import sounddevice as sd
        sd.play(audio_data, sample_rate)
        sd.wait()
        return True
    except ImportError:
        return False
    except Exception as e:
        print(f"Sounddevice error: {e}")
        return False

def play_audio_with_winsound(audio_data, sample_rate=_sample_rate):
    """Play audio using winsound backend (Windows only)"""
    if sys.platform != "win32":
        return False
    try:
        import winsound, wave, tempfile, uuid

        temp_dir = tempfile.gettempdir()
        temp_filename = os.path.join(temp_dir, f"notification_{uuid.uuid4().hex}.wav")

        try:
            with wave.open(temp_filename, "w") as wav_file:
                wav_file.setnchannels(1)
                wav_file.setsampwidth(2)
                wav_file.setframerate(sample_rate)
                audio_int16 = (audio_data * 32767).astype(np.int16)
                wav_file.writeframes(audio_int16.tobytes())

            winsound.PlaySound(temp_filename, winsound.SND_FILENAME)

        finally:
            try:
                if os.path.exists(temp_filename):
                    os.unlink(temp_filename)
            except:
                pass

        return True
    except ImportError:
        return False
    except Exception as e:
        print(f"Winsound error: {e}")
        return False

def play_notification_sound(volume=50):
    """Play notification sound with specified volume"""
    if volume == 0:
        return

    audio_data = _get_cached_waveform(volume)
    if len(audio_data) == 0:
        return

    audio_backends = [play_audio_with_pygame, play_audio_with_sounddevice, play_audio_with_winsound]
    for backend in audio_backends:
        try:
            if backend(audio_data):
                return
        except Exception:
            continue

    print("All audio backends failed, using terminal beep")
    print("\a")

def play_notification_async(volume=50):
    """Play notification sound asynchronously (non-blocking)"""
    def play_sound():
        try:
            play_notification_sound(volume)
        except Exception as e:
            print(f"Error playing notification sound: {e}")

    threading.Thread(target=play_sound, daemon=True).start()

def notify_video_completion(video_path=None, volume=50):
    """Notify about completed video generation"""
    play_notification_async(volume)

for vol in (25, 50, 75, 100):
    _get_cached_waveform(vol)