/
/
/
1"""
2Setup flow for the Plex Connect plugin.
3
4Each instance links one Music Assistant player to one configured Plex music provider.
5Both choices are derived from live runtime state (the loaded plex instances and the
6currently known players), so the form is built when the flow runs rather than declared
7as a static set of entries.
8"""
9
10from __future__ import annotations
11
12from typing import TYPE_CHECKING, Any
13
14from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
15from music_assistant_models.enums import ConfigEntryType, ProviderType
16
17from music_assistant.helpers.config_entries import PLAYBACK_TARGET_TYPES
18from music_assistant.models.setup_flow import AbortFlow, SetupFlowError
19
20from . import CONF_MASS_PLAYER_ID, CONF_PLEX_PROVIDER_ID
21
22if TYPE_CHECKING:
23 from music_assistant_models.config_entries import ConfigValueType
24
25 from music_assistant.mass import MusicAssistant
26 from music_assistant.models.setup_flow import SetupSession
27
28PLEX_DOMAIN = "plex"
29
30
31async def run_setup(session: SetupSession) -> None:
32 """
33 Run the Plex Connect setup flow: pick the Plex provider and the player to expose.
34
35 :param session: The setup session driving the flow.
36 """
37 plex_options = _plex_provider_options(session.mass)
38 if not plex_options:
39 # the engine's depends_on check only guarantees an enabled plex config exists,
40 # not that the instance actually loaded
41 raise AbortFlow("no_plex_provider")
42 player_options = _player_options(session.mass)
43 if not player_options:
44 raise AbortFlow("no_players")
45 setup_data = dict(session.context.setup_data)
46 errors: dict[str, str] | None = None
47 while True:
48 # setup_data (reconfigure) wins over the instance's stored option values, which
49 # still hold the selection on installs made before this flow existed
50 prefill: dict[str, Any] = {**session.context.values, **setup_data}
51 submitted = await session.form(
52 [
53 _select_entry(
54 CONF_PLEX_PROVIDER_ID, plex_options, prefill.get(CONF_PLEX_PROVIDER_ID)
55 ),
56 _select_entry(
57 CONF_MASS_PLAYER_ID, player_options, prefill.get(CONF_MASS_PLAYER_ID)
58 ),
59 ],
60 step_id="user",
61 errors=errors,
62 last_step=True,
63 )
64 setup_data.update(submitted)
65 try:
66 await session.finish(setup_data)
67 return
68 except SetupFlowError as err:
69 errors = {"base": err.translation_key or str(err)}
70
71
72def _plex_provider_options(mass: MusicAssistant) -> list[ConfigValueOption]:
73 """Return a picker option for every loaded Plex music provider instance."""
74 return [
75 ConfigValueOption(provider.instance_id, title=provider.name)
76 for provider in mass.get_provider_instances(
77 PLEX_DOMAIN, return_unavailable=True, provider_type=ProviderType.MUSIC
78 )
79 ]
80
81
82def _player_options(mass: MusicAssistant) -> list[ConfigValueOption]:
83 """Return a picker option for every available and enabled Music Assistant player."""
84 return [
85 ConfigValueOption(player.player_id, title=player.display_name)
86 for player in sorted(
87 mass.players.all_players(False, False),
88 key=lambda player: player.display_name.lower(),
89 )
90 if player.type in PLAYBACK_TARGET_TYPES
91 ]
92
93
94def _select_entry(
95 key: str, options: list[ConfigValueOption], prefill: ConfigValueType
96) -> ConfigEntry:
97 """
98 Build the required single-select form entry for the given key.
99
100 :param key: The setup data key the selection is collected under.
101 :param options: The available options (never empty).
102 :param prefill: Previously selected value, used when it is still a valid option.
103 """
104 selected = prefill if any(option.value == prefill for option in options) else options[0].value
105 return ConfigEntry(
106 key=key,
107 type=ConfigEntryType.STRING,
108 required=True,
109 options=options,
110 default_value=selected,
111 value=selected,
112 )
113