Spaces:
Sleeping
Sleeping
Upload folder using huggingface_hub (#15)
Browse files- Upload folder using huggingface_hub (75853892c9819544b883b9622fbb073c9d930cd9)
Co-authored-by: nicanor zousko <zousko-stark@users.noreply.huggingface.co>
- main.py +26 -3
- quality_control.py +235 -0
main.py
CHANGED
|
@@ -1001,10 +1001,33 @@ class MedSigClipWrapper:
|
|
| 1001 |
try:
|
| 1002 |
if specific_results:
|
| 1003 |
top_label_text = specific_results[0]['label']
|
| 1004 |
-
|
| 1005 |
|
| 1006 |
-
#
|
| 1007 |
-
#
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1008 |
import explainability
|
| 1009 |
engine = explainability.ExplainabilityEngine(self)
|
| 1010 |
|
|
|
|
| 1001 |
try:
|
| 1002 |
if specific_results:
|
| 1003 |
top_label_text = specific_results[0]['label']
|
| 1004 |
+
logger.info(f"Generating Medical Explanation for: {top_label_text}")
|
| 1005 |
|
| 1006 |
+
# --- QUALITY CONTROL GATE (Gate 1 & 2) ---
|
| 1007 |
+
# Added per user request: Verify quality before deep explanation/analysis
|
| 1008 |
+
# Ideally this should be even earlier, but performing it here ensures we have the image object ready.
|
| 1009 |
+
from quality_control import QualityControlEngine
|
| 1010 |
+
qc_engine = QualityControlEngine()
|
| 1011 |
+
qc_result = qc_engine.run_quality_check(image)
|
| 1012 |
+
|
| 1013 |
+
enhanced_result['image_quality'] = {
|
| 1014 |
+
"quality_score": qc_result['quality_score'],
|
| 1015 |
+
"passed": qc_result['passed'],
|
| 1016 |
+
"reasons": qc_result['reasons'],
|
| 1017 |
+
"metrics": qc_result['metrics']
|
| 1018 |
+
}
|
| 1019 |
+
|
| 1020 |
+
if not qc_result['passed']:
|
| 1021 |
+
logger.warning(f"⛔ Quality Control Failed: {qc_result['reasons']}")
|
| 1022 |
+
# We return early, preserving the basic Classification but flagging the Quality Failure
|
| 1023 |
+
# ensuring the Frontend shows the error.
|
| 1024 |
+
enhanced_result['diagnosis'] = "Analyse Refusée (Qualité Insuffisante)"
|
| 1025 |
+
enhanced_result['confidence'] = 0.0
|
| 1026 |
+
enhanced_result['quality_failure_reasons'] = qc_result['reasons']
|
| 1027 |
+
# Stop explainability
|
| 1028 |
+
return localized_result
|
| 1029 |
+
|
| 1030 |
+
# If QC Passed, Proceed to Explanation
|
| 1031 |
import explainability
|
| 1032 |
engine = explainability.ExplainabilityEngine(self)
|
| 1033 |
|
quality_control.py
ADDED
|
@@ -0,0 +1,235 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
|
| 2 |
+
import numpy as np
|
| 3 |
+
import cv2
|
| 4 |
+
import pydicom
|
| 5 |
+
import logging
|
| 6 |
+
from typing import Dict, Any, List, Tuple, Union
|
| 7 |
+
from PIL import Image
|
| 8 |
+
|
| 9 |
+
logger = logging.getLogger("ElephMind-QC")
|
| 10 |
+
|
| 11 |
+
class QualityControlEngine:
|
| 12 |
+
"""
|
| 13 |
+
Advanced Quality Control Engine (Gatekeeper).
|
| 14 |
+
Implements the 9-Point QC Checklist.
|
| 15 |
+
|
| 16 |
+
Metrics:
|
| 17 |
+
1. Structural (DICOM)
|
| 18 |
+
2. Intensity (Contrast)
|
| 19 |
+
3. Blur (Laplacian)
|
| 20 |
+
4. Noise (SNR)
|
| 21 |
+
5. Saturation (Clipping)
|
| 22 |
+
6. Spatial (Aspect Ratio)
|
| 23 |
+
|
| 24 |
+
Decision:
|
| 25 |
+
QC Score = Weighted Sum
|
| 26 |
+
Threshold >= 0.75 -> PASS
|
| 27 |
+
"""
|
| 28 |
+
|
| 29 |
+
def __init__(self):
|
| 30 |
+
# Weights defined by user
|
| 31 |
+
self.weights = {
|
| 32 |
+
"structure": 0.30, # Weight 3 (Normalized approx)
|
| 33 |
+
"blur": 0.20, # Weight 2
|
| 34 |
+
"contrast": 0.20, # Weight 2
|
| 35 |
+
"noise": 0.10, # Weight 1
|
| 36 |
+
"saturation": 0.10,
|
| 37 |
+
"spatial": 0.10
|
| 38 |
+
}
|
| 39 |
+
# Thresholds
|
| 40 |
+
self.thresholds = {
|
| 41 |
+
"blur_var": 100.0, # Laplacian Variance < 100 -> Blurry
|
| 42 |
+
"contrast_std": 10.0, # Std Dev < 10 -> Low Contrast
|
| 43 |
+
"entropy": 4.0, # Entropy < 4.0 -> Low Info
|
| 44 |
+
"snr_min": 2.0, # Signal-to-Noise Ratio < 2.0 -> Noisy
|
| 45 |
+
"saturation_max": 0.05, # >5% pixels at min/max -> Saturated
|
| 46 |
+
"aspect_min": 0.5, # Too thin
|
| 47 |
+
"aspect_max": 2.0 # Too wide
|
| 48 |
+
}
|
| 49 |
+
|
| 50 |
+
def evaluate_dicom(self, dataset: pydicom.dataset.FileDataset) -> Dict[str, Any]:
|
| 51 |
+
"""
|
| 52 |
+
Gate 1: Structural DICOM Check.
|
| 53 |
+
"""
|
| 54 |
+
reasons = []
|
| 55 |
+
passed = True
|
| 56 |
+
|
| 57 |
+
try:
|
| 58 |
+
# 1. Pixel Data Presence
|
| 59 |
+
if not hasattr(dataset, "PixelData") or dataset.PixelData is None:
|
| 60 |
+
return {"passed": False, "score": 0.0, "reasons": ["CRITICAL: Missing PixelData"]}
|
| 61 |
+
|
| 62 |
+
# 2. Dimensions
|
| 63 |
+
rows = getattr(dataset, "Rows", 0)
|
| 64 |
+
cols = getattr(dataset, "Columns", 0)
|
| 65 |
+
if rows <= 0 or cols <= 0:
|
| 66 |
+
return {"passed": False, "score": 0.0, "reasons": ["CRITICAL: Invalid Dimensions (Rows/Cols <= 0)"]}
|
| 67 |
+
|
| 68 |
+
# 3. Transfer Syntax (Compression check - basic)
|
| 69 |
+
# If we can read pixel_array, it's usually mostly fine, preventing crash is handled in processor.
|
| 70 |
+
# Here we just check logical validity.
|
| 71 |
+
|
| 72 |
+
pass
|
| 73 |
+
except Exception as e:
|
| 74 |
+
return {"passed": False, "score": 0.0, "reasons": [f"CRITICAL: DICOM Corrupt ({str(e)})"]}
|
| 75 |
+
|
| 76 |
+
return {"passed": True, "score": 1.0, "reasons": []}
|
| 77 |
+
|
| 78 |
+
def compute_metrics(self, image: np.ndarray) -> Dict[str, float]:
|
| 79 |
+
"""
|
| 80 |
+
Compute raw metrics for the image (H, W) or (H, W, C).
|
| 81 |
+
Image input should be uint8 0-255 or float.
|
| 82 |
+
"""
|
| 83 |
+
metrics = {}
|
| 84 |
+
|
| 85 |
+
# Ensure Grayscale for calculation
|
| 86 |
+
if len(image.shape) == 3:
|
| 87 |
+
gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
|
| 88 |
+
else:
|
| 89 |
+
gray = image
|
| 90 |
+
|
| 91 |
+
# 1. Blur (Variance of Laplacian)
|
| 92 |
+
metrics['blur_var'] = cv2.Laplacian(gray, cv2.CV_64F).var()
|
| 93 |
+
|
| 94 |
+
# 2. Intensity / Contrast
|
| 95 |
+
metrics['std_dev'] = np.std(gray)
|
| 96 |
+
# Entropy
|
| 97 |
+
hist, _ = np.histogram(gray, bins=256, range=(0, 256))
|
| 98 |
+
prob = hist / (np.sum(hist) + 1e-8)
|
| 99 |
+
prob = prob[prob > 0]
|
| 100 |
+
metrics['entropy'] = -np.sum(prob * np.log2(prob))
|
| 101 |
+
|
| 102 |
+
# 3. Noise (Simple SNR estimate)
|
| 103 |
+
# Signal = Mean, Noise = Std(High Pass)
|
| 104 |
+
# Simple High Pass: Image - Blurred
|
| 105 |
+
blurred = cv2.GaussianBlur(gray, (5, 5), 0)
|
| 106 |
+
noise_img = gray.astype(float) - blurred.astype(float)
|
| 107 |
+
noise_std = np.std(noise_img) + 1e-8
|
| 108 |
+
signal_mean = np.mean(gray)
|
| 109 |
+
metrics['snr'] = signal_mean / noise_std
|
| 110 |
+
|
| 111 |
+
# 4. Saturation
|
| 112 |
+
# % pixels at 0 or 255
|
| 113 |
+
n_pixels = gray.size
|
| 114 |
+
n_sat = np.sum(gray <= 5) + np.sum(gray >= 250)
|
| 115 |
+
metrics['saturation_pct'] = n_sat / n_pixels
|
| 116 |
+
|
| 117 |
+
# 5. Spatial
|
| 118 |
+
h, w = gray.shape
|
| 119 |
+
metrics['aspect_ratio'] = w / h
|
| 120 |
+
|
| 121 |
+
return metrics
|
| 122 |
+
|
| 123 |
+
def run_quality_check(self, image_input: Union[Image.Image, np.ndarray, pydicom.dataset.FileDataset]) -> Dict[str, Any]:
|
| 124 |
+
"""
|
| 125 |
+
Main Entry Point.
|
| 126 |
+
Returns: {
|
| 127 |
+
"passed": bool,
|
| 128 |
+
"quality_score": float (0-1),
|
| 129 |
+
"reasons": List[str],
|
| 130 |
+
"metrics": Dict
|
| 131 |
+
}
|
| 132 |
+
"""
|
| 133 |
+
reasons = []
|
| 134 |
+
scores = {}
|
| 135 |
+
|
| 136 |
+
# --- PHASE 1: DICOM STRUCTURE (If DICOM) ---
|
| 137 |
+
dicom_score = 1.0
|
| 138 |
+
if isinstance(image_input, pydicom.dataset.FileDataset):
|
| 139 |
+
res_struct = self.evaluate_dicom(image_input)
|
| 140 |
+
if not res_struct['passed']:
|
| 141 |
+
return {
|
| 142 |
+
"passed": False,
|
| 143 |
+
"quality_score": 0.0,
|
| 144 |
+
"reasons": res_struct['reasons'],
|
| 145 |
+
"metrics": {}
|
| 146 |
+
}
|
| 147 |
+
# Convert to numpy for image analysis using standard processor logic (simplified here or assume pre-converted)
|
| 148 |
+
# ideally the caller passes the converted image.
|
| 149 |
+
# If input is DICOM, we assume we can't analyze image metrics easily here without converting.
|
| 150 |
+
# To simplify integration: Check DICOM Structure, then rely on caller to pass Image object for Visual QC.
|
| 151 |
+
# For this implementation, we assume input is PIL Image or Numpy Array for Visual QC.
|
| 152 |
+
pass
|
| 153 |
+
|
| 154 |
+
# Prepare Image
|
| 155 |
+
if isinstance(image_input, Image.Image):
|
| 156 |
+
img_np = np.array(image_input)
|
| 157 |
+
elif isinstance(image_input, np.ndarray):
|
| 158 |
+
img_np = image_input
|
| 159 |
+
else:
|
| 160 |
+
# If strictly DICOM passed without conversion capability, we only did struct check
|
| 161 |
+
return {"passed": True, "quality_score": 1.0, "reasons": [], "metrics": {}}
|
| 162 |
+
|
| 163 |
+
# --- PHASE 2: VISUAL METRICS ---
|
| 164 |
+
m = self.compute_metrics(img_np)
|
| 165 |
+
|
| 166 |
+
# 1. Blur Check
|
| 167 |
+
# Sigmoid-like soft score or Hard Threshold? User implies Hard Rules composed into Score.
|
| 168 |
+
# "Structure: weight 3, Blur: weight 2..."
|
| 169 |
+
# Let's assign 0 or 1 per category based on threshold, then weight.
|
| 170 |
+
|
| 171 |
+
# Blur
|
| 172 |
+
if m['blur_var'] < self.thresholds['blur_var']:
|
| 173 |
+
scores['blur'] = 0.0
|
| 174 |
+
reasons.append("Image Floue (Netteté insuffisante)")
|
| 175 |
+
else:
|
| 176 |
+
scores['blur'] = 1.0
|
| 177 |
+
|
| 178 |
+
# Contrast / Intensity
|
| 179 |
+
if m['std_dev'] < self.thresholds['contrast_std'] or m['entropy'] < self.thresholds['entropy']:
|
| 180 |
+
scores['contrast'] = 0.0
|
| 181 |
+
reasons.append("Contraste Insuffisant (Image plate/sombre)")
|
| 182 |
+
else:
|
| 183 |
+
scores['contrast'] = 1.0
|
| 184 |
+
|
| 185 |
+
# Noise
|
| 186 |
+
if m['snr'] < self.thresholds['snr_min']:
|
| 187 |
+
scores['noise'] = 0.0
|
| 188 |
+
reasons.append("Bruit Excessif (SNR faible)")
|
| 189 |
+
else:
|
| 190 |
+
scores['noise'] = 1.0
|
| 191 |
+
|
| 192 |
+
# Saturation
|
| 193 |
+
if m['saturation_pct'] > self.thresholds['saturation_max']:
|
| 194 |
+
scores['saturation'] = 0.0
|
| 195 |
+
reasons.append("Saturation Excessive (>5% clipping)")
|
| 196 |
+
else:
|
| 197 |
+
scores['saturation'] = 1.0
|
| 198 |
+
|
| 199 |
+
# Spatial
|
| 200 |
+
if not (self.thresholds['aspect_min'] <= m['aspect_ratio'] <= self.thresholds['aspect_max']):
|
| 201 |
+
scores['spatial'] = 0.0
|
| 202 |
+
reasons.append(f"Format Anatomique Invalide (Ratio {m['aspect_ratio']:.2f})")
|
| 203 |
+
else:
|
| 204 |
+
scores['spatial'] = 1.0
|
| 205 |
+
|
| 206 |
+
# Structural (Implicitly 1 if we got here with an image)
|
| 207 |
+
scores['structure'] = 1.0
|
| 208 |
+
|
| 209 |
+
# --- PHASE 3: GLOBAL SCORE ---
|
| 210 |
+
# QC_score = Sum(w * s)
|
| 211 |
+
final_score = (
|
| 212 |
+
self.weights['structure'] * scores.get('structure', 1.0) +
|
| 213 |
+
self.weights['blur'] * scores.get('blur', 1.0) +
|
| 214 |
+
self.weights['contrast'] * scores.get('contrast', 1.0) +
|
| 215 |
+
self.weights['noise'] * scores.get('noise', 1.0) +
|
| 216 |
+
self.weights['saturation'] * scores.get('saturation', 1.0) +
|
| 217 |
+
self.weights['spatial'] * scores.get('spatial', 1.0)
|
| 218 |
+
)
|
| 219 |
+
|
| 220 |
+
# Normalize weights sum just in case
|
| 221 |
+
total_weight = sum(self.weights.values())
|
| 222 |
+
final_score = final_score / total_weight
|
| 223 |
+
|
| 224 |
+
# DECISION
|
| 225 |
+
is_passed = final_score >= 0.75
|
| 226 |
+
|
| 227 |
+
status = "PASSED" if is_passed else "REJECTED"
|
| 228 |
+
logger.info(f"QC Evaluation: {status} (Score: {final_score:.2f}) - Reasons: {reasons}")
|
| 229 |
+
|
| 230 |
+
return {
|
| 231 |
+
"passed": is_passed,
|
| 232 |
+
"quality_score": round(final_score, 2),
|
| 233 |
+
"reasons": reasons,
|
| 234 |
+
"metrics": m
|
| 235 |
+
}
|