nvti commited on
Commit
3a975fb
·
1 Parent(s): c97f76b

feat(video): add intake quality gate

Browse files
README.md CHANGED
@@ -91,6 +91,23 @@ Each analysis run creates a folder under `runs/<run_id>/`:
91
  mock mode. JSON artifacts are validated before they are written, including required fields, supported
92
  enum values, score ranges, frame/timestamp ordering, and final report shape.
93
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
94
  `annotated_video.mp4` is not implemented yet. The mocked renderer currently returns the original input video path and writes `annotated_video_placeholder.json`.
95
 
96
  ## Replacing Mock Steps
 
91
  mock mode. JSON artifacts are validated before they are written, including required fields, supported
92
  enum values, score ranges, frame/timestamp ordering, and final report shape.
93
 
94
+ `video_manifest.json` is produced from OpenCV metadata and sampled-frame quality checks. It includes
95
+ FPS, duration, frame count, sampled frame count, resolution, codec/container when available,
96
+ brightness, blur score, warning labels, and `analysis_allowed`.
97
+
98
+ Video quality warning labels:
99
+
100
+ - `too_short`
101
+ - `too_long`
102
+ - `too_dark`
103
+ - `too_blurry`
104
+ - `fps_too_low`
105
+ - `resolution_too_low`
106
+ - `video_decode_failed`
107
+
108
+ Decode failures set `analysis_allowed=false`; the Gradio UI surfaces capture guidance instead of
109
+ coach-style feedback when analysis is blocked.
110
+
111
  `annotated_video.mp4` is not implemented yet. The mocked renderer currently returns the original input video path and writes `annotated_video_placeholder.json`.
112
 
113
  ## Replacing Mock Steps
app.py CHANGED
@@ -12,10 +12,40 @@ sys.path.insert(0, str(Path(__file__).parent / "src"))
12
  from pozify.pipeline import run_pipeline
13
 
14
 
 
 
 
 
 
 
 
 
 
 
 
15
  def _pretty_json(value: dict[str, Any]) -> str:
16
  return json.dumps(value, ensure_ascii=False, indent=2)
17
 
18
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
19
  def analyze_video(
20
  video_path: str | None,
21
  goal: str,
@@ -38,7 +68,24 @@ def analyze_video(
38
  )
39
 
40
  report = result["final_report"]
 
41
  summary = report["coach_summary"]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
42
  markdown = f"""## Scan Summary
43
 
44
  - **Exercise:** {report["exercise"]["exercise"]} ({report["exercise"]["confidence"]:.0%} confidence)
@@ -59,6 +106,8 @@ def analyze_video(
59
 
60
  ### Next Session Plan
61
  {chr(10).join(f"- {item}" for item in summary["next_session_plan"])}
 
 
62
  """
63
 
64
  artifact_path = Path(result["run_dir"]) / "final_report.json"
 
12
  from pozify.pipeline import run_pipeline
13
 
14
 
15
+ QUALITY_GUIDANCE = {
16
+ "too_short": "Record at least 10 seconds so the set contains enough movement context.",
17
+ "too_long": "Keep the clip under 60 seconds for the MVP analyzer.",
18
+ "too_dark": "Use brighter, even lighting and keep the body visible against the background.",
19
+ "too_blurry": "Stabilize the camera and avoid fast panning or heavy motion blur.",
20
+ "fps_too_low": "Use a camera mode with at least 15 FPS.",
21
+ "resolution_too_low": "Record at 480x360 or higher so joint positions are readable.",
22
+ "video_decode_failed": "Upload a playable video file; the current file could not be decoded.",
23
+ }
24
+
25
+
26
  def _pretty_json(value: dict[str, Any]) -> str:
27
  return json.dumps(value, ensure_ascii=False, indent=2)
28
 
29
 
30
+ def _quality_markdown(video_manifest: dict[str, Any]) -> str:
31
+ warnings = video_manifest["quality_warnings"]
32
+ if not warnings:
33
+ return "## Video Quality\n\nNo quality warnings detected."
34
+
35
+ warning_items = "\n".join(f"- `{warning}`: {QUALITY_GUIDANCE[warning]}" for warning in warnings)
36
+ status = (
37
+ "Analysis is blocked until the video can be decoded reliably."
38
+ if not video_manifest["analysis_allowed"]
39
+ else "Analysis completed, but capture quality may affect downstream feedback."
40
+ )
41
+ return f"""## Video Quality
42
+
43
+ {status}
44
+
45
+ {warning_items}
46
+ """
47
+
48
+
49
  def analyze_video(
50
  video_path: str | None,
51
  goal: str,
 
68
  )
69
 
70
  report = result["final_report"]
71
+ video_quality = _quality_markdown(report["video_manifest"])
72
  summary = report["coach_summary"]
73
+ if not report["video_manifest"]["analysis_allowed"]:
74
+ markdown = f"""{video_quality}
75
+
76
+ ## Run
77
+
78
+ - **Run ID:** `{report["run_id"]}`
79
+ - **Saved report:** `{Path(result["run_dir"]) / "final_report.json"}`
80
+ """
81
+ artifact_path = Path(result["run_dir"]) / "final_report.json"
82
+ return (
83
+ result["annotated_video_path"],
84
+ markdown,
85
+ _pretty_json(report),
86
+ str(artifact_path),
87
+ )
88
+
89
  markdown = f"""## Scan Summary
90
 
91
  - **Exercise:** {report["exercise"]["exercise"]} ({report["exercise"]["confidence"]:.0%} confidence)
 
106
 
107
  ### Next Session Plan
108
  {chr(10).join(f"- {item}" for item in summary["next_session_plan"])}
109
+
110
+ {video_quality}
111
  """
112
 
113
  artifact_path = Path(result["run_dir"]) / "final_report.json"
pyproject.toml CHANGED
@@ -6,6 +6,8 @@ readme = "README.md"
6
  requires-python = ">=3.11"
7
  dependencies = [
8
  "gradio>=4.44.0",
 
 
9
  ]
10
 
11
  [project.optional-dependencies]
 
6
  requires-python = ">=3.11"
7
  dependencies = [
8
  "gradio>=4.44.0",
9
+ "numpy>=1.26.0",
10
+ "opencv-python-headless>=4.10.0",
11
  ]
12
 
13
  [project.optional-dependencies]
src/pozify/contracts.py CHANGED
@@ -34,6 +34,12 @@ class VideoManifest:
34
  duration_sec: float
35
  total_frames: int
36
  sampled_frames: int
 
 
 
 
 
 
37
  quality_warnings: list[str]
38
  analysis_allowed: bool
39
 
@@ -273,6 +279,12 @@ def _validate_video_manifest(value: Any, path: str) -> None:
273
  "duration_sec",
274
  "total_frames",
275
  "sampled_frames",
 
 
 
 
 
 
276
  "quality_warnings",
277
  "analysis_allowed",
278
  },
@@ -286,6 +298,16 @@ def _validate_video_manifest(value: Any, path: str) -> None:
286
  _require_int(payload["sampled_frames"], f"{path}.sampled_frames", minimum=0)
287
  if payload["sampled_frames"] > payload["total_frames"]:
288
  raise ContractValidationError(f"{path}.sampled_frames must be <= total_frames")
 
 
 
 
 
 
 
 
 
 
289
  _require_string_list(payload["quality_warnings"], f"{path}.quality_warnings")
290
  _require_bool(payload["analysis_allowed"], f"{path}.analysis_allowed")
291
 
 
34
  duration_sec: float
35
  total_frames: int
36
  sampled_frames: int
37
+ width: int
38
+ height: int
39
+ codec: str | None
40
+ container: str | None
41
+ brightness_mean: float | None
42
+ blur_laplacian_var: float | None
43
  quality_warnings: list[str]
44
  analysis_allowed: bool
45
 
 
279
  "duration_sec",
280
  "total_frames",
281
  "sampled_frames",
282
+ "width",
283
+ "height",
284
+ "codec",
285
+ "container",
286
+ "brightness_mean",
287
+ "blur_laplacian_var",
288
  "quality_warnings",
289
  "analysis_allowed",
290
  },
 
298
  _require_int(payload["sampled_frames"], f"{path}.sampled_frames", minimum=0)
299
  if payload["sampled_frames"] > payload["total_frames"]:
300
  raise ContractValidationError(f"{path}.sampled_frames must be <= total_frames")
301
+ _require_int(payload["width"], f"{path}.width", minimum=0)
302
+ _require_int(payload["height"], f"{path}.height", minimum=0)
303
+ if payload["codec"] is not None:
304
+ _require_type(payload["codec"], str, f"{path}.codec")
305
+ if payload["container"] is not None:
306
+ _require_type(payload["container"], str, f"{path}.container")
307
+ if payload["brightness_mean"] is not None:
308
+ _require_number(payload["brightness_mean"], f"{path}.brightness_mean", minimum=0)
309
+ if payload["blur_laplacian_var"] is not None:
310
+ _require_number(payload["blur_laplacian_var"], f"{path}.blur_laplacian_var", minimum=0)
311
  _require_string_list(payload["quality_warnings"], f"{path}.quality_warnings")
312
  _require_bool(payload["analysis_allowed"], f"{path}.analysis_allowed")
313
 
src/pozify/steps/pose_landmarker.py CHANGED
@@ -1,6 +1,7 @@
1
  from __future__ import annotations
2
 
3
  from pozify.contracts import PoseFrame, PoseSequence, VideoManifest
 
4
 
5
 
6
  LANDMARK_NAMES = [
@@ -38,8 +39,7 @@ def _mock_landmarks(frame_index: int) -> dict[str, dict[str, float]]:
38
 
39
  def run(manifest: VideoManifest) -> PoseSequence:
40
  frames: list[PoseFrame] = []
41
- step = max(1, int(manifest.total_frames / 12))
42
- for frame_index in range(0, manifest.total_frames, step):
43
  frames.append(
44
  PoseFrame(
45
  frame_index=frame_index,
@@ -60,4 +60,3 @@ def run(manifest: VideoManifest) -> PoseSequence:
60
  smoothing_method="none",
61
  pose_valid_ratio=1.0,
62
  )
63
-
 
1
  from __future__ import annotations
2
 
3
  from pozify.contracts import PoseFrame, PoseSequence, VideoManifest
4
+ from pozify.steps.video_qc import sample_frame_indices
5
 
6
 
7
  LANDMARK_NAMES = [
 
39
 
40
  def run(manifest: VideoManifest) -> PoseSequence:
41
  frames: list[PoseFrame] = []
42
+ for frame_index in sample_frame_indices(manifest.total_frames):
 
43
  frames.append(
44
  PoseFrame(
45
  frame_index=frame_index,
 
60
  smoothing_method="none",
61
  pose_valid_ratio=1.0,
62
  )
 
src/pozify/steps/video_qc.py CHANGED
@@ -1,24 +1,156 @@
1
  from __future__ import annotations
2
 
3
  from pathlib import Path
 
 
 
4
 
5
  from pozify.contracts import VideoManifest
6
 
7
 
8
- def run(video_path: str | None) -> VideoManifest:
9
- warnings: list[str] = []
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
10
  if video_path is None:
11
- warnings.append("no_video_uploaded_mock_mode")
12
- elif not Path(video_path).exists():
13
- warnings.append("video_path_not_found_mock_mode")
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
14
 
 
15
  return VideoManifest(
16
  video_path=video_path,
17
- fps=30.0,
18
- duration_sec=12.0,
19
- total_frames=360,
20
- sampled_frames=180,
 
 
 
 
 
 
21
  quality_warnings=warnings,
22
- analysis_allowed=True,
23
  )
24
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
  from __future__ import annotations
2
 
3
  from pathlib import Path
4
+ from typing import Iterator
5
+
6
+ import cv2
7
 
8
  from pozify.contracts import VideoManifest
9
 
10
 
11
+ MIN_DURATION_SEC = 10.0
12
+ MAX_DURATION_SEC = 60.0
13
+ MIN_FPS = 15.0
14
+ MIN_WIDTH = 480
15
+ MIN_HEIGHT = 360
16
+ MIN_BRIGHTNESS = 45.0
17
+ MIN_BLUR_LAPLACIAN_VAR = 50.0
18
+ DEFAULT_SAMPLE_COUNT = 12
19
+
20
+ HARD_FAILURE_WARNINGS = {"video_decode_failed"}
21
+
22
+
23
+ def _decode_fourcc(value: float) -> str | None:
24
+ code = int(value)
25
+ if code <= 0:
26
+ return None
27
+ chars = [chr((code >> 8 * index) & 0xFF) for index in range(4)]
28
+ decoded = "".join(chars).strip()
29
+ return decoded or None
30
+
31
+
32
+ def _container_from_path(video_path: str | None) -> str | None:
33
  if video_path is None:
34
+ return None
35
+ suffix = Path(video_path).suffix.lower().lstrip(".")
36
+ return suffix or None
37
+
38
+
39
+ def sample_frame_indices(total_frames: int, sample_count: int = DEFAULT_SAMPLE_COUNT) -> list[int]:
40
+ if total_frames <= 0 or sample_count <= 0:
41
+ return []
42
+ if total_frames <= sample_count:
43
+ return list(range(total_frames))
44
+
45
+ last_index = total_frames - 1
46
+ return sorted(
47
+ {
48
+ round(index * last_index / (sample_count - 1))
49
+ for index in range(sample_count)
50
+ }
51
+ )
52
+
53
+
54
+ def sample_video_frames(
55
+ video_path: str,
56
+ sample_count: int = DEFAULT_SAMPLE_COUNT,
57
+ ) -> Iterator[tuple[int, object]]:
58
+ capture = cv2.VideoCapture(video_path)
59
+ try:
60
+ if not capture.isOpened():
61
+ return
62
+
63
+ total_frames = int(capture.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
64
+ for frame_index in sample_frame_indices(total_frames, sample_count):
65
+ capture.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
66
+ ok, frame = capture.read()
67
+ if ok and frame is not None:
68
+ yield frame_index, frame
69
+ finally:
70
+ capture.release()
71
+
72
 
73
+ def _empty_manifest(video_path: str | None, warnings: list[str]) -> VideoManifest:
74
  return VideoManifest(
75
  video_path=video_path,
76
+ fps=0.0,
77
+ duration_sec=0.0,
78
+ total_frames=0,
79
+ sampled_frames=0,
80
+ width=0,
81
+ height=0,
82
+ codec=None,
83
+ container=_container_from_path(video_path),
84
+ brightness_mean=None,
85
+ blur_laplacian_var=None,
86
  quality_warnings=warnings,
87
+ analysis_allowed=False,
88
  )
89
 
90
+
91
+ def _brightness_and_blur(frame: object) -> tuple[float, float]:
92
+ gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
93
+ brightness = float(gray.mean())
94
+ blur = float(cv2.Laplacian(gray, cv2.CV_64F).var())
95
+ return brightness, blur
96
+
97
+
98
+ def run(video_path: str | None) -> VideoManifest:
99
+ if video_path is None or not Path(video_path).exists():
100
+ return _empty_manifest(video_path, ["video_decode_failed"])
101
+
102
+ capture = cv2.VideoCapture(video_path)
103
+ try:
104
+ if not capture.isOpened():
105
+ return _empty_manifest(video_path, ["video_decode_failed"])
106
+
107
+ fps = float(capture.get(cv2.CAP_PROP_FPS) or 0.0)
108
+ total_frames = int(capture.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
109
+ width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
110
+ height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
111
+ codec = _decode_fourcc(capture.get(cv2.CAP_PROP_FOURCC) or 0.0)
112
+ finally:
113
+ capture.release()
114
+
115
+ if total_frames <= 0 or fps <= 0:
116
+ return _empty_manifest(video_path, ["video_decode_failed"])
117
+
118
+ sampled_metrics = [_brightness_and_blur(frame) for _, frame in sample_video_frames(video_path)]
119
+ if not sampled_metrics:
120
+ return _empty_manifest(video_path, ["video_decode_failed"])
121
+
122
+ brightness_values = [brightness for brightness, _ in sampled_metrics]
123
+ blur_values = [blur for _, blur in sampled_metrics]
124
+ brightness_mean = round(sum(brightness_values) / len(brightness_values), 2)
125
+ blur_laplacian_var = round(sum(blur_values) / len(blur_values), 2)
126
+ duration_sec = round(total_frames / fps, 3)
127
+
128
+ warnings: list[str] = []
129
+ if duration_sec < MIN_DURATION_SEC:
130
+ warnings.append("too_short")
131
+ if duration_sec > MAX_DURATION_SEC:
132
+ warnings.append("too_long")
133
+ if brightness_mean < MIN_BRIGHTNESS:
134
+ warnings.append("too_dark")
135
+ if blur_laplacian_var < MIN_BLUR_LAPLACIAN_VAR:
136
+ warnings.append("too_blurry")
137
+ if fps < MIN_FPS:
138
+ warnings.append("fps_too_low")
139
+ if width < MIN_WIDTH or height < MIN_HEIGHT:
140
+ warnings.append("resolution_too_low")
141
+
142
+ return VideoManifest(
143
+ video_path=video_path,
144
+ fps=round(fps, 3),
145
+ duration_sec=duration_sec,
146
+ total_frames=total_frames,
147
+ sampled_frames=len(sampled_metrics),
148
+ width=width,
149
+ height=height,
150
+ codec=codec,
151
+ container=_container_from_path(video_path),
152
+ brightness_mean=brightness_mean,
153
+ blur_laplacian_var=blur_laplacian_var,
154
+ quality_warnings=warnings,
155
+ analysis_allowed=not any(warning in HARD_FAILURE_WARNINGS for warning in warnings),
156
+ )
tests/test_pipeline_contracts.py CHANGED
@@ -6,6 +6,9 @@ import tempfile
6
  import unittest
7
  from pathlib import Path
8
 
 
 
 
9
  sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
10
 
11
  from pozify import pipeline
@@ -32,12 +35,18 @@ EXPECTED_ARTIFACT_KEYS = {
32
  ],
33
  "video_manifest.json": [
34
  "analysis_allowed",
 
 
 
 
35
  "duration_sec",
36
  "fps",
 
37
  "quality_warnings",
38
  "sampled_frames",
39
  "total_frames",
40
  "video_path",
 
41
  ],
42
  "pose_sequence.json": ["frames", "normalized", "pose_valid_ratio", "smoothing_method"],
43
  "exercise_classification.json": [
@@ -87,6 +96,32 @@ class PipelineContractTests(unittest.TestCase):
87
  pipeline.RUNS_DIR = self.original_runs_dir
88
  self.temp_dir.cleanup()
89
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
90
  def _assert_pipeline_artifacts(self, result: dict[str, object]) -> None:
91
  run_dir = Path(str(result["run_dir"]))
92
  manifest_path = run_dir / "manifest.json"
@@ -124,16 +159,20 @@ class PipelineContractTests(unittest.TestCase):
124
  self._assert_pipeline_artifacts(result)
125
  report = result["final_report"]
126
  self.assertEqual(report["exercise"]["exercise"], "push_up")
127
- self.assertEqual(report["video_manifest"]["quality_warnings"], ["no_video_uploaded_mock_mode"])
 
128
 
129
  def test_pipeline_runs_end_to_end_with_fixture_video_path(self) -> None:
130
- fixture = Path(__file__).parent / "fixtures" / "sample.mp4"
131
  result = pipeline.run_pipeline(video_path=str(fixture), profile_input=PROFILE_INPUT, mock=True)
132
 
133
  self._assert_pipeline_artifacts(result)
134
  report = result["final_report"]
135
  self.assertEqual(report["video_manifest"]["video_path"], str(fixture))
136
  self.assertEqual(report["video_manifest"]["quality_warnings"], [])
 
 
 
137
 
138
  def test_contract_validation_rejects_missing_required_field(self) -> None:
139
  payload = {
 
6
  import unittest
7
  from pathlib import Path
8
 
9
+ import cv2
10
+ import numpy as np
11
+
12
  sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
13
 
14
  from pozify import pipeline
 
35
  ],
36
  "video_manifest.json": [
37
  "analysis_allowed",
38
+ "blur_laplacian_var",
39
+ "brightness_mean",
40
+ "codec",
41
+ "container",
42
  "duration_sec",
43
  "fps",
44
+ "height",
45
  "quality_warnings",
46
  "sampled_frames",
47
  "total_frames",
48
  "video_path",
49
+ "width",
50
  ],
51
  "pose_sequence.json": ["frames", "normalized", "pose_valid_ratio", "smoothing_method"],
52
  "exercise_classification.json": [
 
96
  pipeline.RUNS_DIR = self.original_runs_dir
97
  self.temp_dir.cleanup()
98
 
99
+ def _write_video(
100
+ self,
101
+ filename: str,
102
+ *,
103
+ fps: float = 30.0,
104
+ duration_sec: float = 10.0,
105
+ size: tuple[int, int] = (640, 480),
106
+ ) -> Path:
107
+ path = Path(self.temp_dir.name) / filename
108
+ writer = cv2.VideoWriter(
109
+ str(path),
110
+ cv2.VideoWriter_fourcc(*"mp4v"),
111
+ fps,
112
+ size,
113
+ )
114
+ self.assertTrue(writer.isOpened())
115
+ width, height = size
116
+ for frame_index in range(int(fps * duration_sec)):
117
+ frame = np.full((height, width, 3), 130, dtype=np.uint8)
118
+ offset = frame_index % 120
119
+ cv2.rectangle(frame, (40 + offset, 80), (260 + offset, 300), (245, 245, 245), -1)
120
+ cv2.line(frame, (0, frame_index % height), (width - 1, height - 1), (20, 20, 20), 3)
121
+ writer.write(frame)
122
+ writer.release()
123
+ return path
124
+
125
  def _assert_pipeline_artifacts(self, result: dict[str, object]) -> None:
126
  run_dir = Path(str(result["run_dir"]))
127
  manifest_path = run_dir / "manifest.json"
 
159
  self._assert_pipeline_artifacts(result)
160
  report = result["final_report"]
161
  self.assertEqual(report["exercise"]["exercise"], "push_up")
162
+ self.assertEqual(report["video_manifest"]["quality_warnings"], ["video_decode_failed"])
163
+ self.assertFalse(report["video_manifest"]["analysis_allowed"])
164
 
165
  def test_pipeline_runs_end_to_end_with_fixture_video_path(self) -> None:
166
+ fixture = self._write_video("sample.mp4")
167
  result = pipeline.run_pipeline(video_path=str(fixture), profile_input=PROFILE_INPUT, mock=True)
168
 
169
  self._assert_pipeline_artifacts(result)
170
  report = result["final_report"]
171
  self.assertEqual(report["video_manifest"]["video_path"], str(fixture))
172
  self.assertEqual(report["video_manifest"]["quality_warnings"], [])
173
+ self.assertTrue(report["video_manifest"]["analysis_allowed"])
174
+ self.assertEqual(report["video_manifest"]["width"], 640)
175
+ self.assertEqual(report["video_manifest"]["height"], 480)
176
 
177
  def test_contract_validation_rejects_missing_required_field(self) -> None:
178
  payload = {
tests/test_video_qc.py ADDED
@@ -0,0 +1,145 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
+ import tempfile
4
+ import unittest
5
+ from pathlib import Path
6
+
7
+ import cv2
8
+ import numpy as np
9
+
10
+ import sys
11
+
12
+ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
13
+
14
+ from pozify.steps import video_qc
15
+
16
+
17
+ class VideoQCTests(unittest.TestCase):
18
+ def setUp(self) -> None:
19
+ self.temp_dir = tempfile.TemporaryDirectory()
20
+
21
+ def tearDown(self) -> None:
22
+ self.temp_dir.cleanup()
23
+
24
+ def _write_video(
25
+ self,
26
+ filename: str,
27
+ *,
28
+ fps: float = 30.0,
29
+ duration_sec: float = 10.0,
30
+ size: tuple[int, int] = (640, 480),
31
+ mode: str = "valid",
32
+ ) -> Path:
33
+ path = Path(self.temp_dir.name) / filename
34
+ writer = cv2.VideoWriter(
35
+ str(path),
36
+ cv2.VideoWriter_fourcc(*"mp4v"),
37
+ fps,
38
+ size,
39
+ )
40
+ self.assertTrue(writer.isOpened())
41
+
42
+ width, height = size
43
+ for frame_index in range(max(1, int(fps * duration_sec))):
44
+ if mode == "dark":
45
+ frame = np.full((height, width, 3), 8, dtype=np.uint8)
46
+ elif mode == "blurry":
47
+ frame = np.full((height, width, 3), 150, dtype=np.uint8)
48
+ else:
49
+ frame = np.full((height, width, 3), 135, dtype=np.uint8)
50
+ offset = frame_index % max(1, width // 4)
51
+ cv2.rectangle(frame, (40 + offset, 80), (220 + offset, 300), (245, 245, 245), -1)
52
+ cv2.line(frame, (0, frame_index % height), (width - 1, height - 1), (10, 10, 10), 3)
53
+ cv2.putText(
54
+ frame,
55
+ str(frame_index),
56
+ (30, 420),
57
+ cv2.FONT_HERSHEY_SIMPLEX,
58
+ 2.0,
59
+ (255, 255, 255),
60
+ 4,
61
+ )
62
+ writer.write(frame)
63
+
64
+ writer.release()
65
+ return path
66
+
67
+ def test_valid_video_produces_real_metadata(self) -> None:
68
+ path = self._write_video("valid.mp4")
69
+
70
+ manifest = video_qc.run(str(path))
71
+
72
+ self.assertTrue(manifest.analysis_allowed)
73
+ self.assertEqual(manifest.quality_warnings, [])
74
+ self.assertEqual(manifest.width, 640)
75
+ self.assertEqual(manifest.height, 480)
76
+ self.assertEqual(manifest.total_frames, 300)
77
+ self.assertAlmostEqual(manifest.fps, 30.0, places=1)
78
+ self.assertAlmostEqual(manifest.duration_sec, 10.0, places=1)
79
+ self.assertGreater(manifest.sampled_frames, 0)
80
+ self.assertEqual(manifest.container, "mp4")
81
+ self.assertIsNotNone(manifest.codec)
82
+ self.assertIsNotNone(manifest.brightness_mean)
83
+ self.assertIsNotNone(manifest.blur_laplacian_var)
84
+
85
+ def test_invalid_video_sets_decode_failure_and_blocks_analysis(self) -> None:
86
+ manifest = video_qc.run(str(Path(self.temp_dir.name) / "missing.mp4"))
87
+
88
+ self.assertFalse(manifest.analysis_allowed)
89
+ self.assertEqual(manifest.quality_warnings, ["video_decode_failed"])
90
+ self.assertEqual(manifest.total_frames, 0)
91
+ self.assertEqual(manifest.sampled_frames, 0)
92
+
93
+ def test_short_low_resolution_low_fps_video_reports_warnings(self) -> None:
94
+ path = self._write_video("short_low.mp4", fps=10.0, duration_sec=2.0, size=(320, 240))
95
+
96
+ manifest = video_qc.run(str(path))
97
+
98
+ self.assertTrue(manifest.analysis_allowed)
99
+ self.assertIn("too_short", manifest.quality_warnings)
100
+ self.assertIn("fps_too_low", manifest.quality_warnings)
101
+ self.assertIn("resolution_too_low", manifest.quality_warnings)
102
+
103
+ def test_long_video_reports_warning(self) -> None:
104
+ original_min_duration = video_qc.MIN_DURATION_SEC
105
+ original_max_duration = video_qc.MAX_DURATION_SEC
106
+ video_qc.MIN_DURATION_SEC = 0.0
107
+ video_qc.MAX_DURATION_SEC = 1.0
108
+ try:
109
+ path = self._write_video("long.mp4", duration_sec=2.0)
110
+
111
+ manifest = video_qc.run(str(path))
112
+ finally:
113
+ video_qc.MIN_DURATION_SEC = original_min_duration
114
+ video_qc.MAX_DURATION_SEC = original_max_duration
115
+
116
+ self.assertTrue(manifest.analysis_allowed)
117
+ self.assertIn("too_long", manifest.quality_warnings)
118
+
119
+ def test_dark_video_reports_warning(self) -> None:
120
+ path = self._write_video("dark.mp4", mode="dark")
121
+
122
+ manifest = video_qc.run(str(path))
123
+
124
+ self.assertTrue(manifest.analysis_allowed)
125
+ self.assertIn("too_dark", manifest.quality_warnings)
126
+
127
+ def test_blurry_video_reports_warning(self) -> None:
128
+ path = self._write_video("blurry.mp4", mode="blurry")
129
+
130
+ manifest = video_qc.run(str(path))
131
+
132
+ self.assertTrue(manifest.analysis_allowed)
133
+ self.assertIn("too_blurry", manifest.quality_warnings)
134
+
135
+ def test_sample_frame_indices_are_bounded_and_ordered(self) -> None:
136
+ indices = video_qc.sample_frame_indices(total_frames=100, sample_count=5)
137
+
138
+ self.assertEqual(indices, sorted(indices))
139
+ self.assertEqual(indices[0], 0)
140
+ self.assertEqual(indices[-1], 99)
141
+ self.assertEqual(len(indices), 5)
142
+
143
+
144
+ if __name__ == "__main__":
145
+ unittest.main()
uv.lock CHANGED
@@ -574,6 +574,24 @@ wheels = [
574
  { url = "https://files.pythonhosted.org/packages/15/ce/e5ec180bc41812edcd8daeb8639d205622c0e8c02259d8ab25a0201b3c2a/numpy-2.4.6-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:2803abfebfc990042cd494d8ce2d5f82e9d847af6d35ec486923aa19dbad5e73", size = 12504263, upload-time = "2026-05-18T23:37:09.715Z" },
575
  ]
576
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
577
  [[package]]
578
  name = "orjson"
579
  version = "3.11.9"
@@ -804,6 +822,8 @@ version = "0.1.0"
804
  source = { virtual = "." }
805
  dependencies = [
806
  { name = "gradio" },
 
 
807
  ]
808
 
809
  [package.optional-dependencies]
@@ -814,6 +834,8 @@ dev = [
814
  [package.metadata]
815
  requires-dist = [
816
  { name = "gradio", specifier = ">=4.44.0" },
 
 
817
  { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.5.0" },
818
  ]
819
  provides-extras = ["dev"]
 
574
  { url = "https://files.pythonhosted.org/packages/15/ce/e5ec180bc41812edcd8daeb8639d205622c0e8c02259d8ab25a0201b3c2a/numpy-2.4.6-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:2803abfebfc990042cd494d8ce2d5f82e9d847af6d35ec486923aa19dbad5e73", size = 12504263, upload-time = "2026-05-18T23:37:09.715Z" },
575
  ]
576
 
577
+ [[package]]
578
+ name = "opencv-python-headless"
579
+ version = "4.13.0.92"
580
+ source = { registry = "https://pypi.org/simple" }
581
+ dependencies = [
582
+ { name = "numpy" },
583
+ ]
584
+ wheels = [
585
+ { url = "https://files.pythonhosted.org/packages/79/42/2310883be3b8826ac58c3f2787b9358a2d46923d61f88fedf930bc59c60c/opencv_python_headless-4.13.0.92-cp37-abi3-macosx_13_0_arm64.whl", hash = "sha256:1a7d040ac656c11b8c38677cc8cccdc149f98535089dbe5b081e80a4e5903209", size = 46247192, upload-time = "2026-02-05T07:01:35.187Z" },
586
+ { url = "https://files.pythonhosted.org/packages/2d/1e/6f9e38005a6f7f22af785df42a43139d0e20f169eb5787ce8be37ee7fcc9/opencv_python_headless-4.13.0.92-cp37-abi3-macosx_14_0_x86_64.whl", hash = "sha256:3e0a6f0a37994ec6ce5f59e936be21d5d6384a4556f2d2da9c2f9c5dc948394c", size = 32568914, upload-time = "2026-02-05T07:01:51.989Z" },
587
+ { url = "https://files.pythonhosted.org/packages/21/76/9417a6aef9def70e467a5bf560579f816148a4c658b7d525581b356eda9e/opencv_python_headless-4.13.0.92-cp37-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:5c8cfc8e87ed452b5cecb9419473ee5560a989859fe1d10d1ce11ae87b09a2cb", size = 33703709, upload-time = "2026-02-05T10:24:46.469Z" },
588
+ { url = "https://files.pythonhosted.org/packages/92/ce/bd17ff5772938267fd49716e94ca24f616ff4cb1ff4c6be13085108037be/opencv_python_headless-4.13.0.92-cp37-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:0525a3d2c0b46c611e2130b5fdebc94cf404845d8fa64d2f3a3b679572a5bd22", size = 56016764, upload-time = "2026-02-05T10:26:48.904Z" },
589
+ { url = "https://files.pythonhosted.org/packages/8f/b4/b7bcbf7c874665825a8c8e1097e93ea25d1f1d210a3e20d4451d01da30aa/opencv_python_headless-4.13.0.92-cp37-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:eb60e36b237b1ebd40a912da5384b348df8ed534f6f644d8e0b4f103e272ba7d", size = 35010236, upload-time = "2026-02-05T10:28:11.031Z" },
590
+ { url = "https://files.pythonhosted.org/packages/4b/33/b5db29a6c00eb8f50708110d8d453747ca125c8b805bc437b289dbdcc057/opencv_python_headless-4.13.0.92-cp37-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:0bd48544f77c68b2941392fcdf9bcd2b9cdf00e98cb8c29b2455d194763cf99e", size = 60391106, upload-time = "2026-02-05T10:30:14.236Z" },
591
+ { url = "https://files.pythonhosted.org/packages/fb/c3/52cfea47cd33e53e8c0fbd6e7c800b457245c1fda7d61660b4ffe9596a7f/opencv_python_headless-4.13.0.92-cp37-abi3-win32.whl", hash = "sha256:a7cf08e5b191f4ebb530791acc0825a7986e0d0dee2a3c491184bd8599848a4b", size = 30812232, upload-time = "2026-02-05T07:02:29.594Z" },
592
+ { url = "https://files.pythonhosted.org/packages/4a/90/b338326131ccb2aaa3c2c85d00f41822c0050139a4bfe723cfd95455bd2d/opencv_python_headless-4.13.0.92-cp37-abi3-win_amd64.whl", hash = "sha256:77a82fe35ddcec0f62c15f2ba8a12ecc2ed4207c17b0902c7a3151ae29f37fb6", size = 40070414, upload-time = "2026-02-05T07:02:26.448Z" },
593
+ ]
594
+
595
  [[package]]
596
  name = "orjson"
597
  version = "3.11.9"
 
822
  source = { virtual = "." }
823
  dependencies = [
824
  { name = "gradio" },
825
+ { name = "numpy" },
826
+ { name = "opencv-python-headless" },
827
  ]
828
 
829
  [package.optional-dependencies]
 
834
  [package.metadata]
835
  requires-dist = [
836
  { name = "gradio", specifier = ">=4.44.0" },
837
+ { name = "numpy", specifier = ">=1.26.0" },
838
+ { name = "opencv-python-headless", specifier = ">=4.10.0" },
839
  { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.5.0" },
840
  ]
841
  provides-extras = ["dev"]