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