tests/test_face_detect.py
Rama master · réplica de lectura del repositorio original.
import hashlib
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from peach import face_detect
from peach.face_detect import (
MODEL_SHA256,
FaceModelUnavailable,
ensure_model,
)
class EnsureModelTests(unittest.TestCase):
"""模型是二进制资产，不进 Git；取一次就得能复现同一份。"""
def test_a_matching_local_model_is_reused_without_network(self):
with tempfile.TemporaryDirectory() as temp:
target = Path(temp).resolve() / "yunet.onnx"
payload = b"model-bytes"
target.write_bytes(payload)
with patch.object(face_detect, "MODEL_SHA256",
hashlib.sha256(payload).hexdigest()), \
patch.object(face_detect, "read_url") as opener:
self.assertEqual(ensure_model(target), target)
opener.assert_not_called()
def test_a_half_written_model_is_replaced_not_trusted(self):
"""半个下载同样留下一个文件。
只看「文件在不在」的话，下一步 `FaceDetectorYN_create` 会抛一个和网络
毫无关系的错，排查会从模型格式开始找——而真正的原因是上一次没下完。
"""
with tempfile.TemporaryDirectory() as temp:
target = Path(temp).resolve() / "yunet.onnx"
target.write_bytes(b"half")
with self.assertRaises(FaceModelUnavailable):
ensure_model(target, allow_download=False)
self.assertFalse(target.exists(), "校验不过的文件必须删掉，不能留着下次再骗一次")
def test_a_download_whose_digest_differs_is_refused(self):
with tempfile.TemporaryDirectory() as temp:
target = Path(temp).resolve() / "yunet.onnx"
with patch.object(face_detect, "read_url", return_value=b"not-the-model"):
with self.assertRaises(FaceModelUnavailable) as caught:
ensure_model(target)
self.assertIn("校验不符", str(caught.exception))
self.assertFalse(target.exists())
def test_the_pinned_digest_is_a_real_sha256(self):
# 定版是这条复用记录的全部意义：地址钉在 opencv_zoo 的一个提交上，哈希再核一遍内容。
self.assertEqual(len(MODEL_SHA256), 64)
self.assertTrue(all(ch in "0123456789abcdef" for ch in MODEL_SHA256))
self.assertIn("media.githubusercontent.com", face_detect.MODEL_URL,
"opencv_zoo 用 Git LFS，raw 域名只会回一个指针文件")
self.assertIn(face_detect.ZOO_COMMIT, face_detect.MODEL_URL)
class ModelLoadingTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.model = Path(temporary.name).resolve() / "中文模型" / "检出.onnx"
self.model.parent.mkdir()
self.payload = b"verified-yunet-model"
self.model.write_bytes(self.payload)
digest = patch.object(face_detect, "MODEL_SHA256", hashlib.sha256(self.payload).hexdigest())
digest.start()
self.addCleanup(digest.stop)
def test_unicode_model_path_is_read_as_onnx_bytes(self):
import cv2
import numpy
with patch.object(cv2, "FaceDetectorYN_create") as create:
detector = face_detect.FaceDetector(self.model, score=0.72, nms=0.12, allow_download=False)
args = create.call_args.args
self.assertEqual(args[0], "ONNX")
self.assertIsInstance(args[1], numpy.ndarray)
self.assertEqual((args[1].dtype, args[1].ndim, args[1].tobytes()),
(numpy.dtype("uint8"), 1, self.payload))
self.assertEqual((args[2].dtype, args[2].size), (numpy.dtype("uint8"), 0))
self.assertEqual(args[3:], ((320, 320), face_detect.READ_FLOOR, 0.12, 5000))
self.assertEqual(detector.model_path, self.model)
self.assertIs(detector._detector, create.return_value)
def test_model_read_error_after_verification_reaches_the_caller(self):
import cv2
failure = PermissionError(13, "model read denied")
with patch.object(Path, "read_bytes", side_effect=[self.payload, failure]), \
patch.object(cv2, "FaceDetectorYN_create") as create, \
self.assertRaises(PermissionError) as caught:
face_detect.FaceDetector(self.model, allow_download=False)
self.assertIs(caught.exception, failure)
create.assert_not_called()
def test_an_onnx_import_error_is_not_reported_as_no_faces(self):
import cv2
failure = cv2.error("invalid ONNX buffer")
with patch.object(cv2, "FaceDetectorYN_create", side_effect=failure), \
self.assertRaises(cv2.error) as caught:
face_detect.FaceDetector(self.model, allow_download=False)
self.assertIs(caught.exception, failure)
def test_mismatched_model_bytes_never_reach_opencv(self):
import cv2
self.model.write_bytes(b"partial-model")
with patch.object(cv2, "FaceDetectorYN_create") as create, self.assertRaises(FaceModelUnavailable):
face_detect.FaceDetector(self.model, allow_download=False)
create.assert_not_called()
class BundledModelTests(unittest.TestCase):
"""随包那份先认：在且哈希对得上就不联网、不往数据根写；对不上就退回原来的取法。"""
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name).resolve()
self.bundle = self.root / "program" / "tools" / "face-models"
self.bundle.mkdir(parents=True)
self.data = self.root / "data" / "yunet" / MODEL_PATH_NAME
self.payload = b"bundled-yunet"
for patcher in (
patch.dict(face_detect.os.environ, {face_detect.BUNDLE_ENV: str(self.bundle)}),
patch.object(face_detect, "MODEL_PATH", self.data),
patch.object(face_detect, "MODEL_SHA256", hashlib.sha256(self.payload).hexdigest()),
patch.object(face_detect, "_verified", {}),
patch.object(face_detect, "read_url", side_effect=AssertionError("随包有模型时不该联网"))):
patcher.start()
self.addCleanup(patcher.stop)
def test_a_matching_bundled_model_is_used_offline_without_touching_the_data_root(self):
(self.bundle / MODEL_PATH_NAME).write_bytes(self.payload)
seen = []
before = face_detect.downloads
with patch.object(face_detect, "observer", lambda *args: seen.append(args)):
self.assertEqual(ensure_model(), self.bundle / MODEL_PATH_NAME)
self.assertFalse(self.data.exists(), "随包那份不往数据根复制")
self.assertEqual((seen, face_detect.downloads), ([], before), "随包命中不是联网取到的")
def test_a_mismatched_bundled_model_is_kept_and_the_data_root_copy_is_used(self):
(self.bundle / MODEL_PATH_NAME).write_bytes(b"tampered")
self.data.parent.mkdir(parents=True)
self.data.write_bytes(self.payload)
with self.assertLogs("peach.face_detect", "WARNING"):
self.assertEqual(ensure_model(), self.data)
self.assertEqual((self.bundle / MODEL_PATH_NAME).read_bytes(), b"tampered", "程序目录那份归安装包管，不删")
def test_without_any_copy_the_download_fallback_still_runs(self):
with patch.object(face_detect, "read_url", return_value=self.payload) as read:
self.assertEqual(ensure_model(), self.data)
read.assert_called_once()
def test_a_verified_bundled_model_is_hashed_once_per_process(self):
(self.bundle / MODEL_PATH_NAME).write_bytes(self.payload)
with patch.object(face_detect, "_digest", wraps=face_detect._digest) as digest:
for _ in range(3):
face_detect.bundled_model(MODEL_PATH_NAME, face_detect.MODEL_SHA256)
self.assertEqual(digest.call_count, 1)
def test_a_mismatched_bundled_model_is_hashed_and_reported_once_per_process(self):
"""活动中心轮询状态时，一份坏掉的随包模型不能每次都重算 37 MB、刷一条警告。"""
(self.bundle / MODEL_PATH_NAME).write_bytes(b"tampered")
with patch.object(face_detect, "_digest", wraps=face_detect._digest) as digest, \
self.assertLogs("peach.face_detect", "WARNING") as logs:
for _ in range(3):
self.assertIsNone(face_detect.bundled_model(MODEL_PATH_NAME, face_detect.MODEL_SHA256))
self.assertEqual((digest.call_count, len(logs.records)), (1, 1))
def test_the_frozen_desktop_package_reads_its_program_directory(self):
program = self.root / "Peach"
with patch.dict(face_detect.os.environ, {face_detect.BUNDLE_ENV: ""}), \
patch.object(face_detect.sys, "frozen", True, create=True), \
patch.object(face_detect.sys, "executable", str(program / "Peach.exe")):
self.assertEqual(face_detect.bundled_dir(), program.resolve() / "tools" / "face-models")
with patch.dict(face_detect.os.environ, {face_detect.BUNDLE_ENV: ""}):
self.assertIsNone(face_detect.bundled_dir(), "源码运行没有随包目录")
MODEL_PATH_NAME = face_detect.MODEL_PATH.name
class DetectorContractTests(unittest.TestCase):
def test_scripts_do_not_reach_for_the_removed_cascade_api(self):
"""OpenCV 5 的 wheel 里没有 `CascadeClassifier`，也没有级联 XML。
两个脚本此前一起门都进不去，954 张封面 0 个 sidecar——「人脸取景没生效」
的成因就在这里，不在取景公式上。
"""
root = Path(__file__).resolve().parents[1]
for name in ("detect_cover_faces.py", "detect_avatar_faces.py"):
source = (root / "scripts" / name).read_text(encoding="utf-8")
self.assertNotIn("CascadeClassifier", source, name)
self.assertNotIn("haarcascades", source, name)
self.assertIn("FaceDetector()", source, name)
def test_normalised_face_is_clamped_into_the_frame(self):
detector = face_detect.FaceDetector.__new__(face_detect.FaceDetector)
face = face_detect.Face(cx=0.8, cy=0.2, width=0.25, height=0.3, score=0.9)
self.assertAlmostEqual(face.area, 0.075, places=6)
self.assertIsInstance(detector, face_detect.FaceDetector)
class _Model:
"""假的 YuNet：按送检尺寸回不同的框，用来看几个尺度的读数怎么合成一个分。
`reads` 的键是送检时的长边，值是 `(cx, cy, w, h, score)`，坐标按那一档的像素给。
"""
def __init__(self, reads):
self._reads = reads
self._size = None
def setInputSize(self, size): # noqa: N802 - 对齐 OpenCV 的拼法
self._size = size
def detect(self, _image):
import numpy
cols, rows = self._size
rows_out = []
for box in self._reads.get(max(cols, rows), []):
cx, cy, width, height, score = box
rows_out.append([cx - width / 2, cy - height / 2, width, height]
+ [0.0] * 10 + [score])
if not rows_out:
return 1, None
return 1, numpy.array(rows_out, dtype="float32")
class _Cv2:
"""只提供 `detect` 用得到的那一个函数。"""
@staticmethod
def resize(image, size):
import numpy
cols, rows = size
return numpy.zeros((rows, cols, 3), dtype=image.dtype)
def _detector_over(reads, *, score=face_detect.DEFAULT_SCORE):
import numpy
detector = face_detect.FaceDetector.__new__(face_detect.FaceDetector)
detector._cv2 = _Cv2()
detector.score = score
detector._detector = _Model(reads)
return detector, numpy.zeros((1440, 2560, 3), dtype="uint8")
class DetectAcrossScalesTests(unittest.TestCase):
"""一张脸的分是它在几个尺度上读数的中位数，不是最高的那一次。"""
def test_a_face_read_on_every_scale_keeps_its_middling_score(self):
detector, image = _detector_over({
320: [(160, 90, 40, 50, 0.90)],
640: [(320, 180, 80, 100, 0.85)],
1280: [(640, 360, 160, 200, 0.30)],
})
faces = detector.detect(image)
self.assertEqual(len(faces), 1)
self.assertEqual(faces[0].score, 0.85)
self.assertAlmostEqual(faces[0].cx, 0.5, places=2)
def test_a_box_that_only_one_scale_believes_is_dropped(self):
"""实测封面 `SRN-104`：罩住整个身体的框在 320 上 0.85，另两档只有 0.31、0.49。
取最高分它和真正的脸打平，再按面积一比就赢了；取中位数它根本进不了名单。
"""
detector, image = _detector_over({
320: [(160, 90, 150, 160, 0.85)],
640: [(320, 180, 300, 320, 0.31)],
1280: [(640, 360, 600, 640, 0.49)],
})
self.assertEqual(detector.detect(image), [])
def test_a_face_missing_from_one_scale_still_counts(self):
detector, image = _detector_over({
320: [(160, 90, 40, 50, 0.88)],
640: [(320, 180, 80, 100, 0.91)],
})
faces = detector.detect(image)
self.assertEqual(len(faces), 1)
self.assertEqual(faces[0].score, 0.88)
def test_a_small_image_is_only_sent_once(self):
"""三档都缩不下去时只检一次，那一次的读数就是分——补 0 会把小图的脸全判掉。"""
import numpy
detector = face_detect.FaceDetector.__new__(face_detect.FaceDetector)
detector._cv2 = _Cv2()
detector.score = face_detect.DEFAULT_SCORE
detector._detector = _Model({200: [(100, 60, 40, 50, 0.72)]})
faces = detector.detect(numpy.zeros((120, 200, 3), dtype="uint8"))
self.assertEqual([face.score for face in faces], [0.72])
def test_two_faces_side_by_side_stay_two_faces(self):
detector, image = _detector_over({
320: [(80, 90, 40, 50, 0.80), (240, 90, 40, 50, 0.70)],
640: [(160, 180, 80, 100, 0.82), (480, 180, 80, 100, 0.74)],
1280: [(320, 360, 160, 200, 0.81), (960, 360, 160, 200, 0.72)],
})
faces = detector.detect(image)
self.assertEqual(sorted(face.score for face in faces), [0.72, 0.81])
class MainFaceTests(unittest.TestCase):
"""挑主角那张脸：分数先卡一道，再在剩下的里取最大。"""
@staticmethod
def face(cx, cy, width, height, score):
return face_detect.Face(cx=cx, cy=cy, width=width, height=height, score=score)
def test_a_big_but_weak_box_loses_to_the_real_face(self):
"""实测 `performer-8218`：大出一倍多的那个框罩在胸口，分只有 0.798。
只按面积挑，圆头像就取景在胸口——脸在画面上四分之一处。
"""
chest = self.face(0.389, 0.482, 0.427, 0.313, 0.798)
head = self.face(0.389, 0.262, 0.202, 0.170, 0.928)
self.assertEqual(face_detect.main_face([chest, head]), head)
def test_a_small_sharp_bystander_does_not_beat_the_subject(self):
"""实测 `performer-8540`：右上角还检出一个 0.066×0.052 的小框。
分数相当时仍按面积挑，否则背景里那张小而清晰的脸会抢走取景。
"""
subject = self.face(0.608, 0.207, 0.215, 0.152, 0.930)
bystander = self.face(0.694, 0.050, 0.066, 0.052, 0.925)
self.assertEqual(face_detect.main_face([subject, bystander]), subject)
def test_no_face_is_no_face(self):
self.assertIsNone(face_detect.main_face([]))
if __name__ == "__main__":
unittest.main()