/
/
/
1"""
2Setup flow for the Hue Entertainment provider.
3
4Pairs Music Assistant with a Hue bridge: the user enters the bridge IP and presses the
5physical link button on top of the bridge; Music Assistant then registers an app user
6(username + clientkey) and stores it - together with the bridge id (for mDNS IP tracking)
7- as setup data. The bridge only accepts registration for ~30s after the button press, so
8the pairing step runs in a retry loop.
9"""
10
11from __future__ import annotations
12
13from typing import TYPE_CHECKING
14
15from hue_entertainment import HueEntertainmentAPI
16from music_assistant_models.config_entries import ConfigEntry
17from music_assistant_models.enums import ConfigEntryType
18from music_assistant_models.errors import LoginFailed
19
20from music_assistant.models.setup_flow import SetupFlowError, StepExpiredError
21
22from .constants import (
23 CONF_BRIDGE_HOST,
24 CONF_BRIDGE_ID,
25 CONF_CLIENTKEY,
26 CONF_USERNAME,
27 HUE_DEVICE_TYPE,
28)
29
30if TYPE_CHECKING:
31 from music_assistant_models.config_entries import ConfigValueType
32
33 from music_assistant.models.setup_flow import SetupSession
34
35# a touch above the bridge's own ~30s link-button window, so the library's
36# "button not pressed" timeout surfaces before the step deadline does
37_PAIR_TIMEOUT = 35.0
38
39
40async def run_setup(session: SetupSession) -> None:
41 """
42 Run the Hue bridge pairing flow.
43
44 :param session: The setup session driving the flow.
45 """
46 host_default = str(session.context.setup_data.get(CONF_BRIDGE_HOST) or "")
47 errors: dict[str, str] | None = None
48 while True:
49 values = await session.form(
50 [
51 ConfigEntry(key="pair_intro", type=ConfigEntryType.LABEL),
52 ConfigEntry(
53 key=CONF_BRIDGE_HOST,
54 type=ConfigEntryType.STRING,
55 required=True,
56 value=host_default or None,
57 ),
58 ],
59 step_id="user",
60 errors=errors,
61 )
62 host = str(values[CONF_BRIDGE_HOST]).strip()
63 host_default = host
64 api = HueEntertainmentAPI(host)
65 try:
66 credentials = await session.progress_until(
67 api.pair(device_type=HUE_DEVICE_TYPE),
68 step_id="pairing",
69 text="press_button",
70 expires_in=_PAIR_TIMEOUT,
71 )
72 bridge_id = await _fetch_bridge_id(host, credentials["username"])
73 except StepExpiredError, TimeoutError:
74 errors = {"base": "button_not_pressed"}
75 continue
76 except LoginFailed as err:
77 errors = {"base": err.translation_key or str(err)}
78 continue
79 except Exception as err:
80 errors = {"base": str(err)}
81 continue
82 finally:
83 await api.close()
84 collected: dict[str, ConfigValueType] = {
85 CONF_BRIDGE_HOST: host,
86 CONF_USERNAME: str(credentials["username"]),
87 CONF_CLIENTKEY: str(credentials["clientkey"]),
88 }
89 if bridge_id:
90 collected[CONF_BRIDGE_ID] = bridge_id
91 try:
92 await session.finish(collected)
93 return
94 except SetupFlowError as err:
95 errors = {"base": err.translation_key or str(err)}
96
97
98async def _fetch_bridge_id(host: str, username: str) -> str | None:
99 """Fetch the bridge id (for mDNS IP tracking); best-effort, never fatal."""
100 api_authed = HueEntertainmentAPI(host, username)
101 try:
102 return await api_authed.get_bridge_id()
103 except Exception:
104 return None
105 finally:
106 await api_authed.close()
107