Jinsi ya kupunguza gharama za uchambuzi wa skrini na sampuli ya Keyframe Adaptive

Niliunda chombo changu cha uchambuzi wa skrini kwa sababu moja: nilitaka kitu cha kawaida ambacho kinaweza kuangalia rekodi ya skrini, kuhitimisha kile kilichotokea kama mchakato wa kazi, na kisha kurekebisha ufahamu huo katika vifaa vya automatisering (n8n mtiririko, orodha ya hatua, muhtasari wa muundo, pipeline nzima). Sehemu ya gharama kubwa sio kuzalisha JSON au kuonyesha ripoti. Ni hatua ya ufahamu wa multimodal - kila frame ya ziada unayotuma kwa mfano ni pesa halisi. kumbukumbu ya screen ni usambazaji wa kuingia wa hali mbaya zaidi kwa sampuli ya kiwango cha kudumu: muda mrefu wa UI ya static, kisha micro-bursts ghafla ambapo mtumiaji anapiga, kuandika wahusika wawili, ufungaji wa kushuka, flashes ya modals, au swaps ya tab. Kwa hivyo niliacha kutibu viwambo kama "mahali data" na nilianza kutibu kama bajeti. Nini kilikuwa sahihi kwanza (ukosefu ambao ulimfanya mabadiliko) Kipimo changu cha kwanza kilikuwa cha dhahiri: sample kila Nth frame. Toleo hilo lilishindwa kwa njia mbili, na zote mbili zilikuwa zinaonekana katika bidhaa. Uhariri wa kawaida kwenye dashboard yangu ni rekodi ya dakika 6-12. Katika rekodi nyingi hizi, mtumiaji huacha kufikiri, anaona ukurasa, au cursor tu ameketi huko. sampuli ya usawa kwa furaha huchoma ram juu ya dakika hizo za kitu chochote. uchambuzi wa gharama hupanda linearly na urefu wa video hata wakati maudhui ya habari haina. Failure #1: it wasted frames on dead air. Matatizo mabaya zaidi (yaani yale niliyosababisha kuandika tena sampuli ya sampuli) ilikuwa rekodi ya mfupi ya kazi ya utawala ambapo upungufu wa ruhusa muhimu ulifunguliwa na kufunguliwa haraka. sampuli ya pamoja ilichukua ramani tu kabla na baada ya upungufu - hivyo mfano haukuwahi kuona uchaguzi wa ruhusa. muhtasari wa output ulionekana uhakika na uongo: ilionesha mabadiliko tofauti ya mipangilio kwa sababu iliona hali ya ukurasa inayotokana lakini sio UI ya muda mrefu ambayo ilisababisha. Failure #2: it missed the “blink-and-you-miss-it” UI moments. Mchanganyiko huo - kulipa kwa muda wa static wakati bado kukosa hatua halisi - ulikuwa hauhitaji. nilihitaji sampler ambayo inachukua ramani ambapo video inabadilika na huacha kulipa kodi kwenye UI ya static. Wazo la msingi: kutumia mifumo ambapo taarifa ni Njia ya kipekee ni "sample kila Nth frame." Inaonekana haki. Kumbukumbu ya screen sio sinema. Ni karibu na kitabu: hali ndefu za kudumu zilizotajwa na mabadiliko ya muda mfupi. sampuli ya pamoja inahakikisha matokeo mabaya mawili: Wewe ni juu ya kulipa kwa UI imara. Unafundisha mlipuko mfupi ambapo mtiririko wa kazi kweli hutokea. Moja ya mifano (nitaitumia mara moja na kuendelea): sampuli moja kwa moja ni kama kulipa timu zote bonuses sawa bila kujali athari. Hatua ni kupima mabadiliko kwa bei nafuu, kuunganisha muda katika makundi, na kutenga bajeti ya kifungo imara kwa makundi hayo. Usanifu wa Runtime: Ambapo sampler iko Kwa uendeshaji, mfumo umegawanyika katika vipande viwili vya ushirikiano: Huduma ya uchambuzi wa Python (iliyoandaliwa kama huduma ya Cloud Run) ambayo huchukua video iliyowekwa, huchagua mifumo ya keyframes, huendesha uchambuzi wa multimodal, na huunda matumizi ya matokeo ya muundo. Programu ya Next.js ambayo inapokea matokeo kupitia webhook na inaendelea (na huendesha UI ya dashboard). Sampler anaishi ndani ya analyzer, kabla ya simu ya mfano wa gharama kubwa. Kitu ambacho si dhahiri: sampler sio "utengenezaji mzuri." Ni uso wa udhibiti. Inabadilisha "video hii ni muda gani?" kwa "upelelezi kiasi gani unataka kulipa?" Mchambuzi hutoa matokeo nyuma kwa programu ya Next.js kupitia webhook iliyosajiliwa—HMAC-SHA256 juu ya bytes sahihi za JSON, kuthibitishwa na Mstari wa chini: Roho inawakilisha thamani yako ( ), sio dict ya Python ambayo inaweza kurekodiwa tena na maktaba ya HTTP. Hiyo kutofautiana inazalisha makosa ya ukaguzi wa mara kwa mara ambayo inahisi kama ghosts. Lakini kuunganisha webhook ni chapisho jingine - kile kinachohitajika hapa ni kile kinachotokea Picha hii inachukuliwa na Analyst. timingSafeEqual data=body_bytes kabla ya Adaptive Keyframe Sampling: Score → Sehemu → Allocate → Pick Indices Sampler ni mzunguko wa hatua nne: Mabadiliko ya thamani kwa bei nafuu kwa frame (au kwa hatua ya frame). Weka mstari wa muda katika "kiwango kikubwa cha kudumu" na "kiwango kikubwa cha mabadiliko". Kutoa bajeti ya kifungo cha kifungo juu ya vipengele na guardrails. Chagua viashiria vya mchoro wa betri ndani ya kila sehemu. Muundo huu ni kile kinachomfanya uwe wa vitendo. Tathmini ni ya bei nafuu. Usambazaji ni linear. Usambazaji ni wa utabiri. Hatua ya uchimbaji ni ya kiufundi. Hatua ya 1 — Scoring: mabadiliko ya macho ya bei nafuu Mimi ni dhaifu kuhusu kudumisha alama ya bei nafuu. Ikiwa alama ni gharama kubwa sana, mimi ni tu kuhamisha akaunti mapema katika pipeline. Simu ya msingi ya kuaminika zaidi kwa rekodi za skrini ni : frame difference energy Mabadiliko ya Grayscale. Tazama tofauti kamili kati ya ramani za mfululizo. Kuchukua wastani wa picha ya diff (kwa chaguo normalize). Hii inahusisha: Harakati ya Cursor Kuandika (kuvuta caret na updates ya maandishi) Dropdowns na modals Mabadiliko ya ukurasa mabadiliko ya hali ya hewa Sio kamili, lakini ni haraka na ina uhusiano mzuri na "kila kitu kilichotokea." Hatua ya 2 — Uumbaji wa Segment: Wabadilisha mtiririko wa alama ya sauti katika runs Picha hii ilipigwa tokea ktk veranda za moja ya vyumba vya Manyara Serena Lodge. Hivyo mimi segmenta kwa kutumia mashine rahisi ya hali na hysteresis: Kuhifadhi kiwango cha wastani wa rolling. Mabadiliko kwenye sehemu ya "hot" wakati alama ya rolling inakua juu ya kiwango cha chini. Mabadiliko ya kurudi kwenye "baridi" wakati inaanguka chini ya kiwango cha chini. Kuweka urefu mdogo wa sehemu ili si kuunda mamia ya micro-segments. Hii sio uchunguzi wa mabadiliko ya kitaaluma. Ni uhandisi: tabia ya imara, upungufu wa chini wa tuning, na matokeo ya utabiri. Hatua ya 3 - Usambazaji wa bajeti na vifaa vya kuhifadhi Usambazaji wa kipekee wa kimsingi sio ya kutosha. Hiyo haifai juu ya kupunguzwa na inaweza kuua sehemu fupi. Kwa hiyo, mchanganyiko wangu una sheria tatu: Kila sehemu inapata ardhi (min_frames_per_segment). Hakuna sehemu inayoweza kufikia kiwango cha kiwango (max_frames_per_segment). Bajeti iliyobaki hutolewa kwa kiwango cha thamani kwa sekta ya utumiaji. Hii hufanya usambazaji wa imara na kuzuia ugonjwa. Hatua ya 4 - Uchaguzi wa index ndani ya sehemu Mara baada ya sehemu imepewa ramani za K, napenda kuchagua viashiria vya K vilivyoenea katika sehemu hii: Daima ikiwa ni pamoja na mwanzo wa sehemu (msingi wa mabadiliko). Daima kuingiza mwisho wa sehemu (maana ya hali ya mwisho). Kujaza sehemu nyingine kwa vichwa vingi. Ikiwa ninahitaji uaminifu zaidi baadaye, ninaweza kuingilia msimamo kwa kiwango cha ndani cha alama, lakini kuchagua kwa usawa ni msingi mzuri na inachukua msimamo wa moja kwa moja. Utekelezaji kamili wa uendeshaji (tathmini + mchanganyiko + usambazaji + uchimbaji) Ili kufanya chapisho hili iwezekanavyo, hapa ni script moja ya Python ambayo unaweza kuendesha dhidi ya MP4 yeyote. Utegemezi wa: Maelezo ya Python Nume ya Kuendesha Run: python adaptive_keyframes.py --video input.mp4 --budget 60 --out ./keyframes Hapa ni : adaptive_keyframes.py import argparse import os from dataclasses import dataclass from typing import List, Tuple import cv2 import numpy as np @dataclass class Segment: start: int # inclusive frame index end: int # exclusive frame index score: float @property def length(self) -> int: return max(0, self.end - self.start) def frame_diff_score(prev_bgr: np.ndarray, curr_bgr: np.ndarray) -> float: """Cheap per-frame change score in [0, 1] (roughly). Uses grayscale mean absolute difference normalized by 255. """ prev_gray = cv2.cvtColor(prev_bgr, cv2.COLOR_BGR2GRAY) curr_gray = cv2.cvtColor(curr_bgr, cv2.COLOR_BGR2GRAY) diff = cv2.absdiff(prev_gray, curr_gray) return float(diff.mean() / 255.0) def compute_scores( cap: cv2.VideoCapture, stride: int = 1, max_frames: int | None = None, ) -> Tuple[List[float], int]: """Return (scores, total_frames_read). scores[i] is the change score between frame i and i+stride (based on sampled reads). """ scores: List[float] = [] ok, prev = cap.read() if not ok: return scores, 0 frame_idx = 1 frames_read = 1 while True: # Skip stride-1 frames between comparisons. for _ in range(stride - 1): ok = cap.grab() if not ok: return scores, frames_read frame_idx += 1 frames_read += 1 if max_frames is not None and frames_read >= max_frames: return scores, frames_read ok, curr = cap.read() if not ok: return scores, frames_read frames_read += 1 s = frame_diff_score(prev, curr) scores.append(s) prev = curr frame_idx += 1 if max_frames is not None and frames_read >= max_frames: return scores, frames_read def segment_scores( scores: List[float], window: int = 8, hot_thresh: float = 0.030, cold_thresh: float = 0.020, min_len: int = 12, ) -> List[Segment]: """Convert per-step scores into segments with a utility score. Uses a rolling mean with hysteresis to avoid segment flicker. """ if not scores: return [] # Rolling mean via cumulative sum. x = np.array(scores, dtype=np.float32) c = np.cumsum(np.insert(x, 0, 0.0)) def roll_mean(i: int) -> float: j0 = max(0, i - window + 1) n = i - j0 + 1 return float((c[i + 1] - c[j0]) / n) segments: List[Segment] = [] state_hot = False seg_start = 0 seg_scores: List[float] = [] for i in range(len(scores)): rm = roll_mean(i) if state_hot: seg_scores.append(scores[i]) if rm < cold_thresh: # Close hot segment at i+1 seg_end = i + 1 if seg_end - seg_start hot_thresh: # Close cold segment seg_end = i + 1 cold_score = float(np.mean(scores[seg_start:seg_end]) if seg_end > seg_start else 0.0) segments.append(Segment(seg_start, seg_end, cold_score)) # Start hot state_hot = True seg_start = seg_end seg_scores = [] # Close tail tail_end = len(scores) if tail_end > seg_start: tail_score = float(np.mean(scores[seg_start:tail_end])) segments.append(Segment(seg_start, tail_end, tail_score)) # Merge very short segments to keep output stable. merged: List[Segment] = [] for seg in segments: if not merged: merged.append(seg) continue if seg.length < min_len: prev = merged[-1] combined = Segment(prev.start, seg.end, (prev.score * prev.length + seg.score * seg.length) / max(1, (prev.length + seg.length))) merged[-1] = combined else: merged.append(seg) # One more pass: ensure non-empty and strictly increasing. cleaned: List[Segment] = [] for seg in merged: if seg.length 0: cleaned.append(seg) return cleaned def allocate_frames( segments: List[Segment], budget: int, min_frames_per_segment: int = 1, max_frames_per_segment: int = 30, ) -> List[int]: """Allocate keyframes to segments using floor + proportional + cap.""" if budget = budget: alloc = [0] * n for i in range(budget): alloc[i % n] += 1 return alloc utilities = np.array([max(0.0, s.score) for s in segments], dtype=np.float64) total_u = float(utilities.sum()) alloc = [min_frames_per_segment] * n remaining = budget - min_total if total_u == 0.0: raw = np.full(n, remaining / n, dtype=np.float64) else: raw = utilities * (remaining / total_u) # Add integer parts. for i in range(n): add = int(raw[i]) alloc[i] = min(max_frames_per_segment, alloc[i] + add) allocated = sum(alloc) # Distribute leftover by fractional parts, respecting caps. if allocated < budget: frac = raw - np.floor(raw) order = np.argsort(-frac) # descending fractional idx = 0 safety = 0 while allocated budget: order = np.argsort(utilities) # ascending utility idx = 0 safety = 0 while allocated > budget and safety 0 and alloc[i] > min_frames_per_segment: alloc[i] -= 1 allocated -= 1 idx += 1 safety += 1 return alloc def pick_indices_for_segment(seg: Segment, k: int) -> List[int]: """Pick k indices in [seg.start, seg.end] over the score-step domain. Note: scores are defined between frames; we later map these to actual frames. """ if k <= 0 or seg.length <= 0: return [] if k == 1: return [seg.start] # Evenly spaced across [start, end-1] xs = np.linspace(seg.start, seg.end - 1, num=k) idxs = sorted({int(round(x)) for x in xs}) # Ensure exactly k by filling gaps if rounding collapsed points. while len(idxs) = 2: candidates.append((a + b) // 2) if not candidates: # Fall back: walk forward. x = idxs[-1] if x + 1 = k: break idxs = sorted(idxs) # Trim if we overshot. return idxs[:k] def select_keyframe_indices(segments: List[Segment], alloc: List[int], stride: int = 1) -> List[int]: """Return concrete frame indices (0-based) to extract from the video.""" chosen: List[int] = [] for seg, k in zip(segments, alloc): step_idxs = pick_indices_for_segment(seg, k) # Map score-step domain to frame indices. # score i corresponds to diff between frame i and i+stride; # picking frame i is a reasonable representative. for si in step_idxs: chosen.append(si * stride) chosen = sorted(set(chosen)) return chosen def extract_frames(video_path: str, frame_indices: List[int], out_dir: str) -> None: os.makedirs(out_dir, exist_ok=True) cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise RuntimeError(f"Failed to open video: {video_path}") frame_set = set(frame_indices) max_idx = max(frame_set) if frame_set else -1 idx = 0 saved = 0 while idx <= max_idx: ok, frame = cap.read() if not ok: break if idx in frame_set: path = os.path.join(out_dir, f"frame_{idx:06d}.jpg") ok2 = cv2.imwrite(path, frame) if not ok2: raise RuntimeError(f"Failed to write: {path}") saved += 1 idx += 1 cap.release() if saved == 0 and frame_indices: raise RuntimeError("No frames were saved; check indices and video decoding") def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--video", required=True, help="Path to input video") ap.add_argument("--out", required=True, help="Output directory for keyframes") ap.add_argument("--budget", type=int, default=60, help="Total keyframes to extract") ap.add_argument("--stride", type=int, default=2, help="Compare every Nth frame for scoring") ap.add_argument("--window", type=int, default=8, help="Rolling window for segmentation") ap.add_argument("--hot", type=float, default=0.030, help="Enter hot segment threshold") ap.add_argument("--cold", type=float, default=0.020, help="Exit hot segment threshold") args = ap.parse_args() cap = cv2.VideoCapture(args.video) if not cap.isOpened(): raise RuntimeError(f"Failed to open video: {args.video}") scores, frames_read = compute_scores(cap, stride=args.stride) cap.release() segments = segment_scores(scores, window=args.window, hot_thresh=args.hot, cold_thresh=args.cold) alloc = allocate_frames(segments, budget=args.budget, min_frames_per_segment=1, max_frames_per_segment=max(2, args.budget)) keyframes = select_keyframe_indices(segments, alloc, stride=args.stride) # Keep within a hard limit (rounding/uniqueness can change count). if len(keyframes) > args.budget: keyframes = keyframes[: args.budget] extract_frames(args.video, keyframes, args.out) print(f"frames_read={frames_read}") print(f"scores={len(scores)} segments={len(segments)}") print(f"budget={args.budget} selected={len(keyframes)}") if segments: hot_share = sum(1 for s in segments if s.score > args.hot) / len(segments) print(f"segment_hot_share={hot_share:.2f}") if __name__ == "__main__": main() Kifungu hiki ni kwa makusudi moja kwa moja: Hii inafanya orodha ya viwango vya kudumu. Hii inahakikisha kwamba kamwe huna zaidi ya bajeti yako. Inaandika faili za frame deterministic kwa uchambuzi wa chini. Katika huduma yangu ya uzalishaji, viwambo vilivyochapishwa vinahifadhiwa kwenye wito wa multimodal (Gemini kupitia katika utegemezi wangu), na faida ya uchambuzi inayotokana inapelekwa nyuma kwa programu ya Next.js kupitia webhook iliyosajiliwa. google-generativeai Maoni ya maonyesho ya vitendo (chakula ambacho kina umuhimu baada ya demo ya kwanza) Mara baada ya kuwa na fomu ya kazi, faida hutoka kwa tabia ya tuning chini ya rekodi mbaya ya dunia halisi. 1) Chagua hatua ambayo inafanana na maudhui yako Ikiwa unashughulikia kila picha kwenye rekodi ya 30 FPS, utapata harakati ndogo za cursor kila mahali. Hiyo sio mbaya daima, lakini inaweza kuimarisha ishara yako ya "mabadiliko". Hatua ya 2–5 ni hatua nzuri ya kuanza kwa rekodi za screen. Wewe bado unaathiri mabadiliko ya UI, lakini unaweza kuzuia sauti ya sub-frame. 2) Hysteresis kuzuia flipping sehemu Kutumia na ya Pamoja na kiwango kimoja, utashuka kati ya joto / baridi mara kwa mara karibu na kukata. hot_thresh cold_thresh Hysteresis inakupa sehemu za imara na hufanya bajeti kuendesha kwa utabiri. 3) Ardhi na vifungo sio chaguo Bila ardhi, vipande vidogo vinaweza kupunguzwa hadi nusu na utakuwa ukipoteza micro-burst sahihi uliyopenda. Bila kifungo, sehemu moja ya muda mrefu ya "kuhudumia" inaweza kuchukua bajeti yako yote na utapoteza mazingira kutoka kwa mchakato wa kazi. 4) Daima kuandika bajeti unayotuma Kama wewe kuandika Tuma kwa , maktaba ya HTTP inaweza serialize na utaratibu tofauti wa kichwa / nafasi kuliko kile ulichosajiliwa. ambayo hutoa makosa ya ukaguzi wa mara kwa mara ambayo yanaonekana kama ghosts. json.dumps(payload) json=payload Kuanzisha mlinzi juu ya msalaba ( ) hupunguza darasa hilo la makosa. Hii ni muhimu hapa kwa sababu faida ya uchambuzi - moja ambayo inahifadhi nyayo zako zilizochaguliwa kwa uangalifu - ni artefact ya gharama kubwa zaidi katika pipeline. Ikiwa inachukuliwa na kushindwa kwa saini, umepoteza bajeti yote ya ram. data=body_bytes Kwa nini muundo huu wa viwango katika uzalishaji Sampler inafanya kazi kwa sababu inaheshimu vikwazo viwili vigumu: Kiwango cha gharama na ramani zilizochaguliwa, sio urefu wa video. Mara baada ya bajeti imewekwa, hatua ya gharama kubwa ya multimodal ina kiwango cha juu. Tarehe ya uendeshaji inaendelea kutabiriwa. Scoring ni kupita linear; mchanganyiko ni kupita linear; usambazaji ni linear na vigezo vidogo vya kudumu. Jambo muhimu zaidi: sampler hufanya mfumo wangu kutenda kama kitu ninachoweza kuendesha. Badala ya kujadili kama video ya dakika ya 12 ni "kwa muda mrefu sana," niliamua ni kiasi gani cha picha ninapenda kununua. Hiyo ndiyo sababu ya michezo ya kompyuta ya mbwa ajabu hii ni katika mafanikio makubwa kama hayo.