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
|
import shutil
|
import subprocess
|
import sys
|
import tempfile
|
import textwrap
|
import unittest
|
from unittest import mock
|
import zlib
|
|
import numpy as np
|
from PIL import Image, ImageDraw
|
|
|
MODULE_PATH = Path(__file__).resolve().parents[1] / "extract_ppt_slides.py"
|
SPEC = importlib.util.spec_from_file_location("extract_ppt_slides", MODULE_PATH)
|
assert SPEC and SPEC.loader
|
slides = importlib.util.module_from_spec(SPEC)
|
sys.modules[SPEC.name] = slides
|
SPEC.loader.exec_module(slides)
|
|
|
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,
|
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:
|
digest = hashlib.sha256()
|
with path.open("rb") as handle:
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
digest.update(chunk)
|
return digest.hexdigest().upper()
|
|
|
def pdf_rgb_pages(path: Path):
|
data = path.read_bytes()
|
objects = {
|
int(match.group(1)): match.group(2)
|
for match in re.finditer(rb"(?m)^(\d+) 0 obj\n(.*?)\nendobj\n", data, re.DOTALL)
|
}
|
pages_body = objects[2]
|
kids_match = re.search(rb"/Kids \[(.*?)\]", pages_body, re.DOTALL)
|
assert kids_match
|
page_ids = [int(value) for value in re.findall(rb"(\d+) 0 R", kids_match.group(1))]
|
result = []
|
for page_id in page_ids:
|
page = objects[page_id]
|
media = re.search(rb"/MediaBox \[0 0 (\d+) (\d+)\]", page)
|
image_ref = re.search(rb"/Im0 (\d+) 0 R", page)
|
assert media and image_ref
|
image_body = objects[int(image_ref.group(1))]
|
length = int(re.search(rb"/Length (\d+)", image_body).group(1))
|
marker = b"stream\n"
|
start = image_body.index(marker) + len(marker)
|
compressed = image_body[start : start + length]
|
result.append(((int(media.group(1)), int(media.group(2))), zlib.decompress(compressed)))
|
return result
|
|
|
class ShortReadStream:
|
def __init__(self, chunks):
|
self.chunks = list(chunks)
|
|
def read(self, size):
|
if not self.chunks:
|
return b""
|
chunk = self.chunks.pop(0)
|
if len(chunk) <= size:
|
return chunk
|
self.chunks.insert(0, chunk[size:])
|
return chunk[:size]
|
|
|
class ReadExactTests(unittest.TestCase):
|
def test_short_reads_clean_eof_and_partial_eof(self):
|
self.assertEqual(slides.read_exact(ShortReadStream([b"a", b"bc", b"def"]), 6), b"abcdef")
|
self.assertIsNone(slides.read_exact(ShortReadStream([]), 6))
|
with self.assertRaisesRegex(slides.SlideExtractionError, "半帧 EOF"):
|
slides.read_exact(ShortReadStream([b"abc"]), 6)
|
|
|
class SegmentAndClassifierTests(unittest.TestCase):
|
def test_anchor_drift_emits_only_old_and_new_plateaus(self):
|
detector = slides.StableSegmentDetector()
|
emitted = []
|
arrays = [np.full((8, 8), 20, np.uint8)] * 3
|
arrays += [np.full((8, 8), value, np.uint8) for value in (22, 24, 26, 28, 30, 32)]
|
arrays += [np.full((8, 8), 80, np.uint8)] * 3
|
for index, array in enumerate(arrays):
|
value = detector.add(feature(index, array, score=index / 100, sharpness=index))
|
if value:
|
emitted.append(value)
|
final = detector.flush()
|
if final:
|
emitted.append(final)
|
self.assertEqual([round(float(np.mean(item.feature))) for item in emitted], [20, 80])
|
|
def test_animation_increase_decrease_and_equal_score(self):
|
classifier = slides.PageClassifier()
|
base = np.zeros((8, 8), np.uint8)
|
added = base.copy()
|
added[:, :2] = 20
|
self.assertEqual(classifier.consume(candidate(1, base, 0.20)), "new_page")
|
self.assertEqual(classifier.consume(candidate(2, added, 0.30)), "animation_updated")
|
self.assertEqual(classifier.pages[0].current_timestamp, 2)
|
reduced = base.copy()
|
self.assertEqual(classifier.consume(candidate(3, reduced, 0.10)), "animation_retained")
|
self.assertEqual(classifier.pages[0].current_timestamp, 2)
|
self.assertEqual(
|
classifier.consume(candidate(4, added, 0.30 + slides.CONTENT_SCORE_TOLERANCE / 2)),
|
"animation_updated",
|
)
|
self.assertEqual(classifier.pages[0].current_timestamp, 4)
|
|
def test_a_b_a_keeps_first_order(self):
|
classifier = slides.PageClassifier()
|
a = np.zeros((8, 8), np.uint8)
|
b = np.full((8, 8), 255, np.uint8)
|
classifier.consume(candidate(1, a, 0.2, 0))
|
classifier.consume(candidate(2, b, 0.2, (1 << 64) - 1))
|
action = classifier.consume(candidate(3, a, 0.2, 0))
|
self.assertEqual(action, "historical_duplicate")
|
self.assertEqual(len(classifier.pages), 2)
|
self.assertEqual([page.first_seen_order for page in classifier.pages], [0, 1])
|
|
def test_recent_animation_has_priority_over_historical_duplicate(self):
|
classifier = slides.PageClassifier()
|
historical = np.zeros((8, 8), np.uint8)
|
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(), 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):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
video = root / "vfr-nonzero.mkv"
|
subprocess.run(
|
[
|
shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error",
|
"-f", "lavfi", "-i", "color=red:size=64x64:rate=2:duration=2",
|
"-f", "lavfi", "-i", "color=green:size=64x64:rate=3:duration=2",
|
"-f", "lavfi", "-i", "color=blue:size=64x64:rate=4:duration=2",
|
"-filter_complex", "[0:v][1:v][2:v]concat=n=3:v=1:a=0,setpts=PTS+5/TB[v]",
|
"-map", "[v]", "-fps_mode", "vfr", "-c:v", "ffv1", str(video),
|
],
|
check=True,
|
)
|
start_time = float(
|
subprocess.check_output(
|
[
|
shutil.which("ffprobe"), "-v", "error", "-select_streams", "v:0",
|
"-show_entries", "stream=start_time", "-of", "default=nw=1:nk=1", str(video),
|
],
|
text=True,
|
).strip()
|
)
|
self.assertGreaterEqual(start_time, 5.0)
|
|
scan_process = subprocess.Popen(
|
slides._scan_command(shutil.which("ffmpeg"), video, 64, 64),
|
stdin=subprocess.DEVNULL,
|
stdout=subprocess.PIPE,
|
stderr=subprocess.PIPE,
|
)
|
assert scan_process.stdout is not None
|
scanned = []
|
while True:
|
payload = slides.read_exact(scan_process.stdout, 64 * 64 * 3)
|
if payload is None:
|
break
|
scanned.append(np.frombuffer(payload, np.uint8).reshape((64, 64, 3)))
|
scan_process.stdout.close()
|
stderr = scan_process.stderr.read() if scan_process.stderr else b""
|
if scan_process.stderr:
|
scan_process.stderr.close()
|
self.assertEqual(scan_process.wait(), 0, stderr.decode(errors="replace"))
|
self.assertGreaterEqual(len(scanned), 5)
|
|
extracted = []
|
for timestamp in (0.0, 2.0, 4.0):
|
path = root / f"at-{timestamp:.0f}.png"
|
slides._run_file_command(
|
slides._extract_command(shutil.which("ffmpeg"), video, timestamp, path),
|
root / f"at-{timestamp:.0f}.stderr.log",
|
"VFR 定位",
|
)
|
with Image.open(path) as image:
|
extracted.append(np.asarray(image.convert("RGB"), np.uint8))
|
|
for sequence in ((scanned[0], scanned[2], scanned[4]), tuple(extracted)):
|
means = [np.mean(item, axis=(0, 1)) for item in sequence]
|
self.assertEqual([int(np.argmax(value)) for value in means], [0, 1, 2])
|
|
@unittest.skipUnless(shutil.which("ffmpeg"), "FFmpeg is required")
|
def test_large_stderr_does_not_deadlock_and_nonzero_is_reported(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
ok_log = root / "ok.log"
|
slides._run_file_command(
|
[sys.executable, "-c", "import sys; sys.stderr.write('x'*300000)"],
|
ok_log,
|
"桩命令",
|
)
|
self.assertEqual(ok_log.stat().st_size, 300000)
|
bad_log = root / "bad.log"
|
with self.assertRaisesRegex(slides.SlideExtractionError, "exit=7"):
|
slides._run_file_command(
|
[sys.executable, "-c", "import sys; sys.stderr.write('boom'); raise SystemExit(7)"],
|
bad_log,
|
"桩命令",
|
)
|
|
def test_keyboard_interrupt_reaps_file_process_and_preserves_type(self):
|
events = []
|
handles = []
|
|
class FakeProcess:
|
stdout = None
|
|
def __init__(self):
|
self.wait_calls = 0
|
self.running = True
|
|
def wait(self, timeout=None):
|
self.wait_calls += 1
|
events.append(("wait", timeout))
|
if self.wait_calls == 1:
|
raise KeyboardInterrupt()
|
self.running = False
|
return 0
|
|
def poll(self):
|
return None if self.running else 0
|
|
def terminate(self):
|
events.append(("terminate", None))
|
|
def kill(self):
|
events.append(("kill", None))
|
|
def make_process(*args, **kwargs):
|
handles.append(kwargs["stderr"])
|
return FakeProcess()
|
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"stub")
|
output = root / "result"
|
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=[page]
|
), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
|
slides.subprocess, "Popen", side_effect=make_process
|
):
|
with self.assertRaises(KeyboardInterrupt):
|
slides.extract_slides(str(source), str(output))
|
self.assertFalse(output.exists())
|
self.assertFalse(any(root.glob(".result.staging-*")))
|
self.assertTrue(handles[0].closed)
|
self.assertEqual(events[1][0], "terminate")
|
self.assertEqual(events[2][0], "wait")
|
|
def test_keyboard_interrupt_reaps_scan_process_and_preserves_type(self):
|
events = []
|
|
class Stdout:
|
def read(self, size):
|
raise KeyboardInterrupt()
|
|
def close(self):
|
events.append("close")
|
|
class FakeProcess:
|
def __init__(self):
|
self.stdout = Stdout()
|
self.running = True
|
|
def wait(self, timeout=None):
|
events.append("wait")
|
self.running = False
|
return 0
|
|
def poll(self):
|
return None if self.running else 0
|
|
def terminate(self):
|
events.append("terminate")
|
|
def kill(self):
|
events.append("kill")
|
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"stub")
|
output = root / "result"
|
with mock.patch.object(
|
slides.subprocess, "Popen", return_value=FakeProcess()
|
), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"), mock.patch.object(
|
slides, "probe_video", return_value=slides.VideoInfo(640, 360, 10)
|
):
|
with self.assertRaises(KeyboardInterrupt):
|
slides.extract_slides(str(source), str(output))
|
self.assertFalse(output.exists())
|
self.assertFalse(any(root.glob(".result.staging-*")))
|
self.assertEqual(events[:3], ["close", "terminate", "wait"])
|
|
def test_stderr_open_failure_is_wrapped_with_stage(self):
|
with tempfile.TemporaryDirectory() as raw, mock.patch.object(
|
slides.Path, "open", side_effect=OSError("denied")
|
):
|
with self.assertRaisesRegex(slides.SlideExtractionError, "桩阶段失败：denied"):
|
slides._run_file_command(["fake"], Path(raw) / "stderr.log", "桩阶段")
|
|
def test_cleanup_error_does_not_replace_keyboard_interrupt(self):
|
class FakeProcess:
|
stdout = None
|
|
def wait(self, timeout=None):
|
raise KeyboardInterrupt()
|
|
def poll(self):
|
return None
|
|
def terminate(self):
|
raise OSError("cleanup denied")
|
|
with tempfile.TemporaryDirectory() as raw, mock.patch.object(
|
slides.subprocess, "Popen", return_value=FakeProcess()
|
):
|
with self.assertRaises(KeyboardInterrupt):
|
slides._run_file_command(["fake"], Path(raw) / "stderr.log", "桩阶段")
|
|
def test_stderr_close_error_preserves_interrupt_in_both_helpers(self):
|
class CloseFailingHandle:
|
def close(self):
|
raise OSError("stderr close denied")
|
|
def flush(self):
|
return None
|
|
class FileProcess:
|
stdout = None
|
|
def __init__(self):
|
self.calls = 0
|
|
def wait(self, timeout=None):
|
self.calls += 1
|
if self.calls == 1:
|
raise KeyboardInterrupt()
|
return 0
|
|
def poll(self):
|
return None
|
|
def terminate(self):
|
return None
|
|
class ScanStdout:
|
def read(self, size):
|
raise KeyboardInterrupt()
|
|
def close(self):
|
return None
|
|
class ScanProcess:
|
def __init__(self):
|
self.stdout = ScanStdout()
|
|
def wait(self, timeout=None):
|
return 0
|
|
def poll(self):
|
return None
|
|
def terminate(self):
|
return None
|
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
|
slides.subprocess, "Popen", return_value=FileProcess()
|
):
|
with self.assertRaises(KeyboardInterrupt):
|
slides._run_file_command(["fake"], root / "file.stderr.log", "文件阶段")
|
with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
|
slides.subprocess, "Popen", return_value=ScanProcess()
|
), mock.patch.object(slides, "_which_or_error", return_value="ffmpeg"):
|
with self.assertRaises(KeyboardInterrupt):
|
slides.scan_video(Path("x.mp4"), root, slides.VideoInfo(640, 360, 10))
|
|
def test_stderr_close_error_is_staged_on_normal_and_ordinary_failure(self):
|
class CloseFailingHandle:
|
def close(self):
|
raise OSError("stderr close denied")
|
|
def flush(self):
|
return None
|
|
class FakeProcess:
|
stdout = None
|
|
def __init__(self, code):
|
self.code = code
|
|
def wait(self, timeout=None):
|
return self.code
|
|
def poll(self):
|
return self.code
|
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
|
slides.subprocess, "Popen", return_value=FakeProcess(0)
|
):
|
with self.assertRaisesRegex(slides.SlideExtractionError, "文件阶段关闭 stderr 失败"):
|
slides._run_file_command(["fake"], root / "normal.log", "文件阶段")
|
with mock.patch.object(slides.Path, "open", return_value=CloseFailingHandle()), mock.patch.object(
|
slides.subprocess, "Popen", return_value=FakeProcess(7)
|
), mock.patch.object(slides, "_tail", return_value="boom"):
|
with self.assertRaisesRegex(slides.SlideExtractionError, "文件阶段失败（exit=7）"):
|
slides._run_file_command(["fake"], root / "failed.log", "文件阶段")
|
|
|
class PdfTests(unittest.TestCase):
|
def test_pdf_rgb_content_and_order(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
paths = []
|
for index, color in enumerate(((255, 20, 20), (20, 40, 255))):
|
path = root / f"{index}.png"
|
Image.new("RGB", (17 + index, 11 + index), color).save(path)
|
paths.append(path)
|
pdf = root / "slides.pdf"
|
slides.write_image_pdf(paths, pdf)
|
decoded = pdf_rgb_pages(pdf)
|
self.assertEqual([item[0] for item in decoded], [(17, 11), (18, 12)])
|
for path, (_, rgb) in zip(paths, decoded):
|
with Image.open(path) as image:
|
self.assertEqual(hashlib.sha256(rgb).digest(), hashlib.sha256(image.convert("RGB").tobytes()).digest())
|
pdfinfo = shutil.which("pdfinfo")
|
if pdfinfo and Path(pdfinfo).suffix.lower() not in {".cmd", ".bat"}:
|
result = subprocess.run([pdfinfo, str(pdf)], capture_output=True, text=True, check=True)
|
self.assertIn("Pages: 2", result.stdout)
|
|
def test_twenty_pages_keep_one_decoded_image_active(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
paths = []
|
for index in range(20):
|
path = root / f"{index:02d}.png"
|
Image.new("RGB", (12, 8), (index, index * 2, index * 3)).save(path)
|
paths.append(path)
|
active = 0
|
maximum = 0
|
|
class CountedOpen:
|
def __init__(self, path):
|
self.image = Image.open(path)
|
|
def __enter__(self):
|
nonlocal active, maximum
|
active += 1
|
maximum = max(maximum, active)
|
return self.image
|
|
def __exit__(self, exc_type, exc, traceback):
|
nonlocal active
|
self.image.close()
|
active -= 1
|
|
pdf = root / "twenty.pdf"
|
slides.write_image_pdf(paths, pdf, image_opener=CountedOpen)
|
self.assertEqual(maximum, 1)
|
self.assertEqual(active, 0)
|
self.assertEqual(len(pdf_rgb_pages(pdf)), 20)
|
|
|
class SafetyAndIntegrationTests(unittest.TestCase):
|
def test_existing_output_is_rejected_before_external_process(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"not a video")
|
for kind in ("directory", "file"):
|
with self.subTest(kind=kind):
|
output = root / f"existing-{kind}"
|
if kind == "directory":
|
output.mkdir()
|
else:
|
output.write_bytes(b"occupied")
|
with mock.patch.object(slides.subprocess, "Popen") as popen:
|
with self.assertRaisesRegex(slides.SlideExtractionError, "默认不覆盖"):
|
slides.extract_slides(str(source), str(output))
|
popen.assert_not_called()
|
|
def test_live_and_broken_output_links_are_rejected_by_lexical_precheck(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"stub")
|
for link_kind in ("live", "broken"):
|
with self.subTest(link_kind=link_kind):
|
output = slides._absolute_lexical(root / f"{link_kind}-link")
|
original_lexists = slides._lexists
|
|
def controlled_lexists(path):
|
return True if path == output else original_lexists(path)
|
|
with mock.patch.object(slides, "_lexists", side_effect=controlled_lexists), mock.patch.object(
|
slides.subprocess, "Popen"
|
) as popen:
|
with self.assertRaisesRegex(slides.SlideExtractionError, "默认不覆盖"):
|
slides.extract_slides(str(source), str(output))
|
popen.assert_not_called()
|
|
def test_input_symlink_is_rejected_before_external_process(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = slides._absolute_lexical(root / "source-link.mp4")
|
with mock.patch.object(slides, "_is_symlink", side_effect=lambda path: path == source), mock.patch.object(
|
slides.subprocess, "Popen"
|
) as popen:
|
with self.assertRaisesRegex(slides.SlideExtractionError, "拒绝符号链接"):
|
slides.extract_slides(str(source), str(root / "result"))
|
popen.assert_not_called()
|
|
def test_baseexception_removes_staging_and_leaves_no_formal_output(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw)
|
source = root / "source.mp4"
|
source.write_bytes(b"stub")
|
output = root / "result"
|
with mock.patch.object(slides, "probe_video", side_effect=KeyboardInterrupt()):
|
with self.assertRaises(KeyboardInterrupt):
|
slides.extract_slides(str(source), str(output))
|
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)
|
source = root / "source.mp4"
|
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=[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")):
|
with self.assertRaisesRegex(slides.SlideExtractionError, "提交正式输出目录失败：race"):
|
slides.extract_slides(str(source), str(output))
|
self.assertFalse(output.exists())
|
self.assertFalse(any(root.glob(".result.staging-*")))
|
|
@unittest.skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg is required")
|
def test_chinese_path_synthetic_video_end_to_end(self):
|
with tempfile.TemporaryDirectory() as raw:
|
root = Path(raw) / "中文路径"
|
root.mkdir()
|
frame_root = root / "frames"
|
frame_root.mkdir()
|
size = (640, 360)
|
|
def page_a(complete: bool):
|
image = Image.new("RGB", size, "white")
|
draw = ImageDraw.Draw(image)
|
draw.rectangle((0, 0, 639, 65), fill=(26, 78, 145))
|
draw.rectangle((55, 100, 585, 125), fill=(30, 30, 30))
|
if complete:
|
draw.rectangle((55, 165, 470, 190), fill=(30, 30, 30))
|
draw.rectangle((55, 230, 540, 255), fill=(30, 30, 30))
|
draw.rectangle((570, 320, 580, 330), fill=(220, 20, 20))
|
return image
|
|
def page_b():
|
image = Image.new("RGB", size, (18, 34, 68))
|
draw = ImageDraw.Draw(image)
|
draw.rectangle((65, 75, 575, 285), fill=(230, 190, 45))
|
draw.rectangle((110, 125, 530, 155), fill=(18, 34, 68))
|
return image
|
|
frames = [page_a(False)] * 3
|
frames += [Image.new("RGB", size, (127, 127, 127))]
|
frames += [page_a(True)] * 3
|
frames += [Image.new("RGB", size, (90, 20, 130))]
|
frames += [page_b()] * 3
|
frames += [Image.new("RGB", size, (20, 150, 90))]
|
frames += [page_a(True)] * 3
|
for index, image in enumerate(frames):
|
image.save(frame_root / f"frame_{index:03d}.png")
|
|
source = root / "合成会议录屏.mp4"
|
subprocess.run(
|
[
|
shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error", "-framerate", "1",
|
"-i", str(frame_root / "frame_%03d.png"), "-c:v", "libx264rgb", "-crf", "0", "-pix_fmt", "rgb24", str(source),
|
],
|
check=True,
|
)
|
before = (source.stat().st_size, source.stat().st_ctime_ns, source.stat().st_mtime_ns, sha256(source))
|
output = root / "验收输出"
|
result = subprocess.run(
|
[sys.executable, str(MODULE_PATH), str(source), "--output", str(output)],
|
capture_output=True,
|
text=True,
|
timeout=120,
|
)
|
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
|
self.assertEqual(
|
{entry.name for entry in output.iterdir()},
|
{"pages", "合成会议录屏.slides.pdf"},
|
)
|
pngs = sorted((output / "pages").glob("slide_*.png"))
|
self.assertEqual(len(pngs), 2, result.stdout)
|
for png in pngs:
|
with Image.open(png) as image:
|
self.assertEqual(image.size, size)
|
expected = [page_a(True), page_b()]
|
for actual_path, expected_image in zip(pngs, expected):
|
with Image.open(actual_path) as actual:
|
self.assertEqual(
|
hashlib.sha256(actual.convert("RGB").tobytes()).digest(),
|
hashlib.sha256(expected_image.tobytes()).digest(),
|
)
|
decoded = pdf_rgb_pages(output / "合成会议录屏.slides.pdf")
|
self.assertEqual(len(decoded), 2)
|
for png, (_, rgb) in zip(pngs, decoded):
|
with Image.open(png) as image:
|
self.assertEqual(hashlib.sha256(rgb).digest(), hashlib.sha256(image.convert("RGB").tobytes()).digest())
|
after = (source.stat().st_size, source.stat().st_ctime_ns, source.stat().st_mtime_ns, sha256(source))
|
self.assertEqual(after, before)
|
self.assertFalse(any(root.glob(".验收输出.staging-*")))
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
