332 lines
14 KiB
Python
332 lines
14 KiB
Python
"""High-level async controller for the Shining Mask.
|
|
|
|
Example::
|
|
|
|
import asyncio
|
|
from mask import ShiningMask
|
|
|
|
async def main():
|
|
async with ShiningMask() as mask:
|
|
await mask.set_brightness(80)
|
|
await mask.show_image(3)
|
|
await mask.set_text("HELLO", color=(0, 255, 0))
|
|
|
|
asyncio.run(main())
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Optional, Sequence, Tuple
|
|
|
|
import bitmap, protocol
|
|
from constants import TextMode, UploadKind
|
|
from exceptions import UploadError
|
|
from transport import MaskTransport
|
|
|
|
log = logging.getLogger("shiningmask.mask")
|
|
|
|
Color = Tuple[int, int, int]
|
|
|
|
|
|
class ShiningMask:
|
|
"""Async controller for one Shining Mask over BLE."""
|
|
|
|
def __init__(
|
|
self,
|
|
address: Optional[str] = None,
|
|
*,
|
|
name_prefix: str = "MASK",
|
|
adapter: Optional[str] = None,
|
|
write_with_response: Optional[bool] = None,
|
|
transport: Optional[MaskTransport] = None,
|
|
):
|
|
self.transport = transport or MaskTransport(
|
|
address,
|
|
name_prefix=name_prefix,
|
|
adapter=adapter,
|
|
write_with_response=write_with_response,
|
|
)
|
|
self._upload_lock = asyncio.Lock()
|
|
|
|
# -- lifecycle ------------------------------------------------------------
|
|
|
|
async def connect(self, timeout: float = 15.0, attempts: int = 3):
|
|
await self.transport.connect(timeout=timeout, attempts=attempts)
|
|
return self
|
|
|
|
async def disconnect(self):
|
|
await self.transport.disconnect()
|
|
|
|
async def __aenter__(self):
|
|
await self.connect()
|
|
return self
|
|
|
|
async def __aexit__(self, *exc):
|
|
await self.disconnect()
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
return self.transport.is_connected
|
|
|
|
# -- simple control commands ---------------------------------------------
|
|
|
|
async def set_brightness(self, level: int):
|
|
await self.transport.write_command(protocol.cmd_brightness(level))
|
|
|
|
async def show_image(self, image_id: int):
|
|
"""Show a built-in image (IMAG)."""
|
|
await self.transport.write_command(protocol.cmd_image(image_id))
|
|
|
|
async def play_animation(self, anim_id: int):
|
|
"""Play a built-in animation (ANIM)."""
|
|
await self.transport.write_command(protocol.cmd_animation(anim_id))
|
|
|
|
async def play_diy(self, image_id: int):
|
|
"""Show one uploaded DIY image (PLAY)."""
|
|
await self.transport.write_command(protocol.cmd_play_diy(image_id))
|
|
|
|
async def play_diy_sequence(self, image_ids: Sequence[int]):
|
|
await self.transport.write_command(protocol.cmd_play_diy_sequence(list(image_ids)))
|
|
|
|
async def delete_diy(self, image_ids: Sequence[int]):
|
|
await self.transport.write_command(protocol.cmd_delete_diy(list(image_ids)))
|
|
|
|
async def get_diy_count(self, timeout: float = 5.0) -> Optional[int]:
|
|
"""Ask how many DIY images are stored (CHEC). Returns None if no answer."""
|
|
await self.transport.write_command(protocol.cmd_check_diy_count())
|
|
try:
|
|
_, raw = await self._collect_notification("CHEC", timeout=timeout)
|
|
except Exception:
|
|
return None
|
|
import crypto
|
|
buf = crypto.decrypt(raw)
|
|
# Response frame is [len]['CHEC'][count]; the count sits right after the
|
|
# 4-byte command name (NOT after the total length byte buf[0]).
|
|
idx = 1 + len("CHEC")
|
|
return buf[idx] if len(buf) > idx else None
|
|
|
|
# -- text styling ---------------------------------------------------------
|
|
|
|
async def set_text_mode(self, mode: int):
|
|
await self.transport.write_command(protocol.cmd_text_mode(mode))
|
|
|
|
async def set_text_speed(self, speed: int):
|
|
await self.transport.write_command(protocol.cmd_text_speed(speed))
|
|
|
|
async def set_text_color_effect(self, mode: int, enable: bool = True):
|
|
await self.transport.write_command(protocol.cmd_text_color_effect(mode, enable))
|
|
|
|
async def set_foreground_color(self, r: int, g: int, b: int, enable: bool = True):
|
|
await self.transport.write_command(protocol.cmd_text_fg_color(r, g, b, enable))
|
|
|
|
async def set_background_color(self, r: int, g: int, b: int, enable: bool = True):
|
|
await self.transport.write_command(protocol.cmd_text_bg_color(r, g, b, enable))
|
|
|
|
# -- text / bitmap upload -------------------------------------------------
|
|
|
|
async def set_text(
|
|
self,
|
|
text: str,
|
|
*,
|
|
color: Color = (255, 255, 255),
|
|
mode: Optional[int] = TextMode.SCROLL_LEFT,
|
|
speed: Optional[int] = None,
|
|
clear_effects: bool = True,
|
|
timeout: float = 10.0,
|
|
**render_kwargs,
|
|
):
|
|
"""Render ``text`` to a bitmap and upload + display it.
|
|
|
|
``color`` sets the text color. ``mode`` is the scroll style (default
|
|
scroll-left; pass ``TextMode.STEADY`` to hold it still). With
|
|
``clear_effects`` we disable the ``M`` special effect / animated
|
|
background so the mask stops cycling to other animations, and the
|
|
scroll ``mode``/``speed`` are pinned *after* the upload so they stick.
|
|
"""
|
|
bmp, color_array = bitmap.build_text_upload(text, color=color, **render_kwargs)
|
|
await self.upload_bitmap(bmp, color_array, timeout=timeout)
|
|
await self._apply_display_state(color, mode, speed, clear_effects)
|
|
|
|
async def _apply_display_state(self, color, mode, speed, clear_effects):
|
|
"""Pin a clean, steady display state after a bitmap upload."""
|
|
if clear_effects:
|
|
# Turn off the M special-effect/background-image carousel and force a
|
|
# black background so nothing animates behind the bitmap.
|
|
await self.set_text_color_effect(0, enable=False)
|
|
await self.set_background_color(0, 0, 0, enable=True)
|
|
# Color the foreground the way the app does (FC), in addition to the
|
|
# per-column color array already in the upload.
|
|
await self.set_foreground_color(*color, enable=True)
|
|
if mode is not None:
|
|
await self.set_text_mode(mode)
|
|
if speed is not None:
|
|
await self.set_text_speed(speed)
|
|
|
|
async def upload_bitmap(self, bitmap_bytes: bytes, color_array: bytes, *, timeout: float = 10.0):
|
|
"""Upload a raw mask bitmap + color array using the DATS/REOK/DATCP flow."""
|
|
payload = bytes(bitmap_bytes) + bytes(color_array)
|
|
await self._upload(
|
|
payload,
|
|
second_field=len(bitmap_bytes),
|
|
kind=UploadKind.BITMAP,
|
|
timeout=timeout,
|
|
)
|
|
|
|
async def clear_diy(self, timeout: float = 4.0) -> int:
|
|
"""Delete all stored DIY images. Returns how many were removed."""
|
|
n = await self.get_diy_count(timeout=timeout) or 0
|
|
ids = list(range(1, n + 1))
|
|
for i in range(0, len(ids), 10): # DELE takes at most 10 ids
|
|
await self.delete_diy(ids[i:i + 10])
|
|
await asyncio.sleep(0.25)
|
|
return n
|
|
|
|
async def upload_frame(self, data: bytes, slot: int, **kwargs):
|
|
"""Upload one DIY image frame to ``slot`` (see :meth:`upload_raw_image`)."""
|
|
await self.upload_raw_image(bytes(data), index=slot, **kwargs)
|
|
|
|
async def play_frame(self, slot: int):
|
|
"""Show a stored DIY image instantly (alias of :meth:`play_diy`)."""
|
|
await self.play_diy(slot)
|
|
|
|
async def upload_raw_image(self, data: bytes, *, index: int = 1,
|
|
timestamp: int = 0x07C5F9FF, chunk_delay: float = 0.05,
|
|
init_delay: float = 0.15):
|
|
"""Upload a raw DIY image/animation to the mask (fire-and-forget).
|
|
|
|
This is the method that actually works for full-color custom content
|
|
(verified against BishopFox/shining-mask): an image ``DATS`` (toggle
|
|
``0x01``) is sent, then every data chunk is streamed WITHOUT waiting for
|
|
``DATSOK``/``REOK``, then ``DATCP`` with a 4-byte timestamp. The content
|
|
displays immediately and is stored as DIY image(s) that :meth:`play_diy`
|
|
can replay smoothly (no upload logo).
|
|
|
|
``data`` must already be in the mask's internal image format (a
|
|
multi-frame blob is played as an animation). Generating that format for
|
|
arbitrary images is not yet solved -- see the project notes.
|
|
"""
|
|
if self._upload_lock.locked():
|
|
raise UploadError("an upload is already in progress")
|
|
async with self._upload_lock:
|
|
dats_args = len(data).to_bytes(2, "big") + int(index).to_bytes(2, "big") + b"\x01"
|
|
await self.transport.write_command(protocol.encode_command("DATS", dats_args))
|
|
await asyncio.sleep(init_delay)
|
|
for packet in protocol.iter_upload_packets(data): # unpadded, like the app
|
|
await self.transport.write_upload(packet)
|
|
await asyncio.sleep(chunk_delay)
|
|
await self.transport.write_command(
|
|
protocol.encode_command("DATCP", int(timestamp).to_bytes(4, "big"))
|
|
)
|
|
|
|
async def upload_image(self, rgb_bytes: bytes, image_index: int, *, timeout: float = 15.0):
|
|
"""Upload a full-color DIY image (raw RGB) to a storage slot.
|
|
|
|
Experimental: mask-go does not implement image upload, so this follows
|
|
ble-protocol.md only. ``rgb_bytes`` is 3 bytes per pixel; the mask is 16
|
|
rows tall, so length should be ``16 * width * 3``.
|
|
"""
|
|
await self._upload(
|
|
bytes(rgb_bytes),
|
|
second_field=image_index,
|
|
kind=UploadKind.IMAGE,
|
|
timeout=timeout,
|
|
)
|
|
|
|
async def _upload(self, payload: bytes, *, second_field: int, kind: int, timeout: float,
|
|
wait_per_packet: bool = True):
|
|
total_len = len(payload)
|
|
if total_len == 0:
|
|
raise UploadError("nothing to upload (empty payload)")
|
|
if self._upload_lock.locked():
|
|
raise UploadError("an upload is already in progress")
|
|
async with self._upload_lock:
|
|
log.info("upload start: %d bytes (kind=%d)", total_len, kind)
|
|
|
|
await self.transport.write_command(
|
|
protocol.cmd_dats(total_len, second_field, kind)
|
|
)
|
|
await self.transport.wait_for(protocol.RESP_DATS_OK, timeout=timeout)
|
|
|
|
# Image uploads use fixed 100-byte packets (ble-protocol.md); bitmap
|
|
# uploads are unpadded (matching mask-go).
|
|
pad_to = 100 if kind == UploadKind.IMAGE else None
|
|
for packet in protocol.iter_upload_packets(payload, pad_to=pad_to):
|
|
await self.transport.write_upload(packet)
|
|
# wait_per_packet=False streams packets without the per-packet
|
|
# REOK round-trip (much faster, used for animation). The REOKs
|
|
# still arrive and are drained by the DATCPOK wait below.
|
|
if wait_per_packet:
|
|
await self.transport.wait_for(protocol.RESP_REC_OK, timeout=timeout)
|
|
|
|
await self.transport.write_command(protocol.cmd_datcp())
|
|
await self.transport.wait_for(protocol.RESP_DATCP_OK, timeout=timeout)
|
|
log.info("upload complete")
|
|
|
|
# -- faces -----------------------------------------------------------------
|
|
|
|
async def setup_face_mode(self, brightness: int = 80, mono_color: Optional[Color] = None):
|
|
"""Put the mask in a clean state for drawing faces/images.
|
|
|
|
Sets a black background, disables the M effect, and selects steady
|
|
(non-scrolling) display. If ``mono_color`` is given, the whole face is
|
|
colored uniformly via FC (so frames can be uploaded as bitmap-only --
|
|
half the bytes, much faster/more reliable); otherwise FC is disabled so
|
|
per-column colors drive the pixels.
|
|
"""
|
|
await self.set_brightness(brightness)
|
|
await self.set_text_color_effect(0, enable=False) # M effect off
|
|
await self.set_background_color(0, 0, 0, enable=True) # black background
|
|
if mono_color is not None:
|
|
await self.set_foreground_color(*mono_color, enable=True)
|
|
else:
|
|
await self.set_foreground_color(255, 255, 255, enable=False)
|
|
await self.set_text_mode(TextMode.STEADY)
|
|
|
|
async def show_face(self, columns, color: Color = (0, 220, 255), *,
|
|
per_column_colors=None, mono: bool = False,
|
|
fast: bool = True, timeout: float = 8.0):
|
|
"""Display a face/bitmap given as columns (each 16 ints, 0/1).
|
|
|
|
``mono`` uploads just the bitmap (no color array) -- use it after
|
|
``setup_face_mode(mono_color=...)`` for the fastest, most reliable frames
|
|
(half the packets). Otherwise ``color`` (single) or ``per_column_colors``
|
|
(multi) build a color array. With ``fast`` the frame is streamed without
|
|
per-packet acks (for animation).
|
|
"""
|
|
bmp = bitmap.encode_bitmap(columns)
|
|
if mono:
|
|
payload = bytes(bmp) # color comes from FC
|
|
elif per_column_colors is not None:
|
|
payload = bytes(bmp) + bitmap.encode_color_array_per_column(per_column_colors)
|
|
else:
|
|
payload = bytes(bmp) + bitmap.encode_color_array(len(columns), color)
|
|
await self._upload(
|
|
payload,
|
|
second_field=len(bmp),
|
|
kind=UploadKind.BITMAP,
|
|
timeout=timeout,
|
|
wait_per_packet=not fast,
|
|
)
|
|
|
|
# -- audio visualization (experimental) -----------------------------------
|
|
|
|
async def send_audio_frame(self, mode: int, nibbles: Sequence[int]):
|
|
await self.transport.write_audio(protocol.cmd_audio_frame(mode, list(nibbles)))
|
|
|
|
# -- internals ------------------------------------------------------------
|
|
|
|
async def _collect_notification(self, *prefixes: str, timeout: float):
|
|
"""Wait for a matching notification and also return its raw bytes."""
|
|
loop = asyncio.get_running_loop()
|
|
deadline = loop.time() + timeout
|
|
q = self.transport._notify_queue # noqa: SLF001 - intra-package access
|
|
while True:
|
|
remaining = deadline - loop.time()
|
|
if remaining <= 0:
|
|
raise TimeoutError(prefixes)
|
|
token, raw = await asyncio.wait_for(q.get(), timeout=remaining)
|
|
if any(protocol.matches(token, p) for p in prefixes):
|
|
return token, raw
|