diff --git a/src/focus/audio/pipeline.py b/src/focus/audio/pipeline.py index f3aadf0..a5b7582 100644 --- a/src/focus/audio/pipeline.py +++ b/src/focus/audio/pipeline.py @@ -80,6 +80,7 @@ async def _process_loop(self) -> None: target_freq=self.profile.modulation_freq, depth=self.profile.modulation_depth, state=self._mod_state, + band_cutoff_hz=self.profile.modulation_band_hz, ) # 2. Apply spatialization diff --git a/src/focus/cli.py b/src/focus/cli.py index 91ba030..784a6c2 100644 --- a/src/focus/cli.py +++ b/src/focus/cli.py @@ -97,6 +97,12 @@ def show_profiles(): default=None, help="Override modulation depth (0.0-1.0)", ) +@click.option( + "--band", + type=float, + default=None, + help="Override modulated low-band cutoff in Hz (0 = full-spectrum modulation)", +) @click.option( "--prompt", type=str, @@ -163,6 +169,7 @@ def start_session( profile: str, frequency: float | None, depth: float | None, + band: float | None, prompt: str | None, mock: bool, duration: int | None, @@ -188,6 +195,7 @@ def start_session( profile=profile, frequency=frequency, depth=depth, + band=band, prompt=prompt, mock=mock, duration=duration, @@ -205,6 +213,7 @@ def launch_session( profile: str, frequency: float | None = None, depth: float | None = None, + band: float | None = None, prompt: str | None = None, mock: bool = False, duration: int | None = None, @@ -235,13 +244,19 @@ def launch_session( sys.exit(1) # Apply overrides - if frequency or depth or prompt: + if frequency or depth or band is not None or prompt: + if band is None: + band_hz = focus_profile.modulation_band_hz + else: + # 0 (or negative) disables band-limiting -> full-spectrum modulation + band_hz = None if band <= 0 else band focus_profile = FocusProfile( name=focus_profile.name, description=focus_profile.description, prompt=prompt or focus_profile.prompt, modulation_freq=frequency or focus_profile.modulation_freq, modulation_depth=depth or focus_profile.modulation_depth, + modulation_band_hz=band_hz, bpm=focus_profile.bpm, density=focus_profile.density, brightness=focus_profile.brightness, @@ -538,6 +553,7 @@ async def _spectrum_loop(): target_freq=profile.modulation_freq, depth=profile.modulation_depth, state=mod_state, + band_cutoff_hz=profile.modulation_band_hz, ) # Apply fade-in to early chunks diff --git a/src/focus/dsp/entrainment.py b/src/focus/dsp/entrainment.py index 23b0f0d..f64ceb6 100644 --- a/src/focus/dsp/entrainment.py +++ b/src/focus/dsp/entrainment.py @@ -4,17 +4,39 @@ through rapid amplitude modulation in the Beta frequency range (12-20 Hz). """ -from dataclasses import dataclass +from dataclasses import dataclass, field import numpy as np +# scipy is a core dependency, but guard it so the DSP degrades gracefully to +# full-spectrum modulation if it is ever missing (mirrors the numpy/genai +# availability-flag convention used elsewhere in the package). +try: + from scipy.signal import butter, lfilter + + SCIPY_AVAILABLE = True +except ImportError: + SCIPY_AVAILABLE = False + +# Order of the Butterworth low-pass used to isolate the modulation band. +_BAND_FILTER_ORDER = 2 + @dataclass class ModulationState: - """Maintains phase continuity between audio chunks.""" + """Maintains phase (and band-split filter) continuity between audio chunks.""" phase: float = 0.0 + # Low-pass filter state for band-limited modulation. These are threaded back + # into each call so the band split is click-free across chunk boundaries. + _lp_b: np.ndarray | None = field(default=None, repr=False) + _lp_a: np.ndarray | None = field(default=None, repr=False) + _lp_zi: np.ndarray | None = field(default=None, repr=False) + _lp_cutoff: float | None = field(default=None, repr=False) + _lp_sr: int | None = field(default=None, repr=False) + _lp_channels: int | None = field(default=None, repr=False) + def advance(self, samples: int, freq: float, sample_rate: int) -> None: """Advance phase by the given number of samples.""" self.phase += 2.0 * np.pi * freq * samples / sample_rate @@ -22,18 +44,57 @@ def advance(self, samples: int, freq: float, sample_rate: int) -> None: self.phase = self.phase % (2.0 * np.pi) +def _lowpass_band( + audio: np.ndarray, sample_rate: int, cutoff_hz: float, state: ModulationState +) -> np.ndarray: + """Return the low-frequency band of ``audio`` using a stateful Butterworth filter. + + Filter coefficients and delay state live on ``state`` so the split stays + continuous (no clicks) across chunk boundaries. + """ + channels = audio.shape[1] if audio.ndim == 2 else 1 + + needs_init = ( + state._lp_zi is None + or state._lp_cutoff != cutoff_hz + or state._lp_sr != sample_rate + or state._lp_channels != channels + ) + if needs_init: + nyquist = 0.5 * sample_rate + wn = min(max(cutoff_hz / nyquist, 1e-4), 0.99) + b, a = butter(_BAND_FILTER_ORDER, wn, btype="low") + state._lp_b = b + state._lp_a = a + state._lp_cutoff = cutoff_hz + state._lp_sr = sample_rate + state._lp_channels = channels + zi_len = max(len(a), len(b)) - 1 + if audio.ndim == 2: + state._lp_zi = np.zeros((zi_len, channels), dtype=np.float64) + else: + state._lp_zi = np.zeros(zi_len, dtype=np.float64) + + low, state._lp_zi = lfilter(state._lp_b, state._lp_a, audio, axis=0, zi=state._lp_zi) + return low + + def apply_entrainment( audio: np.ndarray, sample_rate: int, target_freq: float = 15.0, - depth: float = 0.3, + depth: float = 0.15, state: ModulationState | None = None, + band_cutoff_hz: float | None = 500.0, ) -> tuple[np.ndarray, ModulationState]: """ Apply amplitude modulation for neural entrainment. - The modulation creates a subtle "tremolo" effect that oscillates at the - target frequency, inducing neural phase locking in the listener. + By default the modulation is *band-limited*: only the low-frequency band + (below ``band_cutoff_hz``) is amplitude-modulated, while the mids/highs that + carry the perceived melody pass through untouched. This keeps the + entrainment pulse working in the background (felt as gentle rhythmic energy) + without the whole mix audibly "throbbing" as a tremolo. Args: audio: Input audio array, shape (samples,) for mono or (samples, channels) for stereo. @@ -43,9 +104,11 @@ def apply_entrainment( Higher frequencies (18-20 Hz) for intense focus, lower (12-14 Hz) for light concentration. depth: Modulation depth from 0.0 (no effect) to 1.0 (full modulation). - Recommended range is 0.2-0.4 for noticeable but non-distracting effect. - state: Optional state object for phase continuity between chunks. + Recommended range is ~0.1-0.2 for a subtle, non-distracting effect. + state: Optional state object for phase (and filter) continuity between chunks. Pass the returned state to subsequent calls to prevent clicks. + band_cutoff_hz: Upper edge of the modulated low band in Hz. Set to ``None`` + (or if scipy is unavailable) to modulate the full spectrum. Returns: Tuple of (modulated_audio, state). The state should be passed to the next @@ -54,7 +117,7 @@ def apply_entrainment( Example: >>> state = ModulationState() >>> for chunk in audio_chunks: - ... modulated, state = apply_entrainment(chunk, 48000, 15.0, 0.3, state) + ... modulated, state = apply_entrainment(chunk, 48000, 15.0, 0.15, state) ... play(modulated) """ if state is None: @@ -66,17 +129,25 @@ def apply_entrainment( t = np.arange(n_samples) / sample_rate phase_array = 2.0 * np.pi * target_freq * t + state.phase - # Create modulation envelope: oscillates between (1-depth) and 1.0 - # Using (1 - depth/2) + (depth/2) * cos(...) gives range [1-depth, 1] - # This ensures we only reduce volume, never amplify + # Create modulation envelope: oscillates between (1-depth) and 1.0. + # (1 - depth) + depth * (0.5 * (1 + cos(...))) gives range [1-depth, 1], + # so we only ever reduce volume, never amplify. modulator = (1.0 - depth) + depth * (0.5 * (1.0 + np.cos(phase_array))) # Reshape modulator for stereo audio if audio.ndim == 2: modulator = modulator[:, np.newaxis] - # Apply modulation - modulated = audio * modulator + use_band = band_cutoff_hz is not None and SCIPY_AVAILABLE and depth > 0.0 and n_samples > 0 + if use_band: + # Split into the low modulation band and the untouched remainder, then + # modulate only the low band and recombine. + low = _lowpass_band(audio, sample_rate, band_cutoff_hz, state) + rest = audio - low + modulated = low * modulator + rest + else: + # Full-spectrum modulation (fallback when band-limiting is disabled). + modulated = audio * modulator # Update state for next chunk state.advance(n_samples, target_freq, sample_rate) diff --git a/src/focus/generation/lyria_client.py b/src/focus/generation/lyria_client.py index 0868c5a..26fcf7d 100644 --- a/src/focus/generation/lyria_client.py +++ b/src/focus/generation/lyria_client.py @@ -3,10 +3,12 @@ This module provides a WebSocket client for the Lyria RealTime API, enabling real-time streaming of AI-generated instrumental music. -Implements proactive session rotation to avoid the 10-minute API limit. +Implements overlapping session rotation to avoid the 10-minute API limit +while keeping track-to-track transitions seamless (no audible gap). """ import asyncio +import contextlib import os import time import warnings @@ -16,50 +18,78 @@ import numpy as np # Session rotation constants -# Lyria has a 10-minute session limit; rotate proactively at 9 minutes +# Lyria has a 10-minute session limit; rotate proactively before that. SESSION_MAX_DURATION_SECONDS = 9 * 60 # 9 minutes -CROSSFADE_DURATION_SECONDS = 3.0 # Seconds to buffer for seamless transition +# Seamless rotation: the next session is opened OVERLAP_SECONDS before the +# current one hits its rotation point, warmed up, then the two live streams are +# equal-power crossfaded over CROSSFADE_DURATION_SECONDS. Because the 10-minute +# cap is per-WebSocket, two concurrent sessions overlap without any audio gap. +OVERLAP_SECONDS = 8.0 +CROSSFADE_DURATION_SECONDS = 4.0 +WARMUP_MIN_SECONDS = 1.5 # Buffered new-session audio before crossfading -def _apply_crossfade(old_audio: np.ndarray, new_audio: np.ndarray, sample_rate: int) -> np.ndarray: - """Apply crossfade between two audio chunks for seamless transition. - Args: - old_audio: Tail of the previous session's audio (stereo, float32) - new_audio: Head of the new session's audio (stereo, float32) - sample_rate: Audio sample rate (e.g., 48000) - - Returns: - Crossfaded audio chunk. - """ - crossfade_samples = int(CROSSFADE_DURATION_SECONDS * sample_rate) - - # Ensure we have enough samples for crossfade - old_samples = min(len(old_audio), crossfade_samples) - new_samples = min(len(new_audio), crossfade_samples) - overlap_samples = min(old_samples, new_samples) +# Fraction of the crossfade window over which the OUTGOING track fades to +# silence. Keeping this below 1.0 makes the last part of the crossfade the +# incoming track alone, which avoids a faint "ghost" of the old track (and two +# basslines stacking/beating) right before the switch completes. +CROSSFADE_OUT_END_FRAC = 0.7 - if overlap_samples == 0: - return new_audio - # Create fade curves (linear for simplicity and low CPU) - fade_out = np.linspace(1.0, 0.0, overlap_samples, dtype=np.float32) - fade_in = np.linspace(0.0, 1.0, overlap_samples, dtype=np.float32) +def _crossfade_gains( + start: int, length: int, total: int, out_end_frac: float = CROSSFADE_OUT_END_FRAC +) -> tuple[np.ndarray, np.ndarray]: + """Crossfade gains for positions ``[start, start+length)`` of ``total``. - # Apply fades to stereo audio - if old_audio.ndim == 2: - fade_out = fade_out[:, np.newaxis] - fade_in = fade_in[:, np.newaxis] + The incoming track rises with an equal-power sine across the full window, + while the outgoing track falls with a cosine that reaches silence by + ``out_end_frac`` of the window (and stays there). With ``out_end_frac == 1.0`` + this is a symmetric equal-power crossfade (``fade_out**2 + fade_in**2 == 1``); + smaller values pull the outgoing track out earlier for a cleaner handover. - # Blend the overlapping region - old_tail = old_audio[-overlap_samples:] * fade_out - new_head = new_audio[:overlap_samples] * fade_in - blended = old_tail + new_head - - # Return: blended overlap + rest of new audio - if overlap_samples < len(new_audio): - return np.concatenate([blended, new_audio[overlap_samples:]], axis=0) - return blended + Returns ``(fade_out, fade_in)``. + """ + idx = np.arange(start, start + length, dtype=np.float64) + frac = np.clip(idx / max(total, 1), 0.0, 1.0) + fade_in = np.sin(frac * (0.5 * np.pi)) + out_frac = np.clip(frac / max(out_end_frac, 1e-6), 0.0, 1.0) + fade_out = np.cos(out_frac * (0.5 * np.pi)) + return fade_out, fade_in + + +class _ChunkBuffer: + """A small FIFO of audio chunks with sample-accurate ``take``.""" + + def __init__(self) -> None: + self._chunks: list[np.ndarray] = [] + self.total: int = 0 + + def add(self, chunk: np.ndarray | None) -> None: + if chunk is not None and len(chunk) > 0: + self._chunks.append(chunk) + self.total += len(chunk) + + def take(self, n: int) -> np.ndarray: + """Pop and return the first ``n`` samples (concatenated) from the buffer.""" + n = min(n, self.total) + out: list[np.ndarray] = [] + got = 0 + while got < n and self._chunks: + head = self._chunks[0] + remaining = n - got + if len(head) <= remaining: + out.append(head) + got += len(head) + self._chunks.pop(0) + else: + out.append(head[:remaining]) + self._chunks[0] = head[remaining:] + got += remaining + self.total -= got + if not out: + return np.zeros((0, 2), dtype=np.float32) + return np.concatenate(out, axis=0) try: @@ -86,24 +116,163 @@ class LyriaConfig: channels: int = 2 +class _LiveSession: + """A single live Lyria WebSocket session with a background reader. + + The session is opened manually (not via ``async with``) so that two + sessions can be held open simultaneously during an overlapping crossfade. + A reader task pumps decoded audio chunks into an ``asyncio.Queue`` so the + consumer can pull from several sessions concurrently without blocking. + """ + + def __init__(self, client: object, config: LyriaConfig, verbose: bool = False) -> None: + self._client = client + self._config = config + self._verbose = verbose + self._cm = None + self._session = None + self._queue: asyncio.Queue | None = None + self._reader_task: asyncio.Task | None = None + self.start_time: float = 0.0 + self.error: Exception | None = None + self._closed = False + + async def open(self) -> None: + """Connect, configure, and start playback; begin buffering audio.""" + with warnings.catch_warnings(): + warnings.filterwarnings("ignore", message="Realtime music generation is experimental") + self._cm = self._client.aio.live.music.connect(model="models/lyria-realtime-exp") + self._session = await self._cm.__aenter__() + + await self._session.set_music_generation_config( + config=types.LiveMusicGenerationConfig( + bpm=self._config.bpm, + temperature=self._config.temperature, + guidance=self._config.guidance, + density=self._config.density, + brightness=self._config.brightness, + ) + ) + await self._session.set_weighted_prompts( + prompts=[types.WeightedPrompt(text=self._config.prompt, weight=1.0)] + ) + await self._session.play() + + self.start_time = time.monotonic() + self._queue = asyncio.Queue(maxsize=64) + self._reader_task = asyncio.create_task(self._read_loop()) + + def _parse(self, message) -> np.ndarray | None: + """Extract a normalized float32 audio chunk from a server message.""" + if ( + hasattr(message, "server_content") + and message.server_content + and hasattr(message.server_content, "audio_chunks") + and message.server_content.audio_chunks + ): + audio_data = message.server_content.audio_chunks[0].data + audio_int16 = np.frombuffer(audio_data, dtype=np.int16) + audio_float = audio_int16.astype(np.float32) / 32768.0 + if self._config.channels == 2 and len(audio_float) >= 2: + audio_float = audio_float.reshape(-1, 2) + return audio_float + return None + + async def _read_loop(self) -> None: + """Pump decoded audio into the queue until the session ends or errors.""" + try: + async for message in self._session.receive(): + audio = self._parse(message) + if audio is not None: + await self._queue.put(audio) + except asyncio.CancelledError: + raise + except Exception as e: # surfaced to the consumer via .error + self.error = e + finally: + # Sentinel so consumers waiting on get() always wake up. + with contextlib.suppress(Exception): + self._queue.put_nowait(None) + + @property + def elapsed(self) -> float: + return time.monotonic() - self.start_time + + def drain_available(self, buffer: _ChunkBuffer) -> bool: + """Move all immediately-available chunks into ``buffer`` (non-blocking). + + Returns True if the end-of-stream sentinel was seen. + """ + ended = False + if self._queue is None: + return True + while True: + try: + chunk = self._queue.get_nowait() + except asyncio.QueueEmpty: + break + if chunk is None: + ended = True + break + buffer.add(chunk) + return ended + + async def get(self) -> np.ndarray | None: + """Await the next audio chunk, or None when the session has ended.""" + if self._queue is None: + return None + return await self._queue.get() + + async def set_prompt(self, prompt: str) -> None: + if self._session: + try: + await self._session.set_weighted_prompts( + prompts=[types.WeightedPrompt(text=prompt, weight=1.0)] + ) + except Exception: + pass # best-effort + + async def close(self) -> None: + if self._closed: + return + self._closed = True + if self._reader_task is not None: + self._reader_task.cancel() + try: + await self._reader_task + except (asyncio.CancelledError, Exception): + pass + self._reader_task = None + if self._session is not None: + try: + await self._session.stop() + except Exception: + pass + if self._cm is not None: + try: + await self._cm.__aexit__(None, None, None) + except Exception: + pass + self._cm = None + self._session = None + + @dataclass class LyriaClient: """Client for Google Lyria RealTime API. - Uses WebSocket connection for real-time streaming music generation. + Uses WebSocket connections for real-time streaming music generation. Audio is streamed as 16-bit PCM at 48kHz. - Implements proactive session rotation to avoid the 10-minute - API limit which causes audio degradation. + Rotates sessions before the 10-minute API limit using an overlapping + dual-session crossfade so track transitions are seamless (no audio gap). """ config: LyriaConfig session_duration: int = SESSION_MAX_DURATION_SECONDS # Configurable rotation time _client: object = field(default=None, init=False, repr=False) - _session: object = field(default=None, init=False, repr=False) + _current: _LiveSession | None = field(default=None, init=False, repr=False) _running: bool = field(default=False, init=False) - _session_start_time: float = field(default=0.0, init=False, repr=False) - _crossfade_buffer: np.ndarray | None = field(default=None, init=False, repr=False) _session_count: int = field(default=0, init=False, repr=False) verbose: bool = field(default=False, init=True) @@ -129,203 +298,154 @@ async def connect(self, api_key: str | None = None) -> None: self._client = genai.Client(api_key=api_key, http_options={"api_version": "v1alpha"}) self._running = True + async def _open_with_retry(self) -> _LiveSession | None: + """Open a new live session, retrying transient errors. + + Returns the session, or None if the caller should fall back to synth + (model unreachable / retries exhausted / non-retryable error). + """ + max_retries = 3 + base_delay = 1.0 + retry_count = 0 + + while self._running and retry_count <= max_retries: + try: + self._session_count += 1 + if self.verbose: + if retry_count > 0: + print(f" [Lyria] Retry attempt {retry_count}/{max_retries}...") + elif self._session_count > 1: + print(f" [Lyria] Starting session #{self._session_count} (rotation)...") + else: + print(" [Lyria] Connecting to live session...") + + session = _LiveSession(self._client, self.config, verbose=self.verbose) + await session.open() + if self.verbose: + print(" [Lyria] Session connected, receiving audio...") + return session + + except Exception as e: + error_msg = str(e) + + # Non-retryable: model not found -> fall back to synth + if "404" in error_msg or "not found" in error_msg.lower(): + if self.verbose: + print(f" ⚠️ Lyria model unreachable ({error_msg})") + print(" 🔄 Falling back to Enhanced Synth engine...") + return None + + is_retryable = any( + indicator in error_msg.lower() + for indicator in [ + "1011", + "service", + "connection", + "websocket", + "timeout", + "closed", + ] + ) + if is_retryable and retry_count < max_retries: + retry_count += 1 + delay = base_delay * (2 ** (retry_count - 1)) + if self.verbose: + shown = error_msg if len(error_msg) <= 80 else f"{error_msg[:80]}..." + print(f" ⚠️ Lyria connection error: {shown}") + print(f" 🔄 Retrying in {delay:.1f}s ({retry_count}/{max_retries})...") + await asyncio.sleep(delay) + continue + + if self.verbose: + if retry_count >= max_retries: + print(f" ❌ Lyria retries exhausted after {max_retries} attempts") + else: + print(f" ❌ Lyria error (non-retryable): {error_msg}") + print(" 🔄 Falling back to Enhanced Synth engine...") + return None + + return None + async def generate_stream(self) -> AsyncIterator[np.ndarray]: - """Generate continuous music stream with automatic session rotation. + """Generate a continuous music stream with seamless session rotation. Yields: Audio chunks as numpy arrays, shape (samples, 2) for stereo, dtype float32, normalized to [-1, 1]. Note: - Proactively rotates sessions at 9 minutes to avoid the 10-minute - API limit which causes audio degradation. Crossfades between - sessions for seamless transitions. + The next session is opened ahead of time and equal-power crossfaded + with the current one, so rotations produce no audible gap. Falls + back to :class:`EnhancedSynthClient` if Lyria is unreachable. """ if not self._client: raise RuntimeError("Client not connected. Call connect() first.") - max_retries = 3 + sample_rate = self.config.sample_rate + rotate_at = self.session_duration - min( + OVERLAP_SECONDS, max(1.0, self.session_duration * 0.15) + ) + use_fallback = False + current = await self._open_with_retry() + if current is None: + use_fallback = True + + try: + while self._running and not use_fallback: + self._current = current + + # --- Steady playback until the rotation point --- + rotate = False + while self._running: + chunk = await current.get() + if chunk is None: # session ended (naturally or via error) + break + if current.elapsed >= rotate_at: + rotate = True + break + yield chunk - # Main loop: rotates sessions when approaching time limit - while self._running and not use_fallback: - retry_count = 0 - base_delay = 1.0 - session_needs_rotation = False + if not self._running: + break - while retry_count <= max_retries and self._running and not session_needs_rotation: - try: - self._session_count += 1 - if self.verbose: - if retry_count > 0: - print(f" [Lyria] Retry attempt {retry_count}/{max_retries}...") - elif self._session_count > 1: - print( - f" [Lyria] Starting session #{self._session_count} (rotation)..." - ) - else: - print(" [Lyria] Connecting to live session...") - - # Connect to Lyria music model - with warnings.catch_warnings(): - warnings.filterwarnings( - "ignore", message="Realtime music generation is experimental" - ) - async with self._client.aio.live.music.connect( - model="models/lyria-realtime-exp", - ) as session: - self._session = session - self._session_start_time = time.monotonic() - retry_count = 0 # Reset on successful connection - - if self.verbose: - print(" [Lyria] Session connected") - - # Configure music generation parameters - await session.set_music_generation_config( - config=types.LiveMusicGenerationConfig( - bpm=self.config.bpm, - temperature=self.config.temperature, - guidance=self.config.guidance, - density=self.config.density, - brightness=self.config.brightness, - ) - ) - - # Set the music style prompt - await session.set_weighted_prompts( - prompts=[types.WeightedPrompt(text=self.config.prompt, weight=1.0)] - ) - - # Start playback - await session.play() - if self.verbose: - print(" [Lyria] Playback started, receiving audio...") - - # Receive audio chunks - async for message in session.receive(): - if not self._running: - return # Clean exit - - # Check session duration for rotation - elapsed = time.monotonic() - self._session_start_time - if elapsed >= self.session_duration: - if self.verbose: - print( - f" [Lyria] Session #{self._session_count} " - f"reached {elapsed:.0f}s, rotating..." - ) - session_needs_rotation = True - break # Exit receive loop to rotate - - # Check for audio data in the message - if ( - hasattr(message, "server_content") - and message.server_content - and hasattr(message.server_content, "audio_chunks") - and message.server_content.audio_chunks - ): - # Get raw audio data from the chunk - audio_data = message.server_content.audio_chunks[0].data - - # Raw 16-bit PCM audio - audio_int16 = np.frombuffer(audio_data, dtype=np.int16) - audio_float = audio_int16.astype(np.float32) / 32768.0 - - # Reshape to stereo (interleaved L/R samples) - if self.config.channels == 2 and len(audio_float) >= 2: - audio_float = audio_float.reshape(-1, 2) - - # Apply crossfade from previous session - if self._crossfade_buffer is not None: - audio_float = _apply_crossfade( - self._crossfade_buffer, - audio_float, - self.config.sample_rate, - ) - self._crossfade_buffer = None # Clear buffer - - # Buffer tail ONLY when approaching rotation - # to avoid crossfading continuous playback - time_until_rotation = self.session_duration - elapsed - if time_until_rotation <= 5.0: # Last 5 seconds - crossfade_samples = int( - CROSSFADE_DURATION_SECONDS * self.config.sample_rate - ) - if len(audio_float) >= crossfade_samples: - tail = audio_float[-crossfade_samples:] - self._crossfade_buffer = tail.copy() - - yield audio_float - - # If session ended cleanly without rotation, we're done - if not session_needs_rotation: - return - - except Exception as e: - error_msg = str(e) - - # Check for non-retryable errors (model not found) - if "404" in error_msg or "not found" in error_msg.lower(): - if self.verbose: - print(f" ⚠️ Lyria model unreachable ({error_msg})") - print(" 🔄 Falling back to Enhanced Synth engine...") + if not rotate: + # Session ended unexpectedly; reconnect (no crossfade + # possible since the old stream is already gone). + await current.close() + current = await self._open_with_retry() + if current is None: use_fallback = True break + continue + + # --- Open the next session and crossfade into it --- + new = await self._open_with_retry() + if new is None: + # Could not start the next session; keep the current one + # playing to the end, then fall back. + async for chunk in self._drain_to_end(current): + if not self._running: + break + yield chunk + await current.close() + use_fallback = True + break - # Check for retryable errors - is_retryable = any( - indicator in error_msg.lower() - for indicator in [ - "1011", - "service", - "connection", - "websocket", - "timeout", - "closed", - ] - ) - - if is_retryable and retry_count < max_retries: - retry_count += 1 - delay = base_delay * (2 ** (retry_count - 1)) - if self.verbose: - print( - f" ⚠️ Lyria connection error: {error_msg[:80]}..." - if len(error_msg) > 80 - else f" ⚠️ Lyria connection error: {error_msg}" - ) - print( - f" 🔄 Retrying in {delay:.1f}s ({retry_count}/{max_retries})..." - ) - await asyncio.sleep(delay) - continue - else: - if self.verbose: - if retry_count >= max_retries: - print(f" ❌ Lyria retries exhausted after {max_retries} attempts") - else: - print(f" ❌ Lyria error (non-retryable): {error_msg}") - print(" 🔄 Falling back to Enhanced Synth engine...") - use_fallback = True + # Route prompt changes to the incoming session from now on. + self._current = new + + async for blended in self._crossfade(current, new, sample_rate): + if not self._running: break + yield blended - # Yield transition audio during session rotation gap - # This fills the silence while the new session connects - if session_needs_rotation and self._crossfade_buffer is not None: - if self.verbose: - print(" [Lyria] Playing transition audio...") - # Apply fade-out to transition buffer and yield it - # This provides audio continuity during connection - fade_samples = len(self._crossfade_buffer) - fade_out = np.linspace(1.0, 0.3, fade_samples, dtype=np.float32) - if self._crossfade_buffer.ndim == 2: - fade_out = fade_out[:, np.newaxis] - transition_audio = self._crossfade_buffer * fade_out - yield transition_audio.astype(np.float32) - # Keep buffer for crossfade with new session's first chunk - # (but attenuated to avoid double-loud join) - self._crossfade_buffer = (self._crossfade_buffer * 0.3).astype(np.float32) + await current.close() + current = new + finally: + if current is not None: + await current.close() + self._current = None # Fallback to EnhancedSynthClient if self._running and use_fallback: @@ -336,15 +456,88 @@ async def generate_stream(self) -> AsyncIterator[np.ndarray]: break yield chunk + async def _drain_to_end(self, session: _LiveSession) -> AsyncIterator[np.ndarray]: + """Yield whatever the session still produces until it ends.""" + while self._running: + chunk = await session.get() + if chunk is None: + return + yield chunk + + async def _crossfade( + self, old: _LiveSession, new: _LiveSession, sample_rate: int + ) -> AsyncIterator[np.ndarray]: + """Warm up ``new`` while ``old`` keeps playing, then equal-power crossfade. + + Yields the (old audio, then blended, then new audio) so playback stays + continuous across the whole handover. + """ + xfade_n = int(CROSSFADE_DURATION_SECONDS * sample_rate) + warmup_n = int(WARMUP_MIN_SECONDS * sample_rate) + + new_buf = _ChunkBuffer() + new_ended = False + + # --- Warmup: keep old playing in real time while new fills its buffer --- + while self._running and new_buf.total < warmup_n and not new_ended: + new_ended = new.drain_available(new_buf) + if new_ended or new_buf.total >= warmup_n: + break + old_chunk = await old.get() + if old_chunk is None: + break + yield old_chunk + + # --- Crossfade: blend equal amounts of old and new, sample-accurate --- + old_buf = _ChunkBuffer() + pos = 0 + old_ended = False + while self._running and pos < xfade_n: + new_ended = new.drain_available(new_buf) or new_ended + + if old_buf.total == 0 and not old_ended: + old_chunk = await old.get() + if old_chunk is None: + old_ended = True + else: + old_buf.add(old_chunk) + if new_buf.total == 0 and not new_ended: + new_chunk = await new.get() + if new_chunk is None: + new_ended = True + else: + new_buf.add(new_chunk) + + if old_buf.total == 0 or new_buf.total == 0: + # One side is exhausted; stop blending and let the tail below run. + break + + avail = min(old_buf.total, new_buf.total, xfade_n - pos) + old_seg = old_buf.take(avail) + new_seg = new_buf.take(avail) + fade_out, fade_in = _crossfade_gains(pos, avail, xfade_n) + if old_seg.ndim == 2: + fade_out = fade_out[:, np.newaxis] + fade_in = fade_in[:, np.newaxis] + blended = old_seg * fade_out + new_seg * fade_in + yield blended.astype(np.float32) + pos += avail + + # --- Emit any already-buffered new audio so we continue seamlessly --- + new.drain_available(new_buf) + if new_buf.total > 0: + yield new_buf.take(new_buf.total) + async def stop(self) -> None: """Stop the generation stream.""" self._running = False - if self._session: + current = self._current + if current is not None: try: - await self._session.stop() + await current.close() except Exception: pass - self._session = None + self._current = None async def set_prompt(self, new_prompt: str) -> None: """Change the music generation prompt during playback. @@ -352,13 +545,9 @@ async def set_prompt(self, new_prompt: str) -> None: Args: new_prompt: The new prompt to guide music generation. """ - if self._session: - try: - await self._session.set_weighted_prompts( - prompts=[types.WeightedPrompt(text=new_prompt, weight=1.0)] - ) - except Exception: - pass # Ignore errors, prompt change is best-effort + current = self._current + if current is not None: + await current.set_prompt(new_prompt) class EnhancedSynthClient: diff --git a/src/focus/profiles.py b/src/focus/profiles.py index daf03ad..277051a 100644 --- a/src/focus/profiles.py +++ b/src/focus/profiles.py @@ -22,6 +22,10 @@ class FocusProfile: modulation_freq: float # Hz (12-20 for Beta waves) modulation_depth: float # 0.0-1.0 + # Upper edge (Hz) of the low band that gets amplitude-modulated. Only this + # band pulses, so the effect stays subtle; None modulates the full spectrum. + modulation_band_hz: float | None = 500.0 + # Optional Lyria parameters bpm: int | None = None density: float | None = None # 0.0-1.0 @@ -43,7 +47,7 @@ class FocusProfile: "subtle evolving textures, hypnotic rhythm" ), modulation_freq=18.0, # High Beta for intense focus - modulation_depth=0.35, + modulation_depth=0.16, bpm=120, density=0.4, brightness=0.3, @@ -62,7 +66,7 @@ class FocusProfile: "cozy atmosphere, slow tempo" ), modulation_freq=12.0, # Low Beta for relaxed attention - modulation_depth=0.25, + modulation_depth=0.13, bpm=85, density=0.3, brightness=0.5, @@ -78,7 +82,7 @@ class FocusProfile: "predictable progression, subtle pink noise undertone" ), modulation_freq=15.0, # Mid Beta - modulation_depth=0.40, # Slightly stronger modulation + modulation_depth=0.20, # Slightly stronger modulation bpm=128, density=0.5, brightness=0.4, @@ -94,7 +98,7 @@ class FocusProfile: "dreamy textures, organic sounds" ), modulation_freq=10.0, # Alpha-Beta border for creative state - modulation_depth=0.20, + modulation_depth=0.12, bpm=90, density=0.25, brightness=0.6, @@ -110,7 +114,7 @@ class FocusProfile: "dynamic tension, forward momentum, bold synthesis" ), modulation_freq=20.0, # High Beta for maximum alertness - modulation_depth=0.40, + modulation_depth=0.20, bpm=140, density=0.6, brightness=0.7, @@ -130,7 +134,7 @@ class FocusProfile: "positive energy, gentle euphoria, inspiring textures" ), modulation_freq=14.0, # Mid Beta for balanced, positive focus - modulation_depth=0.30, + modulation_depth=0.15, bpm=110, density=0.45, brightness=0.8, diff --git a/tests/test_entrainment.py b/tests/test_entrainment.py index a0bc4c5..b378d96 100644 --- a/tests/test_entrainment.py +++ b/tests/test_entrainment.py @@ -66,11 +66,32 @@ def test_zero_depth_no_change(self, mono_audio, sample_rate): output, _ = apply_entrainment(mono_audio, sample_rate, depth=0.0) np.testing.assert_array_almost_equal(output, mono_audio) - def test_output_amplitude_reduced(self, mono_audio, sample_rate): - """With depth > 0, output should never exceed input amplitude.""" - output, _ = apply_entrainment(mono_audio, sample_rate, depth=0.5) + def test_output_amplitude_reduced_full_spectrum(self, mono_audio, sample_rate): + """Full-spectrum modulation only ever attenuates (never amplifies).""" + output, _ = apply_entrainment(mono_audio, sample_rate, depth=0.5, band_cutoff_hz=None) assert np.max(np.abs(output)) <= np.max(np.abs(mono_audio)) + 1e-6 + def test_output_amplitude_bounded_band_limited(self, mono_audio, sample_rate): + """Band-limited modulation may overshoot slightly but stays well bounded.""" + output, _ = apply_entrainment(mono_audio, sample_rate, depth=0.5, band_cutoff_hz=500.0) + # A phase-shifted low band can nudge the peak up a touch; the downstream + # limiter handles it. Guard against runaway gain only. + assert np.max(np.abs(output)) <= np.max(np.abs(mono_audio)) * 1.1 + + def test_high_frequency_preserved_band_limited(self, sample_rate): + """A tone above the modulation band passes nearly unmodulated.""" + tone = create_test_tone(5000.0, 1.0, sample_rate, channels=1) + band_out, _ = apply_entrainment( + tone, sample_rate, target_freq=15.0, depth=0.5, band_cutoff_hz=500.0 + ) + full_out, _ = apply_entrainment( + tone, sample_rate, target_freq=15.0, depth=0.5, band_cutoff_hz=None + ) + band_diff = np.sqrt(np.mean((band_out - tone) ** 2)) + full_diff = np.sqrt(np.mean((full_out - tone) ** 2)) + # Band-limited leaves the high tone far more intact than full-spectrum. + assert band_diff < full_diff * 0.1 + def test_returns_state_for_continuity(self, mono_audio, sample_rate): _, state = apply_entrainment(mono_audio, sample_rate) assert isinstance(state, ModulationState) diff --git a/tests/test_rotation.py b/tests/test_rotation.py new file mode 100644 index 0000000..6436db3 --- /dev/null +++ b/tests/test_rotation.py @@ -0,0 +1,87 @@ +"""Tests for the seamless session-rotation helpers in the Lyria client.""" + +import numpy as np + +from focus.generation.lyria_client import ( + CROSSFADE_OUT_END_FRAC, + _ChunkBuffer, + _crossfade_gains, +) + + +class TestCrossfadeGains: + """Crossfade gain curves.""" + + def test_endpoints(self): + # Start of the crossfade: full old, no new. + fade_out, fade_in = _crossfade_gains(0, 1, 1000) + assert fade_out[0] == 1.0 + assert abs(fade_in[0]) < 1e-9 + + # End of the crossfade: no old, full new. + fade_out, fade_in = _crossfade_gains(1000, 1, 1000) + assert abs(fade_out[0]) < 1e-9 + assert abs(fade_in[0] - 1.0) < 1e-9 + + def test_outgoing_fades_out_before_the_end(self): + """Default curve silences the outgoing track by out_end_frac (no ghost).""" + total = 1000 + fade_out, _ = _crossfade_gains(0, total, total) + end_idx = int(CROSSFADE_OUT_END_FRAC * total) + # Outgoing has reached (near) silence by the cutover point... + assert fade_out[end_idx] < 1e-6 + # ...and stays there for the rest of the crossfade. + assert np.all(fade_out[end_idx:] < 1e-6) + # ...while it was still audible earlier. + assert fade_out[end_idx // 2] > 0.3 + + def test_symmetric_is_equal_power(self): + """With out_end_frac=1.0 the curve is a symmetric equal-power crossfade.""" + total = 4096 + fade_out, fade_in = _crossfade_gains(0, total, total, out_end_frac=1.0) + power = fade_out**2 + fade_in**2 + np.testing.assert_allclose(power, 1.0, atol=1e-9) + + def test_monotonic_and_contiguous(self): + # Two consecutive segments should tile the curve monotonically. + fo_a, fi_a = _crossfade_gains(0, 100, 300) + fo_b, fi_b = _crossfade_gains(100, 100, 300) + assert np.all(np.diff(fo_a) <= 1e-12) # fade-out non-increasing + assert np.all(np.diff(fi_a) >= -1e-12) # fade-in non-decreasing + assert fo_a[-1] >= fo_b[0] + assert fi_a[-1] <= fi_b[0] + + +class TestChunkBuffer: + """Sample-accurate FIFO used to align two live streams.""" + + def _stereo(self, start, n): + col = np.arange(start, start + n, dtype=np.float32) + return np.column_stack([col, col]) + + def test_take_across_chunk_boundaries(self): + buf = _ChunkBuffer() + buf.add(self._stereo(0, 3)) + buf.add(self._stereo(3, 4)) + assert buf.total == 7 + + first = buf.take(5) + assert first.shape == (5, 2) + np.testing.assert_array_equal(first[:, 0], np.arange(0, 5)) + assert buf.total == 2 + + rest = buf.take(10) # asking for more than available + assert rest.shape == (2, 2) + np.testing.assert_array_equal(rest[:, 0], np.arange(5, 7)) + assert buf.total == 0 + + def test_take_empty_returns_zero_length(self): + buf = _ChunkBuffer() + out = buf.take(4) + assert out.shape == (0, 2) + + def test_add_ignores_empty(self): + buf = _ChunkBuffer() + buf.add(None) + buf.add(np.zeros((0, 2), dtype=np.float32)) + assert buf.total == 0