From 1b8c8c1570e2f43c30a318da97f29cfe1c135845 Mon Sep 17 00:00:00 2001
From: Cai <cai@nbcai.cc>
Date: Sun, 16 Aug 2026 18:11:26 +0800
Subject: [PATCH] docs(defense): quantify J-10CE and PL missile value pools
---
dev/project-dev/test/test_extract_ppt_slides.py | 996 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 988 insertions(+), 8 deletions(-)
diff --git a/dev/project-dev/test/test_extract_ppt_slides.py b/dev/project-dev/test/test_extract_ppt_slides.py
index e7301a9..3d28e79 100644
--- a/dev/project-dev/test/test_extract_ppt_slides.py
+++ b/dev/project-dev/test/test_extract_ppt_slides.py
@@ -1,8 +1,11 @@
from __future__ import annotations
+import csv
+from datetime import datetime, timedelta, timezone
import hashlib
import importlib.util
import io
+import json
import os
from pathlib import Path
import re
@@ -10,6 +13,7 @@
import subprocess
import sys
import tempfile
+import textwrap
import unittest
from unittest import mock
import zlib
@@ -26,21 +30,63 @@
SPEC.loader.exec_module(slides)
-def feature(index: int, array: np.ndarray, score: float = 0.2, sharpness: float | None = None):
+def palette(marker_bin: int | None = None, coverage: float = 0.20):
+ histogram = np.full(slides.PALETTE_BIN_COUNT, 1 / slides.PALETTE_BIN_COUNT, np.float32)
+ if marker_bin is not None:
+ histogram.fill((1.0 - coverage) / (slides.PALETTE_BIN_COUNT - 1))
+ histogram[marker_bin] = coverage
+ histogram /= float(np.sum(histogram, dtype=np.float64))
+ return histogram
+
+
+def feature(
+ index: int,
+ array: np.ndarray,
+ score: float = 0.2,
+ sharpness: float | None = None,
+ page_palette: np.ndarray | None = None,
+):
value = int(np.mean(array))
phash = int.from_bytes(hashlib.sha256(array.tobytes()).digest()[:8], "big")
return slides.FrameFeature(
sample_index=index,
relative_seconds=float(index),
gray=array.astype(np.uint8).copy(),
+ palette_histogram=(palette() if page_palette is None else page_palette.copy()),
perceptual_hash=phash,
content_score=score,
sharpness=float(value if sharpness is None else sharpness),
)
-def candidate(timestamp: float, array: np.ndarray, score: float, phash: int = 0):
- return slides.PageCandidate(timestamp, array.astype(np.uint8).copy(), phash, score, 1.0)
+def candidate(
+ timestamp: float,
+ array: np.ndarray,
+ score: float,
+ phash: int = 0,
+ page_palette: np.ndarray | None = None,
+):
+ return slides.PageCandidate(
+ timestamp,
+ array.astype(np.uint8).copy(),
+ palette() if page_palette is None else page_palette.copy(),
+ phash,
+ score,
+ 1.0,
+ )
+
+
+def logical_page(order: int, page_palette: np.ndarray | None = None):
+ page_feature = np.full((8, 8), order % 256, np.uint8)
+ return slides.LogicalPage(
+ logical_anchor_feature=page_feature.copy(),
+ current_feature=page_feature,
+ current_palette=palette() if page_palette is None else page_palette.copy(),
+ current_hash=order,
+ current_timestamp=float(order),
+ current_score=0.2,
+ first_seen_order=order,
+ )
def sha256(path: Path) -> str:
@@ -148,20 +194,915 @@
recent = np.full((8, 8), 4, np.uint8)
classifier.consume(candidate(1, historical, 0.20, 0))
classifier.pages.append(
- slides.LogicalPage(recent.copy(), recent.copy(), 0, 2.0, 0.20, 1)
+ slides.LogicalPage(recent.copy(), recent.copy(), palette(), 0, 2.0, 0.20, 1)
)
action = classifier.consume(candidate(3, historical, 0.22, 0))
self.assertEqual(action, "animation_updated")
self.assertEqual(len(classifier.pages), 2)
+class MainSequenceTests(unittest.TestCase):
+ def test_palette_histogram_distinguishes_dark_blue_from_black(self):
+ image = np.array(
+ [
+ [[16, 16, 48], [16, 16, 48]],
+ [[16, 16, 16], [16, 16, 16]],
+ ],
+ dtype=np.uint8,
+ )
+ histogram = slides._palette_histogram(image)
+ self.assertEqual(histogram.shape, (512,))
+ self.assertTrue(np.all(np.isfinite(histogram)))
+ self.assertAlmostEqual(float(np.sum(histogram)), 1.0, places=6)
+ self.assertEqual(float(histogram[1]), 0.5)
+ self.assertEqual(float(histogram[0]), 0.5)
+
+ def test_selects_old_pages_002_through_038_and_keeps_two_gaps(self):
+ marker_pages = {2, *range(4, 9), *range(10, 39)}
+ pages = [
+ logical_page(index - 1, palette(1) if index in marker_pages else palette())
+ for index in range(1, 54)
+ ]
+ selected = slides.select_main_sequence(pages)
+ self.assertEqual(selected, pages[1:38])
+ self.assertIs(selected[1], pages[2]) # old slide_003: first one-page gap
+ self.assertIs(selected[7], pages[8]) # old slide_009: second one-page gap
+ self.assertEqual([page.first_seen_order for page in selected], list(range(1, 38)))
+
+ def test_conservative_gates_keep_all_candidates(self):
+ cases = {
+ "markers": [palette(1) if index < 7 else palette() for index in range(10)],
+ "share": [palette(1) if 3 <= index <= 10 else palette() for index in range(20)],
+ "density": [
+ palette(1) if index in {0, 2, 4, 6, 7, 8, 9, 10} else palette()
+ for index in range(11)
+ ],
+ }
+ for name, histograms in cases.items():
+ with self.subTest(name=name):
+ pages = [logical_page(index, value) for index, value in enumerate(histograms)]
+ selected = slides.select_main_sequence(pages)
+ self.assertEqual(selected, pages)
+ self.assertTrue(all(left is right for left, right in zip(selected, pages)))
+
+ def test_all_theme_pages_are_preserved(self):
+ pages = [logical_page(index, palette(7)) for index in range(12)]
+ selected = slides.select_main_sequence(pages)
+ self.assertEqual(selected, pages)
+ self.assertTrue(all(left is right for left, right in zip(selected, pages)))
+
+ def test_invalid_palette_is_explicit_failure(self):
+ invalid = [
+ np.zeros(5, np.float32),
+ np.full(512, np.nan, np.float32),
+ np.zeros(512, np.float32),
+ ]
+ for value in invalid:
+ with self.subTest(shape=value.shape, finite=bool(np.all(np.isfinite(value)))):
+ with self.assertRaises(slides.SlideExtractionError):
+ slides.select_main_sequence([logical_page(0, value)])
+
+
class ProcessAndCommandTests(unittest.TestCase):
def test_command_freezes_time_mapping(self):
scan = slides._scan_command("ffmpeg", Path("x.mp4"), 320, 180)
self.assertIn("setpts=PTS-STARTPTS,fps=fps=1:start_time=0,scale=320:180:flags=bilinear", scan)
+ input_index = scan.index("-i")
+ self.assertLess(scan.index("-threads"), input_index)
+ self.assertEqual(scan[scan.index("-threads") + 1], str(slides.MEDIA_DECODE_THREADS))
+ self.assertEqual(scan[scan.index("-filter_threads") + 1], str(slides.MEDIA_FILTER_THREADS))
extract = slides._extract_command("ffmpeg", Path("x.mp4"), 12.0, Path("x.png"))
self.assertEqual(extract[extract.index("-ss") + 1], "12.000")
+ self.assertLess(extract.index("-threads"), extract.index("-i"))
self.assertEqual(feature(9, np.zeros((2, 2), np.uint8)).relative_seconds, 9.0)
+
+ def test_hardware_commands_place_decoder_and_threads_before_input(self):
+ for codec, decoder in (("h264", "h264_cuvid"), ("hevc", "hevc_cuvid")):
+ with self.subTest(codec=codec):
+ self.assertEqual(slides._decoder_for_codec(codec, "yuv420p"), decoder)
+ command = slides._scan_command(
+ "ffmpeg", Path(f"{codec}.mp4"), 320, 180, decoder
+ )
+ input_index = command.index("-i")
+ for option in ("-hwaccel", "-hwaccel_output_format", "-c:v", "-threads"):
+ self.assertLess(command.index(option), input_index)
+ self.assertEqual(command[command.index("-hwaccel") + 1], "cuda")
+ self.assertEqual(command[command.index("-c:v") + 1], decoder)
+ self.assertEqual(
+ command[command.index("-threads") + 1],
+ str(slides.MEDIA_DECODE_THREADS),
+ )
+ self.assertIn("hwdownload,format=nv12", command[command.index("-vf") + 1])
+ self.assertEqual(slides._decoder_for_codec("h264", "gbrp"), "h264_cuvid")
+ self.assertEqual(slides._decoder_for_codec("hevc", "p010le"), "hevc_cuvid")
+ self.assertIsNone(slides._decoder_for_codec("vp9", "yuv420p"))
+
+ def test_hardware_external_failures_have_typed_boundary(self):
+ class FakeStdout:
+ def __init__(self, payloads):
+ self.payloads = list(payloads)
+
+ def read(self, size):
+ return self.payloads.pop(0) if self.payloads else b""
+
+ def close(self):
+ return None
+
+ class FakeProcess:
+ def __init__(self, payloads, code):
+ self.stdout = FakeStdout(payloads)
+ self.code = code
+
+ def wait(self, timeout=None):
+ return self.code
+
+ def poll(self):
+ return self.code
+
+ def terminate(self):
+ return None
+
+ def kill(self):
+ return None
+
+ info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
+ cases = (
+ ("nonzero", FakeProcess([], 7), "exit=7"),
+ ("zero", FakeProcess([], 0), "未输出任何完整"),
+ ("partial", FakeProcess([b"x", b""], 0), "输出半帧"),
+ )
+ with tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_which_or_error", return_value="ffmpeg"
+ ):
+ root = Path(raw)
+ with mock.patch.object(slides.subprocess, "Popen", side_effect=OSError("start denied")):
+ with self.assertRaisesRegex(slides.HardwareDecodePathError, "启动失败"):
+ slides._scan_video_once(Path("x.mp4"), root, info, decoder="h264_cuvid")
+ for name, process, pattern in cases:
+ with self.subTest(name=name), mock.patch.object(
+ slides.subprocess, "Popen", return_value=process
+ ):
+ with self.assertRaisesRegex(slides.HardwareDecodePathError, pattern):
+ slides._scan_video_once(
+ Path("x.mp4"), root, info, decoder="h264_cuvid"
+ )
+
+ def test_hardware_cleanup_failure_does_not_trigger_cpu_fallback(self):
+ info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
+ hardware_error = slides.HardwareDecodePathError("hardware-root")
+ slides._attach_cleanup_note(hardware_error, "cleanup-failed")
+ with tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_scan_video_once", side_effect=hardware_error
+ ) as scan:
+ with self.assertRaises(slides.HardwareDecodePathError) as caught:
+ slides.scan_video(Path("x.mp4"), Path(raw), info)
+ self.assertIs(caught.exception, hardware_error)
+ self.assertEqual(scan.call_count, 1)
+ self.assertIn("cleanup-failed", "\n".join(caught.exception.__notes__))
+
+ def test_active_errors_keep_identity_when_process_cleanup_fails(self):
+ frame_bytes = 320 * 180 * 3
+
+ class Stream:
+ def __init__(self, error=None, partial=False):
+ self.error = error
+ self.partial = partial
+ self.calls = 0
+
+ def read(self, size):
+ self.calls += 1
+ if self.error is not None:
+ raise self.error
+ if self.partial:
+ return b"x" if self.calls == 1 else b""
+ return b"\0" * frame_bytes if self.calls == 1 else b""
+
+ def close(self):
+ return None
+
+ class FakeProcess:
+ def __init__(self, stream):
+ self.stdout = stream
+
+ def wait(self, timeout=None):
+ return 0
+
+ def poll(self):
+ return None
+
+ def terminate(self):
+ return None
+
+ cleanup = OSError("cleanup-failed")
+ info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
+ cases = (
+ ("business", None, ValueError("business-root")),
+ ("cpu", None, slides.SlideExtractionError("cpu-root")),
+ ("base", KeyboardInterrupt(), None),
+ )
+ for name, read_error, feature_error in cases:
+ root_error = read_error or feature_error
+ stream = Stream(error=read_error)
+ patches = [
+ mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"),
+ mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess(stream)),
+ mock.patch.object(slides, "_attempt_reap", return_value=cleanup),
+ ]
+ if feature_error is not None:
+ patches.append(mock.patch.object(slides, "_gray_feature", side_effect=feature_error))
+ with self.subTest(name=name), tempfile.TemporaryDirectory() as raw:
+ for item in patches:
+ item.start()
+ try:
+ with self.assertRaises(type(root_error)) as caught:
+ slides._scan_video_once(
+ Path("x.mp4"), Path(raw), info, decoder=None
+ )
+ self.assertIs(caught.exception, root_error)
+ self.assertIn("cleanup-failed", "\n".join(caught.exception.__notes__))
+ finally:
+ for item in reversed(patches):
+ item.stop()
+
+ partial_process = FakeProcess(Stream(partial=True))
+ with tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_which_or_error", return_value="ffmpeg"
+ ), mock.patch.object(slides.subprocess, "Popen", return_value=partial_process), mock.patch.object(
+ slides, "_attempt_reap", return_value=cleanup
+ ):
+ with self.assertRaises(slides.HardwareDecodePathError) as caught:
+ slides._scan_video_once(
+ Path("x.mp4"), Path(raw), info, decoder="h264_cuvid"
+ )
+ hardware_error = caught.exception
+ self.assertTrue(getattr(hardware_error, "_media_cleanup_failed", False))
+ self.assertIn("cleanup-failed", "\n".join(hardware_error.__notes__))
+ with tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_scan_video_once", side_effect=hardware_error
+ ) as scan:
+ with self.assertRaises(slides.HardwareDecodePathError) as caught_again:
+ slides.scan_video(Path("x.mp4"), Path(raw), info)
+ self.assertIs(caught_again.exception, hardware_error)
+ self.assertEqual(scan.call_count, 1)
+
+ def test_product_popen_uses_below_normal_creation_flags(self):
+ class FakeProcess:
+ stdout = None
+
+ def wait(self, timeout=None):
+ return 0
+
+ def poll(self):
+ return 0
+
+ with tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_subprocess_creation_flags", return_value=12345
+ ), mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess()) as popen:
+ slides._run_file_command(["tool"], Path(raw) / "stderr.log", "stage")
+ self.assertEqual(popen.call_args.kwargs["creationflags"], 12345)
+
+ def test_fallback_only_handles_hardware_error_and_never_retries_cpu(self):
+ info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
+ page = logical_page(0)
+ success = slides.ScanResult((page,), 10, 2, (("new_page", 1),), None)
+ hardware_error = slides.HardwareDecodePathError("decoder failed")
+ with tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_scan_video_once", side_effect=[hardware_error, success]
+ ) as scan, mock.patch("sys.stderr", new_callable=io.StringIO) as stderr:
+ self.assertEqual(slides.scan_video(Path("x.mp4"), Path(raw), info), [page])
+ self.assertEqual([call.kwargs["decoder"] for call in scan.call_args_list], [
+ "h264_cuvid", None
+ ])
+ self.assertIn("decoder failed", stderr.getvalue())
+
+ cpu_error = slides.SlideExtractionError("cpu root cause")
+ with tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_scan_video_once", side_effect=[hardware_error, cpu_error]
+ ) as scan:
+ with self.assertRaises(slides.SlideExtractionError) as caught:
+ slides.scan_video(Path("x.mp4"), Path(raw), info)
+ self.assertIs(caught.exception, cpu_error)
+ self.assertEqual(scan.call_count, 2)
+
+ def test_algorithm_empty_and_baseexception_never_fall_back(self):
+ info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
+ empty = slides.ScanResult((), 10, 0, (), "h264_cuvid")
+ with tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_scan_video_once", return_value=empty
+ ) as scan:
+ with self.assertRaisesRegex(slides.SlideExtractionError, "未发现"):
+ slides.scan_video(Path("x.mp4"), Path(raw), info)
+ scan.assert_called_once()
+ for error in (RuntimeError("classifier failed"), KeyboardInterrupt(), SystemExit(9)):
+ with self.subTest(error=type(error).__name__), tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_scan_video_once", side_effect=error
+ ) as scan:
+ with self.assertRaises(type(error)) as caught:
+ slides.scan_video(Path("x.mp4"), Path(raw), info)
+ self.assertIs(caught.exception, error)
+ scan.assert_called_once()
+
+ def test_algorithm_exception_inside_hardware_scan_preserves_original_type(self):
+ frame_bytes = 320 * 180 * 3
+
+ class FullFrameStream:
+ def __init__(self):
+ self.done = False
+
+ def read(self, size):
+ if self.done:
+ return b""
+ self.done = True
+ return b"\0" * frame_bytes
+
+ def close(self):
+ return None
+
+ class FakeProcess:
+ stdout = FullFrameStream()
+
+ def wait(self, timeout=None):
+ return 0
+
+ def poll(self):
+ return 0
+
+ root_error = ValueError("feature root cause")
+ info = slides.VideoInfo(320, 180, 10.0, "h264", "yuv420p")
+ with tempfile.TemporaryDirectory() as raw, mock.patch.object(
+ slides, "_which_or_error", return_value="ffmpeg"
+ ), mock.patch.object(slides.subprocess, "Popen", return_value=FakeProcess()), mock.patch.object(
+ slides, "_gray_feature", side_effect=root_error
+ ):
+ with self.assertRaises(ValueError) as caught:
+ slides._scan_video_once(
+ Path("x.mp4"), Path(raw), info, decoder="h264_cuvid"
+ )
+ self.assertIs(caught.exception, root_error)
+
+ @unittest.skipUnless(shutil.which("powershell"), "Windows PowerShell is required")
+ def test_resource_sampler_parser_rejects_cpu_boundaries_and_empty_windows(self):
+ sampler = Path(__file__).resolve().parent / "monitor_media_host_responsiveness.ps1"
+
+ fields = (
+ "timestamp_utc", "previous_valid_timestamp_utc", "interval_seconds",
+ "counter_warmup", "cpu_valid", "host_cpu_percent",
+ "scan_phase_active", "scan_ffmpeg_present", "scan_ffmpeg_exit_observed",
+ "scan_ffmpeg_pid", "scan_ffmpeg_command_line", "decoder",
+ "logical_processor_count", "task_cpu_percent", "task_working_set_bytes",
+ "task_pids", "root_priority", "ffmpeg_priorities", "ffmpeg_command_lines",
+ "gpu_query_status", "gpu_query_raw_rows", "gpu_utilization_percent",
+ "gpu_decoder_utilization_percent", "gpu_memory_used_mib",
+ "nvidia_smi_compute_query_status", "nvidia_smi_compute_raw_rows",
+ "nvidia_smi_compute_task_rows",
+ "gpu_process_memory_query_status", "gpu_process_memory_instances",
+ "task_gpu_dedicated_memory_mib",
+ )
+ scan_command = (
+ "ffmpeg -hwaccel cuda -hwaccel_output_format cuda "
+ "-c:v h264_cuvid -threads 4 "
+ "-filter_threads 2 -filter_complex_threads 2 -i x.mp4 -map 0:v:0 "
+ "-vf hwdownload,format=nv12,fps=fps=1:start_time=0 "
+ "-pix_fmt rgb24 -f rawvideo -"
+ )
+
+ def make_row(current, previous, interval, cpu, *, warmup, present, exited):
+ return {
+ "timestamp_utc": current.isoformat().replace("+00:00", "Z"),
+ "previous_valid_timestamp_utc": (
+ "" if previous is None else previous.isoformat().replace("+00:00", "Z")
+ ),
+ "interval_seconds": f"{interval:.6f}",
+ "counter_warmup": str(warmup).lower(),
+ "cpu_valid": str(not warmup).lower(),
+ "host_cpu_percent": "" if warmup else f"{cpu:.6f}",
+ "scan_phase_active": "true",
+ "scan_ffmpeg_present": str(present).lower(),
+ "scan_ffmpeg_exit_observed": str(exited).lower(),
+ "scan_ffmpeg_pid": "4242",
+ "scan_ffmpeg_command_line": scan_command,
+ "decoder": "h264_cuvid",
+ "logical_processor_count": "44",
+ "task_cpu_percent": "" if warmup else "12.5",
+ "task_working_set_bytes": "123456789",
+ "task_pids": "101;4242",
+ "root_priority": "BelowNormal",
+ "ffmpeg_priorities": (
+ "[]" if not present else '[{"pid":4242,"priority":"BelowNormal"}]'
+ ),
+ "ffmpeg_command_lines": (
+ '[{"pid":4242,"command_line":' + json.dumps(scan_command) + "}]"
+ ),
+ "gpu_query_status": "ok",
+ "gpu_query_raw_rows": '["20, 31, 2048"]',
+ "gpu_utilization_percent": "20",
+ "gpu_decoder_utilization_percent": "31",
+ "gpu_memory_used_mib": "2048",
+ "nvidia_smi_compute_query_status": "ok",
+ "nvidia_smi_compute_raw_rows": '["4242, N/A"]',
+ "nvidia_smi_compute_task_rows": '["4242, N/A"]',
+ "gpu_process_memory_query_status": "ok",
+ "gpu_process_memory_instances": (
+ '[{"pid":4242,"instance_name":"pid_4242_luid_0",'
+ '"counter_path":"path","dedicated_usage_bytes":1048576}]'
+ ),
+ "task_gpu_dedicated_memory_mib": "1",
+ }
+
+ def analyze_case(root, name, intervals, cpu_values, mutate=None, omit_field=None):
+ case = root / name
+ case.mkdir()
+ samples = case / "samples.csv"
+ current = datetime(2026, 8, 4, tzinfo=timezone.utc)
+ rows = [make_row(current, None, 0, 0, warmup=True, present=True, exited=False)]
+ previous = current
+ for interval, cpu in zip(intervals, cpu_values):
+ current += timedelta(seconds=interval)
+ rows.append(make_row(
+ current, previous, interval, cpu,
+ warmup=False, present=True, exited=False,
+ ))
+ previous = current
+ current += timedelta(seconds=0.5)
+ rows.append(make_row(
+ current, previous, 0.5, 20.0,
+ warmup=False, present=False, exited=True,
+ ))
+ if mutate is not None:
+ mutate(rows)
+ output_fields = tuple(field for field in fields if field != omit_field)
+ with samples.open("w", newline="", encoding="utf-8") as handle:
+ writer = csv.DictWriter(handle, fieldnames=output_fields, extrasaction="ignore")
+ writer.writeheader()
+ writer.writerows(rows)
+ evidence = case / "evidence"
+ result = subprocess.run(
+ [
+ shutil.which("powershell"), "-NoProfile", "-File", str(sampler),
+ "-AnalyzeSamplesPath", str(samples), "-EvidenceDirectory", str(evidence),
+ ],
+ capture_output=True,
+ text=True,
+ timeout=60,
+ )
+ self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
+ return json.loads((evidence / "resource_summary.json").read_text(encoding="utf-8-sig"))
+
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ low = analyze_case(root, "low", [0.5] * 22, [84.9] * 22)
+ self.assertTrue(low["sampling_valid"])
+ self.assertGreater(low["window_count"], 0)
+ self.assertTrue(low["cpu_gate_pass"])
+
+ equal = analyze_case(root, "equal", [0.5] * 22, [85.0] * 22)
+ self.assertTrue(equal["sampling_valid"])
+ self.assertFalse(equal["all_windows_host_cpu_strictly_below_85"])
+ self.assertFalse(equal["cpu_gate_pass"])
+
+ exact_ten = analyze_case(
+ root, "exact-ten", [0.5] * 21, [85.0] * 20 + [20.0]
+ )
+ self.assertEqual(exact_ten["maximum_high_cpu_segment_seconds"], 10.0)
+ self.assertFalse(exact_ten["cpu_gate_pass"])
+
+ below_ten = analyze_case(
+ root, "below-ten", [0.49995] * 20 + [0.5], [85.0] * 20 + [20.0]
+ )
+ self.assertEqual(below_ten["maximum_high_cpu_segment_seconds"], 9.999)
+
+ sparse = analyze_case(root, "sparse", [1.0] * 12, [20.0] * 12)
+ self.assertFalse(sparse["sampling_valid"])
+ self.assertTrue(any("window_density_invalid" in item for item in sparse["sampling_problems"]))
+
+ zero_window = analyze_case(root, "zero-window", [0.5] * 8, [20.0] * 8)
+ self.assertFalse(zero_window["sampling_valid"])
+ self.assertEqual(zero_window["window_count"], 0)
+
+ gap = analyze_case(
+ root, "gap", [0.5] * 10 + [1.6] + [0.5] * 10, [20.0] * 21
+ )
+ self.assertFalse(gap["sampling_valid"])
+ self.assertGreater(gap["maximum_observed_interval_seconds"], 1.5)
+
+ reversed_time = analyze_case(
+ root, "time-reversal", [0.5] * 12, [20.0] * 12,
+ mutate=lambda rows: rows[5].__setitem__("timestamp_utc", rows[3]["timestamp_utc"]),
+ )
+ self.assertFalse(reversed_time["sampling_valid"])
+ self.assertTrue(any(
+ "timestamp_not_strictly_increasing" in item
+ for item in reversed_time["sampling_problems"]
+ ))
+
+ interval_mismatch = analyze_case(
+ root, "interval-mismatch", [0.5] * 12, [20.0] * 12,
+ mutate=lambda rows: rows[4].__setitem__("interval_seconds", "9.0"),
+ )
+ self.assertFalse(interval_mismatch["sampling_valid"])
+ self.assertTrue(any(
+ "interval_mismatch" in item for item in interval_mismatch["sampling_problems"]
+ ))
+
+ previous_mismatch = analyze_case(
+ root, "previous-mismatch", [0.5] * 12, [20.0] * 12,
+ mutate=lambda rows: rows[4].__setitem__(
+ "previous_valid_timestamp_utc", rows[1]["timestamp_utc"]
+ ),
+ )
+ self.assertFalse(previous_mismatch["sampling_valid"])
+ self.assertTrue(any(
+ "previous_timestamp_mismatch" in item
+ for item in previous_mismatch["sampling_problems"]
+ ))
+
+ for omitted in (
+ "gpu_query_raw_rows", "task_pids", "ffmpeg_command_lines",
+ "nvidia_smi_compute_task_rows",
+ ):
+ with self.subTest(omitted=omitted):
+ missing_field = analyze_case(
+ root, f"missing-{omitted}", [0.5] * 12, [20.0] * 12,
+ omit_field=omitted,
+ )
+ self.assertFalse(missing_field["sampling_valid"])
+ self.assertIn(
+ f"required_column_missing_{omitted}",
+ missing_field["sampling_problems"],
+ )
+
+ first_not_present = analyze_case(
+ root, "first-coverage", [0.5] * 12, [20.0] * 12,
+ mutate=lambda rows: rows[0].__setitem__("scan_ffmpeg_present", "false"),
+ )
+ self.assertFalse(first_not_present["sampling_valid"])
+ self.assertIn("scan_phase_first_row_not_present", first_not_present["sampling_problems"])
+
+ last_not_exit = analyze_case(
+ root, "last-coverage", [0.5] * 12, [20.0] * 12,
+ mutate=lambda rows: rows[-1].__setitem__("scan_ffmpeg_exit_observed", "false"),
+ )
+ self.assertFalse(last_not_exit["sampling_valid"])
+ self.assertIn("scan_phase_last_row_not_exit_observation", last_not_exit["sampling_problems"])
+
+ argv_rewrites = {
+ "missing-output-format": lambda value: value.replace(
+ "-hwaccel_output_format cuda ", ""
+ ),
+ "wrong-output-format": lambda value: value.replace(
+ "-hwaccel_output_format cuda", "-hwaccel_output_format qsv"
+ ),
+ "late-output-format": lambda value: value.replace(
+ "-hwaccel_output_format cuda ", ""
+ ).replace("-i x.mp4", "-i x.mp4 -hwaccel_output_format cuda"),
+ "missing-hwdownload": lambda value: value.replace(
+ "hwdownload,format=nv12,", ""
+ ),
+ }
+ for name, rewrite in argv_rewrites.items():
+ with self.subTest(argv_case=name):
+ invalid_argv = analyze_case(
+ root, f"invalid-{name}", [0.5] * 12, [20.0] * 12,
+ mutate=lambda rows, transform=rewrite: [
+ row.__setitem__(
+ "scan_ffmpeg_command_line",
+ transform(row["scan_ffmpeg_command_line"]),
+ )
+ for row in rows
+ ],
+ )
+ self.assertFalse(invalid_argv["sampling_valid"])
+ self.assertIn("nvdec_scan_argv_invalid", invalid_argv["sampling_problems"])
+
+ @unittest.skipUnless(
+ os.name == "nt" and shutil.which("powershell") and shutil.which("nvidia-smi"),
+ "Windows PowerShell and nvidia-smi are required",
+ )
+ def test_resource_sampler_run_rejects_forged_nvdec_and_excludes_page_ffmpeg(self):
+ sampler = Path(__file__).resolve().parent / "monitor_media_host_responsiveness.ps1"
+ sampler_text = sampler.read_text(encoding="utf-8")
+ process_start = sampler_text.index("[void]$Process.Start()")
+ self.assertLess(
+ sampler_text.index("$HostCpuCounter = New-Object"), process_start
+ )
+ self.assertLess(sampler_text.index("$HostCpuCounter.NextValue()"), process_start)
+ self.assertLess(
+ sampler_text.index("$GpuMemoryCategory = New-Object"), process_start
+ )
+ self.assertLess(sampler_text.index("$GpuMemoryCategory.GetInstanceNames()"), process_start)
+ self.assertLess(
+ sampler_text.index('$StartInfo.EnvironmentVariables["MBX_MEDIA_SAMPLER_READY"]'),
+ process_start,
+ )
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ fake_ffmpeg = root / "ffmpeg.exe"
+ shutil.copy2(sys.executable, fake_ffmpeg)
+ child = root / "fake_ffmpeg_child.py"
+ child.write_text(
+ "import sys, time\n"
+ "time.sleep(float(sys.argv[sys.argv.index('--sleep') + 1]))\n",
+ encoding="utf-8",
+ )
+ target = root / "stub_media_target.py"
+ target.write_text(
+ textwrap.dedent(
+ f"""
+ import ctypes
+ import os
+ from pathlib import Path
+ import subprocess
+ import sys
+ import time
+
+ sys.stdout.reconfigure(encoding="utf-8")
+ sys.stderr.reconfigure(encoding="utf-8")
+ kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+ kernel32.SetPriorityClass(kernel32.GetCurrentProcess(), 0x00004000)
+ creationflags = 0x00004000
+ if os.environ.get("MBX_MEDIA_SAMPLER_READY") != "1":
+ raise RuntimeError("sampler-ready handshake missing")
+ fake_ffmpeg = Path({str(fake_ffmpeg)!r})
+ child = Path({str(child)!r})
+ scan = subprocess.Popen([
+ str(fake_ffmpeg), str(child), "--sleep", "7.0",
+ "-hwaccel", "cuda", "-c:v", "h264_cuvid",
+ "-threads", "4", "-filter_threads", "2",
+ "-filter_complex_threads", "2", "-i", "fake.mp4",
+ "-map", "0:v:0", "-vf",
+ "hwdownload,format=nv12,setpts=PTS-STARTPTS,"
+ "fps=fps=1:start_time=0,scale=320:180",
+ "-pix_fmt", "rgb24", "-f", "rawvideo", "-",
+ ], creationflags=creationflags)
+ scan.wait()
+ print("警告：NVDEC 不可用；改用受限 CPU。", file=sys.stderr, flush=True)
+ print(
+ "扫描解码路径：NVDEC/CUDA (h264_cuvid)；解码线程=4；滤镜线程=2",
+ flush=True,
+ )
+ time.sleep(1.5)
+ for index in range(24):
+ churn = subprocess.Popen([
+ str(fake_ffmpeg), str(child), "--sleep", "0.04",
+ "-threads", "4", "-i", "fake.mp4",
+ "-frames:v", "1", "-n", f"race-{{index}}.png",
+ ], creationflags=creationflags)
+ churn.wait()
+ page = subprocess.Popen([
+ str(fake_ffmpeg), str(child), "--sleep", "4.0",
+ "-threads", "4", "-i", "fake.mp4",
+ "-frames:v", "1", "-n", "fake.png",
+ ], creationflags=creationflags)
+ page.wait()
+ """
+ ).lstrip(),
+ encoding="utf-8",
+ )
+ source = root / "source.stub"
+ source.write_bytes(b"read-only source fingerprint")
+ evidence = root / "evidence"
+ result = subprocess.run(
+ [
+ shutil.which("powershell"), "-NoProfile", "-File", str(sampler),
+ "-PythonPath", sys.executable,
+ "-ScriptPath", str(target),
+ "-VideoPath", str(source),
+ "-OutputPath", str(root / "unused-output"),
+ "-EvidenceDirectory", str(evidence),
+ ],
+ capture_output=True,
+ text=True,
+ encoding="utf-8",
+ errors="replace",
+ env={
+ **os.environ,
+ "PYTHONIOENCODING": "utf-8",
+ "PYTHONUTF8": "1",
+ "MBX_MEDIA_SAMPLER_TEST_CIM_DELAY_MS": "100",
+ "MBX_MEDIA_SAMPLER_TEST_FAIL_SAMPLE_ONCE": "1",
+ },
+ timeout=90,
+ )
+ self.assertEqual(result.returncode, 3, result.stderr + result.stdout)
+ run_summary = json.loads(
+ (evidence / "run_summary.json").read_text(encoding="utf-8-sig")
+ )
+ self.assertEqual(run_summary["target_exit_code"], 0)
+ self.assertFalse(run_summary["resource_acceptance_pass"])
+ self.assertTrue(
+ run_summary["nvdec_command_log_evidence"],
+ json.dumps(
+ {
+ "run_summary": run_summary,
+ "stdout": (evidence / "stdout.log").read_text("utf-8"),
+ },
+ ensure_ascii=False,
+ ),
+ )
+ self.assertFalse(run_summary["nvdec_scan_argv_evidence"])
+ self.assertGreater(run_summary["command_line_disappearance_race_count"], 0)
+ self.assertEqual(run_summary["injected_sample_failure_count"], 1)
+ self.assertEqual(
+ len(run_summary["sampling_errors"]), 1, run_summary["sampling_errors"]
+ )
+ self.assertIn("injected post-scan sample failure", run_summary["sampling_errors"][0])
+ self.assertTrue(run_summary["source_unchanged"])
+ self.assertIn("NVDEC/CUDA (h264_cuvid)", (evidence / "stdout.log").read_text("utf-8"))
+ self.assertIn("改用受限 CPU", (evidence / "stderr.log").read_text("utf-8"))
+
+ with (evidence / "resource_samples.csv").open(
+ newline="", encoding="utf-8-sig"
+ ) as handle:
+ rows = list(csv.DictReader(handle))
+ resource_summary = json.loads(
+ (evidence / "resource_summary.json").read_text(encoding="utf-8-sig")
+ )
+ self.assertFalse(resource_summary["sampling_valid"], resource_summary)
+ self.assertEqual(
+ resource_summary["sampling_problems"], ["nvdec_scan_argv_invalid"]
+ )
+ self.assertTrue(resource_summary["all_windows_host_cpu_strictly_below_85"])
+ self.assertTrue(resource_summary["window_count"] > 0)
+ self.assertFalse(any(
+ "previous_timestamp" in problem or "interval_mismatch" in problem
+ for problem in resource_summary["sampling_problems"]
+ ))
+ for index, row in enumerate(rows):
+ if index == 0:
+ self.assertEqual(row["previous_valid_timestamp_utc"], "")
+ self.assertAlmostEqual(float(row["interval_seconds"]), 0.0, places=6)
+ continue
+ current = datetime.fromisoformat(row["timestamp_utc"].replace("Z", "+00:00"))
+ previous = datetime.fromisoformat(
+ rows[index - 1]["timestamp_utc"].replace("Z", "+00:00")
+ )
+ reported_previous = datetime.fromisoformat(
+ row["previous_valid_timestamp_utc"].replace("Z", "+00:00")
+ )
+ self.assertLess(abs((reported_previous - previous).total_seconds()), 0.001)
+ self.assertLess(
+ abs(float(row["interval_seconds"]) - (current - previous).total_seconds()),
+ 0.001,
+ )
+ active_indexes = [
+ index for index, row in enumerate(rows)
+ if row["scan_phase_active"].lower() == "true"
+ ]
+ self.assertGreaterEqual(
+ len(active_indexes), 2,
+ json.dumps({"rows": rows, "run_summary": run_summary}, ensure_ascii=False),
+ )
+ self.assertEqual(len({rows[index]["scan_ffmpeg_pid"] for index in active_indexes}), 1)
+ last_active = active_indexes[-1]
+ self.assertEqual(rows[last_active]["scan_ffmpeg_present"].lower(), "false")
+ self.assertEqual(rows[last_active]["scan_ffmpeg_exit_observed"].lower(), "true")
+ later_rows = rows[last_active + 1 :]
+ self.assertTrue(later_rows)
+ self.assertTrue(any("-frames:v" in row["ffmpeg_command_lines"] for row in later_rows))
+ self.assertTrue(all(row["scan_phase_active"].lower() == "false" for row in later_rows))
+
+ @unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
+ def test_h264_h265_nvdec_and_cpu_product_chains_are_equivalent(self):
+ ffmpeg = shutil.which("ffmpeg")
+ capabilities = subprocess.check_output(
+ [ffmpeg, "-hide_banner", "-encoders"], text=True, errors="replace"
+ ) + subprocess.check_output(
+ [ffmpeg, "-hide_banner", "-decoders"], text=True, errors="replace"
+ )
+ required = ("libx264", "libx265", "h264_cuvid", "hevc_cuvid")
+ missing = [name for name in required if name not in capabilities]
+ if missing:
+ self.skipTest(f"missing FFmpeg capabilities: {missing}")
+
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ frames_dir = root / "frames"
+ frames_dir.mkdir()
+ size = (320, 180)
+
+ def page(kind, animated=False):
+ colors = {"a": (245, 245, 240), "b": (20, 45, 85), "c": (232, 210, 70)}
+ image = Image.new("RGB", size, colors[kind])
+ draw = ImageDraw.Draw(image)
+ draw.rectangle((0, 0, 319, 30), fill=(25, 90, 160))
+ draw.rectangle((30, 55, 285, 70), fill=(30, 30, 30))
+ if kind != "a" or animated:
+ draw.rectangle((30, 95, 250, 110), fill=(35, 35, 35))
+ if animated:
+ draw.rectangle((260, 140, 280, 160), fill=(210, 30, 30))
+ return image
+
+ transition_colors = ((100, 100, 100), (120, 25, 130), (20, 150, 90), (180, 65, 20))
+ frames = [page("a")] * 3 + [Image.new("RGB", size, transition_colors[0])]
+ frames += [page("a", True)] * 3 + [Image.new("RGB", size, transition_colors[1])]
+ frames += [page("b")] * 3 + [Image.new("RGB", size, transition_colors[2])]
+ frames += [page("a", True)] * 3 + [Image.new("RGB", size, transition_colors[3])]
+ frames += [page("c")] * 3
+ for index, image in enumerate(frames):
+ image.save(frames_dir / f"frame_{index:03d}.png")
+
+ codec_settings = {
+ "h264": ("libx264", "h264_cuvid"),
+ "hevc": ("libx265", "hevc_cuvid"),
+ }
+ evidence = []
+ for codec, (encoder, decoder) in codec_settings.items():
+ source = root / f"sequence-{codec}.mp4"
+ command = [
+ ffmpeg, "-hide_banner", "-loglevel", "error", "-framerate", "1",
+ "-i", str(frames_dir / "frame_%03d.png"), "-threads", "4",
+ "-c:v", encoder, "-preset", "ultrafast",
+ ]
+ if codec == "h264":
+ command.extend(("-qp", "0"))
+ else:
+ command.extend(("-x265-params", "lossless=1:pools=4"))
+ command.extend(("-pix_fmt", "yuv420p", "-n", str(source)))
+ subprocess.run(command, check=True, timeout=120)
+ probe_staging = root / f"probe-{codec}"
+ probe_staging.mkdir()
+ info = slides.probe_video(source, probe_staging)
+ self.assertEqual(info.codec_name, codec)
+ self.assertEqual(info.pixel_format, "yuv420p")
+
+ hardware_staging = root / f"scan-{codec}-hardware"
+ cpu_staging = root / f"scan-{codec}-cpu"
+ hardware_staging.mkdir()
+ cpu_staging.mkdir()
+ hardware = slides._scan_video_once(
+ source, hardware_staging, info, decoder=decoder
+ )
+ cpu = slides._scan_video_once(source, cpu_staging, info, decoder=None)
+ self.assertEqual(hardware.sample_count, cpu.sample_count)
+ self.assertEqual(hardware.stable_candidate_count, cpu.stable_candidate_count)
+ self.assertEqual(hardware.classification_counts, cpu.classification_counts)
+ self.assertEqual(len(hardware.pages), len(cpu.pages))
+ hardware_selected = slides.select_main_sequence(hardware.pages)
+ cpu_selected = slides.select_main_sequence(cpu.pages)
+ self.assertEqual(
+ [page.first_seen_order for page in hardware_selected],
+ [page.first_seen_order for page in cpu_selected],
+ )
+ self.assertEqual(len(hardware_selected), len(cpu_selected))
+ self.assertGreaterEqual(len(hardware_selected), 3)
+ for hardware_page, cpu_page in zip(hardware_selected, cpu_selected):
+ self.assertAlmostEqual(
+ hardware_page.current_timestamp,
+ cpu_page.current_timestamp,
+ delta=0.001,
+ )
+
+ product_summaries = []
+ for label, selected in (("hardware", hardware_selected), ("cpu", cpu_selected)):
+ output = root / f"product-{codec}-{label}"
+ output.mkdir()
+ pngs = slides.extract_original_pages(source, output, info, selected)
+ pdf = output / "slides.pdf"
+ slides.write_image_pdf(pngs, pdf)
+ png_summary = []
+ for path in pngs:
+ with Image.open(path) as image:
+ png_summary.append((path.name, image.mode, image.size))
+ product_summaries.append((png_summary, pdf_rgb_pages(pdf)))
+ self.assertEqual(product_summaries[0][0], product_summaries[1][0])
+ self.assertEqual(
+ [page[0] for page in product_summaries[0][1]],
+ [page[0] for page in product_summaries[1][1]],
+ )
+ self.assertEqual(
+ len(product_summaries[0][1]), len(product_summaries[1][1])
+ )
+ hardware_command = slides._scan_command(
+ ffmpeg, source, *slides._scan_size(info), decoder
+ )
+ cpu_command = slides._scan_command(
+ ffmpeg, source, *slides._scan_size(info), None
+ )
+ evidence.extend((
+ {
+ "codec": codec,
+ "path": "nvdec",
+ "decoder": decoder,
+ "argv": hardware_command,
+ "samples": hardware.sample_count,
+ "stable_candidates": hardware.stable_candidate_count,
+ "classifications": dict(hardware.classification_counts),
+ "selected_pages": len(hardware_selected),
+ "timestamps": [page.current_timestamp for page in hardware_selected],
+ },
+ {
+ "codec": codec,
+ "path": "cpu",
+ "decoder": None,
+ "argv": cpu_command,
+ "samples": cpu.sample_count,
+ "stable_candidates": cpu.stable_candidate_count,
+ "classifications": dict(cpu.classification_counts),
+ "selected_pages": len(cpu_selected),
+ "timestamps": [page.current_timestamp for page in cpu_selected],
+ },
+ ))
+ print("MEDIA_HOST_SYNTHETIC_EVIDENCE=" + json.dumps(evidence, ensure_ascii=False))
@unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
def test_vfr_nonzero_pts_scan_and_exact_seek_share_relative_time(self):
@@ -281,10 +1222,9 @@
source = root / "source.mp4"
source.write_bytes(b"stub")
output = root / "result"
- page_feature = np.zeros((8, 8), np.uint8)
- logical_page = slides.LogicalPage(page_feature, page_feature, 0, 1.0, 0.2, 0)
+ page = logical_page(0)
with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
- slides, "scan_video", return_value=[logical_page]
+ slides, "scan_video", return_value=[page]
), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
slides.subprocess, "Popen", side_effect=make_process
):
@@ -575,6 +1515,46 @@
self.assertFalse(output.exists())
self.assertFalse(any(root.glob(".result.staging-*")))
+ def test_main_sequence_filter_runs_once_before_original_extraction(self):
+ with tempfile.TemporaryDirectory() as raw:
+ root = Path(raw)
+ source = root / "source.mp4"
+ source.write_bytes(b"stub")
+ output = root / "result"
+ candidates = [logical_page(index, palette(1)) for index in range(10)]
+ selected = candidates[1:9]
+ with mock.patch.object(
+ slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)
+ ), mock.patch.object(
+ slides, "scan_video", return_value=candidates
+ ), mock.patch.object(
+ slides, "select_main_sequence", return_value=selected
+ ) as sequence_filter, mock.patch.object(
+ slides, "extract_original_pages", return_value=[]
+ ) as original_extraction, mock.patch.object(slides, "write_image_pdf"):
+ self.assertEqual(slides.extract_slides(str(source), str(output)), output)
+ sequence_filter.assert_called_once_with(candidates)
+ self.assertIs(original_extraction.call_args.args[3], selected)
+
+ def test_selected_pages_are_renumbered_continuously(self):
+ with tempfile.TemporaryDirectory() as raw:
+ staging = Path(raw)
+ selected = [logical_page(index, palette(1)) for index in range(3, 6)]
+
+ def write_png(command, stderr_path, stage):
+ destination = Path(command[-1])
+ Image.new("RGB", (64, 36), (20, 40, 80)).save(destination)
+
+ with mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
+ slides, "_run_file_command", side_effect=write_png
+ ):
+ outputs = slides.extract_original_pages(
+ Path("source.mp4"), staging, slides.VideoInfo(64, 36, 10), selected
+ )
+ self.assertEqual([path.name for path in outputs], [
+ "slide_001.png", "slide_002.png", "slide_003.png"
+ ])
+
def test_commit_conflict_is_wrapped_and_staging_is_removed(self):
with tempfile.TemporaryDirectory() as raw:
root = Path(raw)
@@ -582,7 +1562,7 @@
source.write_bytes(b"stub")
output = root / "result"
with mock.patch.object(slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)), mock.patch.object(
- slides, "scan_video", return_value=[]
+ slides, "scan_video", return_value=[logical_page(0)]
), mock.patch.object(slides, "extract_original_pages", return_value=[]), mock.patch.object(
slides, "write_image_pdf"
), mock.patch.object(slides.os, "rename", side_effect=FileExistsError("race")):
--
Gitblit v1.9.3