music-assistant-server

1"""FullyKiosk Player implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import re
7import time
8from dataclasses import dataclass
9from typing import TYPE_CHECKING, Any, cast
10
11from aiohttp import ClientError, ClientSession, Fingerprint
12from fullykiosk import FullyKiosk
13from fullykiosk.exceptions import FullyKioskError
14from music_assistant_models.config_entries import ConfigEntry
15from music_assistant_models.enums import (
16    ConfigEntryType,
17    IdentifierType,
18    PlaybackState,
19    PlayerFeature,
20)
21from music_assistant_models.errors import PlayerCommandFailed
22
23from music_assistant.constants import (
24    CONF_ENTRY_OUTPUT_CODEC_DEFAULT_MP3,
25    CONF_PASSWORD,
26    CONF_SSL_FINGERPRINT,
27    CONF_USE_SSL,
28    CONF_VERIFY_SSL,
29)
30from music_assistant.models.player import DeviceInfo, Player, PlayerMedia
31from music_assistant.models.setup_flow import SetupFlowError
32
33if TYPE_CHECKING:
34    from music_assistant.models.setup_flow import SetupSession
35
36    from .dashboard import FullyKioskDashboards
37    from .provider import FullyKioskProvider
38
39AUDIOMANAGER_STREAM_MUSIC = 3
40DEFAULT_PORT = 2323
41
42
43@dataclass
44class _FingerprintSessionWrapper:
45    """Proxy ClientSession that enforces a TLS fingerprint."""
46
47    session: ClientSession
48    fingerprint: Fingerprint
49
50    def get(self, *args: Any, **kwargs: Any) -> Any:
51        """Call the wrapped session.get while injecting the fingerprint."""
52        kwargs.setdefault("ssl", self.fingerprint)
53        return self.session.get(*args, **kwargs)
54
55    def __getattr__(self, name: str) -> Any:
56        """Delegate attribute access to the wrapped session."""
57        return getattr(self.session, name)
58
59
60def _build_fingerprint(value: str) -> Fingerprint:
61    """Parse a fingerprint string (sha256 hex) into an aiohttp Fingerprint."""
62    normalized = re.sub(r"[^0-9a-fA-F]", "", value).lower()
63    if not normalized:
64        msg = "Empty fingerprint provided."
65        raise ValueError(msg)
66    if len(normalized) % 2 != 0:
67        msg = "Fingerprint must contain an even number of hex characters."
68        raise ValueError(msg)
69    digest = bytes.fromhex(normalized)
70    return Fingerprint(digest)
71
72
73class FullyKioskPlayer(Player):
74    """FullyKiosk Player implementation."""
75
76    _attr_poll_interval = 10
77
78    def __init__(
79        self,
80        provider: FullyKioskProvider,
81        player_id: str,
82        host: str,
83        port: int = DEFAULT_PORT,
84    ) -> None:
85        """Initialize the FullyKiosk Player."""
86        super().__init__(provider, player_id)
87        self.host = host
88        self.port = port
89        self.fully_kiosk: FullyKiosk | None = None
90        self._attr_needs_setup = True
91        self._attr_setup_reason = "password_required"
92        self._attr_available = False
93        self._attr_name = f"Fully Kiosk ({host})"
94        self._attr_supported_features = {PlayerFeature.PLAY_MEDIA, PlayerFeature.VOLUME_SET}
95
96    @property
97    def needs_poll(self) -> bool:
98        """Poll only once a password has been configured."""
99        return not self._attr_needs_setup
100
101    @property
102    def requires_flow_mode(self) -> bool:
103        """Return if the player requires flow mode."""
104        return True
105
106    async def get_config_entries(self) -> list[ConfigEntry]:
107        """Return all (provider/player specific) Config Entries for the given player (if any)."""
108        return [
109            ConfigEntry(
110                key=CONF_USE_SSL,
111                type=ConfigEntryType.BOOLEAN,
112                default_value=False,
113                advanced=True,
114            ),
115            ConfigEntry(
116                key=CONF_VERIFY_SSL,
117                type=ConfigEntryType.BOOLEAN,
118                default_value=True,
119                advanced=True,
120            ),
121            ConfigEntry(
122                key=CONF_SSL_FINGERPRINT,
123                type=ConfigEntryType.STRING,
124                required=False,
125                advanced=True,
126            ),
127            CONF_ENTRY_OUTPUT_CODEC_DEFAULT_MP3,
128        ]
129
130    async def on_config_updated(self) -> None:
131        """Reconnect to the Fully Kiosk device when player configuration changes."""
132        password = cast("str | None", self.get_setup_value(CONF_PASSWORD) or None)
133        if not password:
134            self.fully_kiosk = None
135            self._attr_needs_setup = True
136            self._attr_setup_reason = "password_required"
137            self._attr_available = False
138            self._dashboards.unregister(self.player_id)
139            self.update_state()
140            return
141        await self._connect()
142
143    async def run_setup_flow(self, session: SetupSession) -> None:
144        """Run the setup flow: collect the Fully Kiosk API password."""
145        entries = [
146            ConfigEntry(key=CONF_PASSWORD, type=ConfigEntryType.SECURE_STRING, required=True)
147        ]
148        errors: dict[str, str] | None = None
149        while True:
150            values = await session.form(entries, step_id="user", errors=errors, last_step=True)
151            try:
152                await session.finish({CONF_PASSWORD: str(values[CONF_PASSWORD])})
153                return
154            except SetupFlowError as err:
155                errors = {"base": err.translation_key or str(err)}
156
157    async def poll(self) -> None:
158        """Poll player for state updates."""
159        if self.fully_kiosk is None:
160            await self._connect()
161            return
162        try:
163            async with asyncio.timeout(15):
164                await self.fully_kiosk.getDeviceInfo()
165        except (TimeoutError, ClientError, FullyKioskError) as err:
166            # do not log err: its repr may include the request URL with the password in the query string
167            self.logger.warning(
168                "Fully Kiosk device %s unavailable (%s) — will retry",
169                self.host,
170                type(err).__name__,
171            )
172            self._attr_available = False
173            self.fully_kiosk = None
174            self._dashboards.unregister(self.player_id)
175            self.update_state()
176            return
177        self._sync_state()
178        self.update_state()
179
180    async def volume_set(self, volume_level: int) -> None:
181        """Send VOLUME_SET command to given player."""
182        if self.fully_kiosk is None:
183            return
184        await self.fully_kiosk.setAudioVolume(volume_level, AUDIOMANAGER_STREAM_MUSIC)
185        self._attr_volume_level = volume_level
186        self.update_state()
187
188    async def stop(self) -> None:
189        """Send STOP command to given player."""
190        if self.fully_kiosk is None:
191            return
192        await self.fully_kiosk.stopSound()
193        self._attr_playback_state = PlaybackState.IDLE
194        self._attr_current_media = None
195        self.update_state()
196
197    async def play(self) -> None:
198        """Handle PLAY command on the player."""
199        raise PlayerCommandFailed("Playback can not be resumed.")
200
201    async def play_media(self, media: PlayerMedia) -> None:
202        """Handle PLAY MEDIA on given player."""
203        if self.fully_kiosk is None:
204            raise PlayerCommandFailed("Fully Kiosk device is not connected.")
205        url = await self.mass.streams.resolve_stream_url(self.player_id, media)
206        await self.fully_kiosk.playSound(url, AUDIOMANAGER_STREAM_MUSIC)
207        self._attr_current_media = media
208        self._attr_elapsed_time = 0
209        self._attr_elapsed_time_last_updated = time.time()
210        self._attr_playback_state = PlaybackState.PLAYING
211        self.update_state()
212
213    async def on_unload(self) -> None:
214        """Handle logic when the player is unloaded from the Player controller."""
215        await super().on_unload()
216        self._dashboards.unregister(self.player_id)
217
218    async def _connect(self) -> None:
219        """Establish a connection to the Fully Kiosk device."""
220        password = cast("str | None", self.get_setup_value(CONF_PASSWORD) or None)
221        if not password:
222            self._attr_needs_setup = True
223            self._attr_setup_reason = "password_required"
224            self._attr_available = False
225            self.fully_kiosk = None
226            self._dashboards.unregister(self.player_id)
227            self.update_state()
228            return
229
230        session_info = self._resolve_http_session()
231        if session_info is None:
232            self._attr_available = False
233            self._dashboards.unregister(self.player_id)
234            self.update_state()
235            return
236        http_session, use_ssl, verify_ssl = session_info
237
238        client = FullyKiosk(
239            http_session,
240            self.host,
241            self.port,
242            password,
243            use_ssl=use_ssl,
244            verify_ssl=verify_ssl,
245        )
246        try:
247            async with asyncio.timeout(15):
248                await client.getDeviceInfo()
249        except (TimeoutError, ClientError, FullyKioskError) as err:
250            # do not log err: its repr may include the request URL with the password in the query string
251            self.logger.warning(
252                "Failed to connect to Fully Kiosk at %s:%s (%s)",
253                self.host,
254                self.port,
255                type(err).__name__,
256            )
257            self.fully_kiosk = None
258            self._attr_available = False
259            self._dashboards.unregister(self.player_id)
260            self.update_state()
261            return
262
263        self.fully_kiosk = client
264        scheme = "https" if use_ssl else "http"
265        address = f"{scheme}://{self.host}:{self.port}"
266        self._attr_device_info = DeviceInfo(
267            model=client.deviceInfo.get("deviceModel", ""),
268            manufacturer=client.deviceInfo.get("deviceManufacturer", ""),
269        )
270        self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, address)
271        self._attr_needs_setup = False
272        self._attr_setup_reason = None
273        self._attr_available = True
274        self._sync_state()
275        self._dashboards.register(self)
276        self.update_state()
277
278    def _resolve_http_session(
279        self,
280    ) -> tuple[ClientSession | _FingerprintSessionWrapper, bool, bool] | None:
281        """Resolve (http_session, use_ssl, verify_ssl) from config, or None if misconfigured."""
282        use_ssl = bool(self.config.get_value(CONF_USE_SSL))
283        fingerprint_value = self.config.get_value(CONF_SSL_FINGERPRINT)
284        fingerprint_raw = fingerprint_value.strip() if isinstance(fingerprint_value, str) else ""
285        if fingerprint_raw and not use_ssl:
286            self.logger.warning(
287                "Fully Kiosk %s: fingerprint validation requires HTTPS to be enabled",
288                self.host,
289            )
290            return None
291        verify_ssl = bool(self.config.get_value(CONF_VERIFY_SSL)) if use_ssl else False
292        if not use_ssl:
293            return self.mass.http_session_no_ssl, use_ssl, verify_ssl
294        if fingerprint_raw:
295            try:
296                fingerprint = _build_fingerprint(fingerprint_raw)
297            except ValueError as err:
298                self.logger.warning(
299                    "Fully Kiosk %s: invalid TLS fingerprint configured: %s", self.host, err
300                )
301                return None
302            return _FingerprintSessionWrapper(self.mass.http_session, fingerprint), use_ssl, True
303        session = self.mass.http_session if verify_ssl else self.mass.http_session_no_ssl
304        return session, use_ssl, verify_ssl
305
306    def _sync_state(self) -> None:
307        """Refresh player attributes from the latest deviceInfo."""
308        if self.fully_kiosk is None:
309            return
310        device_info = self.fully_kiosk.deviceInfo
311        if name := device_info.get("deviceName"):
312            self._attr_name = name
313        for volume_dict in device_info.get("audioVolumes", []):
314            if str(AUDIOMANAGER_STREAM_MUSIC) in volume_dict:
315                self._attr_volume_level = volume_dict[str(AUDIOMANAGER_STREAM_MUSIC)]
316                break
317        if not device_info.get("soundUrlPlaying"):
318            self._attr_playback_state = PlaybackState.IDLE
319
320    @property
321    def _dashboards(self) -> FullyKioskDashboards:
322        """Return the owning provider's dashboard adapter."""
323        return cast("FullyKioskProvider", self.provider).dashboards
324