diff --git a/src/main_stream_service/main_stream_manager.py b/src/main_stream_service/main_stream_manager.py index 11f3554..2c629ff 100644 --- a/src/main_stream_service/main_stream_manager.py +++ b/src/main_stream_service/main_stream_manager.py @@ -8,7 +8,7 @@ from injector import inject from core.config.settings import settings -from main_stream_service.yandex_music_api import YandexMusicAPI +from main_stream_service.yandex_music_api import TrackSource, YandexMusicAPI from ruark_audio_system.exceptions import RuarkDeviceNotFoundError from ruark_audio_system.ruark_r5_controller import RuarkR5Controller from ruark_audio_system.volume_store import RuarkVolumeStore @@ -20,6 +20,8 @@ SILENCE_RESEND_GRACE, STREAM_POLL_INTERVAL, STREAMING_RESTART_DELAY, + TRACK_SOURCE_CACHE_SIZE, + TRACK_SOURCE_TTL, ) from yandex_station.models import Track from yandex_station.station_controls import YandexStationControls @@ -81,6 +83,8 @@ def __init__( self._tasks = [] # Хранение фоновых задач # Ссылка на задачу авто-остановки (Ruark выключен пользователем) self._shutdown_task: asyncio.Task[None] | None = None + self._source_cache: dict[str, tuple[TrackSource, float]] = {} + self._prefetch_tasks: dict[str, asyncio.Task[None]] = {} async def start(self): """Запуск всех стриминговых процессов.""" @@ -132,9 +136,17 @@ async def stop(self): # Отмена всех активных задач for task in self._tasks: task.cancel() + for prefetch_task in self._prefetch_tasks.values(): + prefetch_task.cancel() - await asyncio.gather(*self._tasks, return_exceptions=True) + await asyncio.gather( + *self._tasks, + *self._prefetch_tasks.values(), + return_exceptions=True, + ) self._tasks.clear() + self._prefetch_tasks.clear() + self._source_cache.clear() logger.info("✅ Стриминг остановлен") async def _safe_stop_step( @@ -238,6 +250,7 @@ async def _handle_idle_cycle( track = await self._refresh_track_if_unchanged(track, ctx) await self._resync_after_progress_jump(track, ctx) await self._switch_to_new_track(track, ctx) + self._schedule_prefetch(track) await self._restore_ruark_after_speech(track, ctx) await self._resume_track_if_silent(track, ctx) await self._fade_alice_if_playing(track) @@ -382,6 +395,75 @@ async def _is_ruark_powered_off(self) -> bool: logger.debug(f"Не удалось узнать питание Ruark: {e}") return False + def _get_cached_source(self, track_id: str) -> TrackSource | None: + """Возвращает свежий источник из кеша или None, если протух.""" + cached = self._source_cache.get(track_id) + if not cached: + return None + source, cached_at = cached + if time.monotonic() - cached_at > TRACK_SOURCE_TTL: + del self._source_cache[track_id] + return None + return source + + def _store_source(self, track_id: str, source: TrackSource) -> None: + """Кладёт источник в кеш, вытесняя самую старую запись.""" + if len(self._source_cache) >= TRACK_SOURCE_CACHE_SIZE: + oldest = min( + self._source_cache, + key=lambda key: self._source_cache[key][1], + ) + del self._source_cache[oldest] + self._source_cache[track_id] = (source, time.monotonic()) + + async def _resolve_track_source(self, track_id: str) -> TrackSource | None: + """Источник трека: из кеша или свежим запросом с записью в кеш.""" + cached = self._get_cached_source(track_id) + if cached: + logger.info(f"⚡ Ссылка для {track_id} взята из кеша") + return cached + source = await self._yandex_music_api.get_track_source( + track_id=track_id, + quality=settings.stream_quality, + ) + if source: + self._store_source(track_id, source) + return source + + def _schedule_prefetch(self, track: Track) -> None: + """Фоново предзагружает ссылки соседних треков очереди. + + Следующий трек нужен для мгновенного переключения вперёд, + предыдущий — для возврата назад. Уже закешированные и уже + загружаемые треки пропускаются. + """ + if track.type == "FmRadio": + return + for neighbor_id in (track.next_id, track.prev_id): + if not neighbor_id or neighbor_id == track.id: + continue + if self._get_cached_source(neighbor_id): + continue + if neighbor_id in self._prefetch_tasks: + continue + self._prefetch_tasks[neighbor_id] = asyncio.create_task( + self._prefetch_source(neighbor_id) + ) + + async def _prefetch_source(self, track_id: str) -> None: + """Резолвит и кеширует ссылку трека в фоновой задаче.""" + try: + source = await self._resolve_track_source(track_id) + if source: + logger.info( + f"📦 Предзагружена ссылка трека {track_id} " + f"({source.codec})" + ) + except Exception as e: + logger.debug(f"Предзагрузка трека {track_id} не удалась: {e}") + finally: + self._prefetch_tasks.pop(track_id, None) + async def _resend_current_track( self, track: Track, ctx: "_CycleContext", reason: str ) -> None: @@ -390,10 +472,7 @@ async def _resend_current_track( f"🔁 Ресинк стрима ({reason}): позиция {track.progress:.0f}s" ) resync_started = time.monotonic() - source = await self._yandex_music_api.get_track_source( - track_id=track.id, - quality=settings.stream_quality, - ) + source = await self._resolve_track_source(track.id) if not source: logger.warning("⚠️ Не удалось получить URL трека для ресинка") return @@ -431,10 +510,7 @@ async def _switch_to_new_track( ctx.track_url = await self._station_controls.get_radio_url() logger.info(f"🎵 URL радиостанции: {ctx.track_url}") else: - source = await self._yandex_music_api.get_track_source( - track_id=track.id, - quality=settings.stream_quality, - ) + source = await self._resolve_track_source(track.id) ctx.track_url = source.url if source else None codec = source.codec if source else "mp3" resolve_seconds = time.monotonic() - switch_started diff --git a/src/yandex_station/constants.py b/src/yandex_station/constants.py index cc2aa88..15c03a1 100644 --- a/src/yandex_station/constants.py +++ b/src/yandex_station/constants.py @@ -20,3 +20,8 @@ # после отправки потока, прежде чем считать тишину проблемой, секунды SILENCE_CHECK_INTERVAL = 3.0 SILENCE_RESEND_GRACE = 15.0 + +# Кеш прямых ссылок на треки: время жизни записи (CDN-ссылки +# протухают) и максимум записей, секунды / штуки +TRACK_SOURCE_TTL = 600.0 +TRACK_SOURCE_CACHE_SIZE = 8 diff --git a/src/yandex_station/models.py b/src/yandex_station/models.py index 94b664a..83d2ec4 100644 --- a/src/yandex_station/models.py +++ b/src/yandex_station/models.py @@ -12,3 +12,6 @@ class Track: duration: float progress: float playing: bool + # Соседние треки очереди станции (entityInfo) — для предзагрузки + next_id: str | None = None + prev_id: str | None = None diff --git a/src/yandex_station/station_controls.py b/src/yandex_station/station_controls.py index 0ec1513..c82f342 100644 --- a/src/yandex_station/station_controls.py +++ b/src/yandex_station/station_controls.py @@ -128,6 +128,7 @@ async def get_current_track(self) -> Track | None: player_state = await self.get_player_status() if not player_state: return None + entity = player_state.get("entityInfo") or {} return Track( id=str(player_state.get("id", "0")), title=str(player_state.get("title", "")), @@ -136,11 +137,31 @@ async def get_current_track(self) -> Track | None: duration=float(player_state.get("duration") or 0), progress=float(player_state.get("progress") or 0), playing=bool(player_state.get("playing", False)), + next_id=self._neighbor_track_id(entity, "next"), + prev_id=self._neighbor_track_id(entity, "prev"), ) except Exception as e: logger.error(f"❌ Ошибка при получении текущего трека: {e}") return None + @staticmethod + def _neighbor_track_id(entity: Any, key: str) -> str | None: + """Извлекает id соседнего трека из entityInfo станции. + + Args: + entity (Any): Словарь entityInfo из playerState. + key (str): Ключ соседа: next или prev. + Returns: + str | None: Идентификатор, если сосед — трек. + """ + if not isinstance(entity, dict): + return None + neighbor = entity.get(key) + if not isinstance(neighbor, dict) or neighbor.get("type") != "Track": + return None + neighbor_id = neighbor.get("id") + return str(neighbor_id) if neighbor_id else None + async def get_volume(self) -> float | None: """Получение текущего уровня громкости (0.0–1.0).""" try: diff --git a/tests/test_station_controls.py b/tests/test_station_controls.py new file mode 100644 index 0000000..5e218f6 --- /dev/null +++ b/tests/test_station_controls.py @@ -0,0 +1,72 @@ +from unittest.mock import AsyncMock, MagicMock + +from yandex_station.protobuf_parser import Protobuf +from yandex_station.station_controls import YandexStationControls +from yandex_station.station_ws_control import YandexStationClient + +BASE_PLAYER_STATE = { + "id": "42", + "title": "title", + "type": "Track", + "subtitle": "artist", + "duration": 200, + "progress": 10, +} + + +def make_controls(player_state) -> YandexStationControls: + """Контролы станции с замоканным состоянием плеера.""" + ws_client = AsyncMock(spec=YandexStationClient) + ws_client.get_latest_message.return_value = { + "state": {"playerState": player_state, "playing": True} + } + return YandexStationControls( + ws_client=ws_client, + protobuf=MagicMock(spec=Protobuf), + ) + + +async def test_current_track_carries_neighbor_ids(): + controls = make_controls( + { + **BASE_PLAYER_STATE, + "entityInfo": { + "next": {"id": "43", "type": "Track"}, + "prev": {"id": "41", "type": "Track"}, + }, + } + ) + + track = await controls.get_current_track() + + assert track is not None + assert track.next_id == "43" + assert track.prev_id == "41" + + +async def test_non_track_neighbors_are_ignored(): + controls = make_controls( + { + **BASE_PLAYER_STATE, + "entityInfo": { + "next": {"id": "gen", "type": "Generative"}, + "prev": "мусор", + }, + } + ) + + track = await controls.get_current_track() + + assert track is not None + assert track.next_id is None + assert track.prev_id is None + + +async def test_missing_entity_info_is_safe(): + controls = make_controls(dict(BASE_PLAYER_STATE)) + + track = await controls.get_current_track() + + assert track is not None + assert track.next_id is None + assert track.prev_id is None diff --git a/tests/test_streaming_loop.py b/tests/test_streaming_loop.py index dcc8a8e..500e856 100644 --- a/tests/test_streaming_loop.py +++ b/tests/test_streaming_loop.py @@ -13,6 +13,7 @@ from yandex_station.constants import ( RUARK_IDLE_VOLUME, STREAM_POLL_INTERVAL, + TRACK_SOURCE_TTL, ) from yandex_station.models import Track from yandex_station.station_controls import YandexStationControls @@ -108,6 +109,24 @@ async def fake_wait(timeout): return manager +def set_source_by_id(manager): + """Источник с уникальной ссылкой по id — чтобы отличать треки.""" + + async def resolver(track_id, quality=None): + return TrackSource(url=f"http://track/{track_id}", codec="mp3") + + manager._yandex_music_api.get_track_source.side_effect = resolver + + +def source_calls_for(manager, track_id): + """Вызовы get_track_source для конкретного трека.""" + return [ + call + for call in manager._yandex_music_api.get_track_source.call_args_list + if call.kwargs.get("track_id") == track_id + ] + + async def drive_streaming(manager, until, timeout=1.0): """Крутит streaming() до выполнения условия или таймаута.""" manager._stream_state_running = True @@ -401,6 +420,109 @@ async def test_powered_off_ruark_stops_streaming(fast_sleep, monkeypatch): station.unmute.assert_called() +async def test_next_track_prefetched_in_background(fast_sleep): + """Ссылка следующего трека резолвится фоном, пока играет текущий.""" + track = make_track(next_id="43") + station = make_station_controls(track) + manager = make_manager(station, make_ruark()) + set_source_by_id(manager) + + await drive_streaming( + manager, until=lambda: len(source_calls_for(manager, "43")) > 0 + ) + + assert len(source_calls_for(manager, "43")) == 1 + # Предзагрузка ничего не отправляет — сосед ещё не играет + for call in manager._send_track_to_stream_server.call_args_list: + assert call.args[0] == "http://track/42" + + +async def test_switch_uses_prefetched_link(fast_sleep): + """Переключение вперёд берёт ссылку из кеша, а не резолвит заново.""" + first = make_track(id="42", next_id="43") + second = make_track(id="43", prev_id="42", progress=0.5) + station = make_station_controls(tracks=[first] * 8 + [second]) + manager = make_manager(station, make_ruark()) + set_source_by_id(manager) + send = manager._send_track_to_stream_server + + await drive_streaming( + manager, + until=lambda: any( + call.args[0] == "http://track/43" for call in send.call_args_list + ), + ) + + # Единственный резолв — фоновый префетч, свитч попал в кеш + assert len(source_calls_for(manager, "43")) == 1 + + +async def test_return_to_previous_track_uses_cache(fast_sleep): + """Возврат к прошлому треку не ждёт резолва — ссылка уже в кеше.""" + first = make_track(id="42", next_id="43") + second = make_track(id="43", prev_id="42", progress=0.5) + back = make_track(id="42", next_id="43", progress=1.0) + station = make_station_controls(tracks=[first] * 6 + [second] * 6 + [back]) + manager = make_manager(station, make_ruark()) + set_source_by_id(manager) + send = manager._send_track_to_stream_server + + def returned_to_first(): + sent = [call.args[0] for call in send.call_args_list] + return sent.count("http://track/42") >= 2 + + await drive_streaming(manager, until=returned_to_first) + + # Трек 42 резолвился один раз — при первом свитче + assert len(source_calls_for(manager, "42")) == 1 + + +async def test_resync_reuses_cached_source(fast_sleep): + """Ресинк после перемотки не резолвит ссылку заново.""" + tracks = [make_track(progress=10.0)] * 4 + [make_track(progress=120.0)] + station = make_station_controls(tracks=tracks) + manager = make_manager(station, make_ruark()) + set_source_by_id(manager) + send = manager._send_track_to_stream_server + + def resync_happened(): + return any( + call.kwargs.get("start_position") == pytest.approx(120.0) + for call in send.call_args_list + ) + + await drive_streaming(manager, until=resync_happened) + + # Свитч плюс ресинк — а резолв ссылки только один + assert len(source_calls_for(manager, "42")) == 1 + + +async def test_radio_neighbors_not_prefetched(fast_sleep): + """Для радио предзагрузка соседних треков не запускается.""" + track = make_track( + id="fm_jazz", type="FmRadio", duration=0.0, next_id="43" + ) + station = make_station_controls(track) + manager = make_manager(station, make_ruark(is_playing=False)) + + await drive_streaming(manager, until=lambda: False, timeout=0.15) + + manager._yandex_music_api.get_track_source.assert_not_called() + + +async def test_cached_source_expires_after_ttl(): + """Протухшая запись кеша не используется и удаляется.""" + manager = make_manager(make_station_controls(make_track()), make_ruark()) + stale = TrackSource(url="http://track/old", codec="mp3") + manager._source_cache["42"] = ( + stale, + time.monotonic() - TRACK_SOURCE_TTL - 1.0, + ) + + assert manager._get_cached_source("42") is None + assert "42" not in manager._source_cache + + async def test_stop_survives_unreachable_ruark(): """Недоступный Ruark не прерывает остановку стриминга.""" station = make_station_controls(make_track())