Initial scaffold: capture + vision modules with detect CLI
- 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
This commit is contained in:
@@ -0,0 +1,94 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user