/
/
/
1"""MPD Player Provider implementation."""
2
3from __future__ import annotations
4
5from typing import cast
6
7from music_assistant_models.config_entries import ConfigEntry
8from music_assistant_models.enums import ConfigEntryType
9
10from music_assistant.constants import CONF_ENTRY_MANUAL_DISCOVERY_IPS
11from music_assistant.models.player_provider import PlayerProvider
12
13from .player import MPDPlayer
14
15CONF_MANUAL_IPS = CONF_ENTRY_MANUAL_DISCOVERY_IPS.key
16
17
18def _parse_host_entry(entry: str) -> tuple[str, int]:
19 """
20 Parse a single host entry into a (host, port) tuple.
21
22 Accepted formats:
23 - host
24 - host:port
25
26 Port defaults to 6600 if not specified.
27
28 :param entry: A single host entry string.
29 :return: Tuple of (host, port).
30 """
31 entry = entry.strip()
32 if ":" in entry:
33 host, port_str = entry.rsplit(":", 1)
34 try:
35 return host, int(port_str)
36 except ValueError:
37 return entry, 6600
38 return entry, 6600
39
40
41class MPDPlayerProvider(PlayerProvider):
42 """
43 MPD Player provider.
44
45 One provider instance manages one or more MPD servers. Each server
46 is registered as a separate MA player. Servers are specified as a
47 list of host or host:port entries in the provider configuration.
48 """
49
50 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
51 """Return Config entries to configure this provider."""
52 return (
53 ConfigEntry(
54 key="manual_discovery_ip_addresses",
55 type=ConfigEntryType.STRING,
56 default_value=[],
57 required=True,
58 multi_value=True,
59 ),
60 )
61
62 async def loaded_in_mass(self) -> None:
63 """Sync registered players against the current hosts config."""
64 entries = cast("list[str]", self.config.get_value(CONF_MANUAL_IPS) or [])
65 new_ids = {f"mpd_{h}_{p}" for h, p in (_parse_host_entry(e) for e in entries)}
66
67 for player in self.players:
68 if player.player_id not in new_ids:
69 await self.mass.players.unregister(player.player_id)
70
71 for entry in entries:
72 host, port = _parse_host_entry(entry)
73 player_id = f"mpd_{host}_{port}"
74 if self.mass.players.get_player(player_id):
75 continue
76 player = MPDPlayer(provider=self, player_id=player_id, host=host, port=port)
77 await self.mass.players.register(player)
78
79 async def discover_players(self) -> None:
80 """Register one MPDPlayer per entry in the hosts config."""
81 entries = cast("list[str]", self.config.get_value(CONF_MANUAL_IPS) or [])
82 for entry in entries:
83 host, port = _parse_host_entry(entry)
84 player_id = f"mpd_{host}_{port}"
85 if self.mass.players.get_player(player_id):
86 continue
87 player = MPDPlayer(provider=self, player_id=player_id, host=host, port=port)
88 await self.mass.players.register(player)
89
90 async def remove_player(self, player_id: str) -> None:
91 """Remove a player and persist that removal in the provider config."""
92 host_port: tuple[str, int] | None = None
93 if player := self.mass.players.get_player(player_id):
94 if isinstance(player, MPDPlayer):
95 host_port = (player.host, player.port)
96 if host_port is None:
97 for entry in cast("list[str]", self.config.get_value(CONF_MANUAL_IPS) or []):
98 host, port = _parse_host_entry(entry)
99 if f"mpd_{host}_{port}" == player_id:
100 host_port = (host, port)
101 break
102 if host_port is not None:
103 entries = cast("list[str]", self.config.get_value(CONF_MANUAL_IPS) or [])
104 new_entries = [entry for entry in entries if _parse_host_entry(entry) != host_port]
105 if new_entries != entries:
106 self._update_config_value(CONF_MANUAL_IPS, new_entries)
107 await self.mass.players.unregister(player_id, True)
108