Spaces:
Running
Running
| from __future__ import annotations | |
| from pathlib import Path | |
| from openmusic_analysis.analyzers import ( | |
| BGEM3LyricsAnalyzer, | |
| BGEM3TextEncoder, | |
| ClapAudioEncoder, | |
| ClapGlobalAudioAnalyzer, | |
| ClapTemporalAudioAnalyzer, | |
| LyricsPreprocessor, | |
| ) | |
| from openmusic_analysis.audio import AnalysisContext, AudioDecoder | |
| from openmusic_analysis.domain import AnalysisResponse, TrackReference | |
| from openmusic_analysis.errors import AnalysisError | |
| from openmusic_analysis.registry import ModelRegistry | |
| from openmusic_analysis.runtime import DeviceManager, InferenceGate | |
| from openmusic_analysis.settings import Settings | |
| DEFAULT_AUDIO_REPRESENTATIONS = ("audio.global", "audio.temporal") | |
| class MusicAnalysisService: | |
| def __init__( | |
| self, | |
| *, | |
| registry: ModelRegistry, | |
| decoder: AudioDecoder, | |
| device: str, | |
| ) -> None: | |
| self.registry = registry | |
| self.decoder = decoder | |
| self.device = device | |
| async def analyze( | |
| self, | |
| source_path: str | Path, | |
| *, | |
| lyrics: str | None, | |
| requested_representations: list[str] | None, | |
| track_id: str | None, | |
| content_identity: str | None, | |
| ) -> AnalysisResponse: | |
| if lyrics is not None and not lyrics.strip(): | |
| raise AnalysisError( | |
| "INVALID_LYRICS", "Lyrics must contain non-whitespace text.", status_code=422 | |
| ) | |
| requested = self._resolve_representations(requested_representations, lyrics) | |
| context = AnalysisContext(source_path, self.decoder) | |
| results = {} | |
| for representation in requested: | |
| analyzer = self.registry.analyzer(representation) | |
| if analyzer.input_kind == "audio": | |
| result = await analyzer.analyze(context) | |
| elif analyzer.input_kind == "lyrics": | |
| if lyrics is None: | |
| raise AnalysisError( | |
| "LYRICS_REQUIRED", | |
| f"Representation '{representation}' requires lyrics.", | |
| status_code=422, | |
| ) | |
| result = await analyzer.analyze(lyrics) | |
| else: | |
| raise RuntimeError(f"Unknown analyzer input kind: {analyzer.input_kind}") | |
| results[representation] = result | |
| return AnalysisResponse( | |
| track=TrackReference(track_id=track_id, content_identity=content_identity), | |
| representations=results, | |
| ) | |
| def _resolve_representations( | |
| self, requested: list[str] | None, lyrics: str | None | |
| ) -> list[str]: | |
| if requested is None: | |
| values = list(DEFAULT_AUDIO_REPRESENTATIONS) | |
| if lyrics is not None: | |
| values.append("lyrics.global") | |
| else: | |
| values = list(dict.fromkeys(requested)) | |
| if not values: | |
| raise AnalysisError( | |
| "INVALID_REPRESENTATIONS", | |
| "requested_representations must not be empty.", | |
| status_code=422, | |
| ) | |
| unsupported = [value for value in values if value not in self.registry.representations] | |
| if unsupported: | |
| raise AnalysisError( | |
| "UNSUPPORTED_REPRESENTATION", | |
| f"Unsupported representation(s): {', '.join(unsupported)}.", | |
| status_code=422, | |
| details={"supported": list(self.registry.representations)}, | |
| ) | |
| return values | |
| async def load_models(self) -> None: | |
| seen: set[int] = set() | |
| for representation in self.registry.representations: | |
| analyzer = self.registry.analyzer(representation) | |
| encoder = getattr(analyzer, "encoder", None) | |
| if encoder is not None and id(encoder) not in seen: | |
| seen.add(id(encoder)) | |
| await encoder.ready() | |
| def build_service(settings: Settings | None = None) -> MusicAnalysisService: | |
| settings = settings or Settings.from_env() | |
| device = DeviceManager.select(settings.device) | |
| gate = InferenceGate(settings.inference_concurrency) | |
| clap_encoder = ClapAudioEncoder( | |
| device, gate, batch_size=settings.global_audio.inference_batch_size | |
| ) | |
| text_encoder = BGEM3TextEncoder(device, gate) | |
| analyzers = [ | |
| ClapGlobalAudioAnalyzer(clap_encoder, settings.global_audio), | |
| ClapTemporalAudioAnalyzer(clap_encoder, settings.temporal_audio), | |
| BGEM3LyricsAnalyzer(text_encoder, LyricsPreprocessor(), settings.lyrics), | |
| ] | |
| registry = ModelRegistry(analyzers) | |
| decoder = AudioDecoder( | |
| ffmpeg_binary=settings.ffmpeg_binary, | |
| ffprobe_binary=settings.ffprobe_binary, | |
| canonical_sample_rate=settings.global_audio.sample_rate, | |
| max_audio_seconds=settings.limits.max_audio_seconds, | |
| timeout_seconds=settings.limits.decode_timeout_seconds, | |
| ) | |
| return MusicAnalysisService(registry=registry, decoder=decoder, device=device) | |