music-assistant-server

1"""KION Music provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import zlib
8from collections.abc import AsyncGenerator, Coroutine, Sequence
9from io import BytesIO
10from typing import TYPE_CHECKING, Any
11
12from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
13from music_assistant_models.enums import ConfigEntryType, ImageType, MediaType, ProviderFeature
14from music_assistant_models.errors import (
15    InvalidDataError,
16    LoginFailed,
17    MediaNotFoundError,
18    ProviderUnavailableError,
19    ResourceTemporarilyUnavailable,
20)
21from music_assistant_models.media_items import (
22    Album,
23    Artist,
24    BrowseFolder,
25    ItemMapping,
26    MediaItemImage,
27    MediaItemType,
28    Playlist,
29    ProviderMapping,
30    RecommendationFolder,
31    SearchResults,
32    Track,
33    UniqueList,
34)
35from PIL import Image as PilImage
36
37from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
38from music_assistant.controllers.cache import use_cache
39from music_assistant.helpers.datetime import utc
40from music_assistant.models.music_provider import MusicProvider
41
42from .api_client import KionMusicClient
43from .constants import (
44    BROWSE_INITIAL_TRACKS,
45    COLLECTION_FOLDER_ID,
46    CONF_BASE_URL,
47    CONF_CODECS,
48    CONF_LIKED_TRACKS_MAX_TRACKS,
49    CONF_MY_WAVE_MAX_TRACKS,
50    CONF_QUALITY,
51    CONF_TOKEN,
52    CONF_TRANSPORT,
53    DEFAULT_BASE_URL,
54    DISCOVERY_INITIAL_TRACKS,
55    FOR_YOU_FOLDER_ID,
56    IMAGE_SIZE_MEDIUM,
57    LIKED_TRACKS_PLAYLIST_ID,
58    LISTENING_HISTORY_FOLDER_ID,
59    MY_WAVE_BATCH_SIZE,
60    MY_WAVE_PLAYLIST_ID,
61    MY_WAVES_FOLDER_ID,
62    MY_WAVES_SET_FOLDER_ID,
63    PINNED_ITEMS_FOLDER_ID,
64    PLAYLIST_ID_SPLITTER,
65    QUALITY_BALANCED,
66    QUALITY_EFFICIENT,
67    QUALITY_HIGH,
68    QUALITY_LOSSLESS,
69    RADIO_FOLDER_ID,
70    RADIO_TRACK_ID_SEP,
71    ROTOR_STATION_MY_MIX,
72    TAG_CATEGORY_ACTIVITY,
73    TAG_CATEGORY_ERA,
74    TAG_CATEGORY_GENRES,
75    TAG_CATEGORY_MOOD,
76    TAG_CATEGORY_ORDER,
77    TAG_MIXES,
78    TAG_SEASONAL_MAP,
79    TAG_SLUG_CATEGORY,
80    TRACK_BATCH_SIZE,
81    TRANSPORT_ENCRAW,
82    TRANSPORT_RAW,
83    WAVE_CATEGORY_DISPLAY_ORDER,
84    WAVES_FOLDER_ID,
85    WAVES_LANDING_FOLDER_ID,
86)
87from .parsers import (
88    _get_image_url as get_image_url,
89)
90from .parsers import (
91    get_canonical_provider_name,
92    parse_album,
93    parse_artist,
94    parse_playlist,
95    parse_track,
96)
97from .streaming import KionMusicStreamingManager
98
99if TYPE_CHECKING:
100    from music_assistant_models.streamdetails import StreamDetails
101
102
103def _parse_radio_item_id(item_id: str) -> tuple[str, str | None]:
104    """
105    Extract track_id and optional station_id from provider item_id.
106
107    My Mix tracks use item_id format 'track_id@station_id'. Other tracks use
108    plain track_id.
109
110    :param item_id: Provider item_id (may contain RADIO_TRACK_ID_SEP).
111    :return: (track_id, station_id or None).
112    """
113    if RADIO_TRACK_ID_SEP in item_id:
114        parts = item_id.split(RADIO_TRACK_ID_SEP, 1)
115        return (parts[0], parts[1] if len(parts) > 1 else None)
116    return (item_id, None)
117
118
119# Collection sub-folder browse ids -> (ProviderFeature, library sub_id, label key, English name).
120# The library sub_id ("tracks") and label key ("my_favorites") differ on purpose so the Collection
121# labels stay distinct from the core "media.folder.*" library labels.
122_COLLECTION_SUBFOLDERS: tuple[tuple[ProviderFeature, str, str, str], ...] = (
123    (ProviderFeature.LIBRARY_TRACKS, "tracks", "my_favorites", "My Favorites"),
124    (ProviderFeature.LIBRARY_ARTISTS, "artists", "my_artists", "My Artists"),
125    (ProviderFeature.LIBRARY_ALBUMS, "albums", "my_albums", "My Albums"),
126    (ProviderFeature.LIBRARY_PLAYLISTS, "playlists", "my_playlists", "My Playlists"),
127)
128
129
130def _media_label_key(slug: str) -> str:
131    """Normalize a tag/category slug into its strings.json authoring key (spaces → underscores)."""
132    return slug.replace(" ", "_")
133
134
135class _WaveState:
136    """Per-station mutable state for rotor wave playback."""
137
138    def __init__(self) -> None:
139        self.batch_id: str | None = None
140        self.last_track_id: str | None = None
141        self.seen_track_ids: set[str] = set()
142        self.radio_started_sent: bool = False
143        self.lock: asyncio.Lock = asyncio.Lock()
144
145
146class KionMusicProvider(MusicProvider):
147    """Implementation of a KION Music MusicProvider."""
148
149    _client: KionMusicClient | None = None
150    _streaming: KionMusicStreamingManager | None = None
151    _my_wave_batch_id: str | None = None
152    _my_wave_last_track_id: str | None = None  # last track id for "Load more" (API queue param)
153    _my_wave_playlist_next_cursor: str | None = None  # first_track_id for next playlist page
154    _my_wave_radio_started_sent: bool = False
155    _my_wave_seen_track_ids: set[str]  # Track IDs seen in current My Mix session
156    _my_wave_lock: asyncio.Lock  # Protects My Mix mutable state
157    _wave_states: dict[str, _WaveState]  # Per-station state for tagged wave stations
158    _wave_bg_colors: dict[str, str]  # image_url -> hex bg color for transparent covers
159
160    @property
161    def client(self) -> KionMusicClient:
162        """Return the KION Music client."""
163        if self._client is None:
164            raise ProviderUnavailableError("Provider not initialized")
165        return self._client
166
167    @property
168    def streaming(self) -> KionMusicStreamingManager:
169        """Return the streaming manager."""
170        if self._streaming is None:
171            raise ProviderUnavailableError("Provider not initialized")
172        return self._streaming
173
174    async def get_recommendations(self) -> list[RecommendationFolder]:
175        """
176        Get the available recommendation rows, without items.
177
178        Returns My Mix, Made for you, Chart, New Releases, New Playlists,
179        Top Picks, Mood Mix, Activity Mix and Seasonal Mix rows.
180        """
181        # The seasonal row title carries the current season, derived locally from the month.
182        seasonal_tag = TAG_SEASONAL_MAP.get(utc().month, "autumn")
183        seasonal_name = (
184            self._media_source_name("folder", _media_label_key(seasonal_tag))
185            or seasonal_tag.title()
186        )
187        return [
188            RecommendationFolder(
189                item_id=MY_WAVE_PLAYLIST_ID,
190                provider=self.instance_id,
191                name="My Mix",
192                translation_key=MY_WAVE_PLAYLIST_ID,
193                icon="mdi-waveform",
194            ),
195            RecommendationFolder(
196                item_id="feed",
197                provider=self.instance_id,
198                name="Made for you",
199                translation_key="made_for_you",
200                icon="mdi-account-music",
201            ),
202            RecommendationFolder(
203                item_id="chart",
204                provider=self.instance_id,
205                name="Chart",
206                translation_key="chart",
207                icon="mdi-chart-line",
208            ),
209            RecommendationFolder(
210                item_id="new_releases",
211                provider=self.instance_id,
212                name="New Releases",
213                translation_key="new_releases",
214                icon="mdi-new-box",
215            ),
216            RecommendationFolder(
217                item_id="new_playlists",
218                provider=self.instance_id,
219                name="New Playlists",
220                translation_key="new_playlists",
221                icon="mdi-playlist-star",
222            ),
223            RecommendationFolder(
224                item_id="top_picks",
225                provider=self.instance_id,
226                name="Top Picks",
227                translation_key="top_picks",
228                icon="mdi-star",
229            ),
230            # Mood/Activity rows have a static title; the hourly rotating tag - derived
231            # deterministically, so the items call independently computes the same one -
232            # shows as the row subtitle (cache-only tag-list read, no backend I/O).
233            RecommendationFolder(
234                item_id="mood_mix",
235                provider=self.instance_id,
236                name="Mood Mix",
237                translation_key="mood_mix",
238                subtitle=await self._rotating_row_tag_subtitle("mood"),
239                icon="mdi-emoticon-outline",
240            ),
241            RecommendationFolder(
242                item_id="activity_mix",
243                provider=self.instance_id,
244                name="Activity Mix",
245                translation_key="activity_mix",
246                subtitle=await self._rotating_row_tag_subtitle("activity"),
247                icon="mdi-run",
248            ),
249            RecommendationFolder(
250                item_id="seasonal_mix",
251                provider=self.instance_id,
252                name=f"Seasonal: {seasonal_name}",
253                translation_key="seasonal_mix",
254                translation_params=[seasonal_name],
255                icon="mdi-weather-sunny",
256            ),
257        ]
258
259    async def get_recommendation_items(
260        self, item_id: str
261    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
262        """
263        Get the items for a single recommendation row.
264
265        :param item_id: The item_id of the row, as returned by get_recommendations.
266        """
267        folder: RecommendationFolder | None = None
268        if item_id == MY_WAVE_PLAYLIST_ID:
269            folder = await self._get_my_wave_recommendations()
270        elif item_id == "feed":
271            folder = await self._get_feed_recommendations()
272        elif item_id == "chart":
273            folder = await self._get_chart_recommendations()
274        elif item_id == "new_releases":
275            folder = await self._get_new_releases_recommendations()
276        elif item_id == "new_playlists":
277            folder = await self._get_new_playlists_recommendations()
278        elif item_id == "top_picks":
279            folder = await self._get_top_picks_recommendations()
280        elif item_id == "mood_mix":
281            # the deterministic hourly tag keeps the served items matching the row subtitle
282            if mood_tags := await self._get_valid_tags_for_category("mood"):
283                folder = await self._get_mood_mix_recommendations(
284                    self._rotating_row_tag("mood", mood_tags)
285                )
286        elif item_id == "activity_mix":
287            if activity_tags := await self._get_valid_tags_for_category("activity"):
288                folder = await self._get_activity_mix_recommendations(
289                    self._rotating_row_tag("activity", activity_tags)
290                )
291        elif item_id == "seasonal_mix":
292            folder = await self._get_seasonal_mix_recommendations()
293        if folder is None:
294            return UniqueList()
295        return folder.items
296
297    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
298        """
299        Return Config entries to configure this provider.
300
301        The token is collected by the interactive setup flow (see setup_flow.py); this
302        surface only exposes the genuine playback options.
303        """
304        return (
305            CONF_ENTRY_UNOFFICIAL_PROVIDER,
306            # Quality
307            ConfigEntry(
308                key=CONF_QUALITY,
309                type=ConfigEntryType.STRING,
310                options=[
311                    ConfigValueOption(QUALITY_EFFICIENT),
312                    ConfigValueOption(QUALITY_BALANCED),
313                    ConfigValueOption(QUALITY_HIGH),
314                    ConfigValueOption(QUALITY_LOSSLESS),
315                ],
316                default_value=QUALITY_BALANCED,
317            ),
318            # My Mix maximum tracks (advanced)
319            ConfigEntry(
320                key=CONF_MY_WAVE_MAX_TRACKS,
321                type=ConfigEntryType.INTEGER,
322                range=(10, 1000),
323                default_value=150,
324                required=False,
325                advanced=True,
326            ),
327            # Liked Tracks maximum tracks (advanced)
328            ConfigEntry(
329                key=CONF_LIKED_TRACKS_MAX_TRACKS,
330                type=ConfigEntryType.INTEGER,
331                range=(50, 2000),
332                default_value=500,
333                required=False,
334                advanced=True,
335            ),
336            # Transport mode (advanced)
337            ConfigEntry(
338                key=CONF_TRANSPORT,
339                type=ConfigEntryType.STRING,
340                options=[
341                    ConfigValueOption(TRANSPORT_RAW),
342                    ConfigValueOption(TRANSPORT_ENCRAW),
343                ],
344                default_value=TRANSPORT_RAW,
345                required=False,
346                advanced=True,
347            ),
348            # Custom codecs override (advanced)
349            ConfigEntry(
350                key=CONF_CODECS,
351                type=ConfigEntryType.STRING,
352                default_value="",
353                required=False,
354                advanced=True,
355            ),
356            # API Base URL (advanced)
357            ConfigEntry(
358                key=CONF_BASE_URL,
359                type=ConfigEntryType.STRING,
360                translation_params=[DEFAULT_BASE_URL],
361                default_value=DEFAULT_BASE_URL,
362                required=False,
363                advanced=True,
364            ),
365        )
366
367    async def handle_async_init(self) -> None:
368        """Handle async initialization of the provider."""
369        token = self.get_setup_value(CONF_TOKEN)
370        if not token:
371            raise LoginFailed("No KION Music token provided")
372
373        base_url = self.config.get_value(CONF_BASE_URL, DEFAULT_BASE_URL)
374        self._client = KionMusicClient(str(token), base_url=str(base_url))
375        await self._client.connect()
376        # Suppress kion_music library DEBUG dumps (full API request/response JSON)
377        logging.getLogger("yandex_music").setLevel(self.logger.level + 10)
378        self._streaming = KionMusicStreamingManager(self)
379        # Initialize My Mix duplicate tracking
380        self._my_wave_seen_track_ids = set()
381        self._my_wave_lock = asyncio.Lock()
382        # Initialize per-station wave state dict
383        self._wave_states = {}
384        self._wave_bg_colors = {}
385        self.logger.info("Successfully connected to KION Music")
386
387    async def unload(self, is_removed: bool = False) -> None:
388        """
389        Handle unload/close of the provider.
390
391        :param is_removed: Whether the provider is being removed.
392        """
393        if self._client:
394            await self._client.disconnect()
395        self._client = None
396        self._streaming = None
397        await super().unload(is_removed)
398
399    def get_item_mapping(self, media_type: MediaType | str, key: str, name: str) -> ItemMapping:
400        """
401        Create a generic item mapping.
402
403        :param media_type: The media type.
404        :param key: The item ID.
405        :param name: The item name.
406        :return: An ItemMapping instance.
407        """
408        if isinstance(media_type, str):
409            media_type = MediaType(media_type)
410        return ItemMapping(
411            media_type=media_type,
412            item_id=key,
413            provider=self.instance_id,
414            name=name,
415        )
416
417    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
418        """
419        Browse provider items.
420
421        Root level shows My Mix (personalised radio), For You (picks & mixes),
422        Collection (liked tracks/albums/artists/playlists), Radio (rotor stations
423        by genre/mood/activity/era/local) and AI Mix Sets. Folder labels carry a
424        translation_key so the server localizes them for the connection locale.
425        My Mix tracks use item_id format track_id@station_id for rotor feedback.
426
427        :param path: The path to browse (e.g. provider_id:// or provider_id://waves).
428        """
429        if ProviderFeature.BROWSE not in self.supported_features:
430            raise NotImplementedError
431
432        path_parts = path.split("://")[1].split("/") if "://" in path else []
433        subpath = path_parts[0] if len(path_parts) > 0 else None
434        sub_subpath = path_parts[1] if len(path_parts) > 1 else None
435
436        if subpath == MY_WAVE_PLAYLIST_ID:
437            async with self._my_wave_lock:
438                return await self._browse_my_wave(path, sub_subpath)
439
440        # For You folder (picks + mixes)
441        if subpath == FOR_YOU_FOLDER_ID:
442            return await self._browse_for_you(path, path_parts)
443
444        # Collection folder (library items)
445        if subpath == COLLECTION_FOLDER_ID:
446            return await self._browse_collection(path)
447
448        # Handle picks/ path (mood, activity, era, genres)
449        if subpath == "picks":
450            return await self._browse_picks(path, path_parts)
451
452        # Handle mixes/ path (seasonal collections)
453        if subpath == "mixes":
454            return await self._browse_mixes(path, path_parts)
455
456        # Handle waves/ and radio/ paths (rotor stations by genre/mood/activity)
457        if subpath in (WAVES_FOLDER_ID, RADIO_FOLDER_ID):
458            return await self._browse_waves(path, path_parts)
459
460        # Handle my_waves_set/ path (AI Mix Sets from /landing-blocks/mixes-waves)
461        if subpath == MY_WAVES_SET_FOLDER_ID:
462            return await self._browse_vibe_sets(path, path_parts)
463
464        # Handle waves_landing/ path (Featured Mixes from /landing-blocks/waves)
465        if subpath == WAVES_LANDING_FOLDER_ID:
466            return await self._browse_waves_landing(path, path_parts)
467
468        # Pinned items folder
469        if subpath == PINNED_ITEMS_FOLDER_ID:
470            return await self._browse_pins()
471
472        # Listening history folder
473        if subpath == LISTENING_HISTORY_FOLDER_ID:
474            return await self._browse_history()
475
476        # Handle direct tag subpath (when folder is played by URI, the full path
477        # "picks/category/tag" is lost and only the tag slug arrives as subpath).
478        # Skip the API call for standard top-level folders that are never tag slugs.
479        _known_folders = {
480            "artists",
481            "albums",
482            "tracks",
483            "playlists",
484            LIKED_TRACKS_PLAYLIST_ID,
485            WAVES_FOLDER_ID,
486            RADIO_FOLDER_ID,
487            MY_WAVES_FOLDER_ID,
488            MY_WAVES_SET_FOLDER_ID,
489            WAVES_LANDING_FOLDER_ID,
490            FOR_YOU_FOLDER_ID,
491            COLLECTION_FOLDER_ID,
492            PINNED_ITEMS_FOLDER_ID,
493            LISTENING_HISTORY_FOLDER_ID,
494        }
495        if subpath and subpath not in _known_folders:
496            # Handle direct wave station_id (e.g. "activity:workout") passed when
497            # MA plays a wave station folder using its item_id as the path subpath.
498            # Station IDs have format "category:tag" where category is non-numeric.
499            if ":" in subpath:
500                cat_part = subpath.split(":", 1)[0]
501                if not cat_part.isdigit():
502                    return await self._browse_wave_station(subpath)
503
504            discovered_tags = await self._get_discovered_tag_slugs()
505            if subpath in discovered_tags:
506                return await self._get_tag_playlists_as_browse(subpath)
507
508        if subpath:
509            return await super().browse(path)
510
511        # Each folder carries an English name plus a translation_key; the server localizes the
512        # name for the connection locale at serialization, falling back to the English name.
513        folders: list[BrowseFolder] = []
514        base = path if path.endswith("//") else path.rstrip("/") + "/"
515        # My Mix folder (always enabled — Яндекс «Мой микс»)
516        folders.append(
517            BrowseFolder(
518                item_id=MY_WAVE_PLAYLIST_ID,
519                provider=self.instance_id,
520                path=f"{base}{MY_WAVE_PLAYLIST_ID}",
521                name="My Mix",
522                translation_key=MY_WAVE_PLAYLIST_ID,
523                is_playable=True,
524            )
525        )
526        # For You folder — Picks + Mixes (Яндекс «Для вас»)
527        folders.append(
528            BrowseFolder(
529                item_id=FOR_YOU_FOLDER_ID,
530                provider=self.instance_id,
531                path=f"{base}{FOR_YOU_FOLDER_ID}",
532                name="For You",
533                translation_key=FOR_YOU_FOLDER_ID,
534                is_playable=False,
535            )
536        )
537        # Collection folder — library items (Яндекс «Коллекция»)
538        has_library = any(
539            f in self.supported_features
540            for f in (
541                ProviderFeature.LIBRARY_ARTISTS,
542                ProviderFeature.LIBRARY_ALBUMS,
543                ProviderFeature.LIBRARY_TRACKS,
544                ProviderFeature.LIBRARY_PLAYLISTS,
545            )
546        )
547        if has_library:
548            folders.append(
549                BrowseFolder(
550                    item_id=COLLECTION_FOLDER_ID,
551                    provider=self.instance_id,
552                    path=f"{base}{COLLECTION_FOLDER_ID}",
553                    name="Collection",
554                    translation_key=COLLECTION_FOLDER_ID,
555                    is_playable=False,
556                )
557            )
558        # Radio folder — rotor stations (reuses the shared "Radio" label)
559        folders.append(
560            BrowseFolder(
561                item_id=RADIO_FOLDER_ID,
562                provider=self.instance_id,
563                path=f"{base}{RADIO_FOLDER_ID}",
564                name="Radio",
565                translation_key="radios",
566                is_playable=False,
567            )
568        )
569        # AI Mix Sets — parametric stations from /landing-blocks/mixes-waves
570        folders.append(
571            BrowseFolder(
572                item_id=MY_WAVES_SET_FOLDER_ID,
573                provider=self.instance_id,
574                path=f"{base}{MY_WAVES_SET_FOLDER_ID}",
575                name="AI Mix Sets",
576                translation_key=MY_WAVES_SET_FOLDER_ID,
577                is_playable=False,
578            )
579        )
580        # Pinned items — user-pinned artists/albums/playlists/waves
581        folders.append(
582            BrowseFolder(
583                item_id=PINNED_ITEMS_FOLDER_ID,
584                provider=self.instance_id,
585                path=f"{base}{PINNED_ITEMS_FOLDER_ID}",
586                name="Pinned",
587                translation_key=PINNED_ITEMS_FOLDER_ID,
588                is_playable=False,
589            )
590        )
591        # Listening history — recently played tracks/albums
592        folders.append(
593            BrowseFolder(
594                item_id=LISTENING_HISTORY_FOLDER_ID,
595                provider=self.instance_id,
596                path=f"{base}{LISTENING_HISTORY_FOLDER_ID}",
597                name="Listening History",
598                translation_key=LISTENING_HISTORY_FOLDER_ID,
599                is_playable=False,
600            )
601        )
602        if len(folders) == 1:
603            return await self.browse(folders[0].path)
604        return folders
605
606    def _media_source_name(self, group: str, key: str) -> str | None:
607        """
608        Return the authored English ``name`` for *key* in *group*, or None when not authored.
609
610        :param group: Media translation group (``folder``, ``recommendations`` or ``playlist``).
611        :param key: Authoring key within the group.
612        """
613        return self.mass.translations.get_translation(
614            f"provider.{self.domain}.media.{group}.{key}.name"
615        )
616
617    def _media_label(self, group: str, key: str, fallback: str) -> tuple[str, str | None]:
618        """
619        Map a browse label to the in-code ``name`` and ``translation_key`` for its media item.
620
621        Authored keys return ``(English source name, key)`` — the English name from the
622        provider's ``strings.json`` plus a ``translation_key`` so the server localizes it for
623        the connection locale at serialization. An unauthored key — e.g. a tag discovered from
624        KION's landing API — returns ``(fallback, None)`` so its already-localized name is kept.
625
626        :param group: Media translation group (``folder``, ``recommendations`` or ``playlist``).
627        :param key: Authoring key within the group; also the item's ``translation_key``.
628        :param fallback: English name to use when no string is authored for *key*.
629        """
630        source = self._media_source_name(group, key)
631        if source is None:
632            return fallback, None
633        return source, key
634
635    async def _browse_my_wave(
636        self, path: str, sub_subpath: str | None
637    ) -> list[Track | BrowseFolder]:
638        """
639        Browse My Mix tracks (must be called under _my_wave_lock).
640
641        :param path: Full browse path.
642        :param sub_subpath: Sub-path part ('next' for load more, or track_id cursor).
643        :return: List of Track and optional BrowseFolder for "Load more".
644        """
645        max_tracks_config = int(
646            self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
647        )
648        batch_size_config = MY_WAVE_BATCH_SIZE
649
650        # Effective limit on tracks to collect for this call:
651        # initial browse is capped to BROWSE_INITIAL_TRACKS to avoid marking
652        # extra tracks as "seen" that are never shown to the user.
653        effective_limit = min(
654            BROWSE_INITIAL_TRACKS if sub_subpath != "next" else max_tracks_config,
655            max_tracks_config,
656        )
657
658        # Root my_wave: fetch up to batch_size_config batches so Play adds more tracks.
659        # "Load more" always uses single next batch.
660        max_batches = batch_size_config if sub_subpath != "next" else 1
661
662        # Reset seen tracks on fresh browse (not "load more")
663        if sub_subpath != "next":
664            self._my_wave_seen_track_ids = set()
665
666        queue: str | int | None = None
667        if sub_subpath == "next":
668            queue = self._my_wave_last_track_id
669        elif sub_subpath:
670            queue = sub_subpath
671
672        all_tracks: list[Track | BrowseFolder] = []
673        last_batch_id: str | None = None
674        first_track_id_this_batch: str | None = None
675        total_track_count = 0
676
677        for _ in range(max_batches):
678            if total_track_count >= effective_limit:
679                break
680
681            raw_tracks, batch_id = await self.client.get_my_wave_tracks(queue=queue)
682            if batch_id:
683                self._my_wave_batch_id = batch_id
684                last_batch_id = batch_id
685            if not self._my_wave_radio_started_sent and raw_tracks:
686                sent = await self.client.send_rotor_station_feedback(
687                    ROTOR_STATION_MY_MIX,
688                    "radioStarted",
689                    batch_id=batch_id,
690                )
691                if sent:
692                    self._my_wave_radio_started_sent = True
693            first_track_id_this_batch = None
694            for yt in raw_tracks:
695                if total_track_count >= effective_limit:
696                    break
697
698                track = self._parse_my_wave_track(yt, self._my_wave_seen_track_ids)
699                if track is None:
700                    continue
701                all_tracks.append(track)
702                total_track_count += 1
703
704                track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
705                if first_track_id_this_batch is None:
706                    first_track_id_this_batch = track_id
707
708            if first_track_id_this_batch is not None:
709                self._my_wave_last_track_id = first_track_id_this_batch
710            if (
711                first_track_id_this_batch is None
712                or not batch_id
713                or not raw_tracks
714                or total_track_count >= effective_limit
715            ):
716                break
717            queue = first_track_id_this_batch
718
719        # Only show "Load more" if we haven't reached the limit and there's more data
720        if last_batch_id and total_track_count < max_tracks_config:
721            all_tracks.append(
722                BrowseFolder(
723                    item_id="next",
724                    provider=self.instance_id,
725                    path=f"{path.rstrip('/')}/next",
726                    name="Load more",
727                    translation_key="load_more",
728                    is_playable=False,
729                )
730            )
731        return all_tracks
732
733    def _parse_my_wave_track(self, yt: Any, seen_ids: set[str]) -> Track | None:
734        """
735        Parse a Kion track into a My Mix Track with composite item_id.
736
737        Extracts the track_id, checks for duplicates in the seen_ids set,
738        sets composite item_id (track_id@station_id), and updates provider_mappings.
739        Callers using shared state must hold _my_wave_lock.
740
741        :param yt: Kion track object from rotor station response.
742        :param seen_ids: Set of already-seen track IDs to check and update.
743        :return: Parsed Track with composite item_id, or None if duplicate/invalid.
744        """
745        try:
746            t = parse_track(self, yt)
747        except InvalidDataError as err:
748            self.logger.debug("Error parsing My Mix track: %s", err)
749            return None
750
751        track_id = str(yt.id) if hasattr(yt, "id") and yt.id else getattr(yt, "track_id", None)
752        if not track_id:
753            return t
754
755        if track_id in seen_ids:
756            self.logger.debug("Skipping duplicate My Mix track: %s", track_id)
757            return None
758
759        seen_ids.add(track_id)
760        t.item_id = f"{track_id}{RADIO_TRACK_ID_SEP}{ROTOR_STATION_MY_MIX}"
761        for pm in t.provider_mappings:
762            if pm.provider_instance == self.instance_id:
763                pm.item_id = t.item_id
764                break
765        return t
766
767    @use_cache(3600)
768    async def _validate_tag(self, tag_slug: str) -> bool:
769        """
770        Check if a tag has playlists by calling client.get_tag_playlists().
771
772        :param tag_slug: Tag identifier (e.g. 'chill', '80s').
773        :return: True if the tag has at least one playlist.
774        """
775        try:
776            playlists = await self.client.get_tag_playlists(tag_slug)
777            return len(playlists) > 0
778        except Exception as err:
779            self.logger.debug("Tag validation failed for %s: %s", tag_slug, err)
780            return False
781
782    # allow_expired_cache keeps the items call serving the same (possibly stale) tag
783    # list the rows subtitle was derived from, while a background refresh runs
784    @use_cache(3600, allow_expired_cache=True)
785    async def _get_valid_tags_for_category(self, category: str) -> list[str]:
786        """
787        Get validated tags for a category (only those with playlists).
788
789        Combines hardcoded tags from the category lists with any landing-discovered
790        tags, validates each by calling client.tags(), and returns only those with
791        playlists.
792
793        :param category: Category name ('mood', 'activity', 'era', 'genres').
794        :return: List of valid tag slugs.
795        """
796        category_lists: dict[str, list[str]] = {
797            "mood": list(TAG_CATEGORY_MOOD),
798            "activity": list(TAG_CATEGORY_ACTIVITY),
799            "era": list(TAG_CATEGORY_ERA),
800            "genres": list(TAG_CATEGORY_GENRES),
801        }
802        tags = category_lists.get(category, [])
803
804        # Add landing-discovered tags for this category
805        try:
806            landing_tags = await self.client.get_landing_tags()
807            for slug, _title in landing_tags:
808                cat = TAG_SLUG_CATEGORY.get(slug, "mood")
809                if cat == category and slug not in tags:
810                    tags.append(slug)
811        except Exception as err:
812            self.logger.debug("Landing tag discovery failed: %s", err)
813
814        # Validate tags in parallel with bounded concurrency
815        sem = asyncio.Semaphore(8)
816
817        async def _check(tag: str) -> str | None:
818            async with sem:
819                return tag if await self._validate_tag(tag) else None
820
821        results = await asyncio.gather(*[_check(tag) for tag in tags])
822        return [tag for tag in results if tag is not None]
823
824    @use_cache(3600)
825    async def _get_discovered_tags(self, locale: str) -> list[tuple[str, str]]:
826        """
827        Get all available tags by combining hardcoded tags with landing discovery.
828
829        Starts with all hardcoded tags from category lists, adds landing-discovered
830        tags, validates each via client.tags(), and returns only those with playlists.
831        Results are cached for 1 hour. The locale parameter is included in the cache
832        key so that a locale change invalidates the cached landing titles.
833
834        :param locale: Current metadata locale (used as part of cache key).
835        :return: List of (slug, fallback display name) tuples for tags that have playlists.
836            Hardcoded tags use a derived fallback; landing-discovered tags carry their
837            (already localized) API title. Folder sites attach the translation_key via
838            ``_media_label`` so authored tags localize and discovered ones keep their title.
839        """
840        # Collect all hardcoded tags (non-seasonal)
841        all_tags: dict[str, str] = {}
842        for slug, cat in TAG_SLUG_CATEGORY.items():
843            if cat != "seasonal":
844                all_tags[slug] = slug.title()
845
846        # Add landing-discovered tags
847        try:
848            landing_tags = await self.client.get_landing_tags()
849            for slug, title in landing_tags:
850                if slug not in all_tags:
851                    all_tags[slug] = title
852        except Exception as err:
853            self.logger.debug("Failed to discover tags from landing API: %s", err)
854
855        # Validate tags in parallel with bounded concurrency
856        sem = asyncio.Semaphore(8)
857
858        async def _check(slug: str) -> bool:
859            async with sem:
860                return await self._validate_tag(slug)
861
862        tag_items = list(all_tags.items())
863        results = await asyncio.gather(*[_check(slug) for slug, _ in tag_items])
864        return [
865            (slug, name) for (slug, name), valid in zip(tag_items, results, strict=True) if valid
866        ]
867
868    async def _get_discovered_tag_slugs(self) -> set[str]:
869        """
870        Get set of all valid tag slugs (cached).
871
872        :return: Set of tag slug strings that have playlists.
873        """
874        discovered = await self._get_discovered_tags(self.mass.metadata.locale or "en_US")
875        return {slug for slug, _name in discovered}
876
877    async def _browse_for_you(
878        self, path: str, path_parts: list[str]
879    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
880        """
881        Browse «For You» folder — shows Picks and Mixes sub-folders.
882
883        :param path: Full browse path.
884        :param path_parts: Split path parts after ://.
885        :return: List of sub-folders (Picks, Mixes).
886        """
887        # Strip the for_you segment to build child paths that route to picks/mixes
888        # Path format: ...//for_you  → child paths should be ...//picks, ...//mixes
889        # We build base from the root (before for_you) by dropping the last segment.
890        base_parts = path.split("//", 1)
891        root_base = (base_parts[0] + "//") if len(base_parts) > 1 else path.rstrip("/") + "/"
892
893        if len(path_parts) == 1:
894            return [
895                BrowseFolder(
896                    item_id="picks",
897                    provider=self.instance_id,
898                    path=f"{root_base}picks",
899                    name="Picks",
900                    translation_key="picks",
901                    is_playable=False,
902                ),
903                BrowseFolder(
904                    item_id="mixes",
905                    provider=self.instance_id,
906                    path=f"{root_base}mixes",
907                    name="Mixes",
908                    translation_key="mixes",
909                    is_playable=False,
910                ),
911            ]
912        # Deeper path: delegate to picks or mixes handler via canonical paths
913        return await super().browse(path)
914
915    async def _browse_collection(
916        self, path: str
917    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
918        """
919        Browse «Collection» folder — shows library sub-folders (tracks/artists/albums/playlists).
920
921        :param path: Full browse path.
922        :return: List of library sub-folders.
923        """
924        base_parts = path.split("//", 1)
925        root_base = (base_parts[0] + "//") if len(base_parts) > 1 else path.rstrip("/") + "/"
926
927        folders: list[BrowseFolder] = []
928        for feature, sub_id, label_key, label_name in _COLLECTION_SUBFOLDERS:
929            if feature not in self.supported_features:
930                continue
931            folders.append(
932                BrowseFolder(
933                    item_id=sub_id,
934                    provider=self.instance_id,
935                    path=f"{root_base}{sub_id}",
936                    name=label_name,
937                    translation_key=label_key,
938                    is_playable=True,
939                )
940            )
941        return folders
942
943    async def _browse_pins(self) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
944        """
945        Browse user's pinned items (artists/albums/playlists from Kion Pins).
946
947        Resolves each pin to its full media item via existing single-item lookups.
948        Wave pins are skipped — MA has no native concept for them.
949
950        Pins are resolved concurrently via ``asyncio.gather`` so latency is
951        dominated by the slowest lookup rather than their sum. Individual
952        failures (``MediaNotFoundError`` / ``InvalidDataError``) are skipped
953        without aborting the batch.
954
955        :return: List of resolved media items.
956        """
957        pins_list = await self.client.get_pins()
958        pins = getattr(pins_list, "pins", None) if pins_list else None
959        if not pins:
960            return []
961
962        tasks: list[Coroutine[Any, Any, MediaItemType]] = []
963        pin_descs: list[str] = []
964        for pin in pins:
965            pin_type = getattr(pin, "type", None)
966            data = getattr(pin, "data", None)
967            if data is None:
968                continue
969            if pin_type == "artist_item" and getattr(data, "id", None) is not None:
970                tasks.append(self.get_artist(str(data.id)))
971                pin_descs.append(f"artist:{data.id}")
972            elif pin_type == "album_item" and getattr(data, "id", None) is not None:
973                tasks.append(self.get_album(str(data.id)))
974                pin_descs.append(f"album:{data.id}")
975            elif pin_type == "playlist_item":
976                uid = getattr(data, "uid", None)
977                kind = getattr(data, "kind", None)
978                if uid is not None and kind is not None:
979                    tasks.append(self.get_playlist(f"{uid}:{kind}"))
980                    pin_descs.append(f"playlist:{uid}:{kind}")
981
982        if not tasks:
983            return []
984
985        results = await asyncio.gather(*tasks, return_exceptions=True)
986        items: list[MediaItemType] = []
987        for desc, result in zip(pin_descs, results, strict=True):
988            if isinstance(result, (MediaNotFoundError, InvalidDataError)):
989                self.logger.debug("Skipping pin %s: %s", desc, result)
990            elif isinstance(result, BaseException):
991                raise result
992            else:
993                items.append(result)
994        return items
995
996    async def _browse_history(self) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
997        """
998        Browse user's recent listening history (flattened across days).
999
1000        Filters to ``type == "track"`` entries only — album/playlist context
1001        items in the history feed are dropped. Tracks are de-duplicated by
1002        id and returned in most-recent-first order.
1003
1004        :return: List of recently played Track items.
1005        """
1006        history = await self.client.get_music_history()
1007        tabs = getattr(history, "history_tabs", None) if history else None
1008        if not tabs:
1009            return []
1010
1011        seen_track_ids: set[str] = set()
1012        tracks: list[Track] = []
1013        for tab in tabs:
1014            groups = getattr(tab, "items", None) or []
1015            for group in groups:
1016                history_items = getattr(group, "tracks", None) or []
1017                for hist_item in history_items:
1018                    if getattr(hist_item, "type", None) != "track":
1019                        continue
1020                    full = getattr(getattr(hist_item, "data", None), "full_model", None)
1021                    if full is None or getattr(full, "id", None) is None:
1022                        continue
1023                    track_key = str(full.id)
1024                    if track_key in seen_track_ids:
1025                        continue
1026                    seen_track_ids.add(track_key)
1027                    try:
1028                        tracks.append(parse_track(self, full))
1029                    except InvalidDataError as err:
1030                        self.logger.debug("Skipping history track: %s", err)
1031        return tracks
1032
1033    async def _browse_picks(
1034        self, path: str, path_parts: list[str]
1035    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1036        """
1037        Browse picks folder using hardcoded tags validated against the API.
1038
1039        Tags are sourced from hardcoded category lists and landing API discovery,
1040        then validated via client.tags() to ensure they have playlists.
1041        Only categories with at least one valid tag are shown.
1042
1043        :param path: Full browse path.
1044        :param path_parts: Split path parts after ://.
1045        :return: List of folders or playlists.
1046        """
1047        base = path.rstrip("/") + "/"
1048
1049        # Get validated tags
1050        discovered = await self._get_discovered_tags(self.mass.metadata.locale or "en_US")
1051
1052        # Categorize valid tags, carrying each tag's (slug, fallback display name)
1053        categorized: dict[str, list[tuple[str, str]]] = {}
1054        for slug, fallback_name in discovered:
1055            cat = TAG_SLUG_CATEGORY.get(slug, "mood")
1056            # Skip seasonal tags — they belong in mixes, not picks
1057            if cat == "seasonal":
1058                continue
1059            categorized.setdefault(cat, []).append((slug, fallback_name))
1060
1061        # Sort tags within each category by preferred order
1062        for cat, cat_tags in categorized.items():
1063            order = TAG_CATEGORY_ORDER.get(cat, [])
1064            order_map = {s: i for i, s in enumerate(order)}
1065            cat_tags.sort(key=lambda t: order_map.get(t[0], len(order)))
1066
1067        # picks/ - show category folders (only those with valid tags)
1068        if len(path_parts) == 1:
1069            category_display_order = ["mood", "activity", "era", "genres"]
1070            folders: list[BrowseFolder] = []
1071            for cat in category_display_order:
1072                if cat in categorized:
1073                    name, translation_key = self._media_label("folder", cat, cat.title())
1074                    folders.append(
1075                        BrowseFolder(
1076                            item_id=cat,
1077                            provider=self.instance_id,
1078                            path=f"{base}{cat}",
1079                            name=name,
1080                            translation_key=translation_key,
1081                            is_playable=False,
1082                        )
1083                    )
1084            # Show any extra categories not in the standard order
1085            for cat in categorized:
1086                if cat not in category_display_order:
1087                    name, translation_key = self._media_label("folder", cat, cat.title())
1088                    folders.append(
1089                        BrowseFolder(
1090                            item_id=cat,
1091                            provider=self.instance_id,
1092                            path=f"{base}{cat}",
1093                            name=name,
1094                            translation_key=translation_key,
1095                            is_playable=False,
1096                        )
1097                    )
1098            return folders
1099
1100        category: str | None = path_parts[1] if len(path_parts) > 1 else None
1101        tag: str | None = path_parts[2] if len(path_parts) > 2 else None
1102
1103        self.logger.debug(
1104            "Browse picks: path=%s, category=%s, tag=%s",
1105            path,
1106            category,
1107            tag,
1108        )
1109
1110        # picks/category/ - show valid tag folders for this category
1111        if category and not tag:
1112            category_tags = categorized.get(category, [])
1113            folders = []
1114            for slug, fallback_name in category_tags:
1115                name, translation_key = self._media_label(
1116                    "folder", _media_label_key(slug), fallback_name
1117                )
1118                folders.append(
1119                    BrowseFolder(
1120                        item_id=slug,
1121                        provider=self.instance_id,
1122                        path=f"{base}{slug}",
1123                        name=name,
1124                        translation_key=translation_key,
1125                        is_playable=False,
1126                    )
1127                )
1128            self.logger.debug("Returning %d tag folders for category %s", len(folders), category)
1129            return folders
1130
1131        # picks/category/tag - show playlists for the tag
1132        if tag:
1133            discovered_slugs = {slug for slug, _name in discovered}
1134            if tag in discovered_slugs:
1135                self.logger.debug("Fetching playlists for tag: %s", tag)
1136                return await self._get_tag_playlists_as_browse(tag)
1137
1138        self.logger.debug("No match found, returning empty list")
1139        return []
1140
1141    async def _browse_mixes(
1142        self, path: str, path_parts: list[str]
1143    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1144        """
1145        Browse mixes folder (seasonal collections) using hardcoded tags.
1146
1147        Uses TAG_MIXES directly and validates each tag via client.tags()
1148        to check if it has playlists. Does not depend on landing API discovery.
1149
1150        :param path: Full browse path.
1151        :param path_parts: Split path parts after ://.
1152        :return: List of folders or playlists.
1153        """
1154        base = path.rstrip("/") + "/"
1155
1156        # Validate seasonal tags in parallel (no landing dependency)
1157        sem = asyncio.Semaphore(5)
1158
1159        async def _check(tag: str) -> str | None:
1160            async with sem:
1161                return tag if await self._validate_tag(tag) else None
1162
1163        results = await asyncio.gather(*[_check(t) for t in TAG_MIXES])
1164        available_mixes = [t for t in results if t is not None]
1165
1166        # mixes/ - show seasonal folders (only valid ones)
1167        if len(path_parts) == 1:
1168            folders = []
1169            for t in available_mixes:
1170                name, translation_key = self._media_label("folder", t, t.title())
1171                folders.append(
1172                    BrowseFolder(
1173                        item_id=t,
1174                        provider=self.instance_id,
1175                        path=f"{base}{t}",
1176                        name=name,
1177                        translation_key=translation_key,
1178                        is_playable=False,
1179                    )
1180                )
1181            return folders
1182
1183        # mixes/tag - show playlists for the tag
1184        tag = path_parts[1] if len(path_parts) > 1 else None
1185        if tag and tag in TAG_MIXES:
1186            return await self._get_tag_playlists_as_browse(tag)
1187
1188        return []
1189
1190    def _get_wave_state(self, station_id: str) -> _WaveState:
1191        """
1192        Get or create per-station wave state.
1193
1194        :param station_id: Rotor station ID (e.g. 'genre:rock', 'mood:chill').
1195        :return: _WaveState instance for this station.
1196        """
1197        return self._wave_states.setdefault(station_id, _WaveState())
1198
1199    async def _browse_waves(
1200        self, path: str, path_parts: list[str]
1201    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1202        """
1203        Browse waves folder (rotor stations by genre/mood/activity/epoch/local).
1204
1205        Fetches available stations from the Kion rotor API and groups them by category.
1206
1207        :param path: Full browse path.
1208        :param path_parts: Split path parts after ://.
1209        :return: List of folders or tracks.
1210        """
1211        base = path.rstrip("/") + "/"
1212
1213        # Station names come back from the rotor API already localized; the metadata locale
1214        # selects the API content language (not the folder labels, which use translation_key).
1215        locale = (self.mass.metadata.locale or "en_US").lower()
1216        language = "ru" if locale.startswith("ru") else "en"
1217
1218        all_stations = await self.client.get_wave_stations(language)
1219
1220        # Group stations by category, preserving image_url
1221        categorized: dict[str, list[tuple[str, str, str | None]]] = {}
1222        for station_id, cat_key, station_name, image_url in all_stations:
1223            categorized.setdefault(cat_key, []).append((station_id, station_name, image_url))
1224
1225        # waves/ — show category folders
1226        if len(path_parts) == 1:
1227            folders: list[BrowseFolder] = []
1228            # Personalized stations first — only show if dashboard returns stations
1229            dashboard_stations = await self._get_dashboard_stations_cached()
1230            if dashboard_stations:
1231                name, translation_key = self._media_label("folder", MY_WAVES_FOLDER_ID, "Personal")
1232                folders.append(
1233                    BrowseFolder(
1234                        item_id=MY_WAVES_FOLDER_ID,
1235                        provider=self.instance_id,
1236                        path=f"{base}{MY_WAVES_FOLDER_ID}",
1237                        name=name,
1238                        translation_key=translation_key,
1239                        is_playable=False,
1240                    )
1241                )
1242            # Featured Mixes — only show if landing-blocks/waves returns data
1243            waves_landing = await self._get_waves_landing_cached()
1244            if waves_landing:
1245                name, translation_key = self._media_label(
1246                    "folder", WAVES_LANDING_FOLDER_ID, "Featured Mixes"
1247                )
1248                folders.append(
1249                    BrowseFolder(
1250                        item_id=WAVES_LANDING_FOLDER_ID,
1251                        provider=self.instance_id,
1252                        path=f"{base}{WAVES_LANDING_FOLDER_ID}",
1253                        name=name,
1254                        translation_key=translation_key,
1255                        is_playable=False,
1256                    )
1257                )
1258            for cat in WAVE_CATEGORY_DISPLAY_ORDER:
1259                if cat in categorized:
1260                    name, translation_key = self._media_label("folder", cat, cat.title())
1261                    folders.append(
1262                        BrowseFolder(
1263                            item_id=cat,
1264                            provider=self.instance_id,
1265                            path=f"{base}{cat}",
1266                            name=name,
1267                            translation_key=translation_key,
1268                            is_playable=False,
1269                        )
1270                    )
1271            # Append any categories returned by API that aren't in the predefined order
1272            for cat in categorized:
1273                if cat not in WAVE_CATEGORY_DISPLAY_ORDER:
1274                    name, translation_key = self._media_label("folder", cat, cat.title())
1275                    folders.append(
1276                        BrowseFolder(
1277                            item_id=cat,
1278                            provider=self.instance_id,
1279                            path=f"{base}{cat}",
1280                            name=name,
1281                            translation_key=translation_key,
1282                            is_playable=False,
1283                        )
1284                    )
1285            return folders
1286
1287        category: str | None = path_parts[1] if len(path_parts) > 1 else None
1288        tag: str | None = path_parts[2] if len(path_parts) > 2 else None
1289
1290        # waves/my_waves/ — show personalized stations from dashboard
1291        if category == MY_WAVES_FOLDER_ID and not tag:
1292            return await self._browse_my_waves_stations(path)
1293
1294        # waves/waves_landing/... — redirect to Featured Mixes browse
1295        if category == WAVES_LANDING_FOLDER_ID:
1296            return await self._browse_waves_landing(path, path_parts[1:])
1297
1298        # waves/my_waves/<tag>[/next] — play a specific personal station
1299        # The full station_id has format "genre:allrock", not "my_waves:allrock".
1300        # Resolve by matching against dashboard stations cache.
1301        if category == MY_WAVES_FOLDER_ID and tag:
1302            dashboard_stations = await self._get_dashboard_stations_cached()
1303            for sid, _, _ in dashboard_stations:
1304                sid_tag = sid.split(":", 1)[1] if ":" in sid else sid
1305                if sid_tag == tag:
1306                    return await self._browse_wave_station(sid, path=path)
1307            # Fallback: try tag as direct station_id (e.g. "genre:allrock" passed verbatim)
1308            if ":" in tag:
1309                return await self._browse_wave_station(tag, path=path)
1310            return []
1311
1312        # waves/<category>/ — show station folders with artwork
1313        if category and not tag:
1314            cat_stations = categorized.get(category, [])
1315            folders = []
1316            for station_id, station_name, image_url in cat_stations:
1317                tag_part = station_id.split(":", 1)[1] if ":" in station_id else station_id
1318                station_image: MediaItemImage | None = None
1319                if image_url:
1320                    station_image = MediaItemImage(
1321                        type=ImageType.THUMB,
1322                        path=image_url,
1323                        provider=self.instance_id,
1324                        remotely_accessible=True,
1325                    )
1326                folders.append(
1327                    BrowseFolder(
1328                        item_id=station_id,
1329                        provider=self.instance_id,
1330                        path=f"{base}{tag_part}",
1331                        name=station_name,
1332                        is_playable=True,
1333                        image=station_image,
1334                    )
1335                )
1336            return folders
1337
1338        # waves/<category>/<tag>[/next] — stream tracks from rotor station
1339        if category and tag:
1340            station_id = f"{category}:{tag}"
1341            return await self._browse_wave_station(station_id, path=path)
1342
1343        return []
1344
1345    @use_cache(600)
1346    async def _get_dashboard_stations_cached(self) -> list[tuple[str, str, str | None]]:
1347        """
1348        Get personalized dashboard stations, cached for 10 minutes.
1349
1350        :return: List of (station_id, name, image_url) tuples.
1351        """
1352        return await self.client.get_dashboard_stations()
1353
1354    async def _browse_my_waves_stations(self, path: str) -> list[BrowseFolder]:
1355        """
1356        Browse personalized wave stations from rotor/stations/dashboard.
1357
1358        Names are resolved from the non-personalized station list so that
1359        stations show their actual genre/mood name (e.g. "Рок") rather than
1360        the generic "Мой микс" label that the dashboard API returns.
1361
1362        :param path: Full browse path (used to build sub-paths).
1363        :return: List of playable BrowseFolder items, one per station.
1364        """
1365        stations = await self._get_dashboard_stations_cached()
1366
1367        # Build a name map from the non-personalized list for proper localized names.
1368        locale = (self.mass.metadata.locale or "en_US").lower()
1369        language = "ru" if locale.startswith("ru") else "en"
1370        all_stations = await self.client.get_wave_stations(language)
1371        station_name_map: dict[str, str] = {sid: name for sid, _, name, _ in all_stations}
1372
1373        base = path.rstrip("/") + "/"
1374        folders: list[BrowseFolder] = []
1375        for station_id, fallback_name, image_url in stations:
1376            # Use full station_id (e.g. "genre:rock") in path to avoid collisions
1377            # when two stations share the same tag but differ by category.
1378            # The routing fallback (if ":" in tag) handles this correctly.
1379            name = station_name_map.get(station_id, fallback_name)
1380            station_image: MediaItemImage | None = None
1381            if image_url:
1382                station_image = MediaItemImage(
1383                    type=ImageType.THUMB,
1384                    path=image_url,
1385                    provider=self.instance_id,
1386                    remotely_accessible=True,
1387                )
1388            folders.append(
1389                BrowseFolder(
1390                    item_id=station_id,
1391                    provider=self.instance_id,
1392                    path=f"{base}{station_id}",
1393                    name=name,
1394                    is_playable=True,
1395                    image=station_image,
1396                )
1397            )
1398        return folders
1399
1400    async def _browse_wave_station(
1401        self, station_id: str, path: str = ""
1402    ) -> list[Track | BrowseFolder]:
1403        """
1404        Browse a rotor wave station and return tracks.
1405
1406        Fetches tracks from the rotor station, deduplicates within the current session,
1407        and sends radioStarted feedback on first call. Appends a "Load more" BrowseFolder
1408        at the end so MA can continue fetching the next batch automatically (radio mode).
1409
1410        :param station_id: Rotor station ID (e.g. 'genre:rock', 'mood:chill').
1411        :param path: Current browse path, used to construct the "Load more" next path.
1412        :return: List of Track objects with composite item_id (track_id@station_id),
1413                 followed by a "Load more" BrowseFolder if more tracks are available.
1414        """
1415        state = self._get_wave_state(station_id)
1416        async with state.lock:
1417            max_tracks = int(
1418                self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
1419            )
1420
1421            self.logger.debug(
1422                "Browse wave station: station_id=%s path=%s last_track_id=%s",
1423                station_id,
1424                path,
1425                state.last_track_id,
1426            )
1427            raw_tracks, batch_id = await self.client.get_rotor_station_tracks(
1428                station_id, queue=state.last_track_id
1429            )
1430            if batch_id:
1431                state.batch_id = batch_id
1432
1433            if not state.radio_started_sent and raw_tracks:
1434                sent = await self.client.send_rotor_station_feedback(
1435                    station_id,
1436                    "radioStarted",
1437                    batch_id=batch_id,
1438                )
1439                if sent:
1440                    state.radio_started_sent = True
1441
1442            tracks: list[Track] = []
1443            first_track_id: str | None = None
1444            for yt in raw_tracks:
1445                if len(state.seen_track_ids) >= max_tracks:
1446                    break
1447                track = self._parse_my_wave_track(yt, state.seen_track_ids)
1448                if track is None:
1449                    continue
1450                # Override station_id in composite item_id to reflect this specific station
1451                old_item_id = track.item_id
1452                track_id = old_item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
1453                track.item_id = f"{track_id}{RADIO_TRACK_ID_SEP}{station_id}"
1454                # Keep provider mappings in sync with the new item_id
1455                for pm in getattr(track, "provider_mappings", []):
1456                    if (
1457                        getattr(pm, "item_id", None) == old_item_id
1458                        and getattr(pm, "provider_instance", None) == self.instance_id
1459                    ):
1460                        pm.item_id = track.item_id
1461                if first_track_id is None:
1462                    first_track_id = track_id
1463                tracks.append(track)
1464
1465            if first_track_id is not None:
1466                state.last_track_id = first_track_id
1467
1468            self.logger.debug(
1469                "Wave station %s returned %d tracks: %s",
1470                station_id,
1471                len(tracks),
1472                [t.item_id.split(RADIO_TRACK_ID_SEP, 1)[0] for t in tracks[:5]],
1473            )
1474            result: list[Track | BrowseFolder] = list(tracks)
1475
1476            # Append "Load more" sentinel so MA knows to call browse again for next batch.
1477            # This mirrors the My Mix mechanism and enables continuous radio playback.
1478            if tracks and len(state.seen_track_ids) < max_tracks and path:
1479                # Append /next to the current path (same pattern as _browse_my_wave).
1480                # This makes each "Load more" path unique (e.g. /next/next/next...)
1481                # so MA never serves a cached result for subsequent presses.
1482                result.append(
1483                    BrowseFolder(
1484                        item_id="next",
1485                        provider=self.instance_id,
1486                        path=f"{path.rstrip('/')}/next",
1487                        name="Load more",
1488                        translation_key="load_more",
1489                        is_playable=False,
1490                    )
1491                )
1492
1493            return result
1494
1495    @staticmethod
1496    def _extract_wave_item_cover(item: dict[str, Any]) -> tuple[str | None, str | None]:
1497        """
1498        Extract cover URI and background color from a wave/mix item.
1499
1500        :param item: Wave or mix item dict from the API.
1501        :return: (cover_uri, bg_color) tuple where bg_color is a hex string or None.
1502        """
1503        agent_uri = item.get("agent", {}).get("cover", {}).get("uri", "")
1504        cover_uri = agent_uri or item.get("compact_image_url")
1505        bg_color = item.get("colors", {}).get("average")
1506        return cover_uri, bg_color
1507
1508    @use_cache(3600)
1509    async def _get_mixes_waves_cached(self) -> list[dict[str, Any]] | None:
1510        """
1511        Get AI Wave Set data from /landing-blocks/mixes-waves, cached for 1 hour.
1512
1513        :return: List of mix category dicts from the API, or None on error.
1514        """
1515        return await self.client.get_mixes_waves()
1516
1517    @use_cache(3600)
1518    async def _get_waves_landing_cached(self) -> list[dict[str, Any]] | None:
1519        """
1520        Get Featured Mixes data from /landing-blocks/waves, cached for 1 hour.
1521
1522        :return: List of wave category dicts from the API, or None on error.
1523        """
1524        return await self.client.get_waves_landing()
1525
1526    async def _browse_waves_landing(
1527        self, path: str, path_parts: list[str]
1528    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1529        """
1530        Browse Featured Mixes (from /landing-blocks/waves).
1531
1532        :param path: Full browse path.
1533        :param path_parts: Split path parts after ://.
1534        :return: List of folders or tracks.
1535        """
1536        waves_data = await self._get_waves_landing_cached()
1537        return await self._browse_wave_categories(
1538            path, path_parts, waves_data or [], WAVES_LANDING_FOLDER_ID
1539        )
1540
1541    async def _browse_wave_categories(
1542        self,
1543        path: str,
1544        path_parts: list[str],
1545        categories_data: list[dict[str, Any]],
1546        id_prefix: str,
1547    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1548        """
1549        Browse wave-like category folders and their station items.
1550
1551        Shared logic for both 'my_waves_set' browse trees:
1552        - Level 1 (e.g. my_waves_set/): category folders
1553        - Level 2 (e.g. my_waves_set/ai-sets/): playable station folders with artwork
1554        - Level 3+ (e.g. my_waves_set/ai-sets/genre:rock[/next]): track listing
1555
1556        :param path: Full browse path.
1557        :param path_parts: Split path parts after ://.
1558        :param categories_data: List of category dicts from the API.
1559        :param id_prefix: Prefix for BrowseFolder item_id (e.g. 'my_waves_set').
1560        :return: List of folders or tracks.
1561        """
1562        base = path.rstrip("/") + "/"
1563
1564        if not categories_data:
1565            return []
1566
1567        # Level 1 → category folders
1568        if len(path_parts) == 1:
1569            folders: list[BrowseFolder] = []
1570            for wave_category in categories_data:
1571                cat_id = wave_category.get("id", "")
1572                cat_title = wave_category.get("title", "")
1573                items = wave_category.get("items", [])
1574                if not items or not cat_id:
1575                    continue
1576                display_name = cat_title.capitalize() if cat_title else cat_id.capitalize()
1577                folders.append(
1578                    BrowseFolder(
1579                        item_id=f"{id_prefix}_{cat_id}",
1580                        provider=self.instance_id,
1581                        path=f"{base}{cat_id}",
1582                        name=display_name,
1583                        is_playable=False,
1584                    )
1585                )
1586            return folders
1587
1588        category_id = path_parts[1] if len(path_parts) > 1 else None
1589        if not category_id:
1590            return []
1591
1592        # Level 3+ → stream tracks from rotor station
1593        if len(path_parts) > 2:
1594            station_id = path_parts[2]
1595            return await self._browse_wave_station(station_id, path=path)
1596
1597        # Level 2 → playable station folders with artwork
1598        for wave_category in categories_data:
1599            if wave_category.get("id") == category_id:
1600                items = wave_category.get("items", [])
1601                result: list[BrowseFolder] = []
1602                for item in items:
1603                    station_id = item.get("station_id", "")
1604                    title = item.get("title", "")
1605                    if not station_id or not title:
1606                        continue
1607                    cover_uri, bg_color = self._extract_wave_item_cover(item)
1608                    image: MediaItemImage | None = None
1609                    if cover_uri:
1610                        if cover_uri.startswith("http"):
1611                            img_url: str = cover_uri.replace("%%", IMAGE_SIZE_MEDIUM)
1612                        else:
1613                            raw = get_image_url(cover_uri)
1614                            img_url = "" if raw is None else raw
1615                        if img_url:
1616                            if bg_color:
1617                                # Append bg_color as URL fragment for cache-key uniqueness.
1618                                # MA will call resolve_image() to composite the transparent PNG.
1619                                if len(self._wave_bg_colors) > 200:
1620                                    self._wave_bg_colors.clear()
1621                                img_url = f"{img_url}#{bg_color.lstrip('#')}"
1622                                self._wave_bg_colors[img_url] = bg_color
1623                            image = MediaItemImage(
1624                                type=ImageType.THUMB,
1625                                path=img_url,
1626                                provider=self.instance_id,
1627                                remotely_accessible=bg_color is None,
1628                            )
1629                    result.append(
1630                        BrowseFolder(
1631                            item_id=station_id,
1632                            provider=self.instance_id,
1633                            path=f"{base}{station_id}",
1634                            name=title,
1635                            is_playable=True,
1636                            image=image,
1637                        )
1638                    )
1639                return result
1640
1641        return []
1642
1643    async def _browse_vibe_sets(
1644        self, path: str, path_parts: list[str]
1645    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1646        """
1647        Browse AI Mix Sets (from /landing-blocks/mixes-waves).
1648
1649        :param path: Full browse path.
1650        :param path_parts: Split path parts after ://.
1651        :return: List of folders or tracks.
1652        """
1653        mixes_data = await self._get_mixes_waves_cached()
1654        return await self._browse_wave_categories(
1655            path, path_parts, mixes_data or [], MY_WAVES_SET_FOLDER_ID
1656        )
1657
1658    @use_cache(600)
1659    async def _get_tag_playlists_as_browse(
1660        self, tag_id: str
1661    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1662        """
1663        Get playlists for a tag and return as browse items.
1664
1665        :param tag_id: Tag identifier (e.g. 'chill', '80s').
1666        :return: List of Playlist objects.
1667        """
1668        self.logger.debug("Fetching playlists for tag: %s", tag_id)
1669        playlists = await self.client.get_tag_playlists(tag_id)
1670        self.logger.debug("Got %d playlists for tag %s", len(playlists), tag_id)
1671        result: list[Playlist] = []
1672        for playlist in playlists:
1673            try:
1674                result.append(parse_playlist(self, playlist))
1675            except InvalidDataError as err:
1676                self.logger.debug("Error parsing tag playlist: %s", err)
1677        self.logger.debug("Parsed %d playlists for tag %s", len(result), tag_id)
1678        return result
1679
1680    # Search
1681
1682    @use_cache(3600 * 24 * 14)
1683    async def search(
1684        self, search_query: str, media_types: list[MediaType], limit: int = 5
1685    ) -> SearchResults:
1686        """
1687        Perform search on KION Music.
1688
1689        :param search_query: The search query.
1690        :param media_types: List of media types to search for.
1691        :param limit: Maximum number of results per type.
1692        :return: SearchResults with found items.
1693        """
1694        result = SearchResults()
1695
1696        # Determine search type based on requested media types
1697        # Map MediaType to Kion API search type
1698        type_mapping = {
1699            MediaType.TRACK: "track",
1700            MediaType.ALBUM: "album",
1701            MediaType.ARTIST: "artist",
1702            MediaType.PLAYLIST: "playlist",
1703        }
1704        requested_types = [type_mapping[mt] for mt in media_types if mt in type_mapping]
1705
1706        # Use specific type if only one requested, otherwise search all
1707        search_type = requested_types[0] if len(requested_types) == 1 else "all"
1708
1709        search_result = await self.client.search(search_query, search_type=search_type, limit=limit)
1710        if not search_result:
1711            return result
1712
1713        # Parse tracks
1714        if MediaType.TRACK in media_types and search_result.tracks:
1715            for track in search_result.tracks.results[:limit]:
1716                try:
1717                    result.tracks = [*result.tracks, parse_track(self, track)]
1718                except InvalidDataError as err:
1719                    self.logger.debug("Error parsing track: %s", err)
1720
1721        # Parse albums
1722        if MediaType.ALBUM in media_types and search_result.albums:
1723            for album in search_result.albums.results[:limit]:
1724                try:
1725                    result.albums = [*result.albums, parse_album(self, album)]
1726                except InvalidDataError as err:
1727                    self.logger.debug("Error parsing album: %s", err)
1728
1729        # Parse artists
1730        if MediaType.ARTIST in media_types and search_result.artists:
1731            for artist in search_result.artists.results[:limit]:
1732                try:
1733                    result.artists = [*result.artists, parse_artist(self, artist)]
1734                except InvalidDataError as err:
1735                    self.logger.debug("Error parsing artist: %s", err)
1736
1737        # Parse playlists
1738        if MediaType.PLAYLIST in media_types and search_result.playlists:
1739            for playlist in search_result.playlists.results[:limit]:
1740                try:
1741                    result.playlists = [*result.playlists, parse_playlist(self, playlist)]
1742                except InvalidDataError as err:
1743                    self.logger.debug("Error parsing playlist: %s", err)
1744
1745        return result
1746
1747    # Get single items
1748
1749    @use_cache(3600 * 24 * 30)
1750    async def get_artist(self, prov_artist_id: str) -> Artist:
1751        """
1752        Get artist details by ID, enriched with description and listener stats.
1753
1754        :param prov_artist_id: The provider artist ID.
1755        :return: Artist object.
1756        :raises MediaNotFoundError: If artist not found.
1757        """
1758        artist, about = await asyncio.gather(
1759            self.client.get_artist(prov_artist_id),
1760            self.client.get_artist_about(prov_artist_id),
1761        )
1762        if not artist:
1763            raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
1764        return parse_artist(self, artist, about=about)
1765
1766    @use_cache(3600 * 24 * 30)
1767    async def get_album(self, prov_album_id: str) -> Album:
1768        """
1769        Get album details by ID.
1770
1771        :param prov_album_id: The provider album ID.
1772        :return: Album object.
1773        :raises MediaNotFoundError: If album not found.
1774        """
1775        album = await self.client.get_album(prov_album_id)
1776        if not album:
1777            raise MediaNotFoundError(f"Album {prov_album_id} not found")
1778        return parse_album(self, album)
1779
1780    async def get_track(self, prov_track_id: str) -> Track:
1781        """
1782        Get track details by ID.
1783
1784        Supports composite item_id (track_id@station_id) for My Mix tracks;
1785        only the track_id part is used for the API. Normalizes the ID before
1786        caching to avoid duplicate cache entries.
1787
1788        :param prov_track_id: The provider track ID (or track_id@station_id).
1789        :return: Track object.
1790        :raises MediaNotFoundError: If track not found.
1791        """
1792        track_id, _ = _parse_radio_item_id(prov_track_id)
1793        return await self._get_track_cached(track_id)
1794
1795    @use_cache(3600 * 24 * 30)
1796    async def _get_track_cached(self, track_id: str) -> Track:
1797        """
1798        Get track details by normalized ID (cached).
1799
1800        :param track_id: Normalized track ID (without station suffix).
1801        :return: Track object.
1802        :raises MediaNotFoundError: If track not found.
1803        """
1804        raw_track = await self.client.get_track(track_id)
1805        if not raw_track:
1806            raise MediaNotFoundError(f"Track {track_id} not found")
1807
1808        # Use the already-fetched track object to avoid a duplicate API call
1809        lyrics, lyrics_synced = await self.client.get_track_lyrics_from_track(raw_track)
1810
1811        return parse_track(self, raw_track, lyrics=lyrics, lyrics_synced=lyrics_synced)
1812
1813    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
1814        """
1815        Get playlist details by ID.
1816
1817        Supports virtual playlists MY_WAVE_PLAYLIST_ID (My Mix) and
1818        LIKED_TRACKS_PLAYLIST_ID (Liked Tracks). Real playlists use format "owner_id:kind".
1819
1820        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind",
1821            my_wave, or liked_tracks).
1822        :return: Playlist object.
1823        :raises MediaNotFoundError: If playlist not found.
1824        """
1825        # Virtual playlists - constructed locally (no API call). translation_key localizes
1826        # the name for the connection locale at serialization; the English name is kept (not
1827        # dropped like browse/recommendation folders) because a playable item's name is also
1828        # read outside outbound API serialization (e.g. queue / now-playing metadata).
1829        if prov_playlist_id == MY_WAVE_PLAYLIST_ID:
1830            return Playlist(
1831                item_id=MY_WAVE_PLAYLIST_ID,
1832                provider=self.instance_id,
1833                name="My Mix",
1834                translation_key=MY_WAVE_PLAYLIST_ID,
1835                owner=get_canonical_provider_name(self),
1836                provider_mappings={
1837                    ProviderMapping(
1838                        item_id=MY_WAVE_PLAYLIST_ID,
1839                        provider_domain=self.domain,
1840                        provider_instance=self.instance_id,
1841                        is_unique=True,
1842                    )
1843                },
1844                is_editable=False,
1845            )
1846
1847        if prov_playlist_id == LIKED_TRACKS_PLAYLIST_ID:
1848            return Playlist(
1849                item_id=LIKED_TRACKS_PLAYLIST_ID,
1850                provider=self.instance_id,
1851                name="My Favorites",
1852                translation_key=LIKED_TRACKS_PLAYLIST_ID,
1853                owner=get_canonical_provider_name(self),
1854                provider_mappings={
1855                    ProviderMapping(
1856                        item_id=LIKED_TRACKS_PLAYLIST_ID,
1857                        provider_domain=self.domain,
1858                        provider_instance=self.instance_id,
1859                        is_unique=True,
1860                    )
1861                },
1862                is_editable=False,
1863            )
1864
1865        # Real playlists - use cached method
1866        return await self._get_real_playlist(prov_playlist_id)
1867
1868    @use_cache(3600 * 24 * 30)
1869    async def _get_real_playlist(self, prov_playlist_id: str) -> Playlist:
1870        """
1871        Get real playlist details by ID (cached).
1872
1873        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind").
1874        :return: Playlist object.
1875        :raises MediaNotFoundError: If playlist not found.
1876        """
1877        # Parse the playlist ID (format: owner_id:kind)
1878        if PLAYLIST_ID_SPLITTER in prov_playlist_id:
1879            owner_id, kind = prov_playlist_id.split(PLAYLIST_ID_SPLITTER, 1)
1880        else:
1881            owner_id = str(self.client.user_id)
1882            kind = prov_playlist_id
1883
1884        playlist = await self.client.get_playlist(owner_id, kind)
1885        if not playlist:
1886            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found")
1887        return parse_playlist(self, playlist)
1888
1889    # single_flight=False: this advances the rotor cursor and sends a one-shot "radioStarted"
1890    # feedback, so concurrent callers must each get their own batch instead of sharing one
1891    @use_cache(3600 * 3, allow_expired_cache=True, single_flight=False)
1892    async def _get_my_wave_playlist_tracks(self, page: int) -> list[Track]:
1893        """
1894        Get My Mix tracks for virtual playlist (uses cursor for page > 0).
1895
1896        Fetches MY_WAVE_BATCH_SIZE Rotor API batches per page call to reduce
1897        the number of round-trips when the player controller paginates through pages.
1898
1899        :param page: Page number (0 = first batch, 1+ = next batches via queue cursor).
1900        :return: List of Track objects for this page.
1901        """
1902        async with self._my_wave_lock:
1903            max_tracks_config = int(
1904                self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
1905            )
1906
1907            # Reset seen tracks on first page
1908            if page == 0:
1909                self._my_wave_seen_track_ids = set()
1910
1911            queue: str | int | None = None
1912            if page > 0:
1913                queue = self._my_wave_playlist_next_cursor
1914                if not queue:
1915                    return []
1916
1917            # Check if we've already reached the limit
1918            if len(self._my_wave_seen_track_ids) >= max_tracks_config:
1919                return []
1920
1921            tracks: list[Track] = []
1922            next_cursor: str | None = None
1923
1924            # Fetch MY_WAVE_BATCH_SIZE Rotor API batches per page to reduce API round-trips
1925            for _ in range(MY_WAVE_BATCH_SIZE):
1926                if len(self._my_wave_seen_track_ids) >= max_tracks_config:
1927                    break
1928
1929                raw_tracks, batch_id = await self.client.get_my_wave_tracks(queue=queue)
1930                if batch_id:
1931                    self._my_wave_batch_id = batch_id
1932                if not self._my_wave_radio_started_sent and raw_tracks:
1933                    sent = await self.client.send_rotor_station_feedback(
1934                        ROTOR_STATION_MY_MIX,
1935                        "radioStarted",
1936                        batch_id=batch_id,
1937                    )
1938                    if sent:
1939                        self._my_wave_radio_started_sent = True
1940
1941                if not raw_tracks:
1942                    break
1943
1944                first_track_id_this_batch = None
1945                for yt in raw_tracks:
1946                    if len(self._my_wave_seen_track_ids) >= max_tracks_config:
1947                        break
1948
1949                    track = self._parse_my_wave_track(yt, self._my_wave_seen_track_ids)
1950                    if track is None:
1951                        continue
1952
1953                    tracks.append(track)
1954                    track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
1955                    if first_track_id_this_batch is None:
1956                        first_track_id_this_batch = track_id
1957
1958                if first_track_id_this_batch is not None:
1959                    next_cursor = first_track_id_this_batch
1960                    queue = first_track_id_this_batch
1961                else:
1962                    # All tracks in this batch were duplicates or failed to parse
1963                    break
1964
1965            # Store cursor for next page call (None clears pagination so next call returns [])
1966            self._my_wave_playlist_next_cursor = next_cursor
1967            return tracks
1968
1969    @use_cache(3600 * 3, allow_expired_cache=True)
1970    async def _get_liked_tracks_playlist_tracks(self, page: int) -> list[Track]:
1971        """
1972        Get liked tracks for virtual playlist (sorted in reverse chronological order).
1973
1974        :param page: Page number (0 = all tracks limited by config, >0 = empty for pagination).
1975        :return: List of Track objects.
1976        """
1977        # Liked tracks API returns all tracks at once, so only return tracks on page 0
1978        if page > 0:
1979            return []
1980
1981        max_tracks_config = int(
1982            self.config.get_value(CONF_LIKED_TRACKS_MAX_TRACKS) or 500  # type: ignore[arg-type]
1983        )
1984
1985        # Fetch liked tracks (already sorted in reverse chronological order by api_client)
1986        track_shorts = await self.client.get_liked_tracks()
1987        if not track_shorts:
1988            self.logger.debug("No liked tracks found")
1989            return []
1990
1991        # Apply max tracks limit
1992        track_shorts = track_shorts[:max_tracks_config]
1993
1994        # Fetch full track details in batches
1995        track_ids = [str(ts.track_id) for ts in track_shorts if ts.track_id]
1996
1997        batch_size = TRACK_BATCH_SIZE
1998        full_tracks = []
1999        for i in range(0, len(track_ids), batch_size):
2000            batch_ids = track_ids[i : i + batch_size]
2001            batch_result = await self.client.get_tracks(batch_ids)
2002            full_tracks.extend(batch_result)
2003
2004        # Create track ID to full track mapping by track ID directly
2005        track_map = {}
2006        for t in full_tracks:
2007            if hasattr(t, "id") and t.id:
2008                track_map[str(t.id)] = t
2009
2010        # Parse tracks in the original order (reverse chronological)
2011        tracks = []
2012        for track_id in track_ids:
2013            # track_id may be compound "trackId:albumId", extract base ID for lookup
2014            base_id = track_id.split(":")[0] if ":" in track_id else track_id
2015            found = track_map.get(track_id) or track_map.get(base_id)
2016            if found:
2017                try:
2018                    tracks.append(parse_track(self, found))
2019                except InvalidDataError as err:
2020                    self.logger.debug("Error parsing liked track %s: %s", track_id, err)
2021
2022        self.logger.debug("Liked tracks: fetched %s, parsed %s", len(track_shorts), len(tracks))
2023        return tracks
2024
2025    # Get related items
2026
2027    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
2028    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
2029        """
2030        Get album tracks.
2031
2032        :param prov_album_id: The provider album ID.
2033        :return: List of Track objects.
2034        """
2035        album = await self.client.get_album_with_tracks(prov_album_id)
2036        if not album or not album.volumes:
2037            return []
2038
2039        tracks = []
2040        for volume_index, volume in enumerate(album.volumes):
2041            for track_index, track in enumerate(volume):
2042                try:
2043                    parsed_track = parse_track(self, track)
2044                    parsed_track.disc_number = volume_index + 1
2045                    parsed_track.track_number = track_index + 1
2046                    tracks.append(parsed_track)
2047                except InvalidDataError as err:
2048                    self.logger.debug("Error parsing album track: %s", err)
2049        return tracks
2050
2051    @use_cache(3600 * 3, allow_expired_cache=True)
2052    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
2053        """
2054        Get similar tracks using Kion Rotor station for this track.
2055
2056        Uses rotor station track:{id} so MA radio mode gets Kion recommendations.
2057
2058        :param prov_track_id: Provider track ID (plain or track_id@station_id).
2059        :param limit: Maximum number of tracks to return.
2060        :return: List of similar Track objects.
2061        """
2062        track_id, _ = _parse_radio_item_id(prov_track_id)
2063        station_id = f"track:{track_id}"
2064        raw_tracks, _ = await self.client.get_rotor_station_tracks(station_id, queue=None)
2065        tracks = []
2066        for yt in raw_tracks[:limit]:
2067            try:
2068                tracks.append(parse_track(self, yt))
2069            except InvalidDataError as err:
2070                self.logger.debug("Error parsing similar track: %s", err)
2071        return tracks
2072
2073    @use_cache(3600 * 3)
2074    async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
2075        """
2076        Get artists similar to the given one via Kion artists/similar endpoint.
2077
2078        :param prov_artist_id: Provider artist ID.
2079        :param limit: Maximum number of artists to return.
2080        :return: List of similar Artist objects.
2081        """
2082        raw_artists = await self.client.get_similar_artists(prov_artist_id, limit=limit)
2083        artists: list[Artist] = []
2084        for ya in raw_artists:
2085            try:
2086                artists.append(parse_artist(self, ya))
2087            except InvalidDataError as err:
2088                self.logger.debug("Error parsing similar artist: %s", err)
2089        return artists
2090
2091    @use_cache(600)
2092    async def _get_my_wave_recommendations(self) -> RecommendationFolder | None:
2093        """
2094        Get My Mix recommendation folder with personalized tracks.
2095
2096        :return: RecommendationFolder with My Mix tracks, or None if empty.
2097        """
2098        max_tracks_config = int(
2099            self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
2100        )
2101        batch_size_config = MY_WAVE_BATCH_SIZE
2102
2103        seen_track_ids: set[str] = set()
2104        items: list[Track] = []
2105        queue: str | int | None = None
2106
2107        for _ in range(batch_size_config):
2108            if len(seen_track_ids) >= max_tracks_config:
2109                break
2110
2111            raw_tracks, _ = await self.client.get_my_wave_tracks(queue=queue)
2112            if not raw_tracks:
2113                break
2114
2115            first_track_id_this_batch = None
2116            for yt in raw_tracks:
2117                if len(seen_track_ids) >= max_tracks_config:
2118                    break
2119
2120                track = self._parse_my_wave_track(yt, seen_ids=seen_track_ids)
2121                if track is None:
2122                    continue
2123
2124                items.append(track)
2125                track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
2126                if first_track_id_this_batch is None:
2127                    first_track_id_this_batch = track_id
2128
2129            queue = first_track_id_this_batch
2130            if not queue:
2131                break
2132
2133        if not items:
2134            return None
2135
2136        initial_tracks_limit = DISCOVERY_INITIAL_TRACKS
2137        if len(items) > initial_tracks_limit:
2138            items = items[:initial_tracks_limit]
2139
2140        # Recommendation folders keep their English name (not dropped like browse folders):
2141        # MusicProvider.browse() re-wraps them into plain BrowseFolders for the
2142        # "<provider>://recommendations" listing, where a bare translation_key would resolve
2143        # under the wrong media group. translation_key still localizes the recommendations view.
2144        return RecommendationFolder(
2145            item_id=MY_WAVE_PLAYLIST_ID,
2146            provider=self.instance_id,
2147            name="My Mix",
2148            translation_key=MY_WAVE_PLAYLIST_ID,
2149            items=UniqueList(items),
2150            icon="mdi-waveform",
2151        )
2152
2153    @use_cache(1800)
2154    async def _get_feed_recommendations(self) -> RecommendationFolder | None:
2155        """
2156        Get personalized feed playlists (Playlist of the Day, DejaVu, etc.).
2157
2158        :return: RecommendationFolder with generated playlists, or None if unavailable.
2159        """
2160        feed = await self.client.get_feed()
2161        if not feed or not feed.generated_playlists:
2162            return None
2163        items: list[Playlist] = []
2164        for gen_playlist in feed.generated_playlists:
2165            if gen_playlist.data and gen_playlist.ready:
2166                try:
2167                    items.append(parse_playlist(self, gen_playlist.data))
2168                except InvalidDataError as err:
2169                    self.logger.debug("Error parsing feed playlist: %s", err)
2170        if not items:
2171            return None
2172        return RecommendationFolder(
2173            item_id="feed",
2174            provider=self.instance_id,
2175            name="Made for you",
2176            translation_key="made_for_you",
2177            items=UniqueList(items),
2178            icon="mdi-account-music",
2179        )
2180
2181    @use_cache(3600)
2182    async def _get_chart_recommendations(self) -> RecommendationFolder | None:
2183        """
2184        Get chart tracks (hot tracks of the month).
2185
2186        :return: RecommendationFolder with chart tracks, or None if unavailable.
2187        """
2188        chart_info = await self.client.get_chart()
2189        if not chart_info or not chart_info.chart:
2190            return None
2191        playlist = chart_info.chart
2192        if not playlist.tracks:
2193            return None
2194        # TrackShort objects in chart context have .track (full Track) and .chart (position)
2195        tracks: list[Track] = []
2196        for track_short in playlist.tracks[:20]:
2197            track_obj = getattr(track_short, "track", None)
2198            if not track_obj:
2199                continue
2200            try:
2201                tracks.append(parse_track(self, track_obj))
2202            except InvalidDataError as err:
2203                self.logger.debug("Error parsing chart track: %s", err)
2204        if not tracks:
2205            return None
2206        return RecommendationFolder(
2207            item_id="chart",
2208            provider=self.instance_id,
2209            name="Chart",
2210            translation_key="chart",
2211            items=UniqueList(tracks),
2212            icon="mdi-chart-line",
2213        )
2214
2215    @use_cache(3600)
2216    async def _get_new_releases_recommendations(self) -> RecommendationFolder | None:
2217        """
2218        Get new album releases.
2219
2220        :return: RecommendationFolder with new albums, or None if unavailable.
2221        """
2222        releases = await self.client.get_new_releases()
2223        if not releases or not releases.new_releases:
2224            return None
2225        # new_releases is a list of album IDs (int) — need to batch-fetch full details
2226        album_ids = [str(aid) for aid in releases.new_releases[:20]]
2227        if not album_ids:
2228            return None
2229        full_albums = await self.client.get_albums(album_ids)
2230        if not full_albums:
2231            return None
2232        albums: list[Album] = []
2233        for album in full_albums:
2234            try:
2235                albums.append(parse_album(self, album))
2236            except InvalidDataError as err:
2237                self.logger.debug("Error parsing new release album: %s", err)
2238        if not albums:
2239            return None
2240        return RecommendationFolder(
2241            item_id="new_releases",
2242            provider=self.instance_id,
2243            name="New Releases",
2244            translation_key="new_releases",
2245            items=UniqueList(albums),
2246            icon="mdi-new-box",
2247        )
2248
2249    @use_cache(3600)
2250    async def _get_new_playlists_recommendations(self) -> RecommendationFolder | None:
2251        """
2252        Get new editorial playlists.
2253
2254        :return: RecommendationFolder with new playlists, or None if unavailable.
2255        """
2256        result = await self.client.get_new_playlists()
2257        if not result or not result.new_playlists:
2258            return None
2259        # new_playlists is a list of PlaylistId objects (uid, kind) — fetch full details
2260        playlist_ids = [
2261            f"{pid.uid}:{pid.kind}"
2262            for pid in result.new_playlists[:20]
2263            if hasattr(pid, "uid") and hasattr(pid, "kind")
2264        ]
2265        if not playlist_ids:
2266            return None
2267        full_playlists = await self.client.get_playlists(playlist_ids)
2268        if not full_playlists:
2269            return None
2270        playlists: list[Playlist] = []
2271        for playlist in full_playlists:
2272            try:
2273                playlists.append(parse_playlist(self, playlist))
2274            except InvalidDataError as err:
2275                self.logger.debug("Error parsing new playlist: %s", err)
2276        if not playlists:
2277            return None
2278        return RecommendationFolder(
2279            item_id="new_playlists",
2280            provider=self.instance_id,
2281            name="New Playlists",
2282            translation_key="new_playlists",
2283            items=UniqueList(playlists),
2284            icon="mdi-playlist-star",
2285        )
2286
2287    @use_cache(3600)
2288    async def _get_top_picks_recommendations(self) -> RecommendationFolder | None:
2289        """
2290        Get Top Picks recommendation folder (tag: top).
2291
2292        :return: RecommendationFolder with top playlists, or None if unavailable.
2293        """
2294        playlists = await self.client.get_tag_playlists("top")
2295        if not playlists:
2296            return None
2297        items: list[Playlist] = []
2298        for playlist in playlists[:10]:
2299            try:
2300                items.append(parse_playlist(self, playlist))
2301            except InvalidDataError as err:
2302                self.logger.debug("Error parsing top picks playlist: %s", err)
2303        if not items:
2304            return None
2305        return RecommendationFolder(
2306            item_id="top_picks",
2307            provider=self.instance_id,
2308            name="Top Picks",
2309            translation_key="top_picks",
2310            items=UniqueList(items),
2311            icon="mdi-star",
2312        )
2313
2314    @use_cache(1800)
2315    async def _get_mood_mix_recommendations(self, mood_tag: str) -> RecommendationFolder | None:
2316        """
2317        Get Mood Mix recommendation folder for a specific tag.
2318
2319        :param mood_tag: Preselected mood tag slug.
2320        :return: RecommendationFolder with mood playlists, or None if unavailable.
2321        """
2322        playlists = await self.client.get_tag_playlists(mood_tag)
2323        if not playlists:
2324            self.logger.debug("No playlists for mood tag %s, skipping recommendation", mood_tag)
2325            return None
2326        items: list[Playlist] = []
2327        for playlist in playlists[:8]:
2328            try:
2329                items.append(parse_playlist(self, playlist))
2330            except InvalidDataError as err:
2331                self.logger.debug("Error parsing mood playlist: %s", err)
2332        if not items:
2333            return None
2334        tag_name = self._media_source_name("folder", _media_label_key(mood_tag)) or mood_tag.title()
2335        return RecommendationFolder(
2336            item_id="mood_mix",
2337            provider=self.instance_id,
2338            name=f"Mood Mix: {tag_name}",
2339            translation_key="mood_mix",
2340            translation_params=[tag_name],
2341            items=UniqueList(items),
2342            icon="mdi-emoticon-outline",
2343        )
2344
2345    @use_cache(1800)
2346    async def _get_activity_mix_recommendations(
2347        self, activity_tag: str
2348    ) -> RecommendationFolder | None:
2349        """
2350        Get Activity Mix recommendation folder for a specific tag.
2351
2352        :param activity_tag: Preselected activity tag slug.
2353        :return: RecommendationFolder with activity playlists, or None if unavailable.
2354        """
2355        playlists = await self.client.get_tag_playlists(activity_tag)
2356        if not playlists:
2357            self.logger.debug(
2358                "No playlists for activity tag %s, skipping recommendation", activity_tag
2359            )
2360            return None
2361        items: list[Playlist] = []
2362        for playlist in playlists[:8]:
2363            try:
2364                items.append(parse_playlist(self, playlist))
2365            except InvalidDataError as err:
2366                self.logger.debug("Error parsing activity playlist: %s", err)
2367        if not items:
2368            return None
2369        tag_name = (
2370            self._media_source_name("folder", _media_label_key(activity_tag))
2371            or activity_tag.title()
2372        )
2373        return RecommendationFolder(
2374            item_id="activity_mix",
2375            provider=self.instance_id,
2376            name=f"Activity Mix: {tag_name}",
2377            translation_key="activity_mix",
2378            translation_params=[tag_name],
2379            items=UniqueList(items),
2380            icon="mdi-run",
2381        )
2382
2383    @use_cache(3600 * 6)
2384    async def _get_seasonal_mix_recommendations(self) -> RecommendationFolder | None:
2385        """
2386        Get Seasonal Mix recommendation folder (based on current month).
2387
2388        :return: RecommendationFolder with seasonal playlists, or None if unavailable.
2389        """
2390        # Determine current season tag
2391        current_month = utc().month
2392        seasonal_tag = TAG_SEASONAL_MAP.get(current_month, "autumn")
2393
2394        # Validate the seasonal tag; fall back to autumn if not available
2395        if not await self._validate_tag(seasonal_tag):
2396            seasonal_tag = "autumn"
2397
2398        playlists = await self.client.get_tag_playlists(seasonal_tag)
2399        if not playlists:
2400            return None
2401        items: list[Playlist] = []
2402        for playlist in playlists[:8]:
2403            try:
2404                items.append(parse_playlist(self, playlist))
2405            except InvalidDataError as err:
2406                self.logger.debug("Error parsing seasonal playlist: %s", err)
2407        if not items:
2408            return None
2409        tag_name = (
2410            self._media_source_name("folder", _media_label_key(seasonal_tag))
2411            or seasonal_tag.title()
2412        )
2413        return RecommendationFolder(
2414            item_id="seasonal_mix",
2415            provider=self.instance_id,
2416            name=f"Seasonal: {tag_name}",
2417            translation_key="seasonal_mix",
2418            translation_params=[tag_name],
2419            items=UniqueList(items),
2420            icon="mdi-weather-sunny",
2421        )
2422
2423    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
2424        """
2425        Get playlist tracks.
2426
2427        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind",
2428            my_wave, or liked_tracks).
2429        :param page: Page number for pagination.
2430        :return: List of Track objects.
2431        """
2432        self.logger.debug(
2433            "get_playlist_tracks called: prov_playlist_id=%s, page=%s", prov_playlist_id, page
2434        )
2435
2436        if prov_playlist_id == MY_WAVE_PLAYLIST_ID:
2437            self.logger.debug("Fetching My Mix tracks")
2438            return await self._get_my_wave_playlist_tracks(page)
2439
2440        if prov_playlist_id == LIKED_TRACKS_PLAYLIST_ID:
2441            self.logger.debug("Fetching Liked Tracks for virtual playlist")
2442            result = await self._get_liked_tracks_playlist_tracks(page)
2443            self.logger.debug("Liked Tracks playlist returned %s tracks", len(result))
2444            return result
2445
2446        return await self._get_regular_playlist_tracks(prov_playlist_id, page)
2447
2448    @use_cache(3600 * 3, allow_expired_cache=True)
2449    async def _get_regular_playlist_tracks(self, prov_playlist_id: str, page: int) -> list[Track]:
2450        """
2451        Get the tracks of a regular (non-virtual) playlist.
2452
2453        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind").
2454        :param page: Page number for pagination.
2455        :return: List of Track objects.
2456        """
2457        # KION Music API returns all playlist tracks in one call (no server-side pagination).
2458        # Return empty list for page > 0 so the controller pagination loop terminates.
2459        if page > 0:
2460            return []
2461
2462        # Parse the playlist ID (format: owner_id:kind)
2463        if PLAYLIST_ID_SPLITTER in prov_playlist_id:
2464            owner_id, kind = prov_playlist_id.split(PLAYLIST_ID_SPLITTER, 1)
2465        else:
2466            owner_id = str(self.client.user_id)
2467            kind = prov_playlist_id
2468
2469        playlist = await self.client.get_playlist(owner_id, kind)
2470        if not playlist:
2471            return []
2472
2473        # API sometimes returns playlist without tracks; fetch them explicitly if needed
2474        tracks_list = playlist.tracks or []
2475        track_count = getattr(playlist, "track_count", None) or 0
2476        if not tracks_list and track_count > 0:
2477            self.logger.debug(
2478                "Playlist %s/%s: track_count=%s but no tracks in response, "
2479                "calling fetch_tracks_async",
2480                owner_id,
2481                kind,
2482                track_count,
2483            )
2484            try:
2485                tracks_list = await playlist.fetch_tracks_async()
2486            except Exception as err:
2487                self.logger.warning("fetch_tracks_async failed for %s/%s: %s", owner_id, kind, err)
2488            if not tracks_list:
2489                raise ResourceTemporarilyUnavailable(
2490                    "Playlist tracks not available; try again later"
2491                )
2492
2493        if not tracks_list:
2494            return []
2495
2496        # Kion returns TrackShort objects, we need to fetch full track info
2497        track_ids = [
2498            str(track.track_id) if hasattr(track, "track_id") else str(track.id)
2499            for track in tracks_list
2500            if track
2501        ]
2502        if not track_ids:
2503            return []
2504
2505        # Fetch full track details in batches to avoid timeouts
2506        batch_size = TRACK_BATCH_SIZE
2507        full_tracks = []
2508        for i in range(0, len(track_ids), batch_size):
2509            batch = track_ids[i : i + batch_size]
2510            batch_result = await self.client.get_tracks(batch)
2511            if not batch_result:
2512                self.logger.warning(
2513                    "Received empty result for playlist %s tracks batch %s-%s",
2514                    prov_playlist_id,
2515                    i,
2516                    i + len(batch) - 1,
2517                )
2518                raise ResourceTemporarilyUnavailable(
2519                    "Playlist tracks not fully available; try again later"
2520                )
2521            full_tracks.extend(batch_result)
2522
2523        if track_ids and not full_tracks:
2524            raise ResourceTemporarilyUnavailable("Failed to load track details; try again later")
2525
2526        tracks = []
2527        for track in full_tracks:
2528            try:
2529                tracks.append(parse_track(self, track))
2530            except InvalidDataError as err:
2531                self.logger.debug("Error parsing playlist track: %s", err)
2532        return tracks
2533
2534    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
2535    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
2536        """
2537        Get artist's albums.
2538
2539        :param prov_artist_id: The provider artist ID.
2540        :return: List of Album objects.
2541        """
2542        albums = await self.client.get_artist_albums(prov_artist_id)
2543        result = []
2544        for album in albums:
2545            try:
2546                result.append(parse_album(self, album))
2547            except InvalidDataError as err:
2548                self.logger.debug("Error parsing artist album: %s", err)
2549        return result
2550
2551    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
2552    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
2553        """
2554        Get artist's top tracks.
2555
2556        :param prov_artist_id: The provider artist ID.
2557        :return: List of Track objects.
2558        """
2559        tracks = await self.client.get_artist_tracks(prov_artist_id)
2560        result = []
2561        for track in tracks:
2562            try:
2563                result.append(parse_track(self, track))
2564            except InvalidDataError as err:
2565                self.logger.debug("Error parsing artist track: %s", err)
2566        return result
2567
2568    # Library methods
2569
2570    async def get_library_artists(self) -> AsyncGenerator[Artist]:
2571        """Retrieve library artists from KION Music."""
2572        artists = await self.client.get_liked_artists()
2573        for artist in artists:
2574            try:
2575                yield parse_artist(self, artist)
2576            except InvalidDataError as err:
2577                self.logger.debug("Error parsing library artist: %s", err)
2578
2579    async def get_library_albums(self) -> AsyncGenerator[Album]:
2580        """Retrieve library albums from KION Music."""
2581        batch_size = TRACK_BATCH_SIZE
2582        albums = await self.client.get_liked_albums(batch_size=batch_size)
2583        for album in albums:
2584            try:
2585                yield parse_album(self, album)
2586            except InvalidDataError as err:
2587                self.logger.debug("Error parsing library album: %s", err)
2588
2589    async def get_library_tracks(self) -> AsyncGenerator[Track]:
2590        """Retrieve library tracks from KION Music."""
2591        track_shorts = await self.client.get_liked_tracks()
2592        if not track_shorts:
2593            return
2594
2595        # Fetch full track details in batches
2596        track_ids = [str(ts.track_id) for ts in track_shorts if ts.track_id]
2597        batch_size = TRACK_BATCH_SIZE
2598        for i in range(0, len(track_ids), batch_size):
2599            batch_ids = track_ids[i : i + batch_size]
2600            full_tracks = await self.client.get_tracks(batch_ids)
2601            for track in full_tracks:
2602                try:
2603                    yield parse_track(self, track)
2604                except InvalidDataError as err:
2605                    self.logger.debug("Error parsing library track: %s", err)
2606
2607    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
2608        """
2609        Retrieve library playlists from KION Music.
2610
2611        Includes virtual playlists (My Mix and Liked Tracks if enabled), user-created playlists,
2612        and user-liked editorial playlists (returned by a separate API endpoint).
2613        """
2614        yield await self.get_playlist(MY_WAVE_PLAYLIST_ID)
2615        yield await self.get_playlist(LIKED_TRACKS_PLAYLIST_ID)
2616        seen_ids: set[str] = set()
2617        # User-created playlists
2618        playlists = await self.client.get_user_playlists()
2619        for playlist in playlists:
2620            try:
2621                parsed = parse_playlist(self, playlist)
2622                seen_ids.add(parsed.item_id)
2623                yield parsed
2624            except InvalidDataError as err:
2625                self.logger.debug("Error parsing library playlist: %s", err)
2626        # User-liked editorial playlists (not in users_playlists_list)
2627        liked_playlists = await self.client.get_liked_playlists()
2628        for playlist in liked_playlists:
2629            try:
2630                parsed = parse_playlist(self, playlist)
2631                if parsed.item_id not in seen_ids:
2632                    yield parsed
2633            except InvalidDataError as err:
2634                self.logger.debug("Error parsing liked playlist: %s", err)
2635
2636    # Library edit methods
2637
2638    async def library_add(self, item: MediaItemType) -> bool:
2639        """
2640        Add item to library.
2641
2642        :param item: The media item to add.
2643        :return: True if successful.
2644        """
2645        prov_item_id = self._get_provider_item_id(item)
2646        if not prov_item_id:
2647            return False
2648        track_id, _ = _parse_radio_item_id(prov_item_id)
2649
2650        if item.media_type == MediaType.TRACK:
2651            return await self.client.like_track(track_id)
2652        if item.media_type == MediaType.ALBUM:
2653            return await self.client.like_album(prov_item_id)
2654        if item.media_type == MediaType.ARTIST:
2655            return await self.client.like_artist(prov_item_id)
2656        return False
2657
2658    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
2659        """
2660        Remove item from library.
2661
2662        :param prov_item_id: The provider item ID (may be track_id@station_id for tracks).
2663        :param media_type: The media type.
2664        :return: True if successful.
2665        """
2666        track_id, _ = _parse_radio_item_id(prov_item_id)
2667        if media_type == MediaType.TRACK:
2668            return await self.client.unlike_track(track_id)
2669        if media_type == MediaType.ALBUM:
2670            return await self.client.unlike_album(prov_item_id)
2671        if media_type == MediaType.ARTIST:
2672            return await self.client.unlike_artist(prov_item_id)
2673        return False
2674
2675    def _get_provider_item_id(self, item: MediaItemType) -> str | None:
2676        """Get provider item ID from media item."""
2677        for mapping in item.provider_mappings:
2678            if mapping.provider_instance == self.instance_id:
2679                return mapping.item_id
2680        return item.item_id if item.provider == self.instance_id else None
2681
2682    # Streaming
2683
2684    async def get_stream_details(
2685        self, item_id: str, media_type: MediaType = MediaType.TRACK
2686    ) -> StreamDetails:
2687        """
2688        Get stream details for a track.
2689
2690        :param item_id: The track ID (or track_id@station_id for My Mix).
2691        :param media_type: The media type (should be TRACK).
2692        :return: StreamDetails for the track.
2693        """
2694        return await self.streaming.get_stream_details(item_id)
2695
2696    async def get_audio_stream(
2697        self, streamdetails: StreamDetails, seek_position: int = 0
2698    ) -> AsyncGenerator[bytes]:
2699        """
2700        Return the audio stream for the provider item.
2701
2702        Uses windowed Range-request streaming to prevent Kion CDN drops.
2703        Handles both raw (direct) and encrypted (encraw) transports.
2704
2705        :param streamdetails: Stream details with URL and optional decryption key.
2706        :param seek_position: Seek position in seconds (handled by provider for raw transport).
2707        :return: Async generator yielding audio chunks.
2708        """
2709        async for chunk in self.streaming.get_audio_stream(streamdetails, seek_position):
2710            yield chunk
2711
2712    async def get_rotor_station_tracks(
2713        self, station_id: str, queue: str | int | None = None
2714    ) -> tuple[list[Any], str | None]:
2715        """
2716        Fetch tracks from a rotor station (My Mix, similar, etc.).
2717
2718        Wrapper around client.get_rotor_station_tracks for use by ynison plugin.
2719        """
2720        return await self.client.get_rotor_station_tracks(station_id, queue=queue)
2721
2722    def get_quality(self) -> str:
2723        """
2724        Return the configured audio quality tier (e.g. 'balanced', 'superb').
2725
2726        Mirrors the legacy-value normalization used by the streaming layer:
2727        older configs store the lossless tier as ``"lossless"``, while the
2728        current canonical value is ``QUALITY_LOSSLESS`` (``"superb"``).
2729        External callers (e.g. the ynison plugin wrapper) see the same
2730        normalized value the streaming code would resolve to.
2731        """
2732        quality = str(self.config.get_value(CONF_QUALITY) or "").strip().lower()
2733        if quality == "lossless":
2734            quality = QUALITY_LOSSLESS
2735        return quality
2736
2737    async def resolve_image(self, path: str) -> str | bytes:
2738        """
2739        Resolve wave cover image with background color fill for transparent PNGs.
2740
2741        If the image URL has an associated background color (stored in _wave_bg_colors),
2742        downloads the PNG from Kion CDN and composites it on a solid color background
2743        using Pillow, returning JPEG bytes. Falls back to the original URL on any error.
2744
2745        :param path: Image URL (may include #rrggbb fragment used as cache key).
2746        :return: Composited JPEG bytes, or original path string as fallback.
2747        """
2748        bg_color = self._wave_bg_colors.get(path)
2749        if not bg_color:
2750            return path
2751
2752        # Strip the #color fragment before fetching the actual image
2753        fetch_url = path.split("#", maxsplit=1)[0] if "#" in path else path
2754        try:
2755            async with self.mass.http_session.get(fetch_url) as resp:
2756                resp.raise_for_status()
2757                raw = await resp.read()
2758        except Exception as err:
2759            self.logger.debug("Failed to fetch wave cover %s: %s", fetch_url, err)
2760            return fetch_url
2761
2762        def _composite() -> bytes:
2763            bg_clean = bg_color.lstrip("#")
2764            try:
2765                r = int(bg_clean[0:2], 16)
2766                g = int(bg_clean[2:4], 16)
2767                b = int(bg_clean[4:6], 16)
2768            except ValueError, IndexError:
2769                return raw
2770            fg = PilImage.open(BytesIO(raw)).convert("RGBA")
2771            bg = PilImage.new("RGBA", fg.size, (r, g, b, 255))
2772            bg.paste(fg, mask=fg)
2773            out = BytesIO()
2774            bg.convert("RGB").save(out, "JPEG", quality=92)
2775            return out.getvalue()
2776
2777        try:
2778            return await asyncio.to_thread(_composite)
2779        except Exception as err:
2780            self.logger.debug("Wave cover composite failed for %s: %s", fetch_url, err)
2781            return fetch_url
2782
2783    async def on_played(
2784        self,
2785        media_type: MediaType,
2786        prov_item_id: str,
2787        fully_played: bool,
2788        position: int,
2789        media_item: MediaItemType,
2790        is_playing: bool = False,
2791    ) -> None:
2792        """
2793        Report playback for rotor feedback when the track is from My Mix.
2794
2795        Sends trackStarted when the track is currently playing (is_playing=True).
2796        trackFinished/skip are sent from on_streamed to use accurate seconds_streamed.
2797        """
2798        if media_type != MediaType.TRACK:
2799            return
2800        track_id, station_id = _parse_radio_item_id(prov_item_id)
2801        if not station_id:
2802            return
2803        if is_playing:
2804            if station_id == ROTOR_STATION_MY_MIX:
2805                batch_id = self._my_wave_batch_id
2806            else:
2807                state = self._wave_states.get(station_id)
2808                batch_id = state.batch_id if state else None
2809            await self.client.send_rotor_station_feedback(
2810                station_id,
2811                "trackStarted",
2812                track_id=track_id,
2813                batch_id=batch_id,
2814            )
2815
2816    async def on_streamed(self, streamdetails: StreamDetails) -> None:
2817        """
2818        Report stream completion for My Mix rotor feedback.
2819
2820        Sends trackFinished or skip with actual seconds_streamed so Kion
2821        can improve recommendations.
2822        """
2823        track_id, station_id = _parse_radio_item_id(streamdetails.item_id)
2824        if not station_id:
2825            return
2826        seconds = int(streamdetails.seconds_streamed or 0)
2827        duration = streamdetails.duration or 0
2828        feedback_type = "trackFinished" if duration and seconds >= max(0, duration - 10) else "skip"
2829        if station_id == ROTOR_STATION_MY_MIX:
2830            batch_id = self._my_wave_batch_id
2831        else:
2832            state = self._wave_states.get(station_id)
2833            batch_id = state.batch_id if state else None
2834        await self.client.send_rotor_station_feedback(
2835            station_id,
2836            feedback_type,
2837            track_id=track_id,
2838            total_played_seconds=seconds,
2839            batch_id=batch_id,
2840        )
2841
2842    async def _rotating_row_tag_subtitle(self, category: str) -> str | None:
2843        """
2844        Return the display label of the current rotating tag for a mood/activity row.
2845
2846        Cache-only read of the validated tag list (rows must stay free of backend I/O):
2847        returns None - no subtitle - until an items fetch has warmed that cache.
2848
2849        :param category: Tag category ('mood' or 'activity').
2850        """
2851        # key mirrors the @use_cache key construction on _get_valid_tags_for_category:
2852        # the wrapped function's __name__ (preserved by functools.wraps, so it survives
2853        # renames) plus its positional args, joined by dots
2854        tags, _, found = await self.mass.cache.get_with_freshness(
2855            f"{self._get_valid_tags_for_category.__name__}.{category}",
2856            provider=self.instance_id,
2857            include_expired=True,
2858        )
2859        if not found or not tags:
2860            return None
2861        tag = self._rotating_row_tag(category, tags)
2862        return self._media_source_name("folder", _media_label_key(tag)) or tag.title()
2863
2864    def _rotating_row_tag(self, category: str, valid_tags: list[str]) -> str:
2865        """
2866        Deterministically pick the current hour's tag for a mood/activity row.
2867
2868        Rows and items derive the same tag independently - no shared state, so
2869        concurrent clients (or multiple users on one instance) can never make the
2870        served items mismatch the row subtitle. The pick rotates hourly and
2871        differs per provider instance.
2872
2873        :param category: Tag category the tags belong to.
2874        :param valid_tags: Non-empty list of valid tag slugs to pick from.
2875        """
2876        hour_bucket = int(utc().timestamp()) // 3600
2877        seed = f"{self.instance_id}.{category}.{hour_bucket}".encode()
2878        return sorted(valid_tags)[zlib.crc32(seed) % len(valid_tags)]
2879