File size: 6,116 Bytes
aac350d 23d337e aac350d 23d337e aac350d 23d337e aac350d 23d337e aac350d 23d337e aac350d 23d337e aac350d 23d337e aac350d 23d337e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """
Conflict detector — finds cross-provider disagreements.
Currently detects:
- face_count_mismatch (detectors disagree on number of faces)
- match_disagreement (recognizers disagree on best match for same face)
- quality_disagreement (image-quality providers disagree substantially)
- integrity_disagreement (forensics providers disagree on integrity)
- format_mismatch (metadata providers disagree on format)
"""
from __future__ import annotations
from typing import Dict, List, Protocol
from models.reports import ConflictReport
from providers.base import ProviderResult
# Duck-typed protocols (confidence must not import from normalization)
class _BoxLike(Protocol):
detector: str
confidence: float
class _MatchLike(Protocol):
query_face_index: int
best_match: str | None
recognizer: str
class ConflictDetector:
"""Detects cross-provider disagreements."""
def detect(
self,
results: Dict[str, ProviderResult],
boxes: List[_BoxLike],
matches: List[_MatchLike],
) -> List[ConflictReport]:
conflicts: List[ConflictReport] = []
conflicts.extend(self._detect_face_count_mismatch(results))
conflicts.extend(self._detect_match_disagreement(matches))
conflicts.extend(self._detect_quality_disagreement(results))
conflicts.extend(self._detect_integrity_disagreement(results))
conflicts.extend(self._detect_format_mismatch(results))
return conflicts
# ------------------------------------------------------------------ #
# Individual conflict detectors
# ------------------------------------------------------------------ #
def _detect_face_count_mismatch(self, results: Dict[str, ProviderResult]) -> List[ConflictReport]:
detector_counts: Dict[str, int] = {}
for r in results.values():
if r.success and r.capability.value == "detection":
detector_counts[r.provider] = r.normalized.get("num_faces", 0)
if len(detector_counts) < 2:
return []
counts = list(detector_counts.values())
if max(counts) == min(counts):
return []
return [ConflictReport(
kind="face_count_mismatch",
providers=list(detector_counts.keys()),
description=(
f"Detectors disagree on face count: {detector_counts}"
),
severity="warning",
)]
def _detect_match_disagreement(self, matches: List[_MatchLike]) -> List[ConflictReport]:
by_face: Dict[int, List[_MatchLike]] = {}
for m in matches:
by_face.setdefault(m.query_face_index, []).append(m)
out: List[ConflictReport] = []
for face_idx, face_matches in by_face.items():
if len(face_matches) < 2:
continue
best_matches = {m.best_match for m in face_matches if m.best_match}
if len(best_matches) > 1:
out.append(ConflictReport(
kind="match_disagreement",
providers=[m.recognizer for m in face_matches],
description=(
f"Recognizers disagree on best match for face #{face_idx}: "
f"{[(m.recognizer, m.best_match) for m in face_matches]}"
),
severity="warning",
))
return out
def _detect_quality_disagreement(self, results: Dict[str, ProviderResult]) -> List[ConflictReport]:
"""Flag if image-quality providers disagree on quality by > 0.3."""
quality_scores: Dict[str, float] = {}
for r in results.values():
if r.success and r.capability.value == "image_analysis":
qs = r.normalized.get("quality_score")
if qs is not None:
quality_scores[r.provider] = float(qs)
if len(quality_scores) < 2:
return []
scores = list(quality_scores.values())
if max(scores) - min(scores) > 0.3:
return [ConflictReport(
kind="quality_disagreement",
providers=list(quality_scores.keys()),
description=f"Image-quality providers disagree by >0.3: {quality_scores}",
severity="info",
)]
return []
def _detect_integrity_disagreement(self, results: Dict[str, ProviderResult]) -> List[ConflictReport]:
"""Flag if forensics providers disagree on integrity by > 0.3."""
integrity_scores: Dict[str, float] = {}
for r in results.values():
if r.success and r.capability.value == "forensics":
ii = r.normalized.get("integrity_score")
if ii is not None:
integrity_scores[r.provider] = float(ii)
if len(integrity_scores) < 2:
return []
scores = list(integrity_scores.values())
if max(scores) - min(scores) > 0.3:
return [ConflictReport(
kind="integrity_disagreement",
providers=list(integrity_scores.keys()),
description=f"Forensics providers disagree on integrity: {integrity_scores}",
severity="warning",
)]
return []
def _detect_format_mismatch(self, results: Dict[str, ProviderResult]) -> List[ConflictReport]:
"""Flag if metadata providers disagree on image format."""
formats: Dict[str, str] = {}
for r in results.values():
if r.success and r.capability.value == "metadata":
fmt = r.normalized.get("format")
if fmt:
formats[r.provider] = fmt
if len(formats) < 2:
return []
unique_formats = set(formats.values())
if len(unique_formats) > 1:
return [ConflictReport(
kind="format_mismatch",
providers=list(formats.keys()),
description=f"Metadata providers disagree on format: {formats}",
severity="info",
)]
return []
|