/
/
/
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
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 aiosendspin.models.core import ClientHelloPayload
19 from aiosendspin.models.management import ManagementResultData
20
21
22class SecurityActionError(Exception):
23 """A pairing/management action failure carrying a strings.json alert slug for the UI."""
24
25 def __init__(self, alert_key: str, *, detail: str | None = None) -> None:
26 """Initialize with the alert slug and optional untranslated {0}-placeholder detail."""
27 super().__init__(alert_key if detail is None else f"{alert_key}: {detail}")
28 self.alert_key = alert_key
29 self.detail = detail
30
31
32@dataclass(frozen=True)
33class AlertText:
34 """A strings.json slug and optional {0}-placeholder params for an operator ALERT entry."""
35
36 key: str
37 params: list[str] | None = None
38
39
40_PAIR_ABORT_KEYS = {
41 PairAbortReason.ATTEMPT_TIMEOUT: "pairing_error_timeout",
42 PairAbortReason.CONCURRENT_ATTEMPT: "pairing_error_concurrent",
43 PairAbortReason.LOCKED_OUT: "pairing_error_locked_out",
44 PairAbortReason.METHOD_NOT_SUPPORTED: "pairing_error_method_unsupported",
45 PairAbortReason.PIN_LENGTH_UNACCEPTABLE: "pairing_error_pin_length",
46 PairAbortReason.PIN_MISMATCH: "pairing_error_pin_mismatch",
47 PairAbortReason.USER_CANCELLED: "pairing_error_cancelled",
48}
49
50
51def error_alert(err: Exception) -> AlertText:
52 """Map a pairing or management failure to a localized operator alert."""
53 if isinstance(err, SecurityActionError):
54 return AlertText(err.alert_key, [err.detail] if err.detail is not None else None)
55 if isinstance(err, PairingAbortError):
56 key = _PAIR_ABORT_KEYS.get(err.reason)
57 if key is not None:
58 return AlertText(key)
59 return AlertText("pairing_error_aborted", [err.reason.value])
60 if isinstance(err, TimeoutError):
61 return AlertText("pairing_error_timeout")
62 if isinstance(err, OSError):
63 return AlertText("pairing_error_storage", [str(err)])
64 if isinstance(err, HandshakeAbortedError):
65 return AlertText("pairing_error_handshake")
66 if isinstance(err, PairingError):
67 return AlertText("pairing_error_failed", [str(err)])
68 return AlertText("pairing_error_generic")
69
70
71def alert_entry(text: AlertText) -> ConfigEntry:
72 """Build an ALERT config entry from a localized alert descriptor."""
73 return ConfigEntry(key=text.key, type=ConfigEntryType.ALERT, translation_params=text.params)
74
75
76def action_entry(action: str, *, advanced: bool = False) -> ConfigEntry:
77 """Build an ACTION config entry whose key mirrors its action."""
78 return ConfigEntry(key=action, type=ConfigEntryType.ACTION, action=action, advanced=advanced)
79
80
81def effective_pair_methods(
82 info: ClientHelloPayload | None, config: ManagementResultData | None
83) -> list[PairMethodDescriptor]:
84 """
85 Return the pairing methods the device currently offers.
86
87 A pairing config fetched over a management session on the current connection is
88 authoritative; the hello advertisement cannot reflect config changes until reconnect.
89 Methods the config enables beyond the hello get a synthesized descriptor.
90 """
91 hello_methods = list(info.supported_pair_methods or []) if info is not None else []
92 if config is None:
93 return hello_methods
94 advertised = {descriptor.method: descriptor for descriptor in hello_methods}
95 methods: list[PairMethodDescriptor] = []
96 for method, method_config in (
97 (PairMethod.PAIRING_PSK, config.pairing_psk),
98 (PairMethod.STATIC_PIN, config.static_pin),
99 (PairMethod.DYNAMIC_PIN, config.dynamic_pin),
100 ):
101 if method_config is None or not method_config.enabled:
102 continue
103 descriptor = advertised.get(method) or PairMethodDescriptor(method=method)
104 methods.append(
105 replace(
106 descriptor,
107 locked_out=method_config.locked_out,
108 min_pin_length=method_config.min_pin_length,
109 )
110 )
111 return methods
112
113
114def effective_unpaired_access(
115 info: ClientHelloPayload | None, config: ManagementResultData | None
116) -> bool:
117 """
118 Whether the device currently offers unpaired access.
119
120 A pairing config fetched over a management session on the current connection is
121 authoritative; the hello advertisement cannot reflect config changes until reconnect.
122 """
123 if config is not None and config.unpaired_access is not None:
124 return config.unpaired_access.enabled
125 return info is not None and info.unpaired_access.enabled
126
127
128def bridge_client_id_from_mac(mac: str) -> str:
129 """Generate a Sendspin bridge client ID from a MAC address."""
130 return f"{BRIDGE_PREFIX}{mac.replace(':', '').lower()}"
131
132
133def bridge_client_id_from_uuid(uuid: str) -> str:
134 """Generate a Sendspin bridge client ID from a UUID."""
135 return f"{BRIDGE_PREFIX}{uuid.replace('-', '').lower()}"
136
137
138def mac_from_bridge_client_id(client_id: str) -> str | None:
139 """Extract a MAC address from a Sendspin bridge client ID."""
140 if not client_id.startswith(BRIDGE_PREFIX):
141 return None
142 mac_part = client_id[len(BRIDGE_PREFIX) :]
143 if len(mac_part) != 12:
144 return None
145 if not all(ch in "0123456789abcdefABCDEF" for ch in mac_part):
146 return None
147 # Reconstruct MAC address with colons
148 return ":".join(mac_part[i : i + 2] for i in range(0, 12, 2))
149