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