ilia
6a78c84bcd
- capture/: frame sampling at configurable fps, ffmpeg audio extraction (WAV, whisper-ready) - vision/: changed-pixel motion scoring, stable/moving segmentation, contour + perspective document detection, Laplacian sharpness scoring, optional CLAHE enhancement - pipeline: two-pass detect (motion timeline, then best-frame crop per stable window) writing crops, debug frames, report.json, and motion_scores.csv - CLI: probe / detect / extract-audio subcommands - config.yaml with tunable thresholds; placeholder packages for events, transcribe, ocr, naming, pdf, review_cli - synthetic sample video generator + 16 unit tests
95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
"""Frame-to-frame motion detection and stable/moving segmentation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
|
|
def prepare_gray(image_bgr: np.ndarray, blur_ksize: int = 21) -> np.ndarray:
|
|
"""Grayscale + Gaussian blur, the representation used for differencing."""
|
|
if blur_ksize % 2 == 0:
|
|
blur_ksize += 1
|
|
gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY)
|
|
return cv2.GaussianBlur(gray, (blur_ksize, blur_ksize), 0)
|
|
|
|
|
|
def motion_score(
|
|
prev_gray: np.ndarray, gray: np.ndarray, pixel_threshold: int = 12
|
|
) -> float:
|
|
"""Fraction of pixels (0.0-1.0) whose gray level changed by more than
|
|
pixel_threshold between consecutive frames.
|
|
|
|
A changed-pixel ratio separates localized motion (a hand flipping a page)
|
|
from sensor noise far better than the mean absolute difference, which
|
|
dilutes small moving regions across the whole frame.
|
|
"""
|
|
diff = cv2.absdiff(prev_gray, gray)
|
|
return float(np.count_nonzero(diff > pixel_threshold)) / diff.size
|
|
|
|
|
|
@dataclass
|
|
class Segment:
|
|
kind: str # "stable" | "moving"
|
|
t_start: float
|
|
t_end: float
|
|
# Indices into the sampled score series (inclusive range).
|
|
first_sample: int
|
|
last_sample: int
|
|
|
|
@property
|
|
def duration(self) -> float:
|
|
return self.t_end - self.t_start
|
|
|
|
|
|
def find_segments(
|
|
timestamps: list[float],
|
|
scores: list[float],
|
|
threshold: float,
|
|
stable_min_duration_s: float,
|
|
) -> list[Segment]:
|
|
"""Classify each sample as stable/moving and group into segments.
|
|
|
|
Stable segments shorter than stable_min_duration_s are reclassified as
|
|
moving (they are usually mid-gesture pauses), then adjacent segments of
|
|
the same kind are merged. scores[i] describes the transition into
|
|
timestamps[i], so both lists must be the same length.
|
|
"""
|
|
if len(timestamps) != len(scores):
|
|
raise ValueError("timestamps and scores must have the same length")
|
|
if not timestamps:
|
|
return []
|
|
|
|
labels = ["stable" if s < threshold else "moving" for s in scores]
|
|
|
|
segments: list[Segment] = []
|
|
start = 0
|
|
for i in range(1, len(labels) + 1):
|
|
if i == len(labels) or labels[i] != labels[start]:
|
|
segments.append(
|
|
Segment(
|
|
kind=labels[start],
|
|
t_start=timestamps[start],
|
|
t_end=timestamps[i - 1],
|
|
first_sample=start,
|
|
last_sample=i - 1,
|
|
)
|
|
)
|
|
start = i
|
|
|
|
# Demote too-short stable segments, then merge same-kind neighbors.
|
|
for seg in segments:
|
|
if seg.kind == "stable" and seg.duration < stable_min_duration_s:
|
|
seg.kind = "moving"
|
|
|
|
merged: list[Segment] = []
|
|
for seg in segments:
|
|
if merged and merged[-1].kind == seg.kind:
|
|
merged[-1].t_end = seg.t_end
|
|
merged[-1].last_sample = seg.last_sample
|
|
else:
|
|
merged.append(seg)
|
|
return merged
|