"""Device operations with readback and capability checks.""" from .i18n import tr from dataclasses import asdict, dataclass import time from .protocol import CODECS, MODES, STATES, TRANSPORTS, Command as C, validate_text from .transport import DeviceError @dataclass class Status: state: int mode: int transport: int connected_transport: int codecs: int codec: int resolution: int frequency: int le_state: int transports: int broadcast_public: int broadcast_quality: int broadcast_encrypted: int broadcast_name: str firmware: str gaming: int = -1 @property def codec_name(self): return ' + '.join(name for bit, name in CODECS.items() if self.codec & bit) or '—' @property def quality(self): bits = {1: tr('16 Bit'), 2: tr('24 Bit')}.get(self.resolution, '—') frequency = {1: tr('44,1 kHz'), 2: '48 kHz', 3: '96 kHz'}.get(self.frequency, '—') return f'{bits} / {frequency}' @property def gaming_allowed(self): if self.gaming >= 0: return self.gaming == 1 return self.mode == 1 or (self.state >= 2 and (self.connected_transport in (2, 3) or bool(self.codecs & 4))) def public_dict(self): result = asdict(self) result.update(state_name=tr(STATES.get(self.state, tr('Unbekannt ({value})').format(value=self.state))), mode_name=tr(MODES.get(self.mode, tr('Unbekannt ({value})').format(value=self.mode))), codec_name=self.codec_name, audio_quality=self.quality, transport_name=tr(TRANSPORTS.get(self.transport, tr('Unbekannt ({value})').format(value=self.transport)))) return result class Controller: def __init__(self, transport): self.transport = transport self.firmware = None def snapshot(self) -> Status: q = self.transport.request state = q(C.GET_STATE)[0] mode = q(C.GET_MODE) quality = q(C.GET_QUALITY) codecs = q(C.GET_CODECS)[0] codec = q(C.GET_CODEC)[0] le = q(C.GET_LE_STATE)[0] transports = q(C.GET_TRANSPORTS)[0] if state >= 2 else 0 broadcast = q(C.GET_BROADCAST) name = q(C.GET_NAME).split(b'\0', 1)[0].decode('utf-8', errors='replace') if self.firmware is None: try: self.firmware = '.'.join(map(str, q(C.GET_VERSION)[:3])) except TimeoutError: self.firmware = tr('Unbekannt') # Firmware 3.11 does not respond to GET_GAMING; use events/fallback, # matching the original application's capability logic. event = self.transport.events.get(23, b'') return Status(state, mode[0], mode[1], mode[2] if len(mode) > 2 else mode[1], codecs, codec, *quality[:2], le, transports, *broadcast[:3], name, self.firmware, event[0] if event else -1) def _verify(self, getter, expected, *, timeout=5.0, exact=True, null_padded=False): deadline = time.monotonic() + timeout while True: result = self.transport.request(getter) comparison = result.rstrip(b'\0') if null_padded else result matches = comparison == expected if exact else comparison.startswith(expected) if matches: return if time.monotonic() >= deadline: raise DeviceError(tr('Der Dongle hat die angeforderte Einstellung nicht bestätigt. Bitte den aktuellen Status prüfen.')) time.sleep(0.15) def set_mode(self, mode: int, transport: int | None = None): status = self.snapshot() if mode not in MODES: raise ValueError(tr('Unbekannter Audiomodus.')) if mode == 1 and not status.gaming_allowed: raise ValueError(tr('Gaming ist für die aktuelle Verbindung nicht verfügbar.')) selected = status.transport if transport is None else transport if selected not in (1, 2, 3): raise ValueError(tr('Unbekannter Bluetooth-Transport.')) if transport is not None and selected in (1, 2) and status.transports and not status.transports & selected: raise ValueError(tr('Die Kopfhörer unterstützen diesen Transport nicht.')) payload = bytes((mode, selected)) self.transport.request(C.SET_MODE, payload) self._verify(C.GET_MODE, payload, exact=False) def set_codec(self, codec: int): status = self.snapshot() if status.mode != 0 or status.state < 2: raise ValueError(tr('Codec-Auswahl benötigt verbundene Kopfhörer im Standard-Modus.')) if codec not in CODECS or not status.codecs & codec: raise ValueError(tr('Dieser Codec wird für die aktuelle Verbindung nicht angeboten.')) payload = bytes((codec,)) self.transport.request(C.SET_CODEC, payload) self._verify(C.GET_CODEC, payload) def set_connection(self, connected: bool): self.transport.request(C.SET_CONNECTION, bytes((int(connected),))) deadline = time.monotonic() + 8 while True: state = self.transport.request(C.GET_STATE)[0] if (state >= 2) == connected: return if time.monotonic() >= deadline: raise DeviceError(tr('Verbindung wurde angefordert, aber noch nicht bestätigt. Sind die gekoppelten Kopfhörer eingeschaltet?')) time.sleep(0.25) def set_broadcast(self, *, name=None, password=None, public=None, quality=None, encrypted=None): # Validate the entire change before issuing any setter. name_data = validate_text(name) if name is not None else None key_data = validate_text(password, key=True) if password is not None else None if any(value is not None and not isinstance(value, bool) for value in (public, encrypted)): raise ValueError(tr('Auracast-Schalter benötigen einen booleschen Wert.')) if quality is not None and quality not in (0, 1, 2): raise ValueError(tr('Unbekannte Auracast-Qualität.')) before = self.snapshot() info = [before.broadcast_public, before.broadcast_quality, before.broadcast_encrypted] if public is not None: info[0] = int(public) if quality is not None: info[1] = quality if password is not None: info[2] = int(bool(password)) if encrypted is not None: info[2] = int(encrypted) if info[2] and key_data == b'': raise ValueError(tr('Verschlüsselung benötigt ein Passwort mit 4–16 Zeichen.')) if info[2] and key_data is None: existing = self.transport.request(C.GET_KEY).rstrip(b'\0') if len(existing) < 4: raise ValueError(tr('Bitte zuerst ein Passwort mit 4–16 Zeichen eingeben.')) # Preserve the original settings while changing a key, then write # key, name, and final settings in the order used by the original app. try: if key_data is not None and before.broadcast_encrypted: self.transport.request(C.SET_BROADCAST, bytes((before.broadcast_public, before.broadcast_quality, before.broadcast_encrypted))) if key_data is not None: self.transport.request(C.SET_KEY, key_data) self._verify(C.GET_KEY, key_data, null_padded=True) if name_data is not None: self.transport.request(C.SET_NAME, name_data) if name_data: self._verify(C.GET_NAME, name_data, null_padded=True) else: self.transport.request(C.GET_NAME) self.transport.request(C.SET_BROADCAST, bytes(info)) self._verify(C.GET_BROADCAST, bytes(info)) except Exception as exc: # Do not silently repeat writes or claim atomicity: settings persist # independently on the dongle. raise DeviceError(tr('Auracast wurde nur teilweise oder nicht übernommen. Aktuellen Namen und Passwortschutz prüfen.')) from exc def factory_reset(self, *, confirmed=False): if not confirmed: raise ValueError(tr('Zurücksetzen muss ausdrücklich bestätigt werden; alle Kopplungen werden gelöscht.')) self.transport.request(C.FACTORY_RESET)