# -*- coding: utf-8 -*- """yjm_ai_snapshot_ring.py CCTV 순환녹화 방식의 AI snapshot ring. - 현재 캡처 source를 주기적으로 확인 - 이전 저장 후보와 비교해 일정 변화율 이상일 때만 저장 - latest.jpg, manifest.json, snapshots/*.jpg 관리 - 최근 keep_count개만 유지 주의: Google Drive API 직접 업로드가 아니라 지정 폴더에 저장한다. 지정 폴더가 Google Drive for Desktop 동기화 폴더이면 클라이언트가 자동 업로드한다. """ from __future__ import annotations import json import threading import time from dataclasses import asdict from datetime import datetime from pathlib import Path from typing import Callable, Optional import mss from PIL import Image, ImageChops, ImageStat from yjm_settings import SnapshotSettings class SnapshotRingWorker: def __init__(self, source_manager, logger, status_callback: Optional[Callable[[str], None]] = None): self.source_manager = source_manager self.logger = logger self.status_callback = status_callback or (lambda _msg: None) self.settings = SnapshotSettings() self.running = False self.thread: Optional[threading.Thread] = None self._lock = threading.Lock() self._last_compare_img: Optional[Image.Image] = None self._last_saved_at = 0.0 self._save_seq = 0 def update_settings(self, settings: SnapshotSettings) -> None: with self._lock: self.settings = settings if settings.enabled: self.start() else: self.stop() def start(self) -> None: if self.running: return self.running = True self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() self._status("AI 스냅샷 ring 시작") def stop(self) -> None: self.running = False self._status("AI 스냅샷 ring 중지") def _status(self, msg: str) -> None: try: self.status_callback(msg) except Exception: pass @staticmethod def _fit_width(img: Image.Image, max_width: int) -> Image.Image: max_width = int(max_width or 0) if max_width > 0 and img.width > max_width: new_h = max(1, int(img.height * (max_width / img.width))) return img.resize((max_width, new_h), Image.Resampling.BILINEAR) return img @staticmethod def _compare_image(img: Image.Image) -> Image.Image: # 변화율 계산용 저해상도 grayscale. 너무 크게 비교하지 않는다. small_w = 240 if img.width > small_w: small_h = max(1, int(img.height * (small_w / img.width))) img = img.resize((small_w, small_h), Image.Resampling.BILINEAR) return img.convert("L") @staticmethod def _change_percent(prev: Image.Image, cur: Image.Image) -> float: if prev.size != cur.size: return 100.0 diff = ImageChops.difference(prev, cur) stat = ImageStat.Stat(diff) # 0~255 평균 차이를 0~100%로 환산. UI 작은 변화도 어느 정도 잡는다. mean = float(stat.mean[0]) if stat.mean else 0.0 return max(0.0, min(100.0, mean * 100.0 / 255.0)) def _run(self) -> None: with mss.mss() as sct: while self.running: with self._lock: st = self.settings try: self._tick(sct, st) except Exception as exc: self.logger.debug("snapshot ring tick failed: %s", exc) time.sleep(1.0) time.sleep(max(0.2, float(st.interval_sec or 1.0))) def _tick(self, sct: mss.mss, st: SnapshotSettings) -> None: if not st.enabled: return target = self.source_manager.get_target() if not target: return shot = sct.grab(target.monitor_dict) img = Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX") img = self._fit_width(img, int(st.max_width)) cmp_img = self._compare_image(img) changed = 100.0 if self._last_compare_img is None else self._change_percent(self._last_compare_img, cmp_img) now = time.time() if self._last_compare_img is not None: if changed < float(st.change_threshold_percent): return if now - self._last_saved_at < float(st.min_save_interval_sec): return self._last_compare_img = cmp_img self._last_saved_at = now self._save_snapshot(img, st, changed, target.title) def _save_snapshot(self, img: Image.Image, st: SnapshotSettings, changed: float, title: str) -> None: base = Path(st.folder).expanduser() if not base.is_absolute(): base = Path.cwd() / base snap_dir = base / "snapshots" snap_dir.mkdir(parents=True, exist_ok=True) self._save_seq += 1 ts = datetime.now().strftime("%Y%m%d_%H%M%S") name = f"{ts}_{self._save_seq:04d}.jpg" path = snap_dir / name latest = base / "latest.jpg" quality = max(20, min(95, int(st.jpeg_quality))) img.save(path, format="JPEG", quality=quality, optimize=False) img.save(latest, format="JPEG", quality=quality, optimize=False) files = sorted(snap_dir.glob("*.jpg"), key=lambda p: p.stat().st_mtime, reverse=True) keep = max(1, int(st.keep_count)) for old in files[keep:]: try: old.unlink() except Exception: pass files = sorted(snap_dir.glob("*.jpg"), key=lambda p: p.stat().st_mtime, reverse=True) manifest = { "updated_at": datetime.now().isoformat(timespec="seconds"), "latest": "latest.jpg", "keep_count": keep, "source_title": title, "changed_percent": round(float(changed), 3), "width": img.width, "height": img.height, "snapshots": [ { "file": f"snapshots/{p.name}", "size": p.stat().st_size, "mtime": datetime.fromtimestamp(p.stat().st_mtime).isoformat(timespec="seconds"), } for p in files ], "settings": asdict(st), } (base / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") self._status(f"AI 스냅샷 저장: {name} / 변화율 {changed:.2f}%")