/
/
/
1"""FullyKiosk Player implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from typing import TYPE_CHECKING
8
9from music_assistant_models.enums import PlaybackState, PlayerFeature, PlayerType
10from music_assistant_models.errors import PlayerCommandFailed, PlayerUnavailableError
11
12from music_assistant.constants import CONF_ENTRY_OUTPUT_CODEC_DEFAULT_MP3
13from music_assistant.models.player import DeviceInfo, Player, PlayerMedia
14
15if TYPE_CHECKING:
16 from fullykiosk import FullyKiosk
17 from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
18
19 from .provider import FullyKioskProvider
20
21AUDIOMANAGER_STREAM_MUSIC = 3
22
23
24class FullyKioskPlayer(Player):
25 """FullyKiosk Player implementation."""
26
27 def __init__(
28 self,
29 provider: FullyKioskProvider,
30 player_id: str,
31 fully_kiosk: FullyKiosk,
32 address: str,
33 ) -> None:
34 """Initialize the FullyKiosk Player."""
35 super().__init__(provider, player_id)
36 self.fully_kiosk = fully_kiosk
37 # Set player attributes
38 self._attr_type = PlayerType.PLAYER
39 self._attr_supported_features = {PlayerFeature.VOLUME_SET}
40 self._attr_name = self.fully_kiosk.deviceInfo["deviceName"]
41 self._attr_device_info = DeviceInfo(
42 model=self.fully_kiosk.deviceInfo["deviceModel"],
43 manufacturer=self.fully_kiosk.deviceInfo["deviceManufacturer"],
44 )
45 self._attr_device_info.ip_address = address
46 self._attr_available = True
47 self._attr_needs_poll = True
48 self._attr_poll_interval = 10
49
50 @property
51 def requires_flow_mode(self) -> bool:
52 """Return if the player requires flow mode."""
53 return True
54
55 async def get_config_entries(
56 self,
57 action: str | None = None,
58 values: dict[str, ConfigValueType] | None = None,
59 ) -> list[ConfigEntry]:
60 """Return all (provider/player specific) Config Entries for the given player (if any)."""
61 return [
62 CONF_ENTRY_OUTPUT_CODEC_DEFAULT_MP3,
63 ]
64
65 def set_attributes(self) -> None:
66 """Set/update FullyKiosk player attributes."""
67 self._attr_name = self.fully_kiosk.deviceInfo["deviceName"]
68 for volume_dict in self.fully_kiosk.deviceInfo.get("audioVolumes", []):
69 if str(AUDIOMANAGER_STREAM_MUSIC) in volume_dict:
70 volume = volume_dict[str(AUDIOMANAGER_STREAM_MUSIC)]
71 self._attr_volume_level = volume
72 break
73 current_url = self.fully_kiosk.deviceInfo.get("soundUrlPlaying")
74 if not current_url:
75 self._attr_playback_state = PlaybackState.IDLE
76 self._attr_available = True
77
78 async def volume_set(self, volume_level: int) -> None:
79 """Send VOLUME_SET command to given player."""
80 await self.fully_kiosk.setAudioVolume(volume_level, AUDIOMANAGER_STREAM_MUSIC)
81 self._attr_volume_level = volume_level
82 self.update_state()
83
84 async def stop(self) -> None:
85 """Send STOP command to given player."""
86 await self.fully_kiosk.stopSound()
87 self._attr_playback_state = PlaybackState.IDLE
88 self._attr_current_media = None
89 self.update_state()
90
91 async def play(self) -> None:
92 """Handle PLAY command on the player."""
93 raise PlayerCommandFailed("Playback can not be resumed.")
94
95 async def play_media(self, media: PlayerMedia) -> None:
96 """Handle PLAY MEDIA on given player."""
97 await self.fully_kiosk.playSound(media.uri, AUDIOMANAGER_STREAM_MUSIC)
98 self._attr_current_media = media
99 self._attr_elapsed_time = 0
100 self._attr_elapsed_time_last_updated = time.time()
101 self._attr_playback_state = PlaybackState.PLAYING
102 self.update_state()
103
104 async def poll(self) -> None:
105 """Poll player for state updates."""
106 try:
107 async with asyncio.timeout(15):
108 await self.fully_kiosk.getDeviceInfo()
109 self.set_attributes()
110 self.update_state()
111 except Exception as err:
112 msg = f"Unable to start the FullyKiosk connection ({err!s}"
113 raise PlayerUnavailableError(msg) from err
114