Spaces:
Running
Running
File size: 9,079 Bytes
e7f1d57 |
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 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 |
# Dependencies
from typing import Dict
from utils.logger import get_logger
from config.settings import settings
from config.constants import MetricType
from config.constants import SignalStatus
from config.constants import SIGNAL_THRESHOLDS
# Setup Logging
logger = get_logger(__name__)
class ThresholdManager:
"""
Manage detection thresholds dynamically
Purpose:
--------
Allows runtime adjustment of detection thresholds for:
- A/B testing different sensitivity levels
- Calibration based on real-world performance
- Custom thresholds for specific use cases
- Environment-specific tuning (production vs staging)
Note: Changes are runtime-only and not persisted
"""
def __init__(self):
"""
Initialize Threshold Manager with current settings
"""
self._review_threshold = settings.REVIEW_THRESHOLD
self._signal_thresholds = dict(SIGNAL_THRESHOLDS)
self._metric_weights = dict(settings.get_metric_weights())
logger.info(f"ThresholdManager initialized: review_threshold={self._review_threshold}")
def get_review_threshold(self) -> float:
"""
Get current review threshold
Returns:
--------
{ float } : Current threshold [0.0, 1.0]
"""
return self._review_threshold
def set_review_threshold(self, new_threshold: float) -> bool:
"""
Set new review threshold
Arguments:
----------
new_threshold { float } : New threshold value [0.0, 1.0]
Returns:
--------
{ bool } : Success status
"""
if not (0.0 <= new_threshold <= 1.0):
logger.error(f"Invalid threshold: {new_threshold} (must be between 0.0 and 1.0)")
return False
old_threshold = self._review_threshold
self._review_threshold = new_threshold
logger.info(f"Review threshold changed: {old_threshold:.2f} → {new_threshold:.2f}")
return True
def adjust_sensitivity(self, sensitivity: str) -> bool:
"""
Adjust sensitivity using preset levels
Arguments:
----------
sensitivity { str } : One of 'conservative', 'balanced', 'aggressive'
Returns:
--------
{ bool } : Success status
"""
presets = {'conservative' : 0.75, # Fewer false positives, may miss some AI
'balanced' : 0.65, # Recommended default
'aggressive' : 0.55, # Catch more AI, more false positives
}
if (sensitivity not in presets):
logger.error(f"Invalid sensitivity: {sensitivity}. Must be one of {list(presets.keys())}")
return False
new_threshold = presets[sensitivity]
success = self.set_review_threshold(new_threshold = new_threshold)
if success:
logger.info(f"Sensitivity set to '{sensitivity}' (threshold={new_threshold})")
return success
def get_signal_thresholds(self) -> Dict[SignalStatus, float]:
"""
Get current signal thresholds
Returns:
--------
{ dict } : Signal status → threshold mapping
"""
return self._signal_thresholds.copy()
def set_signal_threshold(self, status: SignalStatus, threshold: float) -> bool:
"""
Set threshold for specific signal status
Arguments:
----------
status { SignalStatus } : Signal status to modify
threshold { float } : New threshold [0.0, 1.0]
Returns:
--------
{ bool } : Success status
"""
if not (0.0 <= threshold <= 1.0):
logger.error(f"Invalid threshold: {threshold}")
return False
old_threshold = self._signal_thresholds.get(status)
self._signal_thresholds[status] = threshold
logger.info(f"Signal threshold for {status.value}: {old_threshold:.2f} → {threshold:.2f}")
return True
def get_metric_weights(self) -> Dict[MetricType, float]:
"""
Get current metric weights
Returns:
--------
{ dict } : Metric type → weight mapping
"""
return self._metric_weights.copy()
def set_metric_weight(self, metric: MetricType, weight: float) -> bool:
"""
Set weight for specific metric
Arguments:
----------
metric { MetricType } : Metric to modify
weight { float } : New weight [0.0, 1.0]
Returns:
--------
{ bool } : Success status
"""
if not (0.0 <= weight <= 1.0):
logger.error(f"Invalid weight: {weight}")
return False
old_weight = self._metric_weights.get(metric, 0.0)
self._metric_weights[metric] = weight
# Validate total weight
total_weight = sum(self._metric_weights.values())
if not (0.99 <= total_weight <= 1.01):
logger.warning(f"Total metric weights = {total_weight:.3f} (should sum to 1.0)")
logger.info(f"Metric weight for {metric.value}: {old_weight:.2f} → {weight:.2f}")
return True
def set_all_metric_weights(self, weights: Dict[MetricType, float]) -> bool:
"""
Set all metric weights at once (ensures sum = 1.0)
Arguments:
----------
weights { dict } : Complete metric weights mapping
Returns:
--------
{ bool } : Success status
"""
# Validate input
if (not all(0.0 <= w <= 1.0 for w in weights.values())):
logger.error("All weights must be between 0.0 and 1.0")
return False
total_weight = sum(weights.values())
if not (0.99 <= total_weight <= 1.01):
logger.error(f"Weights must sum to 1.0, got {total_weight:.3f}")
return False
self._metric_weights = dict(weights)
logger.info(f"All metric weights updated: {self._metric_weights}")
return True
def get_recommendations(self, score: float) -> Dict[str, str]:
"""
Get action recommendations based on score
Arguments:
----------
score { float } : Overall suspicion score [0.0, 1.0]
Returns:
--------
{ dict } : Recommendation details
"""
if (score >= 0.85):
return {"priority" : "HIGH",
"action" : "Immediate manual verification recommended",
"confidence" : "Very high likelihood of AI generation",
"next_steps" : "Forensic analysis, reverse image search, metadata inspection",
}
elif (score >= 0.70):
return {"priority" : "MEDIUM",
"action" : "Manual verification recommended",
"confidence" : "High likelihood of AI generation",
"next_steps" : "Visual inspection, compare with similar authentic images",
}
elif (score >= 0.50):
return {"priority" : "LOW",
"action" : "Optional review",
"confidence" : "Moderate indicators of AI generation",
"next_steps" : "May be heavily edited real photo, check source",
}
else:
return {"priority" : "NONE",
"action" : "No immediate action needed",
"confidence" : "Low likelihood of AI generation",
"next_steps" : "Likely authentic, proceed normally",
}
def get_current_config(self) -> Dict[str, object]:
"""
Get complete current configuration
Returns:
--------
{ dict } : All current threshold and weight settings
"""
return {"review_threshold" : self._review_threshold,
"signal_thresholds" : self._signal_thresholds.copy(),
"metric_weights" : self._metric_weights.copy(),
}
def reset_to_defaults(self) -> None:
"""
Reset all thresholds to default settings
"""
self._review_threshold = settings.REVIEW_THRESHOLD
self._signal_thresholds = dict(SIGNAL_THRESHOLDS)
self._metric_weights = dict(settings.get_metric_weights())
logger.info("All thresholds reset to default values") |