first commit

README.md

@@ -1,12 +1,41 @@
 ---
 title: Xplainer
-emoji: 🔥
-colorFrom: blue
-colorTo: red
+emoji: 📊
+colorFrom: yellow
+colorTo: yellow
 sdk: gradio
-sdk_version: 3.35.2
+sdk_version: 3.34.0
+python_version: 3.7.16
 app_file: app.py
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+This is the official demo for the paper "Xplainer: From X-Ray Observations to Explainable Zero-Shot Diagnosis" (https://arxiv.org/pdf/2303.13391.pdf), which was accepted for publication at MICCAI 2023. 
+
+We propose a new way of explainability for zero-shot diagnosis prediction in the clinical domain. Instead of directly predicting a diagnosis, we prompt the model to classify the existence of descriptive observations, which a radiologist would look for on an X-Ray scan, and use the descriptor probabilities to estimate the likelihood of a diagnosis, making our model explainable by design. For this we leverage BioVil, a pretrained CLIP model for X-rays and apply contrastive observation-based prompting. We evaluate Xplainer on two chest X-ray
+datasets, CheXpert and ChestX-ray14, and demonstrate its effectiveness
+in improving the performance and explainability of zero-shot diagnosis.
+**Authors**: [Chantal Pellegrini][cp], [Matthias Keicher][mk], [Ege Özsoy][eo], [Petra Jiraskova][pj], [Rickmer Braren][rb], [Nassir Navab][nn]
+
+[cp]:https://www.cs.cit.tum.de/camp/members/chantal-pellegrini/
+[eo]:https://www.cs.cit.tum.de/camp/members/ege-oezsoy/
+[mk]:https://www.cs.cit.tum.de/camp/members/matthias-keicher/
+[pj]:https://campus.tum.de/tumonline/ee/ui/ca2/app/desktop/#/pl/ui/$ctx/visitenkarte.show_vcard?$ctx=design=ca2;header=max;lang=de&pPersonenGruppe=3&pPersonenId=46F3A857F258DEE6
+[rb]:https://radiologie.mri.tum.de/de/person/prof-dr-rickmer-f-braren
+[nn]:https://www.cs.cit.tum.de/camp/members/cv-nassir-navab/nassir-navab/
+
+Github: https://github.com/ChantalMP/Xplainer/tree/master
+
+```
+@inproceedings{pellegrini2023xplainer,
+    title={Xplainer: From X-Ray Observations to Explainable Zero-Shot Diagnosis},
+    author={Pellegrini, Chantal and Keicher, Matthias and Özsoy, Ege and Jiraskova, Petra and Braren, Rickmer and Navab, Nassir},
+    booktitle={International Conference on Medical Image Computing and Computer-Assisted Intervention},
+    year={2023},
+    organization={Springer}
+}
+```
+
+### Intended Use
+This model is intended to be used solely for (I) future research on visual-language processing and (II) reproducibility of the experimental results reported in the reference paper.

app.py

@@ -0,0 +1,131 @@
+from pathlib import Path
+
+import gradio as gr
+import numpy as np
+from matplotlib import pyplot as plt
+
+from descriptors import disease_descriptors_chexpert, disease_descriptors_chestxray14
+from model import InferenceModel
+
+
+def plot_bars(model_output):
+    # sort model_output by overall_probability
+    model_output = {k: v for k, v in sorted(model_output.items(), key=lambda item: item[1]['overall_probability'], reverse=True)}
+
+    # Create a figure with as many subplots as there are diseases, arranged vertically
+    fig, axs = plt.subplots(len(model_output), 1, figsize=(10, 5 * len(model_output)))
+    # axs is not iterable if only one subplot is created, so make it a list
+    if len(model_output) == 1:
+        axs = [axs]
+
+    for ax, (disease, data) in zip(axs, model_output.items()):
+        desc_probs = list(data['descriptor_probabilities'].items())
+        # sort descending
+        desc_probs = sorted(desc_probs, key=lambda item: item[1], reverse=True)
+
+        my_probs = [p[1] for p in desc_probs]
+        min_prob = min(my_probs)
+        max_prob = max(my_probs)
+        my_labels = [p[0] for p in desc_probs]
+
+        # Convert probabilities to differences from 0.5
+        diffs = np.abs(np.array(my_probs) - 0.5)
+
+        # Set colors based on sign of difference
+        colors = ['red' if p < 0.5 else 'forestgreen' for p in my_probs]
+
+        # Plot bars with appropriate colors and left offsets
+        left = [p if p < 0.5 else 0.5 for p in my_probs]
+        bars = ax.barh(my_labels, diffs, left=left, color=colors, alpha=0.3)
+
+        for i, bar in enumerate(bars):
+            ax.text(min_prob - 0.04, bar.get_y() + bar.get_height() / 2, my_labels[i], ha='left', va='center', color='black', fontsize=15)
+
+        ax.set_xlim(min(min_prob - 0.05, 0.49), max(max_prob + 0.05, 0.51))
+
+        # Invert the y-axis to show bars with values less than 0.5 to the left of the center
+        ax.invert_yaxis()
+
+        ax.set_yticks([])
+
+        # Add a title for the disease
+        if data['overall_probability'] >= 0.5:
+            ax.set_title(f"{disease} : score of {data['overall_probability']:.2f}")
+        else:
+            ax.set_title(f"No {disease} : score of {data['overall_probability']:.2f}")
+
+        # make title larger and bold
+        ax.title.set_fontsize(15)
+        ax.title.set_fontweight(600)
+
+    # Save the plot
+    plt.tight_layout()  # Adjust subplot parameters to give specified padding
+    file_path = 'plot.png'
+    plt.savefig(file_path)
+    plt.close(fig)
+
+    return file_path
+
+
+def classify_image(inference_model, image_path, diseases_to_predict):
+    descriptors_with_indication = [d + " indicating " + disease for disease, descriptors in diseases_to_predict.items() for d in descriptors]
+    probs, negative_probs = inference_model.get_descriptor_probs(image_path=Path(image_path), descriptors=descriptors_with_indication,
+                                                                 do_negative_prompting=True, demo=True)
+
+    disease_probs, negative_disease_probs = inference_model.get_diseases_probs(diseases_to_predict, pos_probs=probs, negative_probs=negative_probs)
+
+    model_output = {}
+    for idx, disease in enumerate(diseases_to_predict.keys()):
+        model_output[disease] = {
+            'overall_probability': disease_probs[disease],
+            'descriptor_probabilities': {descriptor: probs[f'{descriptor} indicating {disease}'].item() for descriptor in
+                                         diseases_to_predict[disease]}
+        }
+
+    file_path = plot_bars(model_output)
+    return file_path
+
+
+# Define the function you want to wrap
+def process_input(image_path, prompt_names: list, disease_name: str, descriptors: str):
+    diseases_to_predict = {}
+
+    for prompt in prompt_names:
+        if prompt == 'Custom':
+            diseases_to_predict[disease_name] = descriptors.split('\n')
+        else:
+            if prompt in disease_descriptors_chexpert:
+                diseases_to_predict[prompt] = disease_descriptors_chexpert[prompt]
+            else:  # only chestxray14
+                diseases_to_predict[prompt] = disease_descriptors_chestxray14[prompt]
+
+    # classify
+    model = InferenceModel()
+    output = classify_image(model, image_path, diseases_to_predict)
+
+    return output
+
+
+# Define the Gradio interface
+iface = gr.Interface(
+    fn=process_input,
+    examples = [['examples/enlarged_cardiomediastinum.jpg', ['Enlarged Cardiomediastinum'], '', ''],['examples/edema.jpg', ['Edema'], '', ''],
+                ['examples/support_devices.jpg', ['Custom'], 'Pacemaker', 'metalic object\nimplant on the left side of the chest\nimplanted cardiac device']],
+    inputs=[gr.inputs.Image(type="filepath"), gr.inputs.CheckboxGroup(
+        choices=['Enlarged Cardiomediastinum', 'Cardiomegaly', 'Lung Opacity', 'Lung Lesion', 'Edema', 'Consolidation', 'Pneumonia',
+                 'Atelectasis', 'Pneumothorax', 'Pleural Effusion', 'Pleural Other', 'Fracture', 'Support Devices',
+                 'Infiltration', 'Mass', 'Nodule', 'Emphysema', 'Fibrosis', 'Pleural_Thickening', 'Hernia',
+                 'Custom'],
+        default=['Enlarged Cardiomediastinum', 'Cardiomegaly', 'Lung Opacity', 'Lung Lesion', 'Edema', 'Consolidation', 'Pneumonia',
+                 'Atelectasis', 'Pneumothorax', 'Pleural Effusion', 'Pleural Other', 'Fracture', 'Support Devices'],
+        label='Selct to use predefined disease descriptors. Select "Custom" to define your own observations.'),
+            gr.inputs.Textbox(lines=2, placeholder="Name of pathology for which you want to define custom observations", label='Pathology:'),
+            gr.inputs.Textbox(lines=2, placeholder="Add your custom (positive) observations separated by a new line"
+                                                   "\n Note: Each descriptor will automatically be embedded into our prompt format: There is/are (no) <observation> indicating <pathology>"
+                                                   "\n Example:\n\n Opacity\nPleural Effusion\nConsolidation"
+                              , label='Custom Observations:')],
+    outputs=gr.outputs.Image(type="filepath")
+)
+
+# Launch the interface
+iface.launch()

descriptors.py

@@ -0,0 +1,204 @@
+disease_descriptors_chexpert = {
+    "No Finding": [
+        "Clear lung fields",
+        "Normal heart size and shape",
+        "No Abnormal fluid buildup",
+        "No Visible tumors or masses",
+        "No Signs of bone fractures or dislocations"
+    ],
+    "Enlarged Cardiomediastinum": [
+        "Increased width of the heart shadow",
+        "Widened mediastinum",
+        "Abnormal contour of the heart border",
+        "Fluid or air within the pericardium",
+        "Mass within the mediastinum",
+    ],
+    "Cardiomegaly": [
+        "Increased size of the heart shadow",
+        "Enlargement of the heart silhouette",
+        "Increased diameter of the heart border",
+        "Increased cardiothoracic ratio",
+    ],
+    "Lung Opacity": [
+        "Increased density in the lung field",
+        "Whitish or grayish area in the lung field",
+        "Obscured or blurred margins of the lung field",
+        "Loss of normal lung markings within the opacity",
+        "Air bronchograms within the opacity",
+        "Fluid levels within the opacity",
+        "Silhouette sign loss with adjacent structures",
+
+    ],
+    "Lung Lesion": [
+        "Consolidation of lung tissue",
+        "Pleural effusion",
+        "Cavities or abscesses in the lung",
+        "Abnormal opacity or shadow in the lung",
+        "Irregular or blurred margins of the lung",
+
+    ],
+    "Edema": [
+        "Blurry vascular markings in the lungs",
+        "Enlarged heart",
+        "Kerley B lines",
+        "Increased interstitial markings in the lungs",
+        "Widening of interstitial spaces",
+    ],
+    "Consolidation": [
+        "Loss of lung volume",
+        "Increased density of lung tissue",
+        "Obliteration of the diaphragmatic silhouette",
+        "Presence of opacities",
+    ],
+    "Pneumonia": [
+        "Consolidation of lung tissue",
+        "Air bronchograms",
+        "Cavitation",
+        "Interstitial opacities",
+    ],
+    "Atelectasis": [
+        "Increased opacity",
+        "Volume loss of the affected lung region",
+        "Blunting of the costophrenic angle",
+        "Shifting of the mediastinum",
+      ],
+      "Pneumothorax": [
+        "Tracheal deviation",
+        "Deep sulcus sign",
+        "Increased radiolucency",
+        "Flattening of the hemidiaphragm",
+        "Absence of lung markings",
+        "Shifting of the mediastinum"
+      ],
+      "Pleural Effusion": [
+        "Blunting of costophrenic angles",
+        "Opacity in the lower lung fields",
+        "Mediastinal shift",
+        "Reduced lung volume",
+        "Presence of meniscus sign or veil-like appearance"
+      ],
+      "Pleural Other": [
+        "Pleural thickening",
+        "Pleural calcification",
+        "Pleural masses or nodules",
+        "Pleural empyema",
+        "Pleural fibrosis",
+        "Pleural adhesions"
+      ],
+      "Fracture": [
+        "Visible breaks in the continuity of the bone",
+        "Misalignments of bone fragments",
+        "Displacements of bone fragments",
+        "Disruptions of the cortex or outer layer of the bone",
+        "Visible callus or healing tissue",
+        "Fracture lines that are jagged or irregular in shape",
+        "Multiple fracture lines that intersect at different angles"
+      ],
+      "Support Devices": [
+        "Artificial joints or implants",
+        "Pacemakers or cardiac devices",
+        "Stents or other vascular devices",
+        "Prosthetic devices or limbs",
+        "Breast implants",
+        "Radiotherapy markers or seeds"
+      ]
+    }
+
+disease_descriptors_chestxray14 = {
+
+    "No Finding": ["No Finding"],
+    "Cardiomegaly": [
+        "Increased size of the heart shadow",
+        "Enlargement of the heart silhouette",
+        "Increased diameter of the heart border",
+        "Increased cardiothoracic ratio"
+    ],
+    "Edema": [
+        "Blurry vascular markings in the lungs",
+        "Kerley B lines",
+        "Increased interstitial markings in the lungs",
+        "Widening of interstitial spaces"
+    ],
+    "Consolidation": [
+        "Loss of lung volume",
+        "Increased density of lung tissue",
+        "Obliteration of the diaphragmatic silhouette",
+        "Presence of opacities"
+    ],
+    "Pneumonia": [
+        "Consolidation of lung tissue",
+        "Air bronchograms",
+        "Cavitation",
+        "Interstitial opacities"
+    ],
+    "Atelectasis": [
+        "Increased opacity",
+        "Volume loss of the affected lung region",
+        "Displacement of the diaphragm",
+        "Blunting of the costophrenic angle",
+        "Shifting of the mediastinum"
+      ],
+    "Pneumothorax": [
+        "Tracheal deviation",
+        "Deep sulcus sign",
+        "Increased radiolucency",
+        "Flattening of the hemidiaphragm",
+        "Absence of lung markings",
+        "Shifting of the mediastinum"
+      ],
+    "Pleural Effusion": [
+        "Blunting of costophrenic angles",
+        "Opacity in the lower lung fields",
+        "Mediastinal shift",
+        "Reduced lung volume",
+        "Meniscus sign or veil-like appearance"
+      ],
+    "Infiltration": [
+        "Irregular or fuzzy borders around white areas",
+        "Blurring",
+        "Hazy or cloudy areas",
+        "Increased density or opacity of lung tissue",
+        "Air bronchograms",
+      ],
+      "Mass": [
+        "Calcifications or mineralizations",
+        "Shadowing",
+        "Distortion or compression of tissues",
+        "Anomalous structure or irregularity in shape"
+      ],
+      "Nodule": [
+        "Nodular shape that protrudes into a cavity or airway",
+        "Distinct edges or borders",
+        "Calcifications or speckled areas",
+        "Small round oral shaped spots",
+        "White shadows"
+      ],
+      "Emphysema": [
+        "Flattened hemidiaphragm",
+        "Pulmonary bullae",
+        "Hyperlucent lungs",
+        "Horizontalisation of ribs",
+        "Barrel Chest",
+      ],
+      "Fibrosis": [
+        "Reticular shadowing of the lung peripheries",
+        "Volume loss",
+        "Thickened and irregular interstitial markings",
+        "Bronchial dilation",
+        "Shaggy heart borders"
+      ],
+      "Pleural Thickening": [
+        "Thickened pleural line",
+        "Loss of sharpness of the mediastinal border",
+        "Calcifications on the pleura",
+        "Lobulated peripheral shadowing",
+        "Loss of lung volume",
+      ],
+      "Hernia": [
+        "Bulge or swelling in the abdominal wall",
+        "Protrusion of intestine or other abdominal tissue",
+        "Swelling or enlargement of the herniated sac or surrounding tissues",
+        "Retro-cardiac air-fluid level",
+        "Thickening of intestinal folds"
+      ]
+    }

flagged/log.csv

@@ -0,0 +1,2 @@
+image_path,"Selct to use predefined disease descriptors. Select ""Custom"" to define your own observations.",Pathology:,Custom Observations:,output,flag,username,timestamp
+/Users/chantal/Documents/programmieren_gitbackuped/Xplainer/flagged/image_path/tmp4gt9tvuq.png,['Cardiomegaly'],,,/Users/chantal/Documents/programmieren_gitbackuped/Xplainer/flagged/output/tmpy60f1_o9.png,,,2023-06-27 18:09:42.017441

inference.py

@@ -0,0 +1,116 @@
+import argparse
+import gc
+from pathlib import Path
+
+import torch
+from torch.utils.data import DataLoader
+from tqdm import tqdm
+
+from chestxray14 import ChestXray14Dataset
+from chexpert import CheXpertDataset
+from descriptors import disease_descriptors_chexpert, disease_descriptors_chestxray14
+from model import InferenceModel
+from utils import calculate_auroc
+
+torch.multiprocessing.set_sharing_strategy('file_system')
+
+
+def inference_chexpert():
+    split = 'test'
+    dataset = CheXpertDataset(f'data/chexpert/{split}_labels.csv')  # also do test
+    dataloader = DataLoader(dataset, batch_size=1, shuffle=False, collate_fn=lambda x: x, num_workers=0)
+    inference_model = InferenceModel()
+    all_descriptors = inference_model.get_all_descriptors(disease_descriptors_chexpert)
+
+    all_labels = []
+    all_probs_neg = []
+
+    for batch in tqdm(dataloader):
+        batch = batch[0]
+        image_paths, labels, keys = batch
+        image_paths = [Path(image_path) for image_path in image_paths]
+        agg_probs = []
+        agg_negative_probs = []
+        for image_path in image_paths:
+            probs, negative_probs = inference_model.get_descriptor_probs(image_path, descriptors=all_descriptors)
+            agg_probs.append(probs)
+            agg_negative_probs.append(negative_probs)
+        probs = {}  # Aggregated
+        negative_probs = {}  # Aggregated
+        for key in agg_probs[0].keys():
+            probs[key] = sum([p[key] for p in agg_probs]) / len(agg_probs)  # Mean Aggregation
+
+        for key in agg_negative_probs[0].keys():
+            negative_probs[key] = sum([p[key] for p in agg_negative_probs]) / len(agg_negative_probs)  # Mean Aggregation
+
+        disease_probs, negative_disease_probs = inference_model.get_diseases_probs(disease_descriptors_chexpert, pos_probs=probs,
+                                                                                   negative_probs=negative_probs)
+        predicted_diseases, prob_vector_neg_prompt = inference_model.get_predictions_bin_prompting(disease_descriptors_chexpert,
+                                                                                                   disease_probs=disease_probs,
+                                                                                                   negative_disease_probs=negative_disease_probs,
+                                                                                                   keys=keys)
+        all_labels.append(labels)
+        all_probs_neg.append(prob_vector_neg_prompt)
+
+    all_labels = torch.stack(all_labels)
+    all_probs_neg = torch.stack(all_probs_neg)
+
+    # evaluation
+    existing_mask = sum(all_labels, 0) > 0
+    all_labels_clean = all_labels[:, existing_mask]
+    all_probs_neg_clean = all_probs_neg[:, existing_mask]
+    all_keys_clean = [key for idx, key in enumerate(keys) if existing_mask[idx]]
+
+    overall_auroc, per_disease_auroc = calculate_auroc(all_probs_neg_clean, all_labels_clean)
+    print(f"AUROC: {overall_auroc:.5f}\n")
+    for idx, key in enumerate(all_keys_clean):
+        print(f'{key}: {per_disease_auroc[idx]:.5f}')
+
+
+def inference_chestxray14():
+    dataset = ChestXray14Dataset(f'data/chestxray14/Data_Entry_2017_v2020_modified.csv')
+    dataloader = DataLoader(dataset, batch_size=1, shuffle=False, collate_fn=lambda x: x, num_workers=1)
+    inference_model = InferenceModel()
+    all_descriptors = inference_model.get_all_descriptors(disease_descriptors_chestxray14)
+
+    all_labels = []
+    all_probs_neg = []
+    for batch in tqdm(dataloader):
+        batch = batch[0]
+        image_path, labels, keys = batch
+        image_path = Path(image_path)
+        probs, negative_probs = inference_model.get_descriptor_probs(image_path, descriptors=all_descriptors)
+        disease_probs, negative_disease_probs = inference_model.get_diseases_probs(disease_descriptors_chestxray14, pos_probs=probs,
+                                                                                   negative_probs=negative_probs)
+        predicted_diseases, prob_vector_neg_prompt = inference_model.get_predictions_bin_prompting(disease_descriptors_chestxray14,
+                                                                                                   disease_probs=disease_probs,
+                                                                                                   negative_disease_probs=negative_disease_probs,
+                                                                                                   keys=keys)
+        all_labels.append(labels)
+        all_probs_neg.append(prob_vector_neg_prompt)
+        gc.collect()
+
+    all_labels = torch.stack(all_labels)
+    all_probs_neg = torch.stack(all_probs_neg)
+
+    existing_mask = sum(all_labels, 0) > 0
+    all_labels_clean = all_labels[:, existing_mask]
+    all_probs_neg_clean = all_probs_neg[:, existing_mask]
+    all_keys_clean = [key for idx, key in enumerate(keys) if existing_mask[idx]]
+
+    overall_auroc, per_disease_auroc = calculate_auroc(all_probs_neg_clean[:, 1:], all_labels_clean[:, 1:])
+    print(f"AUROC: {overall_auroc:.5f}\n")
+    for idx, key in enumerate(all_keys_clean[1:]):
+        print(f'{key}: {per_disease_auroc[idx]:.5f}')
+
+
+if __name__ == '__main__':
+    # add argument parser
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--dataset', type=str, default='chexpert', help='chexpert or chestxray14')
+    args = parser.parse_args()
+
+    if args.dataset == 'chexpert':
+        inference_chexpert()
+    elif args.dataset == 'chestxray14':
+        inference_chestxray14()

model.py

@@ -0,0 +1,158 @@
+from pathlib import Path
+from typing import List
+
+import torch
+import torch.nn.functional as F
+from health_multimodal.image import get_biovil_resnet_inference
+from health_multimodal.text import get_cxr_bert_inference
+from health_multimodal.vlp import ImageTextInferenceEngine
+
+from utils import cos_sim_to_prob, prob_to_log_prob, log_prob_to_prob
+
+
+class InferenceModel():
+    def __init__(self):
+        self.text_inference = get_cxr_bert_inference()
+        self.image_inference = get_biovil_resnet_inference()
+        self.image_text_inference = ImageTextInferenceEngine(
+            image_inference_engine=self.image_inference,
+            text_inference_engine=self.text_inference,
+        )
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.image_text_inference.to(self.device)
+
+        # caches for faster inference
+        self.text_embedding_cache = {}
+        self.image_embedding_cache = {}
+
+        self.transform = self.image_inference.transform
+
+    def get_similarity_score_from_raw_data(self, image_embedding, query_text: str) -> float:
+        """Compute the cosine similarity score between an image and one or more strings.
+        If multiple strings are passed, their embeddings are averaged before L2-normalization.
+        :param image_path: Path to the input chest X-ray, either a DICOM or JPEG file.
+        :param query_text: Input radiology text phrase.
+        :return: The similarity score between the image and the text.
+        """
+        assert not self.image_text_inference.image_inference_engine.model.training
+        assert not self.image_text_inference.text_inference_engine.model.training
+        if query_text in self.text_embedding_cache:
+            text_embedding = self.text_embedding_cache[query_text]
+        else:
+            text_embedding = self.image_text_inference.text_inference_engine.get_embeddings_from_prompt([query_text], normalize=False)
+            text_embedding = text_embedding.mean(dim=0)
+            text_embedding = F.normalize(text_embedding, dim=0, p=2)
+            self.text_embedding_cache[query_text] = text_embedding
+
+        cos_similarity = image_embedding @ text_embedding.t()
+
+        return cos_similarity.item()
+
+    def process_image(self, image):
+        ''' same code as in image_text_inference.image_inference_engine.get_projected_global_embedding() but adapted to deal with image instances instead of path'''
+
+        transformed_image = self.transform(image)
+        projected_img_emb = self.image_inference.model.forward(transformed_image).projected_global_embedding
+        projected_img_emb = F.normalize(projected_img_emb, dim=-1)
+        assert projected_img_emb.shape[0] == 1
+        assert projected_img_emb.ndim == 2
+        return projected_img_emb[0]
+
+    def get_descriptor_probs(self, image_path: Path, descriptors: List[str], do_negative_prompting=True, demo=False):
+        probs = {}
+        negative_probs = {}
+        if image_path in self.image_embedding_cache:
+            image_embedding = self.image_embedding_cache[image_path]
+        else:
+            image_embedding = self.image_text_inference.image_inference_engine.get_projected_global_embedding(image_path)
+            if not demo:
+                self.image_embedding_cache[image_path] = image_embedding
+
+        # Default get_similarity_score_from_raw_data would load the image every time. Instead we only load once.
+        for desc in descriptors:
+            prompt = f'There are {desc}'
+            score = self.get_similarity_score_from_raw_data(image_embedding, prompt)
+            if do_negative_prompting:
+                neg_prompt = f'There are no {desc}'
+                neg_score = self.get_similarity_score_from_raw_data(image_embedding, neg_prompt)
+
+            pos_prob = cos_sim_to_prob(score)
+
+            if do_negative_prompting:
+                pos_prob, neg_prob = torch.softmax((torch.tensor([score, neg_score]) / 0.5), dim=0)
+                negative_probs[desc] = neg_prob
+
+            probs[desc] = pos_prob
+
+        return probs, negative_probs
+
+    def get_all_descriptors(self, disease_descriptors):
+        all_descriptors = set()
+        for disease, descs in disease_descriptors.items():
+            all_descriptors.update([f"{desc} indicating {disease}" for desc in descs])
+        all_descriptors = sorted(all_descriptors)
+        return all_descriptors
+
+    def get_all_descriptors_only_disease(self, disease_descriptors):
+        all_descriptors = set()
+        for disease, descs in disease_descriptors.items():
+            all_descriptors.update([f"{desc}" for desc in descs])
+        all_descriptors = sorted(all_descriptors)
+        return all_descriptors
+
+    def get_diseases_probs(self, disease_descriptors, pos_probs, negative_probs, prior_probs=None, do_negative_prompting=True):
+        disease_probs = {}
+        disease_neg_probs = {}
+        for disease, descriptors in disease_descriptors.items():
+            desc_log_probs = []
+            desc_neg_log_probs = []
+            for desc in descriptors:
+                desc = f"{desc} indicating {disease}"
+                desc_log_probs.append(prob_to_log_prob(pos_probs[desc]))
+                if do_negative_prompting:
+                    desc_neg_log_probs.append(prob_to_log_prob(negative_probs[desc]))
+            disease_log_prob = sum(sorted(desc_log_probs, reverse=True)) / len(desc_log_probs)
+            if do_negative_prompting:
+                disease_neg_log_prob = sum(desc_neg_log_probs) / len(desc_neg_log_probs)
+            disease_probs[disease] = log_prob_to_prob(disease_log_prob)
+            if do_negative_prompting:
+                disease_neg_probs[disease] = log_prob_to_prob(disease_neg_log_prob)
+
+        return disease_probs, disease_neg_probs
+
+    # Threshold Based
+    def get_predictions(self, disease_descriptors, threshold, disease_probs, keys):
+        predicted_diseases = []
+        prob_vector = torch.zeros(len(keys), dtype=torch.float)  # num of diseases
+        for idx, disease in enumerate(disease_descriptors):
+            if disease == 'No Finding':
+                continue
+            prob_vector[keys.index(disease)] = disease_probs[disease]
+            if disease_probs[disease] > threshold:
+                predicted_diseases.append(disease)
+
+        if len(predicted_diseases) == 0:  # No finding rule based
+            prob_vector[0] = 1.0 - max(prob_vector)
+        else:
+            prob_vector[0] = 1.0 - max(prob_vector)
+
+        return predicted_diseases, prob_vector
+
+    # Negative vs Positive Prompting
+    def get_predictions_bin_prompting(self, disease_descriptors, disease_probs, negative_disease_probs, keys):
+        predicted_diseases = []
+        prob_vector = torch.zeros(len(keys), dtype=torch.float)  # num of diseases
+        for idx, disease in enumerate(disease_descriptors):
+            if disease == 'No Finding':
+                continue
+            pos_neg_scores = torch.tensor([disease_probs[disease], negative_disease_probs[disease]])
+            prob_vector[keys.index(disease)] = pos_neg_scores[0]
+            if torch.argmax(pos_neg_scores) == 0:  # Positive is More likely
+                predicted_diseases.append(disease)
+
+        if len(predicted_diseases) == 0:  # No finding rule based
+            prob_vector[0] = 1.0 - max(prob_vector)
+        else:
+            prob_vector[0] = 1.0 - max(prob_vector)
+
+        return predicted_diseases, prob_vector

pre-requirements.txt

@@ -0,0 +1 @@
+hi-ml-multimodal==0.1.2

requirements.txt

@@ -0,0 +1,5 @@
+scikit-learn==1.0.2
+transformers==4.17.0
+gradio==3.34.0
+pandas==1.3.5
+torch==1.13.0

utils.py

@@ -0,0 +1,40 @@
+from math import log, exp
+
+import numpy as np
+from sklearn.metrics import roc_auc_score
+
+
+def cos_sim_to_prob(sim):
+    return (sim + 1) / 2  # linear transformation to 0 and 1
+
+
+def log_prob_to_prob(log_prob):
+    return exp(log_prob)
+
+
+def prob_to_log_prob(prob):
+    return log(prob)
+
+
+def calculate_auroc(all_disease_probs, gt_diseases):
+    '''
+    Calculates the AUROC (Area Under the Receiver Operating Characteristic curve) for multiple diseases.
+
+    Parameters:
+    all_disease_probs (numpy array): predicted disease labels, a multi-hot vector of shape (N_samples, 14)
+    gt_diseases (numpy array): ground truth disease labels, a multi-hot vector of shape (N_samples, 14)
+
+    Returns:
+    overall_auroc (float): the overall AUROC score
+    per_disease_auroc (numpy array): an array of shape (14,) containing the AUROC score for each disease
+    '''
+
+    per_disease_auroc = np.zeros((gt_diseases.shape[1],))  # num of diseases
+    for i in range(gt_diseases.shape[1]):
+        # Compute the AUROC score for each disease
+        per_disease_auroc[i] = roc_auc_score(gt_diseases[:, i], all_disease_probs[:, i])
+
+    # Compute the overall AUROC score
+    overall_auroc = roc_auc_score(gt_diseases, all_disease_probs, average='macro')
+
+    return overall_auroc, per_disease_auroc