/
/
/
1"""
2Rainy Mood provider for Music Assistant.
3
4Serves a looping rain ambience from rainymood.com as a sound effect, suitable for
5direct playback or as the source for a queue's audio overlay. The remote stream is
6handed straight to the player pipeline - the audio overlay engine takes care of the
7looping, format conversion and volume mixing.
8"""
9
10from __future__ import annotations
11
12from typing import TYPE_CHECKING
13
14from music_assistant_models.enums import ContentType, MediaType, ProviderFeature, StreamType
15from music_assistant_models.errors import MediaNotFoundError
16from music_assistant_models.media_items import AudioFormat, ProviderMapping, SoundEffect
17from music_assistant_models.streamdetails import StreamDetails
18
19from music_assistant.models.music_provider import MusicProvider
20
21if TYPE_CHECKING:
22 from collections.abc import AsyncGenerator
23
24 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
25 from music_assistant_models.provider import ProviderManifest
26
27 from music_assistant.mass import MusicAssistant
28 from music_assistant.models import ProviderInstanceType
29
30SUPPORTED_FEATURES = {
31 ProviderFeature.BROWSE,
32 ProviderFeature.SOUND_EFFECTS,
33}
34
35RAIN_ITEM_ID = "rain"
36RAIN_URL = "https://media.rainymood.com/0.mp3"
37
38
39async def setup(
40 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
41) -> ProviderInstanceType:
42 """Initialize provider(instance) with given configuration."""
43 return RainyMoodProvider(mass, manifest, config, SUPPORTED_FEATURES)
44
45
46class RainyMoodProvider(MusicProvider):
47 """Music provider serving a looping rain ambience from rainymood.com."""
48
49 @property
50 def max_concurrent_streams(self) -> None:
51 """Allow unlimited concurrent upstream source streams."""
52 return None
53
54 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
55 """Return Config entries to configure this provider (none needed)."""
56 return ()
57
58 @property
59 def is_streaming_provider(self) -> bool:
60 """Return True if the provider is a streaming provider."""
61 return True
62
63 async def get_sound_effect(self, prov_sound_effect_id: str) -> SoundEffect:
64 """Get full sound effect details by id."""
65 if prov_sound_effect_id != RAIN_ITEM_ID:
66 raise MediaNotFoundError(f"Unknown sound effect: {prov_sound_effect_id}")
67 return self._build_sound_effect()
68
69 async def get_sound_effects(self) -> AsyncGenerator[SoundEffect]:
70 """Get all sound effect items this provider offers."""
71 yield self._build_sound_effect()
72
73 async def get_stream_details(
74 self, item_id: str, media_type: MediaType = MediaType.TRACK
75 ) -> StreamDetails:
76 """Return the streamdetails to stream the rain sound effect."""
77 if item_id != RAIN_ITEM_ID:
78 raise MediaNotFoundError(f"Unknown sound effect: {item_id}")
79 return StreamDetails(
80 provider=self.instance_id,
81 item_id=item_id,
82 # rainymood serves an MP3 stream; declare it explicitly instead of
83 # leaving the container type to be probed
84 audio_format=AudioFormat(content_type=ContentType.MP3),
85 media_type=MediaType.SOUND_EFFECT,
86 stream_type=StreamType.HTTP,
87 path=RAIN_URL,
88 allow_seek=False,
89 can_seek=False,
90 )
91
92 def _build_sound_effect(self) -> SoundEffect:
93 """Create the SoundEffect item for the rain ambience."""
94 sound_effect = SoundEffect(
95 item_id=RAIN_ITEM_ID,
96 provider=self.instance_id,
97 name="Rain",
98 translation_key=RAIN_ITEM_ID,
99 provider_mappings={
100 ProviderMapping(
101 item_id=RAIN_ITEM_ID,
102 provider_domain=self.domain,
103 provider_instance=self.instance_id,
104 )
105 },
106 )
107 sound_effect.metadata.description = "Looping rain ambience from rainymood.com."
108 return sound_effect
109