/
/
/
1"""FullyKiosk Player provider for Music Assistant."""
2
3from __future__ import annotations
4
5import logging
6from typing import cast
7
8from music_assistant_models.config_entries import ConfigEntry
9from music_assistant_models.enums import ConfigEntryType
10
11from music_assistant.constants import CONF_ENTRY_MANUAL_DISCOVERY_IPS, VERBOSE_LOG_LEVEL
12from music_assistant.models.player_provider import PlayerProvider
13
14from .dashboard import FullyKioskDashboards
15from .player import DEFAULT_PORT, FullyKioskPlayer
16
17CONF_MANUAL_IPS = CONF_ENTRY_MANUAL_DISCOVERY_IPS.key
18
19
20def _parse_host_entry(entry: str) -> tuple[str, int]:
21 """
22 Parse a single host entry into a (host, port) tuple.
23
24 Accepted formats:
25 - host
26 - host:port
27
28 Port defaults to 2323 if not specified.
29
30 :param entry: A single host entry string.
31 :return: Tuple of (host, port).
32 """
33 entry = entry.strip()
34 if ":" in entry:
35 host, port_str = entry.rsplit(":", 1)
36 try:
37 return host, int(port_str)
38 except ValueError:
39 return entry, DEFAULT_PORT
40 return entry, DEFAULT_PORT
41
42
43class FullyKioskProvider(PlayerProvider):
44 """
45 Fully Kiosk Player provider.
46
47 One provider instance manages one or more Fully Kiosk devices. Each device
48 is registered as a separate MA player. Devices are specified as a list of
49 host or host:port entries in the provider configuration; the password (and
50 optional SSL options) are configured on the player itself.
51 """
52
53 dashboards: FullyKioskDashboards
54
55 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
56 """Return Config entries to configure this provider."""
57 return (
58 ConfigEntry(
59 key="manual_discovery_ip_addresses",
60 type=ConfigEntryType.STRING,
61 default_value=[],
62 required=True,
63 multi_value=True,
64 ),
65 )
66
67 async def handle_async_init(self) -> None:
68 """Handle async initialization of the provider."""
69 self.dashboards = FullyKioskDashboards(self)
70 if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
71 logging.getLogger("fullykiosk").setLevel(logging.DEBUG)
72 else:
73 logging.getLogger("fullykiosk").setLevel(self.logger.level + 10)
74
75 async def loaded_in_mass(self) -> None:
76 """Sync registered players against the current hosts config."""
77 entries = cast("list[str]", self.config.get_value(CONF_MANUAL_IPS) or [])
78 new_ids = {f"fully_kiosk_{h}_{p}" for h, p in (_parse_host_entry(e) for e in entries)}
79
80 for player in self.players:
81 if player.player_id not in new_ids:
82 await self.mass.players.unregister(player.player_id)
83
84 for entry in entries:
85 host, port = _parse_host_entry(entry)
86 player_id = f"fully_kiosk_{host}_{port}"
87 if self.mass.players.get_player(player_id):
88 continue
89 player = FullyKioskPlayer(provider=self, player_id=player_id, host=host, port=port)
90 await self.mass.players.register(player)
91
92 async def discover_players(self) -> None:
93 """Register one FullyKioskPlayer per entry in the hosts config."""
94 entries = cast("list[str]", self.config.get_value(CONF_MANUAL_IPS) or [])
95 for entry in entries:
96 host, port = _parse_host_entry(entry)
97 player_id = f"fully_kiosk_{host}_{port}"
98 if self.mass.players.get_player(player_id):
99 continue
100 player = FullyKioskPlayer(provider=self, player_id=player_id, host=host, port=port)
101 await self.mass.players.register(player)
102
103 async def remove_player(self, player_id: str) -> None:
104 """Remove a player and persist that removal in the provider config."""
105 host_port: tuple[str, int] | None = None
106 if player := self.mass.players.get_player(player_id):
107 if isinstance(player, FullyKioskPlayer):
108 host_port = (player.host, player.port)
109 if host_port is None:
110 for entry in cast("list[str]", self.config.get_value(CONF_MANUAL_IPS) or []):
111 host, port = _parse_host_entry(entry)
112 if f"fully_kiosk_{host}_{port}" == player_id:
113 host_port = (host, port)
114 break
115 if host_port is not None:
116 entries = cast("list[str]", self.config.get_value(CONF_MANUAL_IPS) or [])
117 new_entries = [entry for entry in entries if _parse_host_entry(entry) != host_port]
118 if new_entries != entries:
119 self._update_config_value(CONF_MANUAL_IPS, new_entries)
120 await self.mass.players.unregister(player_id, True)
121
122 async def unload(self, is_removed: bool = False) -> None:
123 """Handle unload/close of the provider."""
124 await self.dashboards.unload()
125 await super().unload(is_removed)
126