Spaces:
Running
Running
File size: 4,321 Bytes
330f477 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | from __future__ import annotations
import hashlib
from pathlib import Path
import numpy as np
import pytest
from openmusic_analysis.analyzers.clap import (
ClapGlobalAudioAnalyzer,
ClapTemporalAudioAnalyzer,
)
from openmusic_analysis.analyzers.lyrics import BGEM3LyricsAnalyzer, LyricsPreprocessor
from openmusic_analysis.application import MusicAnalysisService
from openmusic_analysis.audio.decoder import AudioMetadata, DecodedAudio
from openmusic_analysis.errors import AudioDecodeError
from openmusic_analysis.registry import ModelRegistry
from openmusic_analysis.settings import GlobalAudioConfig, LyricsConfig, TemporalAudioConfig
class FakeAudioEncoder:
dimension = 4
loaded = True
def __init__(self) -> None:
self.calls: list[list[np.ndarray]] = []
async def ready(self) -> None:
return None
async def encode(self, windows: list[np.ndarray]) -> np.ndarray:
self.calls.append(windows)
rows = []
for window in windows:
rows.append(
[
float(np.mean(window)),
float(np.std(window)),
float(window[0]),
float(window[-1]) + 1.0,
]
)
return np.asarray(rows, dtype=np.float32)
class FailingAudioEncoder(FakeAudioEncoder):
async def encode(self, windows: list[np.ndarray]) -> np.ndarray:
raise RuntimeError("internal model detail must not leak")
class FakeTextEncoder:
dimension = 6
loaded = True
async def ready(self) -> None:
return None
def count_tokens(self, text: str) -> int:
return len(text.split()) + 2
def split_tokens(self, text: str, max_tokens: int) -> list[str]:
words = text.split()
size = max(1, max_tokens - 2)
return [" ".join(words[index : index + size]) for index in range(0, len(words), size)]
async def encode(self, texts: list[str], batch_size: int) -> np.ndarray:
rows = []
for text in texts:
digest = hashlib.sha256(text.encode("utf-8")).digest()
rows.append([float(value + 1) for value in digest[: self.dimension]])
return np.asarray(rows, dtype=np.float32)
class FakeDecoder:
canonical_sample_rate = 10
def __init__(self, waveform: np.ndarray | None = None) -> None:
self.waveform = (
np.asarray(waveform, dtype=np.float32)
if waveform is not None
else np.linspace(-1.0, 1.0, 95, dtype=np.float32)
)
self.calls = 0
def decode(self, source_path: str | Path) -> DecodedAudio:
self.calls += 1
if Path(source_path).read_bytes().startswith(b"bad"):
raise AudioDecodeError()
return DecodedAudio(
waveform=self.waveform,
sample_rate=self.canonical_sample_rate,
metadata=AudioMetadata(
duration_ms=int(round(self.waveform.size * 1000 / self.canonical_sample_rate)),
source_sample_rate=self.canonical_sample_rate,
source_channels=1,
source_format="fake",
),
)
def make_service(
*,
waveform: np.ndarray | None = None,
audio_encoder: FakeAudioEncoder | None = None,
) -> tuple[MusicAnalysisService, FakeDecoder, FakeAudioEncoder]:
audio_encoder = audio_encoder or FakeAudioEncoder()
decoder = FakeDecoder(waveform)
global_config = GlobalAudioConfig(
sample_rate=10,
window_seconds=2.0,
target_windows=4,
minimum_audio_seconds=1.0,
)
temporal_config = TemporalAudioConfig(
sample_rate=10,
window_seconds=2.0,
hop_seconds=2.0,
max_segments=4,
minimum_audio_seconds=1.0,
)
analyzers = [
ClapGlobalAudioAnalyzer(audio_encoder, global_config),
ClapTemporalAudioAnalyzer(audio_encoder, temporal_config),
BGEM3LyricsAnalyzer(
FakeTextEncoder(),
LyricsPreprocessor(),
LyricsConfig(max_chunk_tokens=12, batch_size=4),
),
]
service = MusicAnalysisService(
registry=ModelRegistry(analyzers), decoder=decoder, device="cpu"
)
return service, decoder, audio_encoder
@pytest.fixture
def service_bundle():
return make_service()
|