/
/
/
1"""Helpers for Sendspin provider."""
2
3from __future__ import annotations
4
5from dataclasses import dataclass, replace
6from typing import TYPE_CHECKING
7
8from aiosendspin.models.core import PairMethodDescriptor
9from aiosendspin.models.types import PairAbortReason, PairMethod
10from aiosendspin.noise.driver import HandshakeAbortedError
11from aiosendspin.noise.pairing import PairingAbortError, PairingError, PairingTimeoutError
12from music_assistant_models.config_entries import ConfigEntry
13from music_assistant_models.enums import ConfigEntryType
14
15from .constants import BRIDGE_PREFIX
16
17if TYPE_CHECKING:
18 from collections.abc import Iterable
19
20 from aiosendspin.models.core import ClientHelloPayload
21 from aiosendspin.models.management import ManagementResultData
22
23
24class SecurityActionError(Exception):
25 """A pairing/management action failure carrying a strings.json alert slug for the UI."""
26
27 def __init__(self, alert_key: str, *, detail: str | None = None) -> None:
28 """Initialize with the alert slug and optional untranslated {0}-placeholder detail."""
29 super().__init__(alert_key if detail is None else f"{alert_key}: {detail}")
30 self.alert_key = alert_key
31 self.detail = detail
32
33
34@dataclass(frozen=True)
35class AlertText:
36 """A strings.json slug and optional {0}-placeholder params for an operator ALERT entry."""
37
38 key: str
39 params: list[str] | None = None
40
41
42_PAIR_ABORT_KEYS = {
43 PairAbortReason.ATTEMPT_TIMEOUT: "pairing_error_timeout",
44 PairAbortReason.CONCURRENT_ATTEMPT: "pairing_error_concurrent",
45 PairAbortReason.METHOD_NOT_SUPPORTED: "pairing_error_method_unsupported",
46 PairAbortReason.PIN_LENGTH_UNACCEPTABLE: "pairing_error_pin_length",
47 PairAbortReason.PIN_MISMATCH: "pairing_error_pin_mismatch",
48 PairAbortReason.USER_CANCELLED: "pairing_error_cancelled",
49}
50
51
52def error_alert(err: Exception) -> AlertText:
53 """Map a pairing or management failure to a localized operator alert."""
54 if isinstance(err, SecurityActionError):
55 return AlertText(err.alert_key, [err.detail] if err.detail is not None else None)
56 if isinstance(err, PairingAbortError):
57 key = _PAIR_ABORT_KEYS.get(err.reason)
58 if key is not None:
59 return AlertText(key)
60 return AlertText("pairing_error_aborted", [err.reason.value])
61 if isinstance(err, TimeoutError | PairingTimeoutError):
62 return AlertText("pairing_error_timeout")
63 if isinstance(err, OSError):
64 return AlertText("pairing_error_storage", [str(err)])
65 if isinstance(err, HandshakeAbortedError):
66 return AlertText("pairing_error_handshake")
67 if isinstance(err, PairingError):
68 return AlertText("pairing_error_failed", [str(err)])
69 return AlertText("pairing_error_generic")
70
71
72def alert_entry(text: AlertText) -> ConfigEntry:
73 """Build an ALERT config entry from a localized alert descriptor."""
74 return ConfigEntry(key=text.key, type=ConfigEntryType.ALERT, translation_params=text.params)
75
76
77def action_entry(action: str, *, advanced: bool = False) -> ConfigEntry:
78 """Build an ACTION config entry whose key mirrors its action."""
79 return ConfigEntry(key=action, type=ConfigEntryType.ACTION, action=action, advanced=advanced)
80
81
82def effective_pair_methods(
83 info: ClientHelloPayload | None, config: ManagementResultData | None
84) -> list[PairMethodDescriptor]:
85 """
86 Return the pairing methods the device currently offers.
87
88 A pairing config fetched over a management session on the current connection is
89 authoritative; the hello advertisement cannot reflect config changes until reconnect.
90 Methods the config enables beyond the hello get a synthesized descriptor.
91 """
92 hello_methods = list(info.supported_pair_methods or []) if info is not None else []
93 if config is None:
94 return hello_methods
95 advertised = {descriptor.method: descriptor for descriptor in hello_methods}
96 methods: list[PairMethodDescriptor] = []
97 for method, method_config in (
98 (PairMethod.PAIRING_PSK, config.pairing_psk),
99 (PairMethod.STATIC_PIN, config.static_pin),
100 (PairMethod.DYNAMIC_PIN, config.dynamic_pin),
101 ):
102 if method_config is None or not method_config.enabled:
103 continue
104 descriptor = advertised.get(method) or PairMethodDescriptor(method=method)
105 methods.append(replace(descriptor, min_pin_length=method_config.min_pin_length))
106 return methods
107
108
109def negotiated_pin_length(descriptor: PairMethodDescriptor | None, server_min: int) -> int:
110 """
111 Return the dynamic PIN length this session will use.
112
113 Mirrors the server's own negotiation so the operator prompt can name the digit count
114 before the device reports it.
115 """
116 client_min = descriptor.min_pin_length if descriptor is not None else None
117 return max(client_min or 0, server_min)
118
119
120def pin_code_format(length: int) -> str:
121 """Return the PAIRING_CODE entry format for a numeric PIN of `length` digits."""
122 if length >= 6 and length % 2 == 0:
123 half = length // 2
124 return f"{'#' * half}-{'#' * half}"
125 return "#" * length
126
127
128def pair_method_descriptor(
129 methods: Iterable[PairMethodDescriptor], method: PairMethod
130) -> PairMethodDescriptor | None:
131 """Return the descriptor for ``method``, or None when the device does not offer it."""
132 return next((d for d in methods if d.method is method), None)
133
134
135def effective_unpaired_access(
136 info: ClientHelloPayload | None, config: ManagementResultData | None
137) -> bool:
138 """
139 Whether the device currently offers unpaired access.
140
141 A pairing config fetched over a management session on the current connection is
142 authoritative; the hello advertisement cannot reflect config changes until reconnect.
143 """
144 if config is not None and config.unpaired_access is not None:
145 return config.unpaired_access.enabled
146 return info is not None and info.unpaired_access.enabled
147
148
149def bridge_client_id_from_mac(mac: str) -> str:
150 """Generate a Sendspin bridge client ID from a MAC address."""
151 return f"{BRIDGE_PREFIX}{mac.replace(':', '').lower()}"
152
153
154def bridge_client_id_from_uuid(uuid: str) -> str:
155 """Generate a Sendspin bridge client ID from a UUID."""
156 return f"{BRIDGE_PREFIX}{uuid.replace('-', '').lower()}"
157
158
159def mac_from_bridge_client_id(client_id: str) -> str | None:
160 """Extract a MAC address from a Sendspin bridge client ID."""
161 if not client_id.startswith(BRIDGE_PREFIX):
162 return None
163 mac_part = client_id[len(BRIDGE_PREFIX) :]
164 if len(mac_part) != 12:
165 return None
166 if not all(ch in "0123456789abcdefABCDEF" for ch in mac_part):
167 return None
168 # Reconstruct MAC address with colons
169 return ":".join(mac_part[i : i + 2] for i in range(0, 12, 2))
170