Spaces:
Running
Render Rhythma sessions as layered ambient audio
Browse filesIntroduce a dedicated session renderer that moves the project beyond a single modulated tone and toward the approved ambient design. The new renderer keeps the existing engine intact while adding a layered path for drone, breath pulse, shimmer, and session envelope so later UI work can present a more immersive session without rewriting the engine again.
Constraint: Task 2 must stay scoped to the engine and its targeted tests
Constraint: Session rendering must always include a breath pulse layer regardless of profile modulation type
Rejected: Reusing the legacy single-layer modulation path as the session renderer | not deep enough for the approved redesign
Confidence: medium
Scope-risk: moderate
Reversibility: clean
Directive: Preserve duration validation and the attack/release envelope when expanding the renderer in later tasks
Tested: python -m py_compile rhythma_engine.py tests/test_rhythma_layered_audio.py tests/test_rhythma_regression.py
Tested: PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 python -c "import pytest; raise SystemExit(pytest.main(['tests/test_rhythma_layered_audio.py','tests/test_rhythma_regression.py','-q','-p','no:cacheprovider']))"
Not-tested: Integration with app.py and session profiles in the live UI; deferred to later tasks
- rhythma_engine.py +105 -4
- tests/test_rhythma_layered_audio.py +78 -0
- tests/test_rhythma_regression.py +13 -4
|
@@ -215,6 +215,110 @@ class RhythmaModulationEngine:
|
|
| 215 |
)
|
| 216 |
return carrier * mod_env
|
| 217 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 218 |
def generate_modulated_wave(self, duration):
|
| 219 |
t, base_carrier = self._generate_base_wave(duration)
|
| 220 |
|
|
@@ -227,10 +331,7 @@ class RhythmaModulationEngine:
|
|
| 227 |
else:
|
| 228 |
modulated = base_carrier
|
| 229 |
|
| 230 |
-
|
| 231 |
-
if max_amp <= 0:
|
| 232 |
-
return modulated
|
| 233 |
-
return 0.9 * modulated / max_amp
|
| 234 |
|
| 235 |
def save_audio(self, duration, file_path=None):
|
| 236 |
if not SOUNDFILE_AVAILABLE:
|
|
|
|
| 215 |
)
|
| 216 |
return carrier * mod_env
|
| 217 |
|
| 218 |
+
def _normalize_audio(self, audio):
|
| 219 |
+
max_amp = np.max(np.abs(audio))
|
| 220 |
+
if max_amp <= 0:
|
| 221 |
+
return audio
|
| 222 |
+
return 0.9 * audio / max_amp
|
| 223 |
+
|
| 224 |
+
def _render_drone_layer(self, t, tone_center, density, config):
|
| 225 |
+
drone = np.zeros_like(t)
|
| 226 |
+
density = float(np.clip(density, 0.0, 1.0))
|
| 227 |
+
harmonic_count = 2 if density < 0.5 else 3
|
| 228 |
+
for index, harmonic_amp in enumerate(config["harmonics"][:harmonic_count], start=1):
|
| 229 |
+
harmonic_freq = tone_center * index
|
| 230 |
+
if harmonic_freq < self.sample_rate / 2:
|
| 231 |
+
drone += harmonic_amp * np.sin(2 * np.pi * harmonic_freq * t)
|
| 232 |
+
|
| 233 |
+
max_amp = np.max(np.abs(drone))
|
| 234 |
+
if max_amp > 0:
|
| 235 |
+
drone = drone / max_amp
|
| 236 |
+
return drone * (0.75 + 0.25 * density)
|
| 237 |
+
|
| 238 |
+
def _render_breath_layer(self, t, tone_center, breath_rate, pattern):
|
| 239 |
+
breath_rate = max(0.02, float(breath_rate))
|
| 240 |
+
breath_freq = max(40.0, tone_center * 0.5)
|
| 241 |
+
carrier = np.sin(2 * np.pi * breath_freq * t)
|
| 242 |
+
pattern_config = self.rhythm_configs.get(pattern, self.config)
|
| 243 |
+
breath_env = 0.5 * (
|
| 244 |
+
signal.square(
|
| 245 |
+
2 * np.pi * breath_rate * t,
|
| 246 |
+
duty=pattern_config["pulse_width"],
|
| 247 |
+
)
|
| 248 |
+
+ 1.0
|
| 249 |
+
)
|
| 250 |
+
|
| 251 |
+
return carrier * breath_env
|
| 252 |
+
|
| 253 |
+
def _render_shimmer_layer(self, t, tone_center, brightness, shimmer):
|
| 254 |
+
brightness = float(np.clip(brightness, 0.0, 1.0))
|
| 255 |
+
shimmer = float(np.clip(shimmer, 0.0, 1.0))
|
| 256 |
+
shimmer_layer = np.zeros_like(t)
|
| 257 |
+
harmonic_levels = [
|
| 258 |
+
(2.0, 0.35 + 0.25 * brightness),
|
| 259 |
+
(3.0, 0.2 + 0.2 * shimmer),
|
| 260 |
+
(4.0, 0.1 + 0.15 * brightness),
|
| 261 |
+
]
|
| 262 |
+
for index, (multiplier, amplitude) in enumerate(harmonic_levels, start=1):
|
| 263 |
+
harmonic_freq = tone_center * multiplier
|
| 264 |
+
if harmonic_freq < self.sample_rate / 2:
|
| 265 |
+
shimmer_layer += amplitude * np.sin(
|
| 266 |
+
2 * np.pi * harmonic_freq * t + (index * np.pi / 7)
|
| 267 |
+
)
|
| 268 |
+
|
| 269 |
+
max_amp = np.max(np.abs(shimmer_layer))
|
| 270 |
+
if max_amp > 0:
|
| 271 |
+
shimmer_layer = shimmer_layer / max_amp
|
| 272 |
+
|
| 273 |
+
shimmer_motion = 0.6 + 0.4 * np.sin(
|
| 274 |
+
2 * np.pi * max(0.1, 1.5 + shimmer * 3.0) * t
|
| 275 |
+
)
|
| 276 |
+
return shimmer_layer * shimmer_motion * (0.2 + 0.8 * brightness)
|
| 277 |
+
|
| 278 |
+
def _build_session_envelope(self, sample_count):
|
| 279 |
+
if sample_count <= 1:
|
| 280 |
+
return np.ones(sample_count)
|
| 281 |
+
|
| 282 |
+
attack_count = max(1, int(sample_count * 0.08))
|
| 283 |
+
release_count = max(1, int(sample_count * 0.12))
|
| 284 |
+
if attack_count + release_count >= sample_count:
|
| 285 |
+
attack_count = max(1, sample_count // 2)
|
| 286 |
+
release_count = sample_count - attack_count
|
| 287 |
+
sustain_count = sample_count - attack_count - release_count
|
| 288 |
+
|
| 289 |
+
attack = np.linspace(0.0, 1.0, attack_count, endpoint=False)
|
| 290 |
+
sustain = np.ones(sustain_count)
|
| 291 |
+
release = np.linspace(1.0, 0.0, release_count, endpoint=True)
|
| 292 |
+
return np.concatenate([attack, sustain, release])
|
| 293 |
+
|
| 294 |
+
def render_session(self, profile, duration):
|
| 295 |
+
sample_count = int(self.sample_rate * duration)
|
| 296 |
+
if duration <= 0 or sample_count < 1:
|
| 297 |
+
raise ValueError("duration must produce at least one sample")
|
| 298 |
+
|
| 299 |
+
tone_center = float(profile.get("tone_center", self.base_freq))
|
| 300 |
+
pattern = profile.get("pattern", self.rhythm_pattern)
|
| 301 |
+
config = self.rhythm_configs.get(pattern, self.config)
|
| 302 |
+
t = np.linspace(0, duration, sample_count, endpoint=False)
|
| 303 |
+
|
| 304 |
+
drone = self._render_drone_layer(t, tone_center, profile.get("density", 0.5), config)
|
| 305 |
+
pulse = self._render_breath_layer(
|
| 306 |
+
t,
|
| 307 |
+
tone_center,
|
| 308 |
+
profile.get("breath_rate", config["mod_freq"] / 8),
|
| 309 |
+
pattern,
|
| 310 |
+
)
|
| 311 |
+
shimmer = self._render_shimmer_layer(
|
| 312 |
+
t,
|
| 313 |
+
tone_center,
|
| 314 |
+
profile.get("brightness", 0.25),
|
| 315 |
+
profile.get("shimmer", 0.1),
|
| 316 |
+
)
|
| 317 |
+
|
| 318 |
+
combined = (0.62 * drone) + (0.25 * pulse) + (0.13 * shimmer)
|
| 319 |
+
combined = combined * self._build_session_envelope(len(t))
|
| 320 |
+
return self._normalize_audio(combined)
|
| 321 |
+
|
| 322 |
def generate_modulated_wave(self, duration):
|
| 323 |
t, base_carrier = self._generate_base_wave(duration)
|
| 324 |
|
|
|
|
| 331 |
else:
|
| 332 |
modulated = base_carrier
|
| 333 |
|
| 334 |
+
return self._normalize_audio(modulated)
|
|
|
|
|
|
|
|
|
|
| 335 |
|
| 336 |
def save_audio(self, duration, file_path=None):
|
| 337 |
if not SOUNDFILE_AVAILABLE:
|
|
@@ -0,0 +1,78 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
import numpy as np
|
| 2 |
+
import pytest
|
| 3 |
+
|
| 4 |
+
from rhythma_engine import RhythmaModulationEngine
|
| 5 |
+
|
| 6 |
+
|
| 7 |
+
def test_render_session_returns_normalized_layered_audio():
|
| 8 |
+
engine = RhythmaModulationEngine()
|
| 9 |
+
profile = {
|
| 10 |
+
"tone_center": 396.0,
|
| 11 |
+
"pattern": "calm",
|
| 12 |
+
"modulation_type": "sine",
|
| 13 |
+
"brightness": 0.25,
|
| 14 |
+
"density": 0.45,
|
| 15 |
+
"shimmer": 0.12,
|
| 16 |
+
"breath_rate": 0.08,
|
| 17 |
+
}
|
| 18 |
+
|
| 19 |
+
audio = engine.render_session(profile, duration=4)
|
| 20 |
+
|
| 21 |
+
assert len(audio) == engine.sample_rate * 4
|
| 22 |
+
assert np.max(np.abs(audio)) <= 0.9 + 1e-9
|
| 23 |
+
assert np.max(np.abs(audio[:400])) < np.max(np.abs(audio[2000:2400]))
|
| 24 |
+
assert np.max(np.abs(audio[-400:])) < np.max(np.abs(audio[2000:2400]))
|
| 25 |
+
|
| 26 |
+
|
| 27 |
+
def test_render_session_is_not_identical_to_single_layer_base_wave():
|
| 28 |
+
engine = RhythmaModulationEngine()
|
| 29 |
+
profile = {
|
| 30 |
+
"tone_center": 639.0,
|
| 31 |
+
"pattern": "focused",
|
| 32 |
+
"modulation_type": "sine",
|
| 33 |
+
"brightness": 0.4,
|
| 34 |
+
"density": 0.35,
|
| 35 |
+
"shimmer": 0.18,
|
| 36 |
+
"breath_rate": 0.12,
|
| 37 |
+
}
|
| 38 |
+
|
| 39 |
+
session_audio = engine.render_session(profile, duration=3)
|
| 40 |
+
legacy_audio = engine.generate_modulated_wave(3)
|
| 41 |
+
|
| 42 |
+
assert not np.allclose(session_audio[:2000], legacy_audio[:2000])
|
| 43 |
+
|
| 44 |
+
|
| 45 |
+
def test_render_session_keeps_breath_layer_pulse_shaped():
|
| 46 |
+
engine = RhythmaModulationEngine()
|
| 47 |
+
pulse_profile = {
|
| 48 |
+
"tone_center": 528.0,
|
| 49 |
+
"pattern": "relaxed",
|
| 50 |
+
"modulation_type": "pulse",
|
| 51 |
+
"brightness": 0.3,
|
| 52 |
+
"density": 0.4,
|
| 53 |
+
"shimmer": 0.15,
|
| 54 |
+
"breath_rate": 0.1,
|
| 55 |
+
}
|
| 56 |
+
chirp_profile = dict(pulse_profile, modulation_type="chirp")
|
| 57 |
+
|
| 58 |
+
pulse_audio = engine.render_session(pulse_profile, duration=2)
|
| 59 |
+
chirp_audio = engine.render_session(chirp_profile, duration=2)
|
| 60 |
+
|
| 61 |
+
assert np.allclose(pulse_audio, chirp_audio)
|
| 62 |
+
|
| 63 |
+
|
| 64 |
+
@pytest.mark.parametrize("duration", [0, -1, 0.5 / RhythmaModulationEngine.SAMPLE_RATE])
|
| 65 |
+
def test_render_session_rejects_invalid_duration(duration):
|
| 66 |
+
engine = RhythmaModulationEngine()
|
| 67 |
+
profile = {
|
| 68 |
+
"tone_center": 528.0,
|
| 69 |
+
"pattern": "calm",
|
| 70 |
+
"modulation_type": "sine",
|
| 71 |
+
"brightness": 0.2,
|
| 72 |
+
"density": 0.3,
|
| 73 |
+
"shimmer": 0.1,
|
| 74 |
+
"breath_rate": 0.08,
|
| 75 |
+
}
|
| 76 |
+
|
| 77 |
+
with pytest.raises(ValueError, match="duration must produce at least one sample"):
|
| 78 |
+
engine.render_session(profile, duration=duration)
|
|
@@ -24,10 +24,19 @@ def test_analyze_input_defaults_to_neutral_when_no_text_is_provided():
|
|
| 24 |
assert result["error"] is None
|
| 25 |
|
| 26 |
|
| 27 |
-
def
|
| 28 |
-
engine = RhythmaModulationEngine(emotional_state="stressed")
|
| 29 |
-
|
| 30 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 31 |
|
| 32 |
assert len(wave) == 4410
|
| 33 |
assert np.max(np.abs(wave)) <= 0.9 + 1e-9
|
|
|
|
| 24 |
assert result["error"] is None
|
| 25 |
|
| 26 |
|
| 27 |
+
def test_render_session_has_expected_length_and_headroom():
|
| 28 |
+
engine = RhythmaModulationEngine(emotional_state="stressed", rhythm_pattern="calm")
|
| 29 |
+
profile = {
|
| 30 |
+
"tone_center": engine.base_freq,
|
| 31 |
+
"pattern": "calm",
|
| 32 |
+
"modulation_type": "sine",
|
| 33 |
+
"brightness": 0.2,
|
| 34 |
+
"density": 0.4,
|
| 35 |
+
"shimmer": 0.1,
|
| 36 |
+
"breath_rate": 0.08,
|
| 37 |
+
}
|
| 38 |
+
|
| 39 |
+
wave = engine.render_session(profile, 0.1)
|
| 40 |
|
| 41 |
assert len(wave) == 4410
|
| 42 |
assert np.max(np.abs(wave)) <= 0.9 + 1e-9
|