"""BLE transport for the Shining Mask, built on bleak (BlueZ on Linux). This layer knows nothing about the command semantics -- it scans, connects, writes raw bytes to a characteristic, and exposes decrypted notification tokens through an asyncio queue. The protocol logic lives in :mod:`shiningmask.mask`. ``bleak`` is imported lazily so the rest of the package (and its tests) work without it installed. """ from __future__ import annotations import asyncio import logging from typing import Optional import constants as C import protocol from exceptions import MaskNotConnected, MaskNotFound, NotificationTimeout log = logging.getLogger("shiningmask.transport") class MaskTransport: """Manage the BLE connection and raw characteristic I/O for one mask.""" def __init__( self, address: Optional[str] = None, *, name_prefix: str = C.DEVICE_NAME_PREFIX, adapter: Optional[str] = None, write_with_response: Optional[bool] = None, upload_with_response: bool = False, ): self.address = address self.name_prefix = name_prefix self.adapter = adapter # None -> let bleak pick write-with/without-response from the char's # advertised properties. True/False force a mode. self.write_with_response = write_with_response # Image-upload chunks use write-WITHOUT-response (fire-and-forget): the # mask's upload characteristic does NOT reply to acknowledged writes, so # write-with-response HANGS waiting for an ACK that never comes. The # corruption risk at a weak signal is mitigated by keeping the mask close # for the one-time upload + the resilient per-frame reconnect. self.upload_with_response = upload_with_response self._client = None # bleak.BleakClient self._notify_queue: "asyncio.Queue[tuple[str, bytes]]" = asyncio.Queue() def _adapter_kwargs(self) -> dict: return {"adapter": self.adapter} if self.adapter else {} # -- connection lifecycle ------------------------------------------------- async def scan(self, timeout: float = 15.0): """Return a bleak device for the first mask seen, or raise MaskNotFound.""" from bleak import BleakScanner prefix = self.name_prefix.upper() service = C.SERVICE_UUID.lower() target = self.address.upper() if self.address else None def _match(device, adv) -> bool: # When a specific address is requested, match only that (an active # filter scan is more reliable than find_device_by_address for weak # / infrequently-advertising devices). if target is not None: return str(device.address or "").upper() == target name = str(adv.local_name or device.name or "") if name.upper().startswith(prefix): return True # Fall back to the advertised vendor service UUID (more reliable than # the name -- some clones advertise an empty/garbled local name). svcs = [s.lower() for s in (adv.service_uuids or [])] return service in svcs or any("fff0" in s for s in svcs) device = await BleakScanner.find_device_by_filter( _match, timeout=timeout, **self._adapter_kwargs() ) if device is None: who = f"address {self.address}" if target else f"name starting with {prefix!r}" raise MaskNotFound(f"no mask matching {who} found in {timeout:.0f}s " "(power it on, bring it closer, and free it from any phone)") return device async def connect(self, timeout: float = 15.0, attempts: int = 3): """Scan, connect, and subscribe to notifications. Retries up to ``attempts`` times -- on a marginal link the scan or GATT connect can intermittently time out. """ from bleak import BleakClient from bleak.exc import BleakError last_exc: Optional[BaseException] = None for i in range(attempts): try: device = await self.scan(timeout=timeout) log.info("connecting to mask %s (attempt %d/%d)", getattr(device, "address", device), i + 1, attempts) self._client = BleakClient(device, **self._adapter_kwargs()) await self._client.connect() # Drain any stale notifications from a previous session. while not self._notify_queue.empty(): self._notify_queue.get_nowait() await self._client.start_notify(C.NOTIFY_CHAR_UUID, self._on_notify) log.info("connected and subscribed to notifications") return except (MaskNotFound, asyncio.TimeoutError, BleakError, OSError) as exc: last_exc = exc log.warning("connect attempt %d/%d failed: %r", i + 1, attempts, exc) try: if self._client is not None: await self._client.disconnect() except Exception: pass self._client = None raise last_exc if last_exc else MaskNotFound("could not connect") async def disconnect(self): if self._client is not None: try: await self._client.disconnect() finally: self._client = None @property def is_connected(self) -> bool: return self._client is not None and self._client.is_connected def _require(self): if not self.is_connected: raise MaskNotConnected("mask is not connected") return self._client # -- notifications -------------------------------------------------------- def _on_notify(self, _char, data: bytearray): token = protocol.parse_notification(bytes(data)) log.debug("notify token=%r raw=%s", token, bytes(data).hex()) self._notify_queue.put_nowait((token, bytes(data))) async def wait_for(self, *expected_prefixes: str, timeout: float = 5.0) -> str: """Wait until a notification whose token starts with one of the prefixes. Non-matching notifications are discarded. Raises NotificationTimeout. """ loop = asyncio.get_running_loop() deadline = loop.time() + timeout while True: remaining = deadline - loop.time() if remaining <= 0: raise NotificationTimeout( f"timed out waiting for {expected_prefixes!r}" ) try: token, _ = await asyncio.wait_for(self._notify_queue.get(), timeout=remaining) except asyncio.TimeoutError: raise NotificationTimeout(f"timed out waiting for {expected_prefixes!r}") if any(protocol.matches(token, p) for p in expected_prefixes): return token log.debug("ignoring notification %r while waiting for %r", token, expected_prefixes) # -- raw writes ----------------------------------------------------------- async def _write(self, uuid: str, data: bytes, *, retries: int = 3, response=None): """write_gatt_char with retries — the marginal BLE link glitches single writes (GATT 'Unlikely Error'); over an 800-write upload that's fatal unless we retry the failed chunk. ``response`` overrides the write mode (None -> the transport default).""" from bleak.exc import BleakError client = self._require() resp = self.write_with_response if response is None else response last = None for i in range(retries): try: await client.write_gatt_char(uuid, bytes(data), response=resp) return except (BleakError, EOFError, OSError) as exc: last = exc await asyncio.sleep(0.05 * (i + 1)) raise last async def write_command(self, block: bytes): """Write an AES-encrypted command block to the command characteristic.""" await self._write(C.CMD_CHAR_UUID, block) async def write_upload(self, packet: bytes): """Write a raw data packet to the upload characteristic (write-without- response by default — the upload char is fire-and-forget and hangs on acknowledged writes). Keep the mask close for a clean one-time upload.""" await self._write(C.UPLOAD_CHAR_UUID, packet, response=self.upload_with_response) async def write_audio(self, block: bytes): """Write an AES-encrypted frame to the audio-visualization characteristic.""" client = self._require() await client.write_gatt_char( C.AUDIO_CHAR_UUID, bytes(block), response=self.write_with_response )