Update app.py
Browse files
app.py
CHANGED
|
@@ -16,6 +16,7 @@ import pandas as pd # Para formatear la salida en tabla
|
|
| 16 |
# --- Configuración ---
|
| 17 |
MODEL_REPO_ID = "google/cxr-foundation"
|
| 18 |
MODEL_DOWNLOAD_DIR = './hf_cxr_foundation_space' # Directorio dentro del contenedor del Space
|
|
|
|
| 19 |
SIMILARITY_DIFFERENCE_THRESHOLD = 0.1
|
| 20 |
POSITIVE_SIMILARITY_THRESHOLD = 0.1
|
| 21 |
print(f"Usando umbrales: Comp Δ={SIMILARITY_DIFFERENCE_THRESHOLD}, Simp τ={POSITIVE_SIMILARITY_THRESHOLD}")
|
|
@@ -30,151 +31,255 @@ criteria_list_negative = [
|
|
| 30 |
"cropped image", "scapulae overlying lungs", "blurred image", "obscuring artifact"
|
| 31 |
]
|
| 32 |
|
| 33 |
-
# --- Funciones Auxiliares ---
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 34 |
def bert_tokenize(text, preprocessor):
|
| 35 |
-
|
|
|
|
|
|
|
| 36 |
if not isinstance(text, str): text = str(text)
|
|
|
|
|
|
|
| 37 |
out = preprocessor(tf.constant([text.lower()]))
|
|
|
|
|
|
|
| 38 |
ids = out['input_word_ids'].numpy().astype(np.int32)
|
| 39 |
masks = out['input_mask'].numpy().astype(np.float32)
|
| 40 |
paddings = 1.0 - masks
|
|
|
|
|
|
|
| 41 |
end_token_idx = (ids == 102)
|
| 42 |
ids[end_token_idx] = 0
|
| 43 |
paddings[end_token_idx] = 1.0
|
|
|
|
|
|
|
|
|
|
| 44 |
if ids.ndim == 2: ids = np.expand_dims(ids, axis=1)
|
| 45 |
if paddings.ndim == 2: paddings = np.expand_dims(paddings, axis=1)
|
|
|
|
|
|
|
| 46 |
expected_shape = (1, 1, 128)
|
| 47 |
if ids.shape != expected_shape:
|
|
|
|
| 48 |
if ids.shape == (1,128): ids = np.expand_dims(ids, axis=1)
|
| 49 |
else: raise ValueError(f"Shape incorrecta para ids: {ids.shape}, esperado {expected_shape}")
|
| 50 |
if paddings.shape != expected_shape:
|
| 51 |
if paddings.shape == (1,128): paddings = np.expand_dims(paddings, axis=1)
|
| 52 |
else: raise ValueError(f"Shape incorrecta para paddings: {paddings.shape}, esperado {expected_shape}")
|
|
|
|
| 53 |
return ids, paddings
|
| 54 |
|
| 55 |
def png_to_tfexample(image_array: np.ndarray) -> tf.train.Example:
|
|
|
|
| 56 |
if image_array.ndim == 3 and image_array.shape[2] == 1:
|
| 57 |
-
image_array = np.squeeze(image_array, axis=2)
|
| 58 |
elif image_array.ndim != 2:
|
| 59 |
-
raise ValueError(f'Array debe ser 2-D. Dimensiones: {image_array.ndim}')
|
|
|
|
| 60 |
image = image_array.astype(np.float32)
|
| 61 |
-
min_val
|
|
|
|
|
|
|
|
|
|
| 62 |
if max_val <= min_val:
|
|
|
|
|
|
|
| 63 |
if image_array.dtype == np.uint8 or (min_val >= 0 and max_val <= 255):
|
| 64 |
-
pixel_array = image.astype(np.uint8)
|
| 65 |
-
|
| 66 |
-
|
|
|
|
|
|
|
| 67 |
else:
|
| 68 |
-
image -= min_val
|
| 69 |
current_max = max_val - min_val
|
|
|
|
| 70 |
if image_array.dtype != np.uint8:
|
| 71 |
image *= 65535 / current_max
|
| 72 |
-
pixel_array = image.astype(np.uint16)
|
|
|
|
| 73 |
else:
|
|
|
|
|
|
|
|
|
|
| 74 |
image *= 255 / current_max
|
| 75 |
-
pixel_array = image.astype(np.uint8)
|
|
|
|
|
|
|
|
|
|
| 76 |
output = io.BytesIO()
|
| 77 |
-
png.Writer(
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 78 |
example = tf.train.Example()
|
| 79 |
features = example.features.feature
|
| 80 |
-
features['image/encoded'].bytes_list.value.append(
|
| 81 |
features['image/format'].bytes_list.value.append(b'png')
|
| 82 |
return example
|
| 83 |
|
| 84 |
def generate_image_embedding(img_np, elixrc_infer, qformer_infer):
|
| 85 |
-
|
|
|
|
|
|
|
|
|
|
| 86 |
try:
|
|
|
|
| 87 |
serialized_img_tf_example = png_to_tfexample(img_np).SerializeToString()
|
| 88 |
elixrc_output = elixrc_infer(input_example=tf.constant([serialized_img_tf_example]))
|
| 89 |
elixrc_embedding = elixrc_output['feature_maps_0'].numpy()
|
|
|
|
|
|
|
|
|
|
| 90 |
qformer_input_img = {
|
| 91 |
'image_feature': elixrc_embedding.tolist(),
|
| 92 |
-
'ids': np.zeros((1, 1, 128), dtype=np.int32).tolist(),
|
| 93 |
-
'paddings': np.ones((1, 1, 128), dtype=np.float32).tolist(),
|
| 94 |
}
|
| 95 |
qformer_output_img = qformer_infer(**qformer_input_img)
|
| 96 |
image_embedding = qformer_output_img['all_contrastive_img_emb'].numpy()
|
|
|
|
|
|
|
| 97 |
if image_embedding.ndim > 2:
|
| 98 |
-
|
| 99 |
-
|
| 100 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 101 |
return image_embedding
|
|
|
|
| 102 |
except Exception as e:
|
| 103 |
-
print(f"Error generando embedding imagen: {e}")
|
|
|
|
|
|
|
| 104 |
|
| 105 |
def calculate_similarities_and_classify(image_embedding, bert_preprocessor, qformer_infer):
|
| 106 |
-
|
|
|
|
| 107 |
if bert_preprocessor is None: raise ValueError("Preprocesador BERT es None.")
|
| 108 |
if qformer_infer is None: raise ValueError("QFormer es None.")
|
|
|
|
| 109 |
detailed_results = {}
|
| 110 |
-
print("\n--- Calculando similitudes ---")
|
|
|
|
| 111 |
for i in range(len(criteria_list_positive)):
|
| 112 |
-
positive_text
|
| 113 |
-
|
| 114 |
-
|
|
|
|
|
|
|
| 115 |
similarity_positive, similarity_negative, difference = None, None, None
|
| 116 |
classification_comp, classification_simp = "ERROR", "ERROR"
|
|
|
|
| 117 |
try:
|
|
|
|
| 118 |
tokens_pos, paddings_pos = bert_tokenize(positive_text, bert_preprocessor)
|
| 119 |
-
|
| 120 |
-
|
|
|
|
|
|
|
|
|
|
| 121 |
if text_embedding_pos.ndim == 1: text_embedding_pos = np.expand_dims(text_embedding_pos, axis=0)
|
| 122 |
|
|
|
|
| 123 |
tokens_neg, paddings_neg = bert_tokenize(negative_text, bert_preprocessor)
|
| 124 |
-
|
| 125 |
-
|
|
|
|
|
|
|
|
|
|
| 126 |
if text_embedding_neg.ndim == 1: text_embedding_neg = np.expand_dims(text_embedding_neg, axis=0)
|
| 127 |
|
| 128 |
-
|
| 129 |
-
if image_embedding.shape[1] !=
|
|
|
|
|
|
|
|
|
|
| 130 |
|
|
|
|
| 131 |
similarity_positive = cosine_similarity(image_embedding, text_embedding_pos)[0][0]
|
| 132 |
similarity_negative = cosine_similarity(image_embedding, text_embedding_neg)[0][0]
|
|
|
|
| 133 |
|
|
|
|
| 134 |
difference = similarity_positive - similarity_negative
|
| 135 |
classification_comp = "PASS" if difference > SIMILARITY_DIFFERENCE_THRESHOLD else "FAIL"
|
| 136 |
classification_simp = "PASS" if similarity_positive > POSITIVE_SIMILARITY_THRESHOLD else "FAIL"
|
| 137 |
-
print(f"
|
|
|
|
| 138 |
except Exception as e:
|
| 139 |
-
print(f" ERROR criterio '{criterion_name}': {e}")
|
|
|
|
|
|
|
|
|
|
|
|
|
| 140 |
detailed_results[criterion_name] = {
|
| 141 |
-
'positive_prompt': positive_text,
|
|
|
|
| 142 |
'similarity_positive': float(similarity_positive) if similarity_positive is not None else None,
|
| 143 |
'similarity_negative': float(similarity_negative) if similarity_negative is not None else None,
|
| 144 |
'difference': float(difference) if difference is not None else None,
|
| 145 |
-
'classification_comparative': classification_comp,
|
|
|
|
| 146 |
}
|
| 147 |
return detailed_results
|
| 148 |
|
| 149 |
# --- Carga Global de Modelos ---
|
|
|
|
| 150 |
print("--- Iniciando carga global de modelos ---")
|
| 151 |
start_time = time.time()
|
| 152 |
models_loaded = False
|
| 153 |
bert_preprocessor_global = None
|
| 154 |
elixrc_infer_global = None
|
| 155 |
qformer_infer_global = None
|
|
|
|
| 156 |
try:
|
| 157 |
-
|
| 158 |
-
if
|
|
|
|
|
|
|
|
|
|
| 159 |
|
|
|
|
| 160 |
os.makedirs(MODEL_DOWNLOAD_DIR, exist_ok=True)
|
| 161 |
print(f"Descargando/verificando modelos en: {MODEL_DOWNLOAD_DIR}")
|
| 162 |
snapshot_download(repo_id=MODEL_REPO_ID, local_dir=MODEL_DOWNLOAD_DIR,
|
| 163 |
allow_patterns=['elixr-c-v2-pooled/*', 'pax-elixr-b-text/*'],
|
| 164 |
-
local_dir_use_symlinks=False
|
| 165 |
print("Modelos descargados/verificados.")
|
| 166 |
|
|
|
|
| 167 |
print("Cargando Preprocesador BERT...")
|
|
|
|
| 168 |
bert_preprocess_handle = "https://tfhub.dev/tensorflow/bert_en_uncased_preprocess/3"
|
| 169 |
bert_preprocessor_global = tf_hub.KerasLayer(bert_preprocess_handle)
|
| 170 |
print("Preprocesador BERT cargado.")
|
| 171 |
|
|
|
|
| 172 |
print("Cargando ELIXR-C...")
|
| 173 |
elixrc_model_path = os.path.join(MODEL_DOWNLOAD_DIR, 'elixr-c-v2-pooled')
|
| 174 |
elixrc_model = tf.saved_model.load(elixrc_model_path)
|
| 175 |
elixrc_infer_global = elixrc_model.signatures['serving_default']
|
| 176 |
print("Modelo ELIXR-C cargado.")
|
| 177 |
|
|
|
|
| 178 |
print("Cargando QFormer (ELIXR-B Text)...")
|
| 179 |
qformer_model_path = os.path.join(MODEL_DOWNLOAD_DIR, 'pax-elixr-b-text')
|
| 180 |
qformer_model = tf.saved_model.load(qformer_model_path)
|
|
@@ -184,167 +289,155 @@ try:
|
|
| 184 |
models_loaded = True
|
| 185 |
end_time = time.time()
|
| 186 |
print(f"--- Modelos cargados globalmente con éxito en {end_time - start_time:.2f} segundos ---")
|
|
|
|
| 187 |
except Exception as e:
|
| 188 |
models_loaded = False
|
| 189 |
-
print(f"--- ERROR CRÍTICO DURANTE LA CARGA GLOBAL DE MODELOS ---")
|
|
|
|
|
|
|
|
|
|
| 190 |
|
| 191 |
# --- Función Principal de Procesamiento para Gradio ---
|
| 192 |
-
def
|
| 193 |
-
"""
|
| 194 |
if not models_loaded:
|
| 195 |
raise gr.Error("Error: Los modelos no se pudieron cargar. La aplicación no puede procesar imágenes.")
|
| 196 |
if image_pil is None:
|
| 197 |
-
|
| 198 |
-
|
| 199 |
-
gr.update(visible=False), # Oculta resultados
|
| 200 |
-
None, # Borra imagen de salida
|
| 201 |
-
gr.update(value="N/A"), # Borra etiqueta
|
| 202 |
-
pd.DataFrame(), # Borra dataframe
|
| 203 |
-
None # Borra JSON
|
| 204 |
-
)
|
| 205 |
|
| 206 |
print("\n--- Iniciando evaluación para nueva imagen ---")
|
| 207 |
start_process_time = time.time()
|
|
|
|
| 208 |
try:
|
|
|
|
|
|
|
| 209 |
img_np = np.array(image_pil.convert('L'))
|
|
|
|
|
|
|
|
|
|
|
|
|
| 210 |
image_embedding = generate_image_embedding(img_np, elixrc_infer_global, qformer_infer_global)
|
|
|
|
|
|
|
|
|
|
|
|
|
| 211 |
detailed_results = calculate_similarities_and_classify(image_embedding, bert_preprocessor_global, qformer_infer_global)
|
| 212 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 213 |
for criterion, details in detailed_results.items():
|
| 214 |
total_count += 1
|
| 215 |
-
|
| 216 |
-
|
| 217 |
-
|
| 218 |
-
|
| 219 |
-
|
| 220 |
-
output_data.append([
|
| 221 |
-
|
| 222 |
-
|
| 223 |
-
|
| 224 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 225 |
if total_count > 0:
|
| 226 |
pass_rate = passed_count / total_count
|
| 227 |
if pass_rate >= 0.85: overall_quality = "Excellent"
|
| 228 |
elif pass_rate >= 0.70: overall_quality = "Good"
|
| 229 |
elif pass_rate >= 0.50: overall_quality = "Fair"
|
| 230 |
else: overall_quality = "Poor"
|
| 231 |
-
quality_label = f"{overall_quality} ({passed_count}/{total_count} passed)"
|
|
|
|
| 232 |
end_process_time = time.time()
|
| 233 |
-
print(f"--- Evaluación completada en {end_process_time - start_process_time:.2f}
|
| 234 |
-
|
| 235 |
-
|
| 236 |
-
|
| 237 |
-
|
| 238 |
-
gr.update(value=quality_label), # Actualiza etiqueta
|
| 239 |
-
df_results, # Actualiza dataframe
|
| 240 |
-
detailed_results # Actualiza JSON
|
| 241 |
-
)
|
| 242 |
except Exception as e:
|
| 243 |
-
print(f"Error durante procesamiento Gradio: {e}")
|
| 244 |
-
|
| 245 |
-
|
| 246 |
-
|
| 247 |
-
def reset_ui():
|
| 248 |
-
print("Reseteando UI...")
|
| 249 |
-
return (
|
| 250 |
-
gr.update(visible=True), # Muestra bienvenida
|
| 251 |
-
gr.update(visible=False), # Oculta resultados
|
| 252 |
-
None, # Borra imagen de entrada
|
| 253 |
-
None, # Borra imagen de salida
|
| 254 |
-
gr.update(value="N/A"), # Borra etiqueta
|
| 255 |
-
pd.DataFrame(), # Borra dataframe
|
| 256 |
-
None # Borra JSON
|
| 257 |
-
)
|
| 258 |
|
| 259 |
-
|
| 260 |
-
|
| 261 |
-
|
| 262 |
-
|
| 263 |
-
|
| 264 |
-
font
|
| 265 |
-
|
| 266 |
-
|
| 267 |
-
|
| 268 |
-
|
| 269 |
-
|
| 270 |
-
|
| 271 |
-
|
| 272 |
-
|
| 273 |
-
|
| 274 |
-
|
| 275 |
-
|
| 276 |
-
block_title_text_color="#ffffff",
|
| 277 |
-
# Bordes
|
| 278 |
-
border_color_accent="#374151",
|
| 279 |
-
border_color_primary="#4b5563",
|
| 280 |
-
# Botones y Elementos Interactivos
|
| 281 |
-
button_primary_background_fill="*primary_600",
|
| 282 |
-
button_primary_text_color="#ffffff",
|
| 283 |
-
button_secondary_background_fill="*neutral_700",
|
| 284 |
-
button_secondary_text_color="#ffffff",
|
| 285 |
-
input_background_fill="#374151",
|
| 286 |
-
input_border_color="#4b5563",
|
| 287 |
-
# Sombras y Radios
|
| 288 |
-
shadow_drop="rgba(0,0,0,0.2) 0px 2px 4px",
|
| 289 |
-
block_shadow="rgba(0,0,0,0.2) 0px 2px 5px",
|
| 290 |
-
)
|
| 291 |
-
|
| 292 |
-
# --- Definir la Interfaz Gradio con Bloques y Tema ---
|
| 293 |
-
with gr.Blocks(theme=dark_theme, title="CXR Quality Assessment") as demo:
|
| 294 |
-
# --- Cabecera ---
|
| 295 |
with gr.Row():
|
| 296 |
-
gr.
|
| 297 |
-
"""
|
| 298 |
-
|
| 299 |
-
|
| 300 |
-
|
| 301 |
-
|
| 302 |
-
|
| 303 |
-
|
| 304 |
-
|
| 305 |
-
with gr.Row(equal_height=False):
|
| 306 |
-
|
| 307 |
-
# --- Columna Izquierda (Carga) ---
|
| 308 |
-
with gr.Column(scale=1, min_width=350):
|
| 309 |
-
gr.Markdown("### 1. Upload Image", elem_id="upload-title")
|
| 310 |
-
input_image = gr.Image(type="pil", label="Upload Chest X-ray", height=300)
|
| 311 |
-
with gr.Row():
|
| 312 |
-
analyze_btn = gr.Button("Analyze Image", variant="primary", scale=2)
|
| 313 |
-
reset_btn = gr.Button("Reset", variant="secondary", scale=1)
|
| 314 |
-
# gr.Examples( examples=[os.path.join("examples", "sample_cxr.png")], inputs=input_image, label="Example CXR" )
|
| 315 |
-
gr.Markdown( "<p style='color:#9ca3af; font-size:0.9em;'>Model loading on startup takes ~1 min. Analysis takes ~15-40 sec.</p>" )
|
| 316 |
-
|
| 317 |
-
# --- Columna Derecha (Bienvenida / Resultados) ---
|
| 318 |
with gr.Column(scale=2):
|
| 319 |
-
|
| 320 |
-
|
| 321 |
-
|
| 322 |
-
|
| 323 |
-
|
| 324 |
-
|
| 325 |
-
|
| 326 |
-
|
| 327 |
-
|
| 328 |
-
|
| 329 |
-
|
| 330 |
-
|
| 331 |
-
|
| 332 |
-
|
| 333 |
-
|
| 334 |
-
|
| 335 |
-
|
| 336 |
-
|
| 337 |
-
|
| 338 |
-
|
| 339 |
-
inputs=[input_image],
|
| 340 |
-
outputs=[ welcome_block, results_block, output_image, output_label, output_dataframe, output_json ]
|
| 341 |
)
|
| 342 |
-
|
| 343 |
-
|
| 344 |
-
|
| 345 |
-
|
|
|
|
|
|
|
| 346 |
)
|
| 347 |
|
| 348 |
# --- Iniciar la Aplicación Gradio ---
|
|
|
|
|
|
|
|
|
|
|
|
|
| 349 |
if __name__ == "__main__":
|
|
|
|
|
|
|
|
|
|
| 350 |
demo.launch(server_name="0.0.0.0", server_port=7860)
|
|
|
|
| 16 |
# --- Configuración ---
|
| 17 |
MODEL_REPO_ID = "google/cxr-foundation"
|
| 18 |
MODEL_DOWNLOAD_DIR = './hf_cxr_foundation_space' # Directorio dentro del contenedor del Space
|
| 19 |
+
# Umbrales
|
| 20 |
SIMILARITY_DIFFERENCE_THRESHOLD = 0.1
|
| 21 |
POSITIVE_SIMILARITY_THRESHOLD = 0.1
|
| 22 |
print(f"Usando umbrales: Comp Δ={SIMILARITY_DIFFERENCE_THRESHOLD}, Simp τ={POSITIVE_SIMILARITY_THRESHOLD}")
|
|
|
|
| 31 |
"cropped image", "scapulae overlying lungs", "blurred image", "obscuring artifact"
|
| 32 |
]
|
| 33 |
|
| 34 |
+
# --- Funciones Auxiliares (Integradas o adaptadas) ---
|
| 35 |
+
# @tf.function(input_signature=[tf.TensorSpec(shape=[None], dtype=tf.string)]) # Puede ayudar rendimiento
|
| 36 |
+
def preprocess_text(text):
|
| 37 |
+
"""Función interna del preprocesador BERT."""
|
| 38 |
+
return bert_preprocessor_global(text)
|
| 39 |
+
|
| 40 |
def bert_tokenize(text, preprocessor):
|
| 41 |
+
"""Tokeniza texto usando el preprocesador BERT cargado globalmente."""
|
| 42 |
+
if preprocessor is None:
|
| 43 |
+
raise ValueError("BERT preprocessor no está cargado.")
|
| 44 |
if not isinstance(text, str): text = str(text)
|
| 45 |
+
|
| 46 |
+
# Ejecutar el preprocesador
|
| 47 |
out = preprocessor(tf.constant([text.lower()]))
|
| 48 |
+
|
| 49 |
+
# Extraer y procesar IDs y máscaras
|
| 50 |
ids = out['input_word_ids'].numpy().astype(np.int32)
|
| 51 |
masks = out['input_mask'].numpy().astype(np.float32)
|
| 52 |
paddings = 1.0 - masks
|
| 53 |
+
|
| 54 |
+
# Reemplazar token [SEP] (102) por 0 y marcar como padding
|
| 55 |
end_token_idx = (ids == 102)
|
| 56 |
ids[end_token_idx] = 0
|
| 57 |
paddings[end_token_idx] = 1.0
|
| 58 |
+
|
| 59 |
+
# Asegurar las dimensiones (B, T, S) -> (1, 1, 128)
|
| 60 |
+
# El preprocesador puede devolver (1, 128), necesitamos (1, 1, 128)
|
| 61 |
if ids.ndim == 2: ids = np.expand_dims(ids, axis=1)
|
| 62 |
if paddings.ndim == 2: paddings = np.expand_dims(paddings, axis=1)
|
| 63 |
+
|
| 64 |
+
# Verificar formas finales
|
| 65 |
expected_shape = (1, 1, 128)
|
| 66 |
if ids.shape != expected_shape:
|
| 67 |
+
# Intentar reajustar si es necesario (puede pasar con algunas versiones)
|
| 68 |
if ids.shape == (1,128): ids = np.expand_dims(ids, axis=1)
|
| 69 |
else: raise ValueError(f"Shape incorrecta para ids: {ids.shape}, esperado {expected_shape}")
|
| 70 |
if paddings.shape != expected_shape:
|
| 71 |
if paddings.shape == (1,128): paddings = np.expand_dims(paddings, axis=1)
|
| 72 |
else: raise ValueError(f"Shape incorrecta para paddings: {paddings.shape}, esperado {expected_shape}")
|
| 73 |
+
|
| 74 |
return ids, paddings
|
| 75 |
|
| 76 |
def png_to_tfexample(image_array: np.ndarray) -> tf.train.Example:
|
| 77 |
+
"""Crea tf.train.Example desde NumPy array (escala de grises)."""
|
| 78 |
if image_array.ndim == 3 and image_array.shape[2] == 1:
|
| 79 |
+
image_array = np.squeeze(image_array, axis=2) # Asegurar 2D
|
| 80 |
elif image_array.ndim != 2:
|
| 81 |
+
raise ValueError(f'Array debe ser 2-D (escala de grises). Dimensiones actuales: {image_array.ndim}')
|
| 82 |
+
|
| 83 |
image = image_array.astype(np.float32)
|
| 84 |
+
min_val = image.min()
|
| 85 |
+
max_val = image.max()
|
| 86 |
+
|
| 87 |
+
# Evitar división por cero si la imagen es constante
|
| 88 |
if max_val <= min_val:
|
| 89 |
+
# Si es constante, tratar como uint8 si el rango original lo permitía,
|
| 90 |
+
# o simplemente ponerla a 0 si es float.
|
| 91 |
if image_array.dtype == np.uint8 or (min_val >= 0 and max_val <= 255):
|
| 92 |
+
pixel_array = image.astype(np.uint8)
|
| 93 |
+
bitdepth = 8
|
| 94 |
+
else: # Caso flotante constante o fuera de rango uint8
|
| 95 |
+
pixel_array = np.zeros_like(image, dtype=np.uint16)
|
| 96 |
+
bitdepth = 16
|
| 97 |
else:
|
| 98 |
+
image -= min_val # Mover mínimo a cero
|
| 99 |
current_max = max_val - min_val
|
| 100 |
+
# Escalar a 16-bit para mayor precisión si no era uint8 originalmente
|
| 101 |
if image_array.dtype != np.uint8:
|
| 102 |
image *= 65535 / current_max
|
| 103 |
+
pixel_array = image.astype(np.uint16)
|
| 104 |
+
bitdepth = 16
|
| 105 |
else:
|
| 106 |
+
# Si era uint8, mantener el rango y tipo
|
| 107 |
+
# La resta del min ya la dejó en [0, current_max]
|
| 108 |
+
# Escalar a 255 si es necesario
|
| 109 |
image *= 255 / current_max
|
| 110 |
+
pixel_array = image.astype(np.uint8)
|
| 111 |
+
bitdepth = 8
|
| 112 |
+
|
| 113 |
+
# Codificar como PNG
|
| 114 |
output = io.BytesIO()
|
| 115 |
+
png.Writer(
|
| 116 |
+
width=pixel_array.shape[1],
|
| 117 |
+
height=pixel_array.shape[0],
|
| 118 |
+
greyscale=True,
|
| 119 |
+
bitdepth=bitdepth
|
| 120 |
+
).write(output, pixel_array.tolist())
|
| 121 |
+
png_bytes = output.getvalue()
|
| 122 |
+
|
| 123 |
+
# Crear tf.train.Example
|
| 124 |
example = tf.train.Example()
|
| 125 |
features = example.features.feature
|
| 126 |
+
features['image/encoded'].bytes_list.value.append(png_bytes)
|
| 127 |
features['image/format'].bytes_list.value.append(b'png')
|
| 128 |
return example
|
| 129 |
|
| 130 |
def generate_image_embedding(img_np, elixrc_infer, qformer_infer):
|
| 131 |
+
"""Genera embedding final de imagen."""
|
| 132 |
+
if elixrc_infer is None or qformer_infer is None:
|
| 133 |
+
raise ValueError("Modelos ELIXR-C o QFormer no cargados.")
|
| 134 |
+
|
| 135 |
try:
|
| 136 |
+
# 1. ELIXR-C
|
| 137 |
serialized_img_tf_example = png_to_tfexample(img_np).SerializeToString()
|
| 138 |
elixrc_output = elixrc_infer(input_example=tf.constant([serialized_img_tf_example]))
|
| 139 |
elixrc_embedding = elixrc_output['feature_maps_0'].numpy()
|
| 140 |
+
print(f" Embedding ELIXR-C shape: {elixrc_embedding.shape}")
|
| 141 |
+
|
| 142 |
+
# 2. QFormer (Imagen)
|
| 143 |
qformer_input_img = {
|
| 144 |
'image_feature': elixrc_embedding.tolist(),
|
| 145 |
+
'ids': np.zeros((1, 1, 128), dtype=np.int32).tolist(), # Texto vacío
|
| 146 |
+
'paddings': np.ones((1, 1, 128), dtype=np.float32).tolist(), # Todo padding
|
| 147 |
}
|
| 148 |
qformer_output_img = qformer_infer(**qformer_input_img)
|
| 149 |
image_embedding = qformer_output_img['all_contrastive_img_emb'].numpy()
|
| 150 |
+
|
| 151 |
+
# Ajustar dimensiones si es necesario
|
| 152 |
if image_embedding.ndim > 2:
|
| 153 |
+
print(f" Ajustando dimensiones embedding imagen (original: {image_embedding.shape})")
|
| 154 |
+
image_embedding = np.mean(
|
| 155 |
+
image_embedding,
|
| 156 |
+
axis=tuple(range(1, image_embedding.ndim - 1))
|
| 157 |
+
)
|
| 158 |
+
if image_embedding.ndim == 1:
|
| 159 |
+
image_embedding = np.expand_dims(image_embedding, axis=0)
|
| 160 |
+
elif image_embedding.ndim == 1:
|
| 161 |
+
image_embedding = np.expand_dims(image_embedding, axis=0) # Asegurar 2D
|
| 162 |
+
|
| 163 |
+
print(f" Embedding final imagen shape: {image_embedding.shape}")
|
| 164 |
+
if image_embedding.ndim != 2:
|
| 165 |
+
raise ValueError(f"Embedding final de imagen no tiene 2 dimensiones: {image_embedding.shape}")
|
| 166 |
return image_embedding
|
| 167 |
+
|
| 168 |
except Exception as e:
|
| 169 |
+
print(f"Error generando embedding de imagen: {e}")
|
| 170 |
+
traceback.print_exc()
|
| 171 |
+
raise # Re-lanzar la excepción para que Gradio la maneje
|
| 172 |
|
| 173 |
def calculate_similarities_and_classify(image_embedding, bert_preprocessor, qformer_infer):
|
| 174 |
+
"""Calcula similitudes y clasifica."""
|
| 175 |
+
if image_embedding is None: raise ValueError("Embedding de imagen es None.")
|
| 176 |
if bert_preprocessor is None: raise ValueError("Preprocesador BERT es None.")
|
| 177 |
if qformer_infer is None: raise ValueError("QFormer es None.")
|
| 178 |
+
|
| 179 |
detailed_results = {}
|
| 180 |
+
print("\n--- Calculando similitudes y clasificando ---")
|
| 181 |
+
|
| 182 |
for i in range(len(criteria_list_positive)):
|
| 183 |
+
positive_text = criteria_list_positive[i]
|
| 184 |
+
negative_text = criteria_list_negative[i]
|
| 185 |
+
criterion_name = positive_text # Usar prompt positivo como clave
|
| 186 |
+
|
| 187 |
+
print(f"Procesando criterio: \"{criterion_name}\"")
|
| 188 |
similarity_positive, similarity_negative, difference = None, None, None
|
| 189 |
classification_comp, classification_simp = "ERROR", "ERROR"
|
| 190 |
+
|
| 191 |
try:
|
| 192 |
+
# 1. Embedding Texto Positivo
|
| 193 |
tokens_pos, paddings_pos = bert_tokenize(positive_text, bert_preprocessor)
|
| 194 |
+
qformer_input_text_pos = {
|
| 195 |
+
'image_feature': np.zeros([1, 8, 8, 1376], dtype=np.float32).tolist(), # Dummy
|
| 196 |
+
'ids': tokens_pos.tolist(), 'paddings': paddings_pos.tolist(),
|
| 197 |
+
}
|
| 198 |
+
text_embedding_pos = qformer_infer(**qformer_input_text_pos)['contrastive_txt_emb'].numpy()
|
| 199 |
if text_embedding_pos.ndim == 1: text_embedding_pos = np.expand_dims(text_embedding_pos, axis=0)
|
| 200 |
|
| 201 |
+
# 2. Embedding Texto Negativo
|
| 202 |
tokens_neg, paddings_neg = bert_tokenize(negative_text, bert_preprocessor)
|
| 203 |
+
qformer_input_text_neg = {
|
| 204 |
+
'image_feature': np.zeros([1, 8, 8, 1376], dtype=np.float32).tolist(), # Dummy
|
| 205 |
+
'ids': tokens_neg.tolist(), 'paddings': paddings_neg.tolist(),
|
| 206 |
+
}
|
| 207 |
+
text_embedding_neg = qformer_infer(**qformer_input_text_neg)['contrastive_txt_emb'].numpy()
|
| 208 |
if text_embedding_neg.ndim == 1: text_embedding_neg = np.expand_dims(text_embedding_neg, axis=0)
|
| 209 |
|
| 210 |
+
# Verificar compatibilidad de dimensiones para similitud
|
| 211 |
+
if image_embedding.shape[1] != text_embedding_pos.shape[1]:
|
| 212 |
+
raise ValueError(f"Dimensión incompatible: Imagen ({image_embedding.shape[1]}) vs Texto Pos ({text_embedding_pos.shape[1]})")
|
| 213 |
+
if image_embedding.shape[1] != text_embedding_neg.shape[1]:
|
| 214 |
+
raise ValueError(f"Dimensión incompatible: Imagen ({image_embedding.shape[1]}) vs Texto Neg ({text_embedding_neg.shape[1]})")
|
| 215 |
|
| 216 |
+
# 3. Calcular Similitudes
|
| 217 |
similarity_positive = cosine_similarity(image_embedding, text_embedding_pos)[0][0]
|
| 218 |
similarity_negative = cosine_similarity(image_embedding, text_embedding_neg)[0][0]
|
| 219 |
+
print(f" Sim (+)={similarity_positive:.4f}, Sim (-)={similarity_negative:.4f}")
|
| 220 |
|
| 221 |
+
# 4. Clasificar
|
| 222 |
difference = similarity_positive - similarity_negative
|
| 223 |
classification_comp = "PASS" if difference > SIMILARITY_DIFFERENCE_THRESHOLD else "FAIL"
|
| 224 |
classification_simp = "PASS" if similarity_positive > POSITIVE_SIMILARITY_THRESHOLD else "FAIL"
|
| 225 |
+
print(f" Diff={difference:.4f} -> Comp: {classification_comp}, Simp: {classification_simp}")
|
| 226 |
+
|
| 227 |
except Exception as e:
|
| 228 |
+
print(f" ERROR procesando criterio '{criterion_name}': {e}")
|
| 229 |
+
traceback.print_exc()
|
| 230 |
+
# Mantener clasificaciones como "ERROR"
|
| 231 |
+
|
| 232 |
+
# Guardar resultados
|
| 233 |
detailed_results[criterion_name] = {
|
| 234 |
+
'positive_prompt': positive_text,
|
| 235 |
+
'negative_prompt': negative_text,
|
| 236 |
'similarity_positive': float(similarity_positive) if similarity_positive is not None else None,
|
| 237 |
'similarity_negative': float(similarity_negative) if similarity_negative is not None else None,
|
| 238 |
'difference': float(difference) if difference is not None else None,
|
| 239 |
+
'classification_comparative': classification_comp,
|
| 240 |
+
'classification_simplified': classification_simp
|
| 241 |
}
|
| 242 |
return detailed_results
|
| 243 |
|
| 244 |
# --- Carga Global de Modelos ---
|
| 245 |
+
# Se ejecuta UNA VEZ al iniciar la aplicación Gradio/Space
|
| 246 |
print("--- Iniciando carga global de modelos ---")
|
| 247 |
start_time = time.time()
|
| 248 |
models_loaded = False
|
| 249 |
bert_preprocessor_global = None
|
| 250 |
elixrc_infer_global = None
|
| 251 |
qformer_infer_global = None
|
| 252 |
+
|
| 253 |
try:
|
| 254 |
+
# Verificar autenticación HF (útil si se usan modelos privados, aunque no es el caso aquí)
|
| 255 |
+
# if HfFolder.get_token() is None:
|
| 256 |
+
# print("Advertencia: No se encontró token de Hugging Face.")
|
| 257 |
+
# else:
|
| 258 |
+
# print("Token de Hugging Face encontrado.")
|
| 259 |
|
| 260 |
+
# Crear directorio si no existe
|
| 261 |
os.makedirs(MODEL_DOWNLOAD_DIR, exist_ok=True)
|
| 262 |
print(f"Descargando/verificando modelos en: {MODEL_DOWNLOAD_DIR}")
|
| 263 |
snapshot_download(repo_id=MODEL_REPO_ID, local_dir=MODEL_DOWNLOAD_DIR,
|
| 264 |
allow_patterns=['elixr-c-v2-pooled/*', 'pax-elixr-b-text/*'],
|
| 265 |
+
local_dir_use_symlinks=False) # Evitar symlinks
|
| 266 |
print("Modelos descargados/verificados.")
|
| 267 |
|
| 268 |
+
# Cargar Preprocesador BERT desde TF Hub
|
| 269 |
print("Cargando Preprocesador BERT...")
|
| 270 |
+
# Usar handle explícito puede ser más robusto en algunos entornos
|
| 271 |
bert_preprocess_handle = "https://tfhub.dev/tensorflow/bert_en_uncased_preprocess/3"
|
| 272 |
bert_preprocessor_global = tf_hub.KerasLayer(bert_preprocess_handle)
|
| 273 |
print("Preprocesador BERT cargado.")
|
| 274 |
|
| 275 |
+
# Cargar ELIXR-C
|
| 276 |
print("Cargando ELIXR-C...")
|
| 277 |
elixrc_model_path = os.path.join(MODEL_DOWNLOAD_DIR, 'elixr-c-v2-pooled')
|
| 278 |
elixrc_model = tf.saved_model.load(elixrc_model_path)
|
| 279 |
elixrc_infer_global = elixrc_model.signatures['serving_default']
|
| 280 |
print("Modelo ELIXR-C cargado.")
|
| 281 |
|
| 282 |
+
# Cargar QFormer (ELIXR-B Text)
|
| 283 |
print("Cargando QFormer (ELIXR-B Text)...")
|
| 284 |
qformer_model_path = os.path.join(MODEL_DOWNLOAD_DIR, 'pax-elixr-b-text')
|
| 285 |
qformer_model = tf.saved_model.load(qformer_model_path)
|
|
|
|
| 289 |
models_loaded = True
|
| 290 |
end_time = time.time()
|
| 291 |
print(f"--- Modelos cargados globalmente con éxito en {end_time - start_time:.2f} segundos ---")
|
| 292 |
+
|
| 293 |
except Exception as e:
|
| 294 |
models_loaded = False
|
| 295 |
+
print(f"--- ERROR CRÍTICO DURANTE LA CARGA GLOBAL DE MODELOS ---")
|
| 296 |
+
print(e)
|
| 297 |
+
traceback.print_exc()
|
| 298 |
+
# Gradio se iniciará, pero la función de análisis fallará.
|
| 299 |
|
| 300 |
# --- Función Principal de Procesamiento para Gradio ---
|
| 301 |
+
def assess_quality(image_pil):
|
| 302 |
+
"""Función que Gradio llamará con la imagen de entrada."""
|
| 303 |
if not models_loaded:
|
| 304 |
raise gr.Error("Error: Los modelos no se pudieron cargar. La aplicación no puede procesar imágenes.")
|
| 305 |
if image_pil is None:
|
| 306 |
+
# Devolver resultados vacíos o un mensaje de error si no hay imagen
|
| 307 |
+
return pd.DataFrame(), "N/A", None # Dataframe vacío, Label vacío, JSON vacío
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 308 |
|
| 309 |
print("\n--- Iniciando evaluación para nueva imagen ---")
|
| 310 |
start_process_time = time.time()
|
| 311 |
+
|
| 312 |
try:
|
| 313 |
+
# 1. Convertir PIL Image a NumPy array (escala de grises)
|
| 314 |
+
# Gradio con type="pil" ya la entrega como objeto PIL
|
| 315 |
img_np = np.array(image_pil.convert('L'))
|
| 316 |
+
print(f"Imagen convertida a NumPy. Shape: {img_np.shape}, Tipo: {img_np.dtype}")
|
| 317 |
+
|
| 318 |
+
# 2. Generar Embedding de Imagen
|
| 319 |
+
print("Generando embedding de imagen...")
|
| 320 |
image_embedding = generate_image_embedding(img_np, elixrc_infer_global, qformer_infer_global)
|
| 321 |
+
print("Embedding de imagen generado.")
|
| 322 |
+
|
| 323 |
+
# 3. Calcular Similitudes y Clasificar
|
| 324 |
+
print("Calculando similitudes y clasificando criterios...")
|
| 325 |
detailed_results = calculate_similarities_and_classify(image_embedding, bert_preprocessor_global, qformer_infer_global)
|
| 326 |
+
print("Clasificación completada.")
|
| 327 |
+
|
| 328 |
+
# 4. Formatear Resultados para Gradio
|
| 329 |
+
output_data = []
|
| 330 |
+
passed_count = 0
|
| 331 |
+
total_count = 0
|
| 332 |
for criterion, details in detailed_results.items():
|
| 333 |
total_count += 1
|
| 334 |
+
sim_pos_str = f"{details['similarity_positive']:.4f}" if details['similarity_positive'] is not None else "N/A"
|
| 335 |
+
sim_neg_str = f"{details['similarity_negative']:.4f}" if details['similarity_negative'] is not None else "N/A"
|
| 336 |
+
diff_str = f"{details['difference']:.4f}" if details['difference'] is not None else "N/A"
|
| 337 |
+
assessment_comp = details['classification_comparative']
|
| 338 |
+
assessment_simp = details['classification_simplified']
|
| 339 |
+
output_data.append([
|
| 340 |
+
criterion,
|
| 341 |
+
sim_pos_str,
|
| 342 |
+
sim_neg_str,
|
| 343 |
+
diff_str,
|
| 344 |
+
assessment_comp,
|
| 345 |
+
assessment_simp
|
| 346 |
+
])
|
| 347 |
+
if assessment_comp == "PASS":
|
| 348 |
+
passed_count += 1
|
| 349 |
+
|
| 350 |
+
# Crear DataFrame
|
| 351 |
+
df_results = pd.DataFrame(output_data, columns=[
|
| 352 |
+
"Criterion", "Sim (+)", "Sim (-)", "Difference", "Assessment (Comp)", "Assessment (Simp)"
|
| 353 |
+
])
|
| 354 |
+
|
| 355 |
+
# Calcular etiqueta de calidad general
|
| 356 |
+
overall_quality = "Error"
|
| 357 |
if total_count > 0:
|
| 358 |
pass_rate = passed_count / total_count
|
| 359 |
if pass_rate >= 0.85: overall_quality = "Excellent"
|
| 360 |
elif pass_rate >= 0.70: overall_quality = "Good"
|
| 361 |
elif pass_rate >= 0.50: overall_quality = "Fair"
|
| 362 |
else: overall_quality = "Poor"
|
| 363 |
+
quality_label = f"{overall_quality} ({passed_count}/{total_count} criteria passed)"
|
| 364 |
+
|
| 365 |
end_process_time = time.time()
|
| 366 |
+
print(f"--- Evaluación completada en {end_process_time - start_process_time:.2f} segundos ---")
|
| 367 |
+
|
| 368 |
+
# Devolver DataFrame, Etiqueta y JSON
|
| 369 |
+
return df_results, quality_label, detailed_results
|
| 370 |
+
|
|
|
|
|
|
|
|
|
|
|
|
|
| 371 |
except Exception as e:
|
| 372 |
+
print(f"Error durante el procesamiento de la imagen en Gradio: {e}")
|
| 373 |
+
traceback.print_exc()
|
| 374 |
+
# Lanzar un gr.Error para mostrarlo en la UI de Gradio
|
| 375 |
+
raise gr.Error(f"Error procesando la imagen: {str(e)}")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 376 |
|
| 377 |
+
|
| 378 |
+
# --- Definir la Interfaz Gradio ---
|
| 379 |
+
css = """
|
| 380 |
+
#quality-label label {
|
| 381 |
+
font-size: 1.1em;
|
| 382 |
+
font-weight: bold;
|
| 383 |
+
}
|
| 384 |
+
"""
|
| 385 |
+
with gr.Blocks(css=css) as demo:
|
| 386 |
+
gr.Markdown(
|
| 387 |
+
"""
|
| 388 |
+
# Chest X-ray Technical Quality Assessment
|
| 389 |
+
Upload a chest X-ray image (PNG, JPG, etc.) to evaluate its technical quality based on 7 standard criteria
|
| 390 |
+
using the ELIXR model family (comparative strategy: Positive vs Negative prompts).
|
| 391 |
+
**Note:** Model loading on startup might take a minute. Processing an image can take 10-30 seconds depending on server load.
|
| 392 |
+
"""
|
| 393 |
+
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 394 |
with gr.Row():
|
| 395 |
+
with gr.Column(scale=1):
|
| 396 |
+
input_image = gr.Image(type="pil", label="Upload Chest X-ray")
|
| 397 |
+
submit_button = gr.Button("Assess Quality", variant="primary")
|
| 398 |
+
# Añadir ejemplos si tienes imágenes de ejemplo
|
| 399 |
+
# Asegúrate de que la carpeta 'examples' exista y contenga las imágenes
|
| 400 |
+
# gr.Examples(
|
| 401 |
+
# examples=[os.path.join("examples", "sample_cxr.png")], # Lista de rutas a ejemplos
|
| 402 |
+
# inputs=input_image
|
| 403 |
+
# )
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 404 |
with gr.Column(scale=2):
|
| 405 |
+
output_label = gr.Label(label="Overall Quality Estimate", elem_id="quality-label")
|
| 406 |
+
output_dataframe = gr.DataFrame(
|
| 407 |
+
headers=["Criterion", "Sim (+)", "Sim (-)", "Difference", "Assessment (Comp)", "Assessment (Simp)"],
|
| 408 |
+
label="Detailed Quality Assessment",
|
| 409 |
+
wrap=True,
|
| 410 |
+
height=350
|
| 411 |
+
)
|
| 412 |
+
output_json = gr.JSON(label="Raw Results (for debugging)")
|
| 413 |
+
|
| 414 |
+
|
| 415 |
+
gr.Markdown(
|
| 416 |
+
f"""
|
| 417 |
+
**Explanation:**
|
| 418 |
+
* **Criterion:** The quality aspect being evaluated (using the positive prompt text).
|
| 419 |
+
* **Sim (+):** Cosine similarity between the image and the *positive* text prompt (e.g., "optimal centering"). Higher is better.
|
| 420 |
+
* **Sim (-):** Cosine similarity between the image and the *negative* text prompt (e.g., "poorly centered"). Lower is better.
|
| 421 |
+
* **Difference:** Sim (+) - Sim (-). A large positive difference indicates the image is much closer to the positive description.
|
| 422 |
+
* **Assessment (Comp):** PASS if Difference > {SIMILARITY_DIFFERENCE_THRESHOLD}, otherwise FAIL. This is the main comparative assessment.
|
| 423 |
+
* **Assessment (Simp):** PASS if Sim (+) > {POSITIVE_SIMILARITY_THRESHOLD}, otherwise FAIL. A simpler check based only on positive similarity.
|
| 424 |
+
"""
|
|
|
|
|
|
|
| 425 |
)
|
| 426 |
+
|
| 427 |
+
# Conectar el botón a la función de procesamiento
|
| 428 |
+
submit_button.click(
|
| 429 |
+
fn=assess_quality,
|
| 430 |
+
inputs=input_image,
|
| 431 |
+
outputs=[output_dataframe, output_label, output_json]
|
| 432 |
)
|
| 433 |
|
| 434 |
# --- Iniciar la Aplicación Gradio ---
|
| 435 |
+
# Al desplegar en Spaces, Gradio se encarga de esto automáticamente.
|
| 436 |
+
# Para ejecutar localmente: demo.launch()
|
| 437 |
+
# Para Spaces, es mejor dejar que HF maneje el launch.
|
| 438 |
+
# demo.launch(share=True) # Para obtener un link público temporal si corres localmente
|
| 439 |
if __name__ == "__main__":
|
| 440 |
+
# share=True solo si quieres un enlace público temporal desde local
|
| 441 |
+
# server_name="0.0.0.0" para permitir conexiones de red local
|
| 442 |
+
# server_port=7860 es el puerto estándar de HF Spaces
|
| 443 |
demo.launch(server_name="0.0.0.0", server_port=7860)
|