"""Renderer Metal natif via pyobjc. Architecture : - MTKView : NSView Metal-managee, callback drawInMTKView - MTLDevice : GPU par defaut - MTLCommandQueue : queue de submission - 2 pipelines : bg_pipeline : fullscreen tri + fbm fragment shader (1 draw call) skel_pipeline : lignes pour le squelette pose (max 16 segments) - Uniforms buffer 56 octets (14 floats) cf scene.metal::SceneUniforms Le renderer ne possede PAS de state thread : il lit le State partage sous lock a chaque frame (60 fps). """ from __future__ import annotations import logging import os import struct import time from pathlib import Path import numpy as np import objc from Cocoa import NSObject, NSColor, NSMakeRect from Metal import ( MTLCreateSystemDefaultDevice, MTLCompileOptions, MTLPrimitiveTypeTriangle, MTLPrimitiveTypeLine, MTLRenderPipelineDescriptor, MTLResourceStorageModeShared, MTLVertexAttributeDescriptor, MTLVertexBufferLayoutDescriptor, MTLVertexDescriptor, MTLVertexFormatFloat, MTLVertexFormatFloat2, MTLVertexFormatFloat3, MTLVertexStepFunctionPerVertex, ) # MTKViewDelegate est un @protocol Obj-C ; pas besoin d'import Python. # pyobjc detecte automatiquement l'implementation par signature. from MetalKit import MTKView # noqa: F401 (utilise par d'autres modules) from .arkit_skeleton import arkit_segments, finger_joint_mask from .hand_display import ( HandPersistenceGate, arkit_2d_fresh, gauge_segments, hand_plausible, hand_size, segment_ok, panel_frame, panel_segments, ) from .hand_slots import route_hands from .mesh_topology import ( BODY_TRIANGLES, FACE_TRIANGLES, build_face_triangles_dynamic, ) from .state import State # Draw the full ARKit 91-joint body (topology-driven) instead of the # MP33-reduced body. Set ARKIT_FULL_SKELETON=0 to fall back to MP33. ARKIT_FULL = os.environ.get("ARKIT_FULL_SKELETON", "1") != "0" LOG = logging.getLogger("renderer") # Triangle primitive constant (Metal MTLPrimitiveType.triangle = 3) MTL_PRIMITIVE_TRIANGLE = 3 # 17 keypoints COCO bones (16 paires) — legacy YOLO fallback COCO_BONES: list[tuple[int, int]] = [ (0, 1), (0, 2), (1, 3), (2, 4), (5, 6), (5, 7), (7, 9), (6, 8), (8, 10), (5, 11), (6, 12), (11, 12), (11, 13), (13, 15), (12, 14), (14, 16), ] def _mediapipe_bones(): """Charge les connexions MediaPipe (body+face+hands) au runtime. Retourne 4 listes : (body_bones, face_bones, lhand_bones, rhand_bones) avec chaque bone = (idx_a, idx_b). Retourne None si mediapipe absent (fallback COCO).""" try: from mediapipe.tasks.python.vision import ( FaceLandmarksConnections as F, HandLandmarksConnections as H, PoseLandmarksConnections as P, ) except ImportError: return None def to_pairs(cs): return [(c.start, c.end) for c in cs] # 35 body body = to_pairs(P.POSE_LANDMARKS) # Visage HAUTE DENSITE : tesselation complete (2556) + contours + # iris + nez. Donne ~2700 segments rien que pour le visage. face = ( to_pairs(F.FACE_LANDMARKS_TESSELATION) # 2556 segs (mesh dense) + to_pairs(F.FACE_LANDMARKS_FACE_OVAL) + to_pairs(F.FACE_LANDMARKS_LIPS) + to_pairs(F.FACE_LANDMARKS_LEFT_EYE) + to_pairs(F.FACE_LANDMARKS_RIGHT_EYE) + to_pairs(F.FACE_LANDMARKS_LEFT_EYEBROW) + to_pairs(F.FACE_LANDMARKS_RIGHT_EYEBROW) + to_pairs(F.FACE_LANDMARKS_LEFT_IRIS) + to_pairs(F.FACE_LANDMARKS_RIGHT_IRIS) + to_pairs(F.FACE_LANDMARKS_NOSE) ) # 21 mains (chacune) hand = to_pairs(H.HAND_CONNECTIONS) return body, face, hand, hand # left+right partagent les connexions # Capacite max du vertex buffer skeleton. # 16384 segs * 2 verts * 5 floats (xyz + conf + pid) * 4 bytes = 640 KB # Couvre 4 personnes : 4×35 body + 4×2700 face + 8×21 hand = ~11000 segs. SKEL_MAX_SEGS = 16384 SKEL_VERT_FLOATS = 5 # x, y, z, conf, person_id # Mesh : ~8192 triangles × 3 verts × 5 floats × 4 = 480 KB. # Couvre 4 personnes : 4×~80 face + 4×~16 body + 8×~18 hand ≈ 600 triangles # pour le hardcode, ou ~4×~150 face Delaunay = ~600 triangles. Marge large. MESH_MAX_TRIS = 8192 MESH_VERT_FLOATS = 5 # identique au skel MESH_MAX_VERTS = 10475 # SMPL-X is the larger family; SMPL (6890) fits inside # struct SceneUniforms : 17 floats packs = 68 octets, padding a 80 (multiple # de 16, regle Metal). On y stocke (time, rms, kp_norm, netz_dev, # lightning_flash, flare, wind_norm, bz_norm, social_rate, pose_alive, # pose_count, width, height, viz_mode, _pad0, _pad1). UNIFORM_FLOATS = 24 # +4 floats : hand_l_x/y, hand_r_x/y + hand_height, hand_openness, hand_speed, hand_dist UNIFORM_SIZE = UNIFORM_FLOATS * 4 # 96 octets, aligne 16 class MetalRenderer(NSObject): """Delegate de MTKView, drive l'integralite du rendu.""" def initWithState_(self, state: State): # noqa: N802 self = objc.super(MetalRenderer, self).init() if self is None: return None self._state = state self._device = MTLCreateSystemDefaultDevice() if self._device is None: raise RuntimeError("Metal non disponible sur ce systeme") LOG.info("device: %s", self._device.name()) self._queue = self._device.newCommandQueue() self._uniforms_buf = self._device.newBufferWithLength_options_( UNIFORM_SIZE, MTLResourceStorageModeShared) # Skeleton : N segs × 2 verts × 5 floats (xyz + conf + pid) self._skel_buf = self._device.newBufferWithLength_options_( SKEL_MAX_SEGS * 2 * SKEL_VERT_FLOATS * 4, MTLResourceStorageModeShared) # Mesh buffer : triangles face/hand/body self._mesh_buf = self._device.newBufferWithLength_options_( MESH_MAX_TRIS * 3 * MESH_VERT_FLOATS * 4, MTLResourceStorageModeShared) self._mp_bones = _mediapipe_bones() # None si pas dispo from .config import VizConfig as _VizConfig _cfg = _VizConfig.from_env() self._hand_conf_min = _cfg.hand_conf_min self._arkit_bone_max = _cfg.arkit_bone_max self._hand_gate = HandPersistenceGate(min_frames=_cfg.hand_persist_frames, grace=_cfg.hand_persist_grace) # Safety knob: invert Vision chirality interpretation. # Chirality validated correct live 2026-07-02; keep False unless a # future iPhone app build flips it. self._hand_swap_lr = _cfg.hand_swap_lr self._arkit_full = _cfg.arkit_full_skeleton self._skel_width = max(1, int(_cfg.skel_line_width)) self._init_skel_cpu_buffer() self._init_mesh_cpu_buffer() self._build_pipelines() self._last_lightning_emit = 0.0 return self # ---- Pipelines ------------------------------------------------- def _build_pipelines(self): src_path = Path(__file__).parent / "shaders" / "scene.metal" source = src_path.read_text() opts = MTLCompileOptions.alloc().init() lib, err = self._device.newLibraryWithSource_options_error_( source, opts, None) if lib is None: raise RuntimeError(f"shader compile failed: {err}") self._lib = lib # --- Background pipeline (no vertex buffer) --- bg = MTLRenderPipelineDescriptor.alloc().init() bg.setVertexFunction_(lib.newFunctionWithName_("bg_vertex")) bg.setFragmentFunction_(lib.newFunctionWithName_("bg_fragment")) bg.colorAttachments().objectAtIndexedSubscript_(0).setPixelFormat_(80) # BGRA8Unorm p, err = self._device.newRenderPipelineStateWithDescriptor_error_(bg, None) if p is None: raise RuntimeError(f"bg pipeline failed: {err}") self._bg_pipe = p # --- Skeleton pipeline (vertex buffer pos+conf) --- # Skeleton vertex layout : pos (float3) + conf (float) + pid (float) # Stride = 5 floats × 4 bytes = 20 bytes vd = MTLVertexDescriptor.vertexDescriptor() a0 = vd.attributes().objectAtIndexedSubscript_(0) a0.setFormat_(MTLVertexFormatFloat3); a0.setOffset_(0); a0.setBufferIndex_(0) a1 = vd.attributes().objectAtIndexedSubscript_(1) a1.setFormat_(MTLVertexFormatFloat); a1.setOffset_(12); a1.setBufferIndex_(0) a2 = vd.attributes().objectAtIndexedSubscript_(2) a2.setFormat_(MTLVertexFormatFloat); a2.setOffset_(16); a2.setBufferIndex_(0) ld = vd.layouts().objectAtIndexedSubscript_(0) ld.setStride_(20); ld.setStepFunction_(MTLVertexStepFunctionPerVertex) sk = MTLRenderPipelineDescriptor.alloc().init() sk.setVertexFunction_(lib.newFunctionWithName_("skel_vertex")) sk.setFragmentFunction_(lib.newFunctionWithName_("skel_fragment")) sk.setVertexDescriptor_(vd) ca = sk.colorAttachments().objectAtIndexedSubscript_(0) ca.setPixelFormat_(80) # BGRA8Unorm ca.setBlendingEnabled_(True) # SrcAlpha, OneMinusSrcAlpha (= 4, 5 dans MTLBlendFactor) ca.setSourceRGBBlendFactor_(4) ca.setDestinationRGBBlendFactor_(5) ca.setSourceAlphaBlendFactor_(4) ca.setDestinationAlphaBlendFactor_(5) p, err = self._device.newRenderPipelineStateWithDescriptor_error_(sk, None) if p is None: raise RuntimeError(f"skel pipeline failed: {err}") self._skel_pipe = p # --- Mesh pipeline (triangles face/hand/body) --- # Vertex layout strictement identique au skel (reutilise vd). vd_mesh = MTLVertexDescriptor.vertexDescriptor() a0 = vd_mesh.attributes().objectAtIndexedSubscript_(0) a0.setFormat_(MTLVertexFormatFloat3); a0.setOffset_(0); a0.setBufferIndex_(0) a1 = vd_mesh.attributes().objectAtIndexedSubscript_(1) a1.setFormat_(MTLVertexFormatFloat); a1.setOffset_(12); a1.setBufferIndex_(0) a2 = vd_mesh.attributes().objectAtIndexedSubscript_(2) a2.setFormat_(MTLVertexFormatFloat); a2.setOffset_(16); a2.setBufferIndex_(0) ld2 = vd_mesh.layouts().objectAtIndexedSubscript_(0) ld2.setStride_(20); ld2.setStepFunction_(MTLVertexStepFunctionPerVertex) mp = MTLRenderPipelineDescriptor.alloc().init() mp.setVertexFunction_(lib.newFunctionWithName_("mesh_vertex")) mp.setFragmentFunction_(lib.newFunctionWithName_("mesh_fragment")) mp.setVertexDescriptor_(vd_mesh) cm = mp.colorAttachments().objectAtIndexedSubscript_(0) cm.setPixelFormat_(80) cm.setBlendingEnabled_(True) # SrcAlpha, OneMinusSrcAlpha — alpha classique pour voile mesh cm.setSourceRGBBlendFactor_(4) cm.setDestinationRGBBlendFactor_(5) cm.setSourceAlphaBlendFactor_(4) cm.setDestinationAlphaBlendFactor_(5) p, err = self._device.newRenderPipelineStateWithDescriptor_error_(mp, None) if p is None: raise RuntimeError(f"mesh pipeline failed: {err}") self._mesh_pipe = p # ---- CPU staging buffers -------------------------------------- def _init_skel_cpu_buffer(self) -> None: """Preallocate the CPU staging buffer for skeleton segments. SKEL_MAX_SEGS * 2 * SKEL_VERT_FLOATS floats : each segment = 2 verts × 5 floats (x, y, z, conf, pid). Idempotent — no-op if already allocated. """ if getattr(self, "_skel_cpu_buf", None) is None: self._skel_cpu_buf = np.zeros(SKEL_MAX_SEGS * 2 * SKEL_VERT_FLOATS, dtype=np.float32) def _init_mesh_cpu_buffer(self) -> None: if getattr(self, "_mesh_cpu_buf", None) is None: self._mesh_cpu_buf = np.zeros( MESH_MAX_VERTS * MESH_VERT_FLOATS, dtype=np.float32, ) # ---- Uniforms helpers ------------------------------------------ def _update_uniforms(self) -> int: s = self._state now = time.monotonic() with s.lock(): # Flash de foudre : decay exponentiel ~600 ms dt_flash = now - s.last_lightning_t flash = max(0.0, 1.0 - dt_flash * 1.7) if dt_flash < 1.0 else 0.0 # Positions des mains (point 0 = poignet) — pour mode hands3d lh_wrist = s.left_hand_kp[0] if s.hands_present else None rh_wrist = s.right_hand_kp[0] if s.hands_present else None hlx = (lh_wrist.x if lh_wrist else 0.5) * 2 - 1 hly = 1 - (lh_wrist.y if lh_wrist else 0.5) * 2 hrx = (rh_wrist.x if rh_wrist else 0.5) * 2 - 1 hry = 1 - (rh_wrist.y if rh_wrist else 0.5) * 2 # Expressive hand features from HandFeatureExtractor hf = getattr(s, "hand_feats", None) or {} hl, hr = hf.get("L"), hf.get("R") ys = [h["cy"] for h in (hl, hr) if h] hand_height = (1.0 - min(ys)) if ys else 0.0 hand_open = max([h["openness"] for h in (hl, hr) if h] or [0.0]) hand_speed = max([h["speed"] for h in (hl, hr) if h] or [0.0]) hand_dist = float(hf.get("dist", 0.0)) uniforms = struct.pack( f"{UNIFORM_FLOATS}f", s.elapsed(), # 1 min(1.0, s.rms * 3.0), # 2 min(1.0, s.swpc_kp / 9.0), # 3 max(-0.1, min(0.1, s.netz_dev)), # 4 flash, # 5 s.swpc_flare_norm, # 6 max(0.0, min(1.0, (s.swpc_wind_speed - 280.0) / 600.0)), # 7 max(-1.0, min(1.0, s.swpc_bz / 15.0)), # 8 min(1.0, s.social_rate / 50.0), # 9 1.0 if s.pose_alive() else 0.0, # 10 float(s.pose_count), # 11 float(s.width), # 12 float(s.height), # 13 float(s.viz_mode), # 14 hlx, hly, hrx, hry, # 15-18 wrist xy (kept) hand_height, hand_open, # 19-20 hand_speed, hand_dist, # 21-22 0.0, 0.0, # 23-24 pad (16-byte align) ) n_segs = self._update_skeleton(s) n_tris = self._update_mesh(s) # Copie via memoryview (pyobjc API : varlist.as_buffer(N)) mv = self._uniforms_buf.contents().as_buffer(UNIFORM_SIZE) mv[:] = uniforms # Log debug toutes les 120 frames (~2s) — confirme que mesh draw if not hasattr(self, "_dbg_n"): self._dbg_n = 0 self._dbg_n += 1 if self._dbg_n % 120 == 0: LOG.info("render: %d segs, %d tris (face=%d hand=%d body=%d)", n_segs, n_tris, len(self._state.persons_face), len(self._state.persons_hands), len(self._state.persons_body)) return n_segs, n_tris def _update_skeleton(self, s: State) -> int: """Remplit self._skel_buf avec les segments visibles. Retourne le nombre de segments (2 verts chacun). Priorise MediaPipe (body 33 + face 478 + 2 mains 21) si disponible et present ; sinon fallback COCO 17 keypoints YOLO.""" _arkit_now = time.perf_counter() use_arkit = (getattr(self, '_arkit_full', ARKIT_FULL) and bool(s.arkit_parents) and arkit_2d_fresh(s.persons_arkit_2d_t, _arkit_now)) # NO early-return when nothing is detected: the panel frames and the # X/Y gauges are permanent screen furniture (user request: indicators # must stay visible with no hands). The overlay loops below iterate # empty collections at negligible cost. buf = self._skel_cpu_buf segs = 0 # Mirror overlays in X to match the (already mirrored) iPhone video # background. Display-only: gesture data is untouched. mirror = bool(getattr(s, "mirror_2d", False)) def push(A, B, conf, pid): """Empile un segment (2 verts) dans le buffer CPU prealloque.""" nonlocal segs if segs >= SKEL_MAX_SEGS: return False ax = A.x * 2.0 - 1.0; ay = 1.0 - A.y * 2.0 bx = B.x * 2.0 - 1.0; by = 1.0 - B.y * 2.0 if mirror: ax = -ax; bx = -bx i = segs * 10 buf[i+0] = ax; buf[i+1] = ay; buf[i+2] = float(A.z); buf[i+3] = conf; buf[i+4] = float(pid) buf[i+5] = bx; buf[i+6] = by; buf[i+7] = float(B.z); buf[i+8] = conf; buf[i+9] = float(pid) segs += 1 return True def push_seg(ax, ay, bx, by, conf, pid): """Like push() but takes raw normalized [0,1] coords.""" nonlocal segs if segs >= SKEL_MAX_SEGS: return False cax = ax * 2.0 - 1.0; cay = 1.0 - ay * 2.0 cbx = bx * 2.0 - 1.0; cby = 1.0 - by * 2.0 if mirror: cax = -cax; cbx = -cbx i = segs * 10 buf[i+0] = cax; buf[i+1] = cay; buf[i+2] = 0.0 buf[i+3] = conf; buf[i+4] = float(pid) buf[i+5] = cbx; buf[i+6] = cby; buf[i+7] = 0.0 buf[i+8] = conf; buf[i+9] = float(pid) segs += 1 return True def push_panel(ax, ay, bx, by, conf, pid): """Like push_seg but NEVER applies mirror. Panel coords are final screen coords [0,1] — they must not be flipped again even when the video background is mirrored. """ nonlocal segs if segs >= SKEL_MAX_SEGS: return False cax = ax * 2.0 - 1.0; cay = 1.0 - ay * 2.0 cbx = bx * 2.0 - 1.0; cby = 1.0 - by * 2.0 i = segs * 10 buf[i+0] = cax; buf[i+1] = cay; buf[i+2] = 0.0 buf[i+3] = conf; buf[i+4] = float(pid) buf[i+5] = cbx; buf[i+6] = cby; buf[i+7] = 0.0 buf[i+8] = conf; buf[i+9] = float(pid) segs += 1 return True # Gate called once per frame on the full persons_hands list. # Used for both the overlay loop and the side panels below. # getattr: unit tests build the renderer via __new__ (no Metal init). _gate = getattr(self, "_hand_gate", None) _hand_draw_flags = (_gate.step(s.persons_hands) if _gate is not None else [True] * len(s.persons_hands)) if use_arkit: parents = s.arkit_parents # Hide ARKit's per-finger joints ("false hands"): the display # shows the filtered Vision hands instead. Mask cached until the # topology changes. fmask = getattr(self, "_finger_mask", None) if fmask is None or len(fmask) != len(s.arkit_joint_names): fmask = np.asarray( finger_joint_mask(s.arkit_joint_names), dtype=bool) self._finger_mask = fmask for pid, arr2d in s.persons_arkit_2d.items(): ts = s.persons_arkit_2d_t.get(pid) if ts is None or (_arkit_now - ts) >= 1.0: continue # skip stale or unknown pid valid = s.persons_arkit_2d_valid.get(pid) if len(fmask): if valid is None: valid = ~fmask elif len(valid) == len(fmask): valid = valid & ~fmask # Bold strokes: Metal lines are 1px, so each bone is drawn # in concentric offset passes (SKEL_LINE_WIDTH approx px). _w = getattr(self, "_skel_width", 1) _o = 1.0 / max(1, s.height or 720) _offsets = [(0.0, 0.0)] for _r in range(1, _w): _d = _r * _o _offsets += [(_d, 0.0), (-_d, 0.0), (0.0, _d), (0.0, -_d)] for (ax, ay, bx, by) in arkit_segments( arr2d, valid, parents, max_len=self._arkit_bone_max ): for _dx, _dy in _offsets: if not push_seg(ax + _dx, ay + _dy, bx + _dx, by + _dy, 1.0, pid): break if self._mp_bones is not None and ( s.persons_body or s.persons_face or s.persons_hands or s.body_present or s.face_present or s.hands_present ): body_bones, face_bones, lhand_bones, _ = self._mp_bones # ----- MULTI-PERSONNE : pid = ID stable du tracker ----- # (track_id persiste entre frames, palette se stabilise) ids_b = s.persons_body_ids or list(range(len(s.persons_body))) ids_f = s.persons_face_ids or list(range(len(s.persons_face))) ids_h = s.persons_hands_ids or list(range(len(s.persons_hands))) for i, body_kp in enumerate([] if use_arkit else s.persons_body): pid = ids_b[i] if i < len(ids_b) else i for a, b in body_bones: if a >= len(body_kp) or b >= len(body_kp): continue A = body_kp[a]; B = body_kp[b] if A.c < 0.15 or B.c < 0.15: continue if not push(A, B, min(A.c, B.c), pid): break for i, face_kp in enumerate(s.persons_face): pid = ids_f[i] if i < len(ids_f) else i for a, b in face_bones: if a >= len(face_kp) or b >= len(face_kp): continue if not push(face_kp[a], face_kp[b], 1.0, pid): break if segs >= SKEL_MAX_SEGS: break # Bold hand strokes: same concentric-offset technique as the # ARKit body (SKEL_LINE_WIDTH), one step thinner so small hands # do not blob. _hw = max(1, getattr(self, "_skel_width", 1) - 1) _ho = 1.0 / max(1, s.height or 720) _hoffs = [(0.0, 0.0)] for _r in range(1, _hw): _hd = _r * _ho _hoffs += [(_hd, 0.0), (-_hd, 0.0), (0.0, _hd), (0.0, -_hd)] for i, hand_kp in enumerate(s.persons_hands): if not _hand_draw_flags[i]: continue pid = ids_h[i] if i < len(ids_h) else i if not hand_plausible(hand_kp, conf_min=self._hand_conf_min): continue size = hand_size(hand_kp) for a, b in lhand_bones: if a >= len(hand_kp) or b >= len(hand_kp): continue A = hand_kp[a]; B = hand_kp[b] if not segment_ok(A, B, size, conf_min=self._hand_conf_min): continue # Decalage palette mains (+5) pour les distinguer _c = min(A.c, B.c) for _dx, _dy in _hoffs: if not push_seg(A.x + _dx, A.y + _dy, B.x + _dx, B.y + _dy, _c, pid + 5): break # ----- FALLBACK single-person si persons_* vides ----- if not (s.persons_body or s.persons_face or s.persons_hands): if s.body_present and not use_arkit: for a, b in body_bones: if a >= len(s.body_kp) or b >= len(s.body_kp): continue A = s.body_kp[a]; B = s.body_kp[b] if A.c < 0.15 or B.c < 0.15: continue if not push(A, B, min(A.c, B.c), 0): break if s.face_present: for a, b in face_bones: if a >= len(s.face_kp) or b >= len(s.face_kp): continue if not push(s.face_kp[a], s.face_kp[b], 1.0, 0): break if s.hands_present: for kp_list in (s.left_hand_kp, s.right_hand_kp): if not any(p.x != 0.0 or p.y != 0.0 for p in kp_list): continue if not hand_plausible(kp_list, conf_min=self._hand_conf_min): continue size = hand_size(kp_list) for a, b in lhand_bones: if a >= len(kp_list) or b >= len(kp_list): continue A = kp_list[a]; B = kp_list[b] if not segment_ok(A, B, size, conf_min=self._hand_conf_min): continue if not push(A, B, min(A.c, B.c), 0): break else: # Fallback COCO 17 (YOLO legacy) for a, b in COCO_BONES: A = s.pose_kp[a]; B = s.pose_kp[b] if A.c < 0.2 or B.c < 0.2: continue if not push(A, B, min(A.c, B.c), 0): break # ---- SIDE PANELS: left/right hand front-view zoomed wireframe ---- # Gated on self._mp_bones (panels need _mp_bones for hand bone topology, # even when use_arkit is True). Hands from s.persons_hands only. if self._mp_bones is not None: _body_b, _face_b, lhand_bones_p, _ = self._mp_bones # Filter hands by persistence gate + plausibility before routing # to panels (a ghost hand must not steal a slot from a real one). keep = [ ok and hand_plausible(h, conf_min=self._hand_conf_min) for h, ok in zip(s.persons_hands, _hand_draw_flags) ] gated_hands = [h for h, k in zip(s.persons_hands, keep) if k] chir_src = getattr(s, "persons_hands_chirality", None) or [] # Keep chirality aligned with the gated hands (same filter mask). if chir_src and len(chir_src) == len(s.persons_hands): gated_chir = [c for c, k in zip(chir_src, keep) if k] else: gated_chir = None asp = float(s.width) / float(s.height) if s.width and s.height else 1.0 # Route to L/R panels. mirror= passed for cx fallback path; # no near_min — panels show far hands as user proximity feedback. slotted = route_hands( gated_hands, gated_chir, mirror=mirror, swap=self._hand_swap_lr, ) left_kp, right_kp = slotted # Panel frame: status drives hue (pid 7/8/9), quality drives # brightness (0.25+0.75*q) and thickness (1..3 stroke passes). # pid 7 conf=0.25+0.75*q → status 0: absent (q=0) # → status 1: detected (q≥0.30) # pid 8 conf=0.25+0.75*q → status 2: armed (near+facing) # pid 9 conf=1.0 (q=1.0) → status 3: pinch engaged # Thickness: pass 1 always; pass 2 at q≥0.50; pass 3 at q≥0.85 or status==3. _slot_status = getattr(s, "gesture_slot_status", [0, 0]) _slot_quality = getattr(s, "gesture_slot_quality", [0.0, 0.0]) _hf_all = getattr(s, "hand_feats", None) or {} for _si, (side_name, h_kp, pid_s) in enumerate(( ("left", left_kp, 5), ("right", right_kp, 6), )): _st = _slot_status[_si] if _si < len(_slot_status) else 0 _q = float(_slot_quality[_si]) if _si < len(_slot_quality) else 0.0 _fconf = 0.25 + 0.75 * _q # Status drives hue (pid selection) _fpid = 9 if _st == 3 else (8 if _st == 2 else 7) _frame_segs = panel_frame(side_name, asp) # Pass 1: always for seg in _frame_segs: push_panel(*seg, _fconf, _fpid) # Pass 2: thicker when quality ≥ 0.50 if _q >= 0.5: for ax, ay, bx, by in _frame_segs: push_panel(ax + 0.001, ay, bx + 0.001, by, _fconf, _fpid) # Pass 3: boldest when quality ≥ 0.85 or pinch engaged if _q >= 0.85 or _st == 3: for ax, ay, bx, by in _frame_segs: push_panel(ax + 0.002, ay, bx + 0.002, by, _fconf, _fpid) # Position gauges (X below / Y outer-side). SC's mod cache # HOLDS the last received value when a hand vanishes, so the # markers must stay visible at that last value too (user: # "X et Y toujours visibles") — live = full conf, held = dim. _slot_key = "L" if side_name == "left" else "R" _slot_hf = _hf_all.get(_slot_key) if not hasattr(self, "_gauge_hold"): self._gauge_hold = {"L": (None, None), "R": (None, None)} _live = _slot_hf is not None if _live: self._gauge_hold[_slot_key] = (_slot_hf["cx"], _slot_hf["cy"]) _g_cx, _g_cy = self._gauge_hold[_slot_key] _mk_conf = 1.0 if _live else 0.45 for ax, ay, bx, by, gconf, gpid in gauge_segments( _g_cx, _g_cy, side_name, asp, mirror, content_pid=pid_s, ): push_panel(ax, ay, bx, by, min(gconf, _mk_conf) if gpid == pid_s else gconf, gpid) # Wrist marker ON the video: the X/Y voice mods are anchored # to the wrist, so its control point stays visible whenever # a value exists (held at the last position when the hand # drops, matching what SC hears). push_seg applies the # video mirror like the overlay. if _g_cx is not None and _g_cy is not None: _mx = 0.015 / asp push_seg(_g_cx - _mx, _g_cy, _g_cx + _mx, _g_cy, _mk_conf, pid_s) push_seg(_g_cx, _g_cy - 0.015, _g_cx, _g_cy + 0.015, _mk_conf, pid_s) if h_kp is not None: for seg in panel_segments( h_kp, side_name, lhand_bones_p, asp, mirror=mirror, conf_min=self._hand_conf_min, ): push_panel(*seg, 1.0, pid_s) if segs == 0: return 0 data = self._skel_cpu_buf[: segs * 2 * SKEL_VERT_FLOATS].tobytes() mv = self._skel_buf.contents().as_buffer(len(data)) mv[:] = data return segs def _update_mesh(self, s: State) -> int: """Remplit self._mesh_buf avec des triangles face/body. Retourne le nombre de triangles ecrits (chacun = 3 vertices). Filtre les triangles dont au moins un sommet a confiance < 0.3. """ if not s.pose_alive(): return 0 if not (s.persons_face or s.persons_hands or s.persons_body): return 0 n_verts = 0 def push_tri(kp_list, i, j, k, pid: int) -> bool: """Pousse un triangle (3 verts). Retourne False si buffer plein ou triangle invalide (confiance basse).""" nonlocal n_verts tris = n_verts // 3 if tris >= MESH_MAX_TRIS or n_verts + 3 > MESH_MAX_VERTS: return False if i >= len(kp_list) or j >= len(kp_list) or k >= len(kp_list): return True # skip mais continue A = kp_list[i]; B = kp_list[j]; C = kp_list[k] if A.c < 0.15 or B.c < 0.15 or C.c < 0.15: return True ax = A.x * 2.0 - 1.0; ay = 1.0 - A.y * 2.0 bx = B.x * 2.0 - 1.0; by = 1.0 - B.y * 2.0 cx = C.x * 2.0 - 1.0; cy = 1.0 - C.y * 2.0 conf = min(A.c, B.c, C.c) fpid = float(pid) base = n_verts * 5 self._mesh_cpu_buf[base + 0] = ax self._mesh_cpu_buf[base + 1] = ay self._mesh_cpu_buf[base + 2] = float(A.z) self._mesh_cpu_buf[base + 3] = conf self._mesh_cpu_buf[base + 4] = fpid self._mesh_cpu_buf[base + 5] = bx self._mesh_cpu_buf[base + 6] = by self._mesh_cpu_buf[base + 7] = float(B.z) self._mesh_cpu_buf[base + 8] = conf self._mesh_cpu_buf[base + 9] = fpid self._mesh_cpu_buf[base + 10] = cx self._mesh_cpu_buf[base + 11] = cy self._mesh_cpu_buf[base + 12] = float(C.z) self._mesh_cpu_buf[base + 13] = conf self._mesh_cpu_buf[base + 14] = fpid n_verts += 3 return True ids_b = s.persons_body_ids or list(range(len(s.persons_body))) ids_f = s.persons_face_ids or list(range(len(s.persons_face))) # Body — SKIPPED whenever ARKit is the body source (iphone-usb: the # ARKit topology arrived). persons_body is then the ARKit->MP33 # conversion with c forced to 1.0, so garbage/stale joints would pass # the confidence filter and paint solid triangles anywhere on screen # (the "blue blob"). The body is the gated ARKit wireframe instead. arkit_body = (getattr(self, "_arkit_full", ARKIT_FULL) and bool(s.arkit_parents)) for i, body_kp in enumerate([] if arkit_body else s.persons_body): pid = ids_b[i] if i < len(ids_b) else i for a, b, c in BODY_TRIANGLES: if not push_tri(body_kp, a, b, c, pid): break if n_verts >= MESH_MAX_VERTS: break # Face — utilise triangulation Delaunay dynamique sur les XY, # fallback sur FACE_TRIANGLES statique si Delaunay echoue. for i, face_kp in enumerate(s.persons_face): pid = ids_f[i] if i < len(ids_f) else i pts_xy = [(kp.x, kp.y) for kp in face_kp] tri_list = build_face_triangles_dynamic(pts_xy) or FACE_TRIANGLES for a, b, c in tri_list: if not push_tri(face_kp, a, b, c, pid): break if n_verts >= MESH_MAX_VERTS: break if n_verts == 0: return 0 # Slice is exact — stale floats beyond n_verts*MESH_VERT_FLOATS never reach the GPU. data = self._mesh_cpu_buf[: n_verts * MESH_VERT_FLOATS].tobytes() mv = self._mesh_buf.contents().as_buffer(len(data)) mv[:] = data return n_verts // 3 # ---- MTKViewDelegate ------------------------------------------ def mtkView_drawableSizeWillChange_(self, view, size): # noqa: N802 with self._state.lock(): self._state.width = int(size.width) self._state.height = int(size.height) def drawInMTKView_(self, view): # noqa: N802 n_segs, n_tris = self._update_uniforms() rpd = view.currentRenderPassDescriptor() drawable = view.currentDrawable() if rpd is None or drawable is None: return cb = self._queue.commandBuffer() enc = cb.renderCommandEncoderWithDescriptor_(rpd) # 1) background fullscreen tri enc.setRenderPipelineState_(self._bg_pipe) enc.setFragmentBuffer_offset_atIndex_(self._uniforms_buf, 0, 0) enc.drawPrimitives_vertexStart_vertexCount_(MTLPrimitiveTypeTriangle, 0, 3) # 2) mesh overlay (triangles face/hand/body) — DESSOUS le skel if n_tris > 0: enc.setRenderPipelineState_(self._mesh_pipe) enc.setVertexBuffer_offset_atIndex_(self._mesh_buf, 0, 0) enc.setVertexBuffer_offset_atIndex_(self._uniforms_buf, 0, 1) enc.drawPrimitives_vertexStart_vertexCount_( MTL_PRIMITIVE_TRIANGLE, 0, n_tris * 3) # 3) skeleton overlay (lignes par-dessus le mesh) if n_segs > 0: enc.setRenderPipelineState_(self._skel_pipe) enc.setVertexBuffer_offset_atIndex_(self._skel_buf, 0, 0) # SceneUniforms a buffer(1) du vertex pour acceder a U.rms etc enc.setVertexBuffer_offset_atIndex_(self._uniforms_buf, 0, 1) enc.drawPrimitives_vertexStart_vertexCount_( MTLPrimitiveTypeLine, 0, n_segs * 2) enc.endEncoding() cb.presentDrawable_(drawable) cb.commit() def device(self): return self._device