206 lines
8.0 KiB
Python
206 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Real-time "motion face": track your face on the webcam and mirror it on the mask.
|
|
|
|
Following the community approach, this does NOT stream raw pixels over BLE (too
|
|
slow). Instead it tracks your expression on the host with OpenCV and triggers the
|
|
pre-uploaded face frames with fast PLAY commands:
|
|
|
|
mouth open -> talk1..talk3 head turn -> look_left / look_right
|
|
smile -> smile eyes shut -> blink else -> neutral
|
|
|
|
Detection uses OpenCV Haar cascades (face / eyes / smile) -- no MediaPipe/dlib,
|
|
so nothing new is installed into g1_env. It's approximate (especially mouth
|
|
openness); for precise lip-sync, install mediapipe in a dedicated env and swap
|
|
the tracker (the mask side is unchanged).
|
|
|
|
Usage:
|
|
python facetrack.py # webcam -> mask, mirrored
|
|
python facetrack.py --show # also open a preview window
|
|
python facetrack.py --no-mask --show # tune detection without the mask
|
|
python facetrack.py --image face.jpg # test detection on one image (offline)
|
|
python facetrack.py --camera 1 --address AA:BB:CC:DD:EE:FF
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import time
|
|
|
|
import cv2
|
|
|
|
import colorface # noqa: F401 (kept so FaceAnimator's default frames import cleanly)
|
|
from faceanim import FaceAnimator
|
|
from mask import ShiningMask
|
|
from exceptions import MaskNotFound
|
|
|
|
_HC = cv2.data.haarcascades
|
|
|
|
|
|
class ExpressionTracker:
|
|
"""Maps a webcam frame to one of the mask's face-frame names."""
|
|
|
|
def __init__(self, *, mirror: bool = True,
|
|
mouth_levels=(0.16, 0.22, 0.28), look_thresh: float = 0.18):
|
|
self.mirror = mirror
|
|
self.mouth_levels = mouth_levels # smile-box height / face height -> 1/2/3
|
|
self.look_thresh = look_thresh
|
|
self._face = cv2.CascadeClassifier(_HC + "haarcascade_frontalface_default.xml")
|
|
self._eye = cv2.CascadeClassifier(_HC + "haarcascade_eye.xml")
|
|
self._smile = cv2.CascadeClassifier(_HC + "haarcascade_smile.xml")
|
|
# smoothing / debounce state
|
|
self._look = 0
|
|
self._blink_frames = 0
|
|
|
|
def detect(self, frame_bgr):
|
|
"""Return (frame_name, annotated_bgr). frame_name is a FaceAnimator key."""
|
|
if self.mirror:
|
|
frame_bgr = cv2.flip(frame_bgr, 1)
|
|
gray = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2GRAY)
|
|
faces = self._face.detectMultiScale(gray, 1.2, 5, minSize=(90, 90))
|
|
if len(faces) == 0:
|
|
return "neutral", frame_bgr
|
|
x, y, w, h = max(faces, key=lambda f: f[2] * f[3])
|
|
cv2.rectangle(frame_bgr, (x, y), (x + w, y + h), (0, 220, 255), 2)
|
|
|
|
# --- head turn -> look (with hysteresis) ---
|
|
fcx = x + w / 2.0
|
|
off = (fcx - frame_bgr.shape[1] / 2.0) / (frame_bgr.shape[1] / 2.0)
|
|
if off < -self.look_thresh:
|
|
self._look = -1
|
|
elif off > self.look_thresh:
|
|
self._look = 1
|
|
elif abs(off) < self.look_thresh * 0.5:
|
|
self._look = 0
|
|
|
|
# --- eyes -> blink (debounced) ---
|
|
eye_roi = gray[y:y + int(h * 0.55), x:x + w]
|
|
eyes = self._eye.detectMultiScale(eye_roi, 1.1, 6,
|
|
minSize=(int(w * 0.12), int(w * 0.12)))
|
|
self._blink_frames = self._blink_frames + 1 if len(eyes) == 0 else 0
|
|
blink = self._blink_frames in (1, 2) # only the first couple of eyeless frames
|
|
|
|
# --- mouth openness via the smile cascade on the lower face ---
|
|
my0, my1 = y + int(h * 0.55), y + h
|
|
mx0, mx1 = x + int(w * 0.15), x + int(w * 0.85)
|
|
mouth_roi = gray[my0:my1, mx0:mx1]
|
|
mouth_level, smiling, box = self._mouth(mouth_roi, h)
|
|
if box is not None:
|
|
bx, by, bw, bh = box
|
|
cv2.rectangle(frame_bgr, (mx0 + bx, my0 + by),
|
|
(mx0 + bx + bw, my0 + by + bh), (0, 255, 0), 1)
|
|
|
|
# --- map to a frame name (priority order) ---
|
|
if blink:
|
|
name = "blink"
|
|
elif mouth_level >= 1:
|
|
name = f"talk{mouth_level}"
|
|
elif smiling:
|
|
name = "smile"
|
|
elif self._look < 0:
|
|
name = "look_left"
|
|
elif self._look > 0:
|
|
name = "look_right"
|
|
else:
|
|
name = "neutral"
|
|
|
|
cv2.putText(frame_bgr, name, (x, max(0, y - 8)),
|
|
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
|
|
return name, frame_bgr
|
|
|
|
def _mouth(self, mouth_roi, face_h):
|
|
if mouth_roi.size == 0:
|
|
return 0, False, None
|
|
smiles = self._smile.detectMultiScale(mouth_roi, 1.7, 18,
|
|
minSize=(int(face_h * 0.18), int(face_h * 0.08)))
|
|
if len(smiles) == 0:
|
|
return 0, False, None
|
|
box = max(smiles, key=lambda s: s[2] * s[3])
|
|
ratio = box[3] / float(face_h) # mouth-box height / face height
|
|
lo, mid, hi = self.mouth_levels
|
|
level = 3 if ratio >= hi else 2 if ratio >= mid else 1 if ratio >= lo else 0
|
|
return level, True, box
|
|
|
|
|
|
async def run(args):
|
|
tracker = ExpressionTracker(mirror=not args.no_mirror)
|
|
|
|
# offline single-image test path
|
|
if args.image:
|
|
img = cv2.imread(args.image)
|
|
if img is None:
|
|
print(f"could not read {args.image}")
|
|
return
|
|
name, annotated = tracker.detect(img)
|
|
print("detected expression:", name)
|
|
if args.show:
|
|
cv2.imshow("facetrack", annotated); cv2.waitKey(0); cv2.destroyAllWindows()
|
|
return
|
|
|
|
cap = cv2.VideoCapture(args.camera)
|
|
if not cap.isOpened():
|
|
print(f"could not open camera {args.camera}")
|
|
return
|
|
|
|
face = None
|
|
mask = None
|
|
if not args.no_mask:
|
|
mask = ShiningMask(address=args.address, name_prefix=args.name_prefix)
|
|
print("connecting to mask ...")
|
|
try:
|
|
await mask.connect(timeout=20.0, attempts=8)
|
|
except MaskNotFound as exc:
|
|
print(f"could not find the mask: {exc}; running preview-only")
|
|
mask = None
|
|
if mask is not None:
|
|
await mask.set_brightness(args.brightness)
|
|
face = FaceAnimator(mask)
|
|
print("loading face frames (one-time) ...")
|
|
await face.load(force=args.reload)
|
|
|
|
print("tracking — move/talk to your camera. Ctrl+C (or 'q' in the window) to stop.")
|
|
min_dt = 1.0 / args.fps
|
|
last = 0.0
|
|
try:
|
|
while True:
|
|
ok, frame = cap.read()
|
|
if not ok:
|
|
break
|
|
name, annotated = tracker.detect(frame)
|
|
now = time.monotonic()
|
|
if face is not None and now - last >= min_dt:
|
|
await face.show(name) # PLAY the matching frame (deduped)
|
|
last = now
|
|
if args.show:
|
|
cv2.imshow("facetrack (q to quit)", annotated)
|
|
if cv2.waitKey(1) & 0xFF == ord("q"):
|
|
break
|
|
else:
|
|
await asyncio.sleep(0.001)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
cap.release()
|
|
if args.show:
|
|
cv2.destroyAllWindows()
|
|
if mask is not None:
|
|
await mask.disconnect()
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("--camera", type=int, default=0)
|
|
ap.add_argument("--show", action="store_true", help="open a preview window with overlay")
|
|
ap.add_argument("--no-mask", action="store_true", help="track/preview only, don't touch the mask")
|
|
ap.add_argument("--no-mirror", action="store_true", help="don't horizontally flip the camera")
|
|
ap.add_argument("--image", help="run detection on a single image file (offline test)")
|
|
ap.add_argument("--fps", type=float, default=10.0, help="max mask updates/sec")
|
|
ap.add_argument("--reload", action="store_true", help="force re-upload of the frame set")
|
|
ap.add_argument("--brightness", type=int, default=95)
|
|
ap.add_argument("--address", help="mask BLE MAC")
|
|
ap.add_argument("--name-prefix", default="MASK")
|
|
asyncio.run(run(ap.parse_args()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|