146 lines
4.8 KiB
Python
146 lines
4.8 KiB
Python
"""Animated talking face for the mask.
|
|
|
|
Runs a background loop that blinks periodically and moves the mouth while
|
|
"speaking" by swapping face frames. Frame rate is bounded by the BLE upload
|
|
(~2-4 fps), so this gives a clearly *talking* face rather than smooth lip-sync.
|
|
|
|
Two ways to drive it:
|
|
|
|
* ``set_speaking(True/False)`` around a TTS utterance -- the loop auto-animates
|
|
the mouth while speaking and closes it when silent.
|
|
* ``set_mouth_level(0..4)`` from live audio amplitude for rough lip-sync (this
|
|
switches off the auto mouth animation and uses your values directly).
|
|
|
|
Example::
|
|
|
|
async with ShiningMask() as mask:
|
|
face = await TalkingFace(mask).start()
|
|
face.set_speaking(True) # ... while Marcus talks ...
|
|
await asyncio.sleep(5)
|
|
face.set_speaking(False)
|
|
await face.stop()
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import random
|
|
import time
|
|
from typing import Optional
|
|
|
|
import faces
|
|
|
|
Color = "tuple[int, int, int]"
|
|
|
|
# mouth shapes cycled while speaking (auto mode): clear open/close so the motion
|
|
# reads as talking even at the low (~2 fps) re-upload frame rate.
|
|
_TALK_CYCLE = [4, 0, 3, 0, 4, 1, 2, 0]
|
|
|
|
|
|
class TalkingFace:
|
|
def __init__(self, mask, *, color=(0, 220, 255), width: int = faces.DISPLAY_WIDTH,
|
|
brightness: int = 80, tick: float = 0.05, fast: bool = False):
|
|
self.mask = mask
|
|
self.color = color
|
|
self.width = width
|
|
self.brightness = brightness
|
|
self.tick = tick
|
|
# Reliable (per-packet ack) uploads actually commit each frame; fast
|
|
# streamed uploads drop frames at weak signal, so default to reliable.
|
|
self._fast = fast
|
|
|
|
self._eyes = True
|
|
self._mouth = 0
|
|
self._speaking = False
|
|
self._auto = True # auto-animate mouth from _speaking
|
|
self._blink = True
|
|
self._last = None
|
|
self._task: Optional[asyncio.Task] = None
|
|
self._stop = False
|
|
|
|
# -- lifecycle ------------------------------------------------------------
|
|
|
|
async def start(self):
|
|
await self.mask.setup_face_mode(brightness=self.brightness)
|
|
await self._render(force=True)
|
|
self._stop = False
|
|
self._task = asyncio.create_task(self._loop())
|
|
return self
|
|
|
|
async def stop(self, *, neutral: bool = True):
|
|
self._stop = True
|
|
if self._task:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._task = None
|
|
if neutral:
|
|
self._eyes, self._mouth = True, 0
|
|
await self._render(force=True)
|
|
|
|
async def __aenter__(self):
|
|
return await self.start()
|
|
|
|
async def __aexit__(self, *exc):
|
|
await self.stop()
|
|
|
|
# -- control --------------------------------------------------------------
|
|
|
|
def set_speaking(self, speaking: bool):
|
|
"""Auto-animate the mouth while speaking (closes when False)."""
|
|
self._auto = True
|
|
self._speaking = bool(speaking)
|
|
|
|
def set_mouth_level(self, level: int):
|
|
"""Drive the mouth directly (0=closed..4=wide) e.g. from audio amplitude."""
|
|
self._auto = False
|
|
self._mouth = max(0, min(4, int(level)))
|
|
|
|
def set_blink(self, enabled: bool):
|
|
self._blink = bool(enabled)
|
|
|
|
def set_color(self, color):
|
|
self.color = color
|
|
self._last = None # force re-render on next tick
|
|
|
|
# -- internals ------------------------------------------------------------
|
|
|
|
async def _render(self, force: bool = False):
|
|
key = (self._eyes, self._mouth, self.color)
|
|
if not force and key == self._last:
|
|
return
|
|
cols = faces.face(mouth=self._mouth, eyes=self._eyes, width=self.width)
|
|
try:
|
|
await self.mask.show_face(cols, color=self.color, fast=self._fast)
|
|
self._last = key
|
|
except Exception:
|
|
# tolerate transient BLE hiccups mid-animation; retry next tick
|
|
self._last = None
|
|
|
|
async def _loop(self):
|
|
next_blink = time.monotonic() + random.uniform(2.0, 5.0)
|
|
cycle_i = 0
|
|
while not self._stop:
|
|
now = time.monotonic()
|
|
|
|
# periodic blink (quick close/open)
|
|
if self._blink and self._eyes and now >= next_blink:
|
|
self._eyes = False
|
|
await self._render()
|
|
await asyncio.sleep(0.12)
|
|
self._eyes = True
|
|
next_blink = now + random.uniform(2.5, 5.5)
|
|
|
|
# mouth
|
|
if self._auto:
|
|
if self._speaking:
|
|
self._mouth = _TALK_CYCLE[cycle_i % len(_TALK_CYCLE)]
|
|
cycle_i += 1
|
|
else:
|
|
self._mouth = 0
|
|
|
|
await self._render()
|
|
await asyncio.sleep(self.tick)
|