Add Yandex Music Connect (Ynison) plugin provider

music_assistant/providers/yandex_ynison/__init__.py

@@ -0,0 +1,205 @@
+"""Yandex Music Connect (Ynison) plugin for Music Assistant."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+
+from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
+from music_assistant_models.enums import ConfigEntryType, ProviderFeature
+
+from .config_helpers import list_yandex_music_instances
+from .constants import (
+    CONF_ALLOW_PLAYER_SWITCH,
+    CONF_DEVICE_ID,
+    CONF_MASS_PLAYER_ID,
+    CONF_OUTPUT_BIT_DEPTH,
+    CONF_OUTPUT_SAMPLE_RATE,
+    CONF_PUBLISH_NAME,
+    CONF_TOKEN,
+    CONF_YM_INSTANCE,
+    DEFAULT_DISPLAY_NAME,
+    OUTPUT_AUTO,
+    PLAYER_ID_AUTO,
+    YM_INSTANCE_OWN,
+)
+from .provider import YandexYnisonProvider
+
+if TYPE_CHECKING:
+    from music_assistant_models.config_entries import ConfigValueType, ProviderConfig
+    from music_assistant_models.provider import ProviderManifest
+
+    from music_assistant.mass import MusicAssistant
+    from music_assistant.models import ProviderInstanceType
+
+SUPPORTED_FEATURES = {ProviderFeature.AUDIO_SOURCE}
+
+
+async def setup(
+    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
+) -> ProviderInstanceType:
+    """Initialize provider(instance) with given configuration."""
+    return YandexYnisonProvider(mass, manifest, config, SUPPORTED_FEATURES)
+
+
+async def get_config_entries(
+    mass: MusicAssistant,
+    instance_id: str | None = None,  # noqa: ARG001 — required by MA callback signature
+    action: str | None = None,  # noqa: ARG001 — required by MA callback signature
+    values: dict[str, ConfigValueType] | None = None,
+) -> tuple[ConfigEntry, ...]:
+    """Return Config entries to setup this provider."""
+    if values is None:
+        values = {}
+
+    # Migrate legacy config keys (renamed in v1.5.0)
+    if "player" in values and CONF_MASS_PLAYER_ID not in values:
+        values[CONF_MASS_PLAYER_ID] = values.pop("player")
+    if "display_name" in values and CONF_PUBLISH_NAME not in values:
+        values[CONF_PUBLISH_NAME] = values.pop("display_name")
+
+    # Discover available yandex_music instances for borrow-mode dropdown
+    ym_instances = list_yandex_music_instances(mass)
+    ym_instance_ids = {inst_id for inst_id, _ in ym_instances}
+
+    # Determine the currently selected source (borrow vs own)
+    selected = cast("str | None", values.get(CONF_YM_INSTANCE))
+    if selected is None:
+        # Preserve existing own-token configs on upgrade (CONF_TOKEN already set
+        # but CONF_YM_INSTANCE absent). Only auto-select borrowing for truly
+        # fresh installs with no stored token and exactly one YM instance.
+        has_manual_token = bool(values.get(CONF_TOKEN))
+        if has_manual_token:
+            selected = YM_INSTANCE_OWN
+        else:
+            selected = ym_instances[0][0] if len(ym_instances) == 1 else YM_INSTANCE_OWN
+    borrowing = selected != YM_INSTANCE_OWN and selected in ym_instance_ids
+
+    # Dynamic label
+    if borrowing:
+        ym_name = next((name for inst_id, name in ym_instances if inst_id == selected), selected)
+        label_text = f"Borrowing credentials from Yandex Music instance '{ym_name}'."
+    elif selected != YM_INSTANCE_OWN:
+        # Referenced YM instance is not currently configured
+        label_text = (
+            "Selected Yandex Music instance is not available. "
+            "Re-select below or fall back to manual token."
+        )
+    else:
+        label_text = (
+            "Using a manually entered Yandex Music token. Token refresh is not "
+            "automatic in this mode — prefer borrowing from a Yandex Music "
+            "instance if possible."
+        )
+
+    # Build dropdown options: one per YM instance + "Use own token" sentinel
+    source_options = [
+        ConfigValueOption(f"Yandex Music: {name}", inst_id) for inst_id, name in ym_instances
+    ]
+    source_options.append(ConfigValueOption("Use own token (manual entry)", YM_INSTANCE_OWN))
+
+    # Guard against a stale selection pointing at a removed YM instance — the
+    # UI would otherwise render with a default that isn't in `options`.
+    dropdown_default = selected if borrowing or selected == YM_INSTANCE_OWN else YM_INSTANCE_OWN
+
+    return (
+        ConfigEntry(
+            key="label_text",
+            type=ConfigEntryType.LABEL,
+            label=label_text,
+        ),
+        ConfigEntry(
+            key=CONF_YM_INSTANCE,
+            type=ConfigEntryType.STRING,
+            label="Yandex Music source",
+            description="Borrow OAuth credentials from a linked Yandex Music provider "
+            "instance. Requires configuring Yandex Music first. Select 'Use own token' "
+            "to enter a music token manually.",
+            options=source_options,
+            default_value=dropdown_default,
+            required=True,
+        ),
+        ConfigEntry(
+            key=CONF_TOKEN,
+            type=ConfigEntryType.SECURE_STRING,
+            label="Yandex Music Token",
+            description="Manually pasted Yandex Music OAuth token. Only needed when "
+            "not borrowing from a Yandex Music instance.",
+            required=not borrowing,
+            hidden=borrowing,
+            value=cast("str", values.get(CONF_TOKEN)) if values else None,
+        ),
+        ConfigEntry(
+            key=CONF_MASS_PLAYER_ID,
+            type=ConfigEntryType.STRING,
+            label="Connected Music Assistant Player",
+            description="The Music Assistant player connected to this Ynison plugin. "
+            "When playback is directed to this device in the Yandex Music app, "
+            "the audio will play on the selected player. "
+            "Set to 'Auto' to automatically select a currently playing player.",
+            default_value=PLAYER_ID_AUTO,
+            options=[
+                ConfigValueOption("Auto (prefer playing player)", PLAYER_ID_AUTO),
+                *(
+                    ConfigValueOption(x.display_name, x.player_id)
+                    for x in sorted(
+                        mass.players.all_players(False, False),
+                        key=lambda p: p.display_name.lower(),
+                    )
+                ),
+            ],
+            required=True,
+        ),
+        ConfigEntry(
+            key=CONF_ALLOW_PLAYER_SWITCH,
+            type=ConfigEntryType.BOOLEAN,
+            label="Allow manual player switching",
+            description="When enabled, you can select this plugin as a source on any player "
+            "to switch playback to that player. When disabled, playback is fixed to the "
+            "configured default player.",
+            default_value=True,
+        ),
+        ConfigEntry(
+            key=CONF_OUTPUT_SAMPLE_RATE,
+            type=ConfigEntryType.STRING,
+            label="Output sample rate",
+            description="Sample rate for PCM output to the player. "
+            "'Auto' selects 44.1 kHz for lossy or 48 kHz for lossless sources.",
+            default_value=OUTPUT_AUTO,
+            options=[
+                ConfigValueOption("Auto (from source quality)", OUTPUT_AUTO),
+                ConfigValueOption("44100 Hz (CD)", "44100"),
+                ConfigValueOption("48000 Hz", "48000"),
+                ConfigValueOption("96000 Hz (Hi-Res)", "96000"),
+            ],
+            advanced=True,
+        ),
+        ConfigEntry(
+            key=CONF_OUTPUT_BIT_DEPTH,
+            type=ConfigEntryType.STRING,
+            label="Output bit depth",
+            description="Bit depth for PCM output to the player. "
+            "'Auto' selects 16-bit for lossy or 24-bit for lossless sources.",
+            default_value=OUTPUT_AUTO,
+            options=[
+                ConfigValueOption("Auto (from source quality)", OUTPUT_AUTO),
+                ConfigValueOption("16-bit", "16"),
+                ConfigValueOption("24-bit", "24"),
+            ],
+            advanced=True,
+        ),
+        ConfigEntry(
+            key=CONF_PUBLISH_NAME,
+            type=ConfigEntryType.STRING,
+            label="Device name in Yandex Music",
+            description="How this device appears in the Yandex Music app.",
+            default_value=DEFAULT_DISPLAY_NAME,
+            advanced=True,
+        ),
+        ConfigEntry(
+            key=CONF_DEVICE_ID,
+            type=ConfigEntryType.STRING,
+            label="Device ID",
+            hidden=True,
+            required=False,
+        ),
+    )

music_assistant/providers/yandex_ynison/auth.py

@@ -0,0 +1,22 @@
+"""Yandex Passport authentication helper.
+
+Delegates Passport interactions to the ``ya-passport-auth`` library. Only one
+helper is exposed: :func:`refresh_music_token`, used in borrow mode to obtain
+a fresh music-scoped OAuth token from a session token (``x_token``) read from
+the linked ``yandex_music`` provider's config.
+"""
+
+from __future__ import annotations
+
+from music_assistant_models.errors import LoginFailed
+from ya_passport_auth import PassportClient, SecretStr
+from ya_passport_auth.exceptions import YaPassportError
+
+
+async def refresh_music_token(x_token: SecretStr) -> SecretStr:
+    """Exchange an x_token for a fresh music-scoped OAuth token."""
+    try:
+        async with PassportClient.create() as client:
+            return await client.refresh_music_token(x_token)
+    except YaPassportError as err:
+        raise LoginFailed(f"Failed to refresh music token: {err}") from err

music_assistant/providers/yandex_ynison/config_helpers.py

@@ -0,0 +1,20 @@
+"""Configuration helpers for the Yandex Ynison plugin."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from music_assistant.mass import MusicAssistant
+
+
+def list_yandex_music_instances(mass: MusicAssistant) -> list[tuple[str, str]]:
+    """List configured yandex_music provider instances as (instance_id, display_name) pairs."""
+    instances: list[tuple[str, str]] = []
+    raw_providers = mass.config.get("providers", {})
+    for instance_id, prov_conf in raw_providers.items():
+        if prov_conf.get("domain") != "yandex_music":
+            continue
+        display_name = prov_conf.get("name") or instance_id
+        instances.append((str(instance_id), str(display_name)))
+    return instances

music_assistant/providers/yandex_ynison/constants.py

@@ -0,0 +1,69 @@
+"""Constants for the Yandex Ynison plugin."""
+
+from __future__ import annotations
+
+from typing import Final
+
+# Ynison WebSocket endpoints
+YNISON_REDIRECT_URL: Final[str] = (
+    "wss://ynison.music.yandex.ru/redirector.YnisonRedirectService/GetRedirectToYnison"
+)
+YNISON_STATE_PATH: Final[str] = "/ynison_state.YnisonStateService/PutYnisonState"
+
+# Origin header required by Ynison
+YNISON_ORIGIN: Final[str] = "https://music.yandex.ru"
+
+# Configuration keys
+CONF_TOKEN: Final[str] = "token"
+CONF_YM_INSTANCE: Final[str] = "ym_instance"
+CONF_MASS_PLAYER_ID: Final[str] = "mass_player_id"
+CONF_PUBLISH_NAME: Final[str] = "publish_name"
+CONF_ALLOW_PLAYER_SWITCH: Final[str] = "allow_player_switch"
+CONF_DEVICE_ID: Final[str] = "device_id"
+CONF_OUTPUT_SAMPLE_RATE: Final[str] = "output_sample_rate"
+CONF_OUTPUT_BIT_DEPTH: Final[str] = "output_bit_depth"
+CONF_FFMPEG_PACING: Final[str] = "ffmpeg_pacing"
+
+# ffmpeg pacing mode values
+PACING_REALTIME: Final[str] = "realtime"
+
+# Special value for "auto" config options
+OUTPUT_AUTO: Final[str] = "auto"
+
+# Sentinel value for CONF_YM_INSTANCE — use own manually entered token
+YM_INSTANCE_OWN: Final[str] = "__own__"
+
+# Player selection
+PLAYER_ID_AUTO: Final[str] = "__auto__"
+
+# yandex_music provider config keys (read via provider.config.get_value)
+YANDEX_MUSIC_CONF_QUALITY: Final[str] = "quality"
+YANDEX_MUSIC_CONF_TOKEN: Final[str] = "token"
+YANDEX_MUSIC_CONF_X_TOKEN: Final[str] = "x_token"
+YANDEX_MUSIC_LOSSLESS_QUALITIES: Final[frozenset[str]] = frozenset({"superb", "lossless"})
+
+# Defaults
+DEFAULT_DISPLAY_NAME: Final[str] = "Music Assistant"
+DEFAULT_APP_NAME: Final[str] = "Music Assistant"
+DEFAULT_APP_VERSION: Final[str] = "1.0.0"
+
+# Device types (from Ynison protobuf DeviceType enum)
+DEVICE_TYPE_WEB: Final[str] = "WEB"
+
+# Reconnect settings — indexed by attempt number; attempts past the tuple
+# saturate at the last entry (so reconnect continues forever at 60 s intervals).
+RECONNECT_DELAYS: Final[tuple[float, ...]] = (2.0, 4.0, 8.0, 16.0, 30.0, 60.0)
+
+# WebSocket timeouts
+WS_CONNECT_TIMEOUT: Final[float] = 15.0
+WS_HEARTBEAT: Final[float] = 30.0
+
+# Ynison error codes that require immediate reconnection
+YNISON_ERROR_REBALANCED: Final[str] = "300100001"
+YNISON_ERROR_NOT_SERVED: Final[str] = "300100002"
+YNISON_RECONNECT_ERROR_CODES: Final[frozenset[str]] = frozenset(
+    {
+        YNISON_ERROR_REBALANCED,
+        YNISON_ERROR_NOT_SERVED,
+    }
+)

music_assistant/providers/yandex_ynison/manifest.json

@@ -0,0 +1,12 @@
+{
+  "type": "plugin",
+  "domain": "yandex_ynison",
+  "stage": "beta",
+  "name": "Yandex Music Connect (Ynison)",
+  "description": "Makes a Music Assistant player appear as a device in the Yandex Music app via the Ynison protocol.",
+  "codeowners": ["@TrudenBoy"],
+  "requirements": ["ya-passport-auth==1.3.0"],
+  "depends_on": "yandex_music",
+  "documentation": "https://music-assistant.io/plugins/yandex-ynison/",
+  "multi_instance": true
+}

music_assistant/providers/yandex_ynison/protocols.py

@@ -0,0 +1,35 @@
+"""Protocol definitions for provider dependencies.
+
+Allows typing of external provider references without importing concrete classes.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from music_assistant_models.enums import MediaType
+    from music_assistant_models.streamdetails import StreamDetails
+
+
+@runtime_checkable
+class YandexMusicProviderLike(Protocol):
+    """Structural interface for the yandex_music MusicProvider.
+
+    Only the subset of methods/properties used by the Ynison plugin.
+    """
+
+    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
+        """Resolve stream details for a track."""
+        ...
+
+    def get_audio_stream(self, stream_details: StreamDetails) -> AsyncGenerator[bytes, None]:
+        """Return async generator of raw audio bytes."""
+        ...
+
+    async def get_rotor_station_tracks(
+        self, station_id: str, queue: str | int | None = None
+    ) -> tuple[list[Any], str | None]:
+        """Fetch tracks from a rotor station for radio queue replenishment."""
+        ...