music-assistant-server

1"""MPD Player implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from typing import TYPE_CHECKING, Any, cast
8
9from mpd import CommandError, FailureResponseCode, MPDError
10from mpd.asyncio import MPDClient
11from music_assistant_models.config_entries import ConfigEntry
12from music_assistant_models.enums import (
13    ConfigEntryType,
14    IdentifierType,
15    PlaybackState,
16    PlayerFeature,
17)
18from music_assistant_models.errors import PlayerCommandFailed
19from music_assistant_models.player import DeviceInfo, PlayerMedia
20
21from music_assistant.constants import (
22    CONF_ENTRY_PREFER_WAV_FOR_LIVE_SOURCES_DEFAULT_ENABLED,
23    CONF_PASSWORD,
24)
25from music_assistant.models.player import Player
26from music_assistant.models.setup_flow import SetupFlowError
27
28from .constants import (
29    CONF_ENTRY_OUTPUT_CODEC_MPD,
30    ELAPSED_POLL_INTERVAL,
31    MPD_STATE_MAP,
32    RECONNECT_DELAY,
33)
34
35if TYPE_CHECKING:
36    from music_assistant.models.setup_flow import SetupSession
37
38    from .provider import MPDPlayerProvider
39
40
41class MPDPlayer(Player):
42    """Represents a single MPD server as a Music Assistant player."""
43
44    _attr_poll_interval = ELAPSED_POLL_INTERVAL
45
46    def __init__(
47        self,
48        provider: MPDPlayerProvider,
49        player_id: str,
50        host: str,
51        port: int = 6600,
52    ) -> None:
53        """
54        Initialize MPDPlayer.
55
56        :param provider: The MPDPlayerProvider instance.
57        :param player_id: Unique player identifier.
58        :param host: Hostname or IP address of the MPD server.
59        :param port: TCP port MPD is listening on.
60        """
61        super().__init__(provider, player_id)
62        self.host = host
63        self.port = port
64
65        # Two separate MPD connections are required:
66        # - _client: for sending commands (play, stop, setvol, etc.)
67        # - _idle_client: dedicated to the blocking idle() loop
68        # They must be separate because idle() monopolises the connection
69        # until a subsystem changes, preventing any other commands from being sent.
70        self._client: MPDClient | None = None
71        self._idle_client: MPDClient | None = None
72        self._idle_task: asyncio.Task[None] | None = None
73        self._reconnect_task_id: str = f"mpd_reconnect_{self.player_id}"
74        self._attr_needs_setup: bool = False
75        self._attr_setup_reason: str | None = None
76
77        self._attr_name = f"MPD ({host})"
78        self._attr_supported_features = {
79            PlayerFeature.PLAY_MEDIA,
80            PlayerFeature.PAUSE,
81            PlayerFeature.SEEK,
82            PlayerFeature.VOLUME_SET,
83        }
84
85        def _schedule_disconnect() -> None:
86            self.mass.call_later(0, self._disconnect, task_id=f"mpd_disconnect_{self.player_id}")
87
88        self._on_unload_callbacks.append(_schedule_disconnect)
89
90    @property
91    def needs_setup(self) -> bool:
92        """
93        Return True if the player requires a password to be configured.
94
95        :return: True when a password is required but has not yet been provided.
96        """
97        return self._attr_needs_setup
98
99    @property
100    def needs_poll(self) -> bool:
101        """
102        Return True when elapsed time polling is required.
103
104        MPD's idle mechanism does not push elapsed time continuously;
105        polling fills this gap during active playback.
106
107        :return: True when playback is active.
108        """
109        return self._attr_playback_state == PlaybackState.PLAYING
110
111    async def get_config_entries(self) -> list[ConfigEntry]:
112        """
113        Return player-level config entries.
114
115        :return: List of ConfigEntry objects for this player.
116        """
117        return [
118            CONF_ENTRY_OUTPUT_CODEC_MPD,
119            CONF_ENTRY_PREFER_WAV_FOR_LIVE_SOURCES_DEFAULT_ENABLED,
120        ]
121
122    async def on_config_updated(self) -> None:
123        """Reconnect to MPD when player configuration changes."""
124        self.password = cast("str | None", self.get_setup_value(CONF_PASSWORD) or None)
125        self._attr_needs_setup = False
126        self._attr_setup_reason = None
127        await self._connect()
128
129    async def run_setup_flow(self, session: SetupSession) -> None:
130        """Run the setup flow: collect the MPD server password."""
131        entries = [
132            ConfigEntry(key=CONF_PASSWORD, type=ConfigEntryType.SECURE_STRING, required=True)
133        ]
134        errors: dict[str, str] | None = None
135        while True:
136            values = await session.form(entries, step_id="user", errors=errors, last_step=True)
137            try:
138                await session.finish({CONF_PASSWORD: str(values[CONF_PASSWORD])})
139                return
140            except SetupFlowError as err:
141                errors = {"base": err.translation_key or str(err)}
142
143    async def poll(self) -> None:
144        """Fetch current MPD state to update elapsed time."""
145        await self._fetch_and_sync_state()
146
147    # ------------------------------------------------------------------
148    # Connection helpers
149    # ------------------------------------------------------------------
150
151    async def _connect(self) -> None:
152        """Establish both MPD connections and start the idle loop."""
153        # A failed attempt or a dead idle loop can leave connected clients behind
154        await self._disconnect()
155        try:
156            self._client = MPDClient()
157            await self._client.connect(self.host, self.port)
158            if self.password:
159                await self._client.password(self.password)
160
161            self._idle_client = MPDClient()
162            await self._idle_client.connect(self.host, self.port)
163            if self.password:
164                await self._idle_client.password(self.password)
165
166            status = await self._client.status()
167            self._attr_available = True
168            self._attr_needs_setup = False
169            self._attr_setup_reason = None
170            self._attr_device_info = DeviceInfo(
171                model=f"MPD {self._client.mpd_version}",
172                manufacturer="Music Player Daemon",
173            )
174            self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, self.host)
175            self.logger.info("Connected to MPD at %s:%s", self.host, self.port)
176            self._idle_task = self.mass.create_task(self._idle_loop())
177            await self._sync_state(status)
178            self.update_state()
179
180        except CommandError as err:
181            await self._disconnect()
182            if err.errno in (FailureResponseCode.PASSWORD, FailureResponseCode.PERMISSION):
183                self.logger.warning(
184                    "Authentication failed for MPD at %s:%s — configure password in player settings",
185                    self.host,
186                    self.port,
187                )
188                self._attr_available = False
189                self._attr_needs_setup = True
190                self._attr_setup_reason = "password_required"
191                self.update_state()
192            else:
193                self.logger.warning("MPD command error at %s:%s: %s", self.host, self.port, err)
194                self._attr_available = False
195                self.update_state()
196                self.reconnect()
197        except (MPDError, OSError) as err:
198            await self._disconnect()
199            self.logger.warning("Failed to connect to MPD at %s:%s: %s", self.host, self.port, err)
200            self._attr_available = False
201            self.update_state()
202            self.reconnect()
203
204    async def _disconnect(self) -> None:
205        """Cancel any pending reconnect, stop the idle loop and disconnect both MPD clients."""
206        self.mass.cancel_timer(self._reconnect_task_id)
207        # Connecting has no timeout, so an attempt may still be in flight and would
208        # otherwise arm a new reconnect after this teardown
209        reconnect_task = self.mass.get_task(self._reconnect_task_id)
210        if reconnect_task is not None and reconnect_task is not asyncio.current_task():
211            self.mass.cancel_task(self._reconnect_task_id)
212        if self._idle_task:
213            self._idle_task.cancel()
214            self._idle_task = None
215        for client in (self._client, self._idle_client):
216            if client:
217                client.disconnect()
218        self._client = None
219        self._idle_client = None
220
221    def reconnect(self) -> None:
222        """Schedule a reconnect attempt, deduplicating any pending reconnect tasks."""
223        self.mass.call_later(RECONNECT_DELAY, self._connect, task_id=self._reconnect_task_id)
224
225    # ------------------------------------------------------------------
226    # Background idle loop
227    # ------------------------------------------------------------------
228
229    async def _idle_loop(self) -> None:
230        """Receive push state changes from MPD and sync to MA."""
231        if self._idle_client is None:
232            return
233        try:
234            # MPD's idle command blocks until a subsystem changes, then yields
235            # the subsystem name. We act on player/mixer/playlist changes.
236            async for subsystem in self._idle_client.idle():
237                if subsystem in ("player", "mixer", "playlist"):
238                    await self._fetch_and_sync_state()
239        except MPDError as err:
240            self.logger.warning("MPD idle loop disconnected: %s", err)
241            self._attr_available = False
242            self.update_state()
243            self.reconnect()
244
245    # ------------------------------------------------------------------
246    # State sync
247    # ------------------------------------------------------------------
248
249    async def _fetch_and_sync_state(self) -> None:
250        """Fetch current MPD status and update MA player state."""
251        if self._client is None:
252            return
253        try:
254            status = await self._client.status()
255            await self._sync_state(status)
256            self.update_state()
257        except MPDError as err:
258            self.logger.warning("Failed to fetch MPD status: %s", err)
259
260    async def _sync_state(self, status: dict[str, Any]) -> None:
261        """
262        Map an MPD status dict onto MA player attributes.
263
264        :param status: Status dict as returned by the MPD ``status`` command.
265        """
266        mpd_state = status.get("state", "stop")
267        self._attr_playback_state = MPD_STATE_MAP.get(mpd_state, PlaybackState.IDLE)
268
269        # Volume: MPD reports -1 when no mixer is available
270        volume_str = status.get("volume", "-1")
271        if volume_str != "-1":
272            self._attr_volume_level = max(0, min(100, int(volume_str)))
273
274        # Elapsed time: MPD reports as a float string, absent when stopped
275        elapsed_str = status.get("elapsed")
276        if elapsed_str is not None:
277            self._attr_elapsed_time = float(elapsed_str)
278            self._attr_elapsed_time_last_updated = time.time()
279        else:
280            self._attr_elapsed_time = 0
281
282    # ------------------------------------------------------------------
283    # Player commands
284    # ------------------------------------------------------------------
285
286    async def play_media(self, media: PlayerMedia) -> None:
287        """
288        Send a play command to MPD using the MA stream URL.
289
290        :param media: Details of the media item to play.
291        """
292        if self._client is None:
293            return
294        url = await self.mass.streams.resolve_stream_url(self.player_id, media)
295        self.logger.debug("Sending stream URL to MPD: %s", url)
296        try:
297            await self._client.clear()
298            await self._client.add(url)
299            await self._client.play(0)
300            self._attr_current_media = media
301            self._attr_playback_state = PlaybackState.PLAYING
302            self.update_state()
303        except MPDError as err:
304            raise PlayerCommandFailed(f"play_media failed: {err}") from err
305
306    async def play(self) -> None:
307        """Resume playback."""
308        if self._client is None:
309            return
310        try:
311            await self._client.pause(0)
312            self._attr_playback_state = PlaybackState.PLAYING
313            self.update_state()
314        except MPDError as err:
315            raise PlayerCommandFailed(f"play failed: {err}") from err
316
317    async def stop(self) -> None:
318        """Stop playback and clear current media."""
319        if self._client is None:
320            return
321        try:
322            await self._client.stop()
323            self._attr_playback_state = PlaybackState.IDLE
324            self._attr_current_media = None
325            self.update_state()
326        except MPDError as err:
327            raise PlayerCommandFailed(f"stop failed: {err}") from err
328
329    async def pause(self) -> None:
330        """Pause playback."""
331        if self._client is None:
332            return
333        try:
334            await self._client.pause(1)
335            self._attr_playback_state = PlaybackState.PAUSED
336            self.update_state()
337        except MPDError as err:
338            raise PlayerCommandFailed(f"pause failed: {err}") from err
339
340    async def volume_set(self, volume_level: int) -> None:
341        """
342        Set the player volume.
343
344        :param volume_level: Volume level (0-100).
345        """
346        if self._client is None:
347            return
348        try:
349            await self._client.setvol(volume_level)
350            self._attr_volume_level = volume_level
351            self.update_state()
352        except MPDError as err:
353            raise PlayerCommandFailed(f"volume_set failed: {err}") from err
354
355    async def seek(self, position: int) -> None:
356        """
357        Seek to a position in the current track.
358
359        :param position: Position in seconds.
360        """
361        if self._client is None:
362            return
363        try:
364            await self._client.seekcur(position)
365        except MPDError as err:
366            raise PlayerCommandFailed(f"seek failed: {err}") from err
367