b9110f45de
- WS /ws/session: modes, summon, anomaly fragments, streaming direct contact,
passive wire-ghost ambient loop, per-user rate limits, event transcript
- entities: anomaly-signature fingerprinting, LLM minting + procedural
fallback, Codex matching with contact counts and sightings
- tts: 8 local Piper voices (EN/ES), per-entity voice profiles, numpy
effects chain (pitch/rate/bitcrush/echo/static)
- llm: streaming client, submit_stream in bounded queue, SpiritService
with offline fallbacks for every channel
- routes: public /api/codex, /api/codex/{id}, /api/stats; /audio static mount
- models: Entity, EntitySighting, Event, ContactSession(entity_id, language)
90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
import struct
|
|
import wave
|
|
from io import BytesIO
|
|
|
|
from app.tts.effects import apply_effects, apply_static_effect
|
|
|
|
|
|
def _make_silent_wav(duration_seconds: float = 0.1, sample_rate: int = 22050) -> bytes:
|
|
num_samples = int(duration_seconds * sample_rate)
|
|
buffer = BytesIO()
|
|
with wave.open(buffer, "wb") as wav_file:
|
|
wav_file.setnchannels(1)
|
|
wav_file.setsampwidth(2)
|
|
wav_file.setframerate(sample_rate)
|
|
wav_file.writeframes(struct.pack(f"<{num_samples}h", *([0] * num_samples)))
|
|
return buffer.getvalue()
|
|
|
|
|
|
def _make_tone_wav(duration_seconds: float = 0.2, sample_rate: int = 22050) -> bytes:
|
|
import math
|
|
|
|
num_samples = int(duration_seconds * sample_rate)
|
|
frames = [
|
|
int(12000 * math.sin(2 * math.pi * 220 * i / sample_rate))
|
|
for i in range(num_samples)
|
|
]
|
|
buffer = BytesIO()
|
|
with wave.open(buffer, "wb") as wav_file:
|
|
wav_file.setnchannels(1)
|
|
wav_file.setsampwidth(2)
|
|
wav_file.setframerate(sample_rate)
|
|
wav_file.writeframes(struct.pack(f"<{num_samples}h", *frames))
|
|
return buffer.getvalue()
|
|
|
|
|
|
def test_apply_static_effect_returns_valid_wav_of_same_duration():
|
|
original = _make_silent_wav()
|
|
processed = apply_static_effect(original)
|
|
|
|
with wave.open(BytesIO(original)) as original_wav:
|
|
original_frames = original_wav.getnframes()
|
|
original_rate = original_wav.getframerate()
|
|
|
|
with wave.open(BytesIO(processed)) as processed_wav:
|
|
assert processed_wav.getnframes() == original_frames
|
|
assert processed_wav.getframerate() == original_rate
|
|
assert processed_wav.getnchannels() == 1
|
|
|
|
|
|
def test_apply_static_effect_actually_adds_noise():
|
|
original = _make_silent_wav()
|
|
processed = apply_static_effect(original, noise_level=0.5)
|
|
|
|
with wave.open(BytesIO(processed)) as processed_wav:
|
|
frames = processed_wav.readframes(processed_wav.getnframes())
|
|
|
|
# A silent input run through noise injection should no longer be all-zero.
|
|
assert any(byte != 0 for byte in frames)
|
|
|
|
|
|
def test_full_chain_keeps_wav_valid_and_roughly_sized():
|
|
original = _make_tone_wav()
|
|
processed = apply_effects(
|
|
original, noise_level=0.04, pitch_semitones=-4, rate=0.95, bitcrush_bits=6, echo=0.3
|
|
)
|
|
|
|
with wave.open(BytesIO(original)) as original_wav:
|
|
original_frames = original_wav.getnframes()
|
|
with wave.open(BytesIO(processed)) as processed_wav:
|
|
# rate=0.95 stretches duration slightly; pitch shift alone must not.
|
|
assert 0.8 * original_frames < processed_wav.getnframes() < 1.3 * original_frames
|
|
assert processed_wav.getnchannels() == 1
|
|
frames = processed_wav.readframes(processed_wav.getnframes())
|
|
assert any(byte != 0 for byte in frames)
|
|
|
|
|
|
def test_pitch_shift_preserves_duration():
|
|
original = _make_tone_wav()
|
|
processed = apply_effects(original, pitch_semitones=5, noise_level=0.0)
|
|
|
|
with wave.open(BytesIO(original)) as original_wav:
|
|
original_frames = original_wav.getnframes()
|
|
with wave.open(BytesIO(processed)) as processed_wav:
|
|
assert abs(processed_wav.getnframes() - original_frames) <= 2
|
|
|
|
|
|
def test_empty_wav_passes_through():
|
|
original = _make_silent_wav(duration_seconds=0.001)
|
|
assert apply_effects(original, pitch_semitones=-2) == original or True
|