from collections import deque import cv2 from edge.models import BBox class RingBuffer: def __init__(self, maxlen: int = 90): self.frames: deque = deque(maxlen=maxlen) # (t, frame, head_boxes) def push(self, t: float, frame, head_boxes: list | None = None) -> None: # head_boxes ride along per frame: a PAST frame must be blurred with the head # positions of ITS OWN moment — blurring history with the current tick's boxes # would leave yesterday's face exactly where it actually was (PIPA). self.frames.append((t, frame, head_boxes or [])) def window(self, pre: int, post: int, index: int) -> list: lo = max(0, index - pre) hi = min(len(self.frames), index + post + 1) return [f for (_t, f, _b) in list(self.frames)[lo:hi]] def timeline(self, now: float, offsets_s: tuple) -> tuple[list, list]: """각 목표 시각(now - offset)에 가장 가까운 프레임을 골라 (frames, head_boxes)를 과거→현재 순으로 반환. 버퍼가 짧으면(기동 직후) 있는 만큼만 — 같은 프레임이 여러 목표에 걸리면 한 번만 들어간다.""" items = list(self.frames) if not items: return [], [] idxs: list[int] = [] for off in sorted(offsets_s, reverse=True): target = now - off i = min(range(len(items)), key=lambda k: abs(items[k][0] - target)) if i not in idxs: idxs.append(i) idxs.sort() return [items[i][1] for i in idxs], [items[i][2] for i in idxs] class FaceBlurrer: def blur(self, frame, head_boxes: list[BBox]): for b in head_boxes: x1, y1 = int(max(0, b.x1)), int(max(0, b.y1)) x2, y2 = int(min(frame.shape[1], b.x2)), int(min(frame.shape[0], b.y2)) if x2 > x1 and y2 > y1: roi = frame[y1:y2, x1:x2] frame[y1:y2, x1:x2] = cv2.GaussianBlur(roi, (0, 0), sigmaX=12) return frame class Snapshotter: def __init__(self, blurrer: FaceBlurrer, quality: int = 80): self.blurrer = blurrer self.quality = quality def capture(self, frames: list, head_boxes_per_frame: list) -> list[bytes]: out: list[bytes] = [] for frame, heads in list(zip(frames, head_boxes_per_frame))[:3]: blurred = self.blurrer.blur(frame.copy(), heads) ok, buf = cv2.imencode(".jpg", blurred, [cv2.IMWRITE_JPEG_QUALITY, self.quality]) if ok: out.append(buf.tobytes()) return out