Spaces:
Running
Running
File size: 3,167 Bytes
4e75170 | 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 | from __future__ import annotations
import numpy as np
from src.types import DetectionResponse, EngineResult
ENGINE_WEIGHTS = {
"fingerprint": 0.45,
"coherence": 0.35,
"sstgnn": 0.20,
}
ENGINE_WEIGHTS_VIDEO = {
"fingerprint": 0.30,
"coherence": 0.50,
"sstgnn": 0.20,
}
ATTRIBUTION_PRIORITY = {
"fingerprint": 1,
"sstgnn": 2,
"coherence": 3,
}
def _normalize_generator(value: str | None) -> str:
if not value:
return "real"
return str(value).strip().lower().replace(" ", "_")
def fuse(results: list[EngineResult], is_video: bool = False) -> tuple[str, float, str]:
"""Return (verdict, confidence_for_verdict, attributed_generator)."""
weights = ENGINE_WEIGHTS_VIDEO if is_video else ENGINE_WEIGHTS
active = [result for result in results if result.verdict != "UNKNOWN"]
if not active:
return "UNKNOWN", 0.5, "unknown_gan"
wf = sum(
result.confidence * weights.get(result.engine, 0.1)
for result in active
if result.verdict == "FAKE"
)
wr = sum(
(1.0 - result.confidence) * weights.get(result.engine, 0.1)
for result in active
if result.verdict == "REAL"
)
denom = wf + wr + 1e-9
fake_prob = float(np.clip(wf / denom, 0.0, 1.0))
verdict = "FAKE" if fake_prob > 0.5 else "REAL"
confidence = fake_prob if verdict == "FAKE" else (1.0 - fake_prob)
generator = "real"
if verdict == "FAKE":
for result in sorted(active, key=lambda r: ATTRIBUTION_PRIORITY.get(r.engine, 9)):
candidate = _normalize_generator(result.attributed_generator)
if candidate and candidate != "real":
generator = candidate
break
if generator == "real":
generator = "unknown_gan"
return verdict, confidence, generator
class Fuser:
"""Compatibility wrapper returning `DetectionResponse` objects."""
def fuse(
self,
results: list[EngineResult],
media_type: str = "image",
total_ms: float = 0.0,
) -> DetectionResponse:
if not results:
return DetectionResponse(
verdict="REAL",
confidence=0.5,
attributed_generator="unknown_gan",
explanation="No engine results available.",
processing_time_ms=round(total_ms, 2),
engine_breakdown=[],
)
verdict, confidence, generator = fuse(results, is_video=(media_type == "video"))
if verdict == "UNKNOWN":
explanation = "No active engine outputs were available."
else:
summary = ", ".join(
f"{result.engine}:{result.verdict}({result.confidence:.2f})"
for result in results
)
explanation = f"Fused {media_type} analysis from engines: {summary}."
return DetectionResponse(
verdict=verdict,
confidence=confidence,
attributed_generator=generator,
explanation=explanation,
processing_time_ms=round(total_ms, 2),
engine_breakdown=results,
)
|