/
/
/
1"""Setup flow for the VBAN Receiver plugin."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6
7from aiovban.enums import VBANSampleRate
8from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
9from music_assistant_models.enums import ConfigEntryType
10
11from music_assistant.constants import CONF_BIND_IP, CONF_BIND_PORT, CONF_ENTRY_WARN_PREVIEW
12from music_assistant.helpers.util import get_ip_addresses
13from music_assistant.models.setup_flow import SetupFlowError
14
15from .constants import (
16 CONF_AUDIO_CHANNELS,
17 CONF_PCM_AUDIO_FORMAT,
18 CONF_PCM_SAMPLE_RATE,
19 CONF_SENDER_HOST,
20 CONF_VBAN_STREAM_NAME,
21 DEFAULT_AUDIO_CHANNELS,
22 DEFAULT_PCM_AUDIO_FORMAT,
23 DEFAULT_PCM_SAMPLE_RATE,
24 DEFAULT_UDP_PORT,
25)
26from .helpers import get_supported_pcm_formats
27
28if TYPE_CHECKING:
29 from music_assistant.models.setup_flow import SetupSession
30
31
32async def run_setup(session: SetupSession) -> None:
33 """
34 Configure the remote VBAN stream and local receiver endpoint.
35
36 :param session: The setup session driving the flow.
37 """
38 setup_data = dict(session.context.setup_data)
39 errors: dict[str, str] | None = None
40 while True:
41 prefill: dict[str, Any] = {**session.context.values, **setup_data}
42 values = await session.form(
43 [
44 CONF_ENTRY_WARN_PREVIEW,
45 ConfigEntry(
46 key=CONF_BIND_PORT,
47 type=ConfigEntryType.INTEGER,
48 required=True,
49 default_value=DEFAULT_UDP_PORT,
50 value=prefill.get(CONF_BIND_PORT),
51 ),
52 ConfigEntry(
53 key=CONF_VBAN_STREAM_NAME,
54 type=ConfigEntryType.STRING,
55 required=True,
56 default_value="Network AUX",
57 value=prefill.get(CONF_VBAN_STREAM_NAME),
58 validate=_validate_stream_name, # type: ignore[arg-type]
59 ),
60 ConfigEntry(
61 key=CONF_SENDER_HOST,
62 type=ConfigEntryType.STRING,
63 required=True,
64 default_value="127.0.0.1",
65 value=prefill.get(CONF_SENDER_HOST),
66 ),
67 ConfigEntry(
68 key=CONF_PCM_AUDIO_FORMAT,
69 type=ConfigEntryType.STRING,
70 required=True,
71 default_value=DEFAULT_PCM_AUDIO_FORMAT,
72 value=prefill.get(CONF_PCM_AUDIO_FORMAT),
73 options=[
74 ConfigValueOption(value, title=value)
75 for value in get_supported_pcm_formats()
76 ],
77 ),
78 ConfigEntry(
79 key=CONF_PCM_SAMPLE_RATE,
80 type=ConfigEntryType.INTEGER,
81 required=True,
82 default_value=DEFAULT_PCM_SAMPLE_RATE,
83 value=prefill.get(CONF_PCM_SAMPLE_RATE),
84 options=[
85 ConfigValueOption(value, title=str(value))
86 for value in _get_vban_sample_rates()
87 ],
88 ),
89 ConfigEntry(
90 key=CONF_AUDIO_CHANNELS,
91 type=ConfigEntryType.INTEGER,
92 required=True,
93 default_value=DEFAULT_AUDIO_CHANNELS,
94 value=prefill.get(CONF_AUDIO_CHANNELS),
95 options=[ConfigValueOption(value, title=str(value)) for value in range(1, 9)],
96 ),
97 ConfigEntry(
98 key=CONF_BIND_IP,
99 type=ConfigEntryType.STRING,
100 required=True,
101 default_value="0.0.0.0",
102 value=prefill.get(CONF_BIND_IP),
103 options=[
104 ConfigValueOption(value, title=value)
105 for value in {"0.0.0.0", *await get_ip_addresses(include_ipv6=True)}
106 ],
107 advanced=True,
108 ),
109 ],
110 step_id="user",
111 errors=errors,
112 last_step=True,
113 )
114 setup_data.update(values)
115 try:
116 await session.finish(setup_data)
117 return
118 except SetupFlowError as err:
119 errors = {"base": err.translation_key or str(err)}
120
121
122def _get_vban_sample_rates() -> list[int]:
123 """Return the supported VBAN sample rates."""
124 return [int(member.split("_")[1]) for member in VBANSampleRate.__members__]
125
126
127def _validate_stream_name(config_value: str) -> bool:
128 """Return whether a VBAN stream name is valid."""
129 try:
130 config_value.encode("ascii")
131 except UnicodeEncodeError:
132 return False
133 return len(config_value) <= 16
134