Auto-format code with Black

experiments/csv_file_generator_iterations/generate_emr_csv_final.py

@@ -16,7 +16,7 @@ SAMPLES_PER_CLASS = 300
 categories = {
     "COVID": IMAGES_DIR / "COVID",
     "NORMAL": IMAGES_DIR / "NORMAL",
-    "VIRAL PNEUMONIA": IMAGES_DIR / "VIRAL PNEUMONIA"
+    "VIRAL PNEUMONIA": IMAGES_DIR / "VIRAL PNEUMONIA",
 }
 
 # Shared ambiguous templates
@@ -36,7 +36,7 @@ shared_diagnosis = [
     "Further tests required to confirm diagnosis.",
     "Findings are borderline; clinical judgment advised.",
     "Observation warranted due to overlapping signs.",
-    "Initial assessment inconclusive."
+    "Initial assessment inconclusive.",
 ]
 
 # Noise sentences
@@ -52,8 +52,8 @@ neutral_noise = [
 
 def random_token():
     prefix = "ID"
-    letters = ''.join(random.choices(string.ascii_uppercase, k=2))
-    digits = ''.join(random.choices(string.digits, k=2))
+    letters = "".join(random.choices(string.ascii_uppercase, k=2))
+    digits = "".join(random.choices(string.digits, k=2))
     return f"{prefix}-{letters}{digits}"
 
 
@@ -97,7 +97,7 @@ def build_emr(label, i):
         intro,
         random.choice(shared_symptoms),
         vitals,
-        random.choice(shared_diagnosis)
+        random.choice(shared_diagnosis),
     ]
 
     # Optionally inject a mild class-specific clue (with low probability)
@@ -122,9 +122,13 @@ def build_emr(label, i):
 # Generate records
 records = []
 for label, img_dir in categories.items():
-    image_files = sorted([f for f in img_dir.glob("*") if f.suffix.lower() in [".png", ".jpg", ".jpeg"]])
+    image_files = sorted(
+        [f for f in img_dir.glob("*") if f.suffix.lower() in [".png", ".jpg", ".jpeg"]]
+    )
     for i in range(SAMPLES_PER_CLASS):
-        image_path = str(random.choice(image_files).relative_to(IMAGES_DIR.parent.parent))
+        image_path = str(
+            random.choice(image_files).relative_to(IMAGES_DIR.parent.parent)
+        )
         text = build_emr(label, i)
         triage = triage_map[label]
         records.append([f"{label}-{i + 1}", image_path, text, triage])

experiments/csv_file_generator_iterations/generate_emr_csv_v1.py

@@ -14,15 +14,11 @@ SAMPLES_PER_CLASS = 300  # 300 * 3 = 900 total
 categories = {
     "COVID": IMAGES_DIR / "COVID",
     "NORMAL": IMAGES_DIR / "NORMAL",
-    "VIRAL PNEUMONIA": IMAGES_DIR / "VIRAL PNEUMONIA"
+    "VIRAL PNEUMONIA": IMAGES_DIR / "VIRAL PNEUMONIA",
 }
 
 # Triage mapping
-triage_map = {
-    "COVID": "high",
-    "NORMAL": "low",
-    "VIRAL PNEUMONIA": "medium"
-}
+triage_map = {"COVID": "high", "NORMAL": "low", "VIRAL PNEUMONIA": "medium"}
 
 # --- Noise Sentences ---
 noise_sentences = [
@@ -43,7 +39,7 @@ noise_sentences = [
     "Patient remains alert and cooperative.",
     "No medication administered at this stage.",
     "Doctor recommends home resr and observation.",
-    "Evaluation ongoing for possible infection."
+    "Evaluation ongoing for possible infection.",
 ]
 
 # --- ambiguity sentences ---
@@ -52,18 +48,14 @@ ambiguous_templates = [
     "Normal oxygen levels observed. Slight wheeze on auscultation.",
     "Patient reports chest discomfort but vitals are stable.",
     "No known exposure. Minor throat irritation present.",
-    "Slight fatigue without other systemic symptoms."
+    "Slight fatigue without other systemic symptoms.",
 ]
 
 # --- Vitals & Symptoms ---
 
 
 def get_oxygen(label):
-    base_ranges = {
-        "COVID": (85, 94),
-        "VIRAL PNEUMONIA": (88, 95),
-        "NORMAL": (96, 99)
-    }
+    base_ranges = {"COVID": (85, 94), "VIRAL PNEUMONIA": (88, 95), "NORMAL": (96, 99)}
     base_min, base_max = base_ranges[label]
     # Apply + or - 1 blur, clamping between 80 and 100
     oxygen = random.randint(base_min - 1, base_max + 1)
@@ -112,7 +104,7 @@ def build_emr(label, i):
             f"{name} ({age}) complains of dry cough for {days} days.",
             f"{name} experiencing low-grade fever and SPO2 at {oxygen}%.",
             f"{name} reports breathlessness. X-ray indicates mild infiltrates.",
-        ]
+        ],
     }
 
     # Diagnosis Observations
@@ -120,18 +112,18 @@ def build_emr(label, i):
         "COVID": [
             "Findings suggest viral respiratory infection.",
             "Signs consistent with COVID-19 infection.",
-            "Clinical features align with COVID diagnosis."
+            "Clinical features align with COVID diagnosis.",
         ],
         "NORMAL": [
             "No signs of respiratory infection.",
             "No abnormal findings detected.",
-            "Checkup results within normal limits."
+            "Checkup results within normal limits.",
         ],
         "VIRAL PNEUMONIA": [
             "X-ray shows patchy infiltrates.",
             "Suspected viral origin of symptoms.",
-            "Clinical signs indicate viral pneumonia."
-        ]
+            "Clinical signs indicate viral pneumonia.",
+        ],
     }
 
     # Construct sentence pool
@@ -158,7 +150,9 @@ for label, img_dir in categories.items():
     )
     for i in range(SAMPLES_PER_CLASS):
         patient_id = f"{label}-{i + 1}"
-        image_path = str(random.choice(image_files).relative_to(IMAGES_DIR.parent.parent))
+        image_path = str(
+            random.choice(image_files).relative_to(IMAGES_DIR.parent.parent)
+        )
         emr_text = build_emr(label, i)
         triage_level = triage_map[label]
         records.append([patient_id, image_path, emr_text, triage_level])

experiments/csv_file_generator_iterations/generate_emr_csv_v2.py

@@ -16,7 +16,7 @@ SAMPLES_PER_CLASS = 300
 categories = {
     "COVID": IMAGES_DIR / "COVID",
     "NORMAL": IMAGES_DIR / "NORMAL",
-    "VIRAL PNEUMONIA": IMAGES_DIR / "VIRAL PNEUMONIA"
+    "VIRAL PNEUMONIA": IMAGES_DIR / "VIRAL PNEUMONIA",
 }
 
 # Shared ambiguous templates
@@ -43,8 +43,8 @@ neutral_noise = [
 # ---Patient random token genrator ---
 def random_token():
     prefix = "ID"
-    letters = ''.join(random.choices(string.ascii_uppercase, k=2))
-    digits = ''.join(random.choices(string.digits, k=2))
+    letters = "".join(random.choices(string.ascii_uppercase, k=2))
+    digits = "".join(random.choices(string.digits, k=2))
     return f"{prefix}-{letters}{digits}"
 
 
@@ -79,27 +79,41 @@ def build_emr(label, i):
     temp = get_temp(label)
     days = get_days()
 
-    general_intro = f"Patient {patient_id}, a {age}, presents with symptoms for {days} days."
+    general_intro = (
+        f"Patient {patient_id}, a {age}, presents with symptoms for {days} days."
+    )
     vitals = f"Temperature recorded at {temp}°F, SPO2 levels at {oxygen}%."
 
     # Label-specific (but fuzzy) symptoms
     symptoms = {
-        "COVID": ["Complains of fatigue and shortness of breath.", "Dry cough with mild fever noted."],
-        "NORMAL": ["No major complaints; here for general checkup.", "Reports good health, no active issues."],
-        "VIRAL PNEUMONIA": ["Persistent cough and mild fever observed.", "Slight wheezing with chest tightness."]
+        "COVID": [
+            "Complains of fatigue and shortness of breath.",
+            "Dry cough with mild fever noted.",
+        ],
+        "NORMAL": [
+            "No major complaints; here for general checkup.",
+            "Reports good health, no active issues.",
+        ],
+        "VIRAL PNEUMONIA": [
+            "Persistent cough and mild fever observed.",
+            "Slight wheezing with chest tightness.",
+        ],
     }
 
     diagnosis = {
         "COVID": ["Viral etiology suspected.", "COVID infection not ruled out."],
         "NORMAL": ["Unlikely presence of infection.", "Clinical impression is benign."],
-        "VIRAL PNEUMONIA": ["Signs may indicate atypical pneumonia.", "Possible viral infection of lower tract."]
+        "VIRAL PNEUMONIA": [
+            "Signs may indicate atypical pneumonia.",
+            "Possible viral infection of lower tract.",
+        ],
     }
 
     body = [
         general_intro,
         random.choice(symptoms[label]),
         vitals,
-        random.choice(diagnosis[label])
+        random.choice(diagnosis[label]),
     ]
 
     # Inject 1–2 ambiguous or neutral sentences
@@ -115,9 +129,13 @@ def build_emr(label, i):
 # Generate records
 records = []
 for label, img_dir in categories.items():
-    image_files = sorted([f for f in img_dir.glob("*") if f.suffix.lower() in [".png", ".jpg", ".jpeg"]])
+    image_files = sorted(
+        [f for f in img_dir.glob("*") if f.suffix.lower() in [".png", ".jpg", ".jpeg"]]
+    )
     for i in range(SAMPLES_PER_CLASS):
-        image_path = str(random.choice(image_files).relative_to(IMAGES_DIR.parent.parent))
+        image_path = str(
+            random.choice(image_files).relative_to(IMAGES_DIR.parent.parent)
+        )
         text = build_emr(label, i)
         triage = triage_map[label]
         records.append([f"{label}-{i + 1}", image_path, text, triage])

experiments/train_optuna.py

@@ -43,11 +43,7 @@ def objective(trial, mode):
         project=f"mediLLM-tune-{mode}",
         name=f"{mode}-trial-{trial.number}-v5-{wandb.util.generate_id()}",
         group="SoftLabelTrials",
-        config={
-            "dataset_version": "softlabels",
-            "dataset_size": 900,
-            "mode": mode
-        }
+        config={"dataset_version": "softlabels", "dataset_size": 900, "mode": mode},
     )
 
     # --- Hyperparameters ---
@@ -85,7 +81,9 @@ def objective(trial, mode):
                 images = images.to(device)
 
             optimizer.zero_grad()
-            outputs = model(input_ids=input_ids, attention_mask=attention_mask, image=images)
+            outputs = model(
+                input_ids=input_ids, attention_mask=attention_mask, image=images
+            )
             loss = criterion(outputs, labels)
             loss.backward()
             optimizer.step()
@@ -108,7 +106,9 @@ def objective(trial, mode):
             if images is not None:
                 images = images.to(device)
 
-            outputs = model(input_ids=input_ids, attention_mask=attention_mask, image=images)
+            outputs = model(
+                input_ids=input_ids, attention_mask=attention_mask, image=images
+            )
             preds = torch.argmax(outputs, dim=1).cpu().numpy()
             all_preds.extend(preds)
             all_labels.extend(labels.cpu().numpy())
@@ -117,21 +117,28 @@ def objective(trial, mode):
     acc = accuracy_score(all_labels, all_preds)
 
     # Log to W&B and Optuna
-    wandb.log({
-        "val_f1_score": f1,
-        "val_accuracy": acc,
-        "lr": lr,
-        "dropout": dropout,
-        "hidden_dim": hidden_dim,
-        "batch_size": batch_size
-    })
+    wandb.log(
+        {
+            "val_f1_score": f1,
+            "val_accuracy": acc,
+            "lr": lr,
+            "dropout": dropout,
+            "hidden_dim": hidden_dim,
+            "batch_size": batch_size,
+        }
+    )
 
     # Confusion Matrix
     cm = confusion_matrix(all_labels, all_preds)
     plt.figure(figsize=(6, 5))
-    sns.heatmap(cm, annot=True, fmt="d", cmap="Blues",
-                xticklabels=["low", "medium", "high"],
-                yticklabels=["low", "medium", "high"])
+    sns.heatmap(
+        cm,
+        annot=True,
+        fmt="d",
+        cmap="Blues",
+        xticklabels=["low", "medium", "high"],
+        yticklabels=["low", "medium", "high"],
+    )
     plt.title(f"Confusion Matrix - {mode} Trial {trial.number}")
     plt.xlabel("Predicted")
     plt.ylabel("True")
@@ -142,8 +149,16 @@ def objective(trial, mode):
 
 def get_args():
     parser = argparse.ArgumentParser(description="Run Optuna hyperparameter search")
-    parser.add_argument("--n_trials", type=int, default=10, help="Number of Optuna trials to run")
-    parser.add_argument("--mode", type=str, choices=["text", "image", "multimodal"], required=True, help="Input mode")
+    parser.add_argument(
+        "--n_trials", type=int, default=10, help="Number of Optuna trials to run"
+    )
+    parser.add_argument(
+        "--mode",
+        type=str,
+        choices=["text", "image", "multimodal"],
+        required=True,
+        help="Input mode",
+    )
     return parser.parse_args()
 
 
@@ -152,16 +167,17 @@ if __name__ == "__main__":
     mode = args.mode
 
     study = optuna.create_study(
-        study_name=f"mediLLM_{mode}_optuna",
-        direction="maximize"
+        study_name=f"mediLLM_{mode}_optuna", direction="maximize"
     )
     with tqdm(total=args.n_trials, desc=f"Optuna Trials [{mode}]") as pbar:
+
         def wrapped_objective(trial):
             try:
                 return objective(trial, mode)
             finally:
                 wandb.finish()
                 pbar.update(1)
+
         study.optimize(wrapped_objective, n_trials=args.n_trials)
 
     print(f"✅ Best F1 score for {mode}: {study.best_value}")
@@ -176,7 +192,7 @@ if __name__ == "__main__":
         "dropout": float(study.best_params["dropout"]),
         "hidden_dim": int(study.best_params["hidden_dim"]),
         "batch_size": int(study.best_params["bs"]),
-        "epochs": 5
+        "epochs": 5,
     }
 
     # Load existing or start new
@@ -210,7 +226,7 @@ if __name__ == "__main__":
         "dropout": float(study.best_params["dropout"]),
         "hidden_dim": int(study.best_params["hidden_dim"]),
         "batch_size": int(study.best_params["bs"]),
-        "epochs": 5
+        "epochs": 5,
     }
 
     # Export to config.yaml

src/data_preprocessing.py

@@ -33,11 +33,7 @@ def preprocess_text(text):
     if not isinstance(text, str):
         raise ValueError("Input text must be a string.")
     return tokenizer(
-        text,
-        truncation=True,
-        padding="max_length",
-        max_length=128,
-        return_tensors="pt"
+        text, truncation=True, padding="max_length", max_length=128, return_tensors="pt"
     )
 
 
@@ -60,9 +56,7 @@ if __name__ == "__main__":
     print("Data loaded successfully.")
     # apply function applies to each row in the 'image_path' column and joins
     # the base directory with the relative path
-    df["image_path"] = df["image_path"].apply(
-        lambda p: os.path.join(base_dir, p)
-    )
+    df["image_path"] = df["image_path"].apply(lambda p: os.path.join(base_dir, p))
     print("Sample record:")
     print(df.iloc[0])
 

src/generate_emr_csv.py

@@ -132,8 +132,7 @@ def generate_dataset():
         )
         for i in range(SAMPLES_PER_CLASS):
             image_path = str(
-                random.choice(image_files)
-                .relative_to(IMAGES_DIR.parent.parent)
+                random.choice(image_files).relative_to(IMAGES_DIR.parent.parent)
             )
             text = build_emr(label, i)
             triage = triage_map[label]
@@ -143,12 +142,7 @@ def generate_dataset():
     random.shuffle(records)
     with open(OUTPUT_FILE, "w", newline="") as f:
         writer = csv.writer(f)
-        writer.writerow([
-            "patient_id",
-            "image_path",
-            "emr_text",
-            "triage_level"
-        ])
+        writer.writerow(["patient_id", "image_path", "emr_text", "triage_level"])
         writer.writerows(records)
 
     print(f"✅ Softlabel EMR dataset generated at {OUTPUT_FILE}")

src/multimodal_model.py

@@ -7,9 +7,7 @@ from transformers import AutoModel  # Pretrained text encoders
 class MediLLMModel(nn.Module):
     def __init__(
         self,
-        text_model_name=(
-            "emilyalsentzer/Bio_ClinicalBERT"
-        ),
+        text_model_name=("emilyalsentzer/Bio_ClinicalBERT"),
         # Bio_ClinicalBERT is a pretrained model on clinical notes,
         # output to 3 classes i.e triage levels
         num_classes=3,
@@ -30,9 +28,7 @@ class MediLLMModel(nn.Module):
             text_model_name
         )  # Automodel returns base model without a classification head,
         # just embeddings
-        self.text_hidden_size = (
-            self.text_encoder.config.hidden_size
-        )
+        self.text_hidden_size = self.text_encoder.config.hidden_size
         # Dimensionality of hidden states i.e embedding vector size returned by
         # the text_encoder for each token, 768 for Bert models
 
@@ -75,16 +71,12 @@ class MediLLMModel(nn.Module):
         self.classifier = nn.Sequential(
             nn.Linear(fusion_dim, hidden_dim),  # Dense layer
             nn.ReLU(),  # Non-linear activation function
-            nn.Dropout(
-                dropout
-            ),  # randomly zeroes 30 percent of neuron outputs
+            nn.Dropout(dropout),  # randomly zeroes 30 percent of neuron outputs
             # to prevent over-fitting
             nn.Linear(hidden_dim, num_classes),  # Final Classification output
         )
 
-    def forward(
-        self, input_ids=None, attention_mask=None, image=None
-    ):
+    def forward(self, input_ids=None, attention_mask=None, image=None):
         # input_ids shape: [batch, seq_length]
         # attention_mask: mask to ignore padding, same shape as input_ids
         # image: [batch, 3, 224, 224]
@@ -122,6 +114,4 @@ class MediLLMModel(nn.Module):
             # -> [batch_size, 2816]
 
         # Return logits for each class, later apply softmax during evaluation
-        return self.classifier(
-            features
-        )
+        return self.classifier(features)

src/train.py

@@ -166,9 +166,7 @@ def train_model(mode="multimodal"):
             )  # Save labels for metric computation
 
         # Calculating classification metrics (Accuracy and F1)
-        acc = accuracy_score(
-            all_labels, all_preds
-        )  # Evaluate full-epoch performance
+        acc = accuracy_score(all_labels, all_preds)  # Evaluate full-epoch performance
         f1 = f1_score(all_labels, all_preds, average="weighted")
         # 1) binary: Binary Classification(F1 score of +ve class only)
         # 2) macro: Computes F1 for each class independently, then averages,
@@ -240,9 +238,7 @@ def train_model(mode="multimodal"):
     )  # Saves the model weights only not total architecture to reuse later
 
     # Plot accuracy
-    plot_path = os.path.join(
-        base_dir, "assets", f"model_training_curve_{mode}.png"
-    )
+    plot_path = os.path.join(base_dir, "assets", f"model_training_curve_{mode}.png")
     plt.plot(train_acc, label="Train Acc")
     plt.plot(val_acc, label="Val Acc")
     plt.legend()

src/triage_dataset.py

@@ -39,9 +39,7 @@ class TriageDataset(Dataset):
                         scale=(0.9, 1.0),
                         interpolation=InterpolationMode.BILINEAR,
                     ),  # Slight zoom-in/out
-                    transforms.RandomRotation(
-                        degrees=10
-                    ),  # + or - 10° rotation
+                    transforms.RandomRotation(degrees=10),  # + or - 10° rotation
                     transforms.ColorJitter(
                         brightness=0.3, contrast=0.3
                     ),  # simulate slight imaging variations

tests/test_generate_emr_csv.py

@@ -71,9 +71,7 @@ def test_total_and_per_class_counts(load_emr_csv):
     assert len(load_emr_csv) == 900, "Total records should be 900"
     counts = Counter(row["triage_level"] for row in load_emr_csv)
     for cls in EXPECTED_CLASSES:
-        assert counts[cls] == EXPECTED_SAMPLES_PER_CLASS, (
-            f"{cls} count mismatch"
-        )
+        assert counts[cls] == EXPECTED_SAMPLES_PER_CLASS, f"{cls} count mismatch"
 
 
 def test_patient_id_format_and_uniqueness(load_emr_csv):
@@ -94,9 +92,7 @@ def test_emr_text_quality(load_emr_csv):
 def test_image_path_format(load_emr_csv):
     for row in load_emr_csv:
         path = row["image_path"]
-        assert path.endswith((".jpg", ".jpeg", ".png")), (
-            f"Invalid image path: {path}"
-        )
+        assert path.endswith((".jpg", ".jpeg", ".png")), f"Invalid image path: {path}"
 
 
 def test_ambiguous_and_noise_injection(load_emr_csv):

tests/test_multimodal_model.py

@@ -44,9 +44,7 @@ def test_text_only(dummy_inputs):
         input_ids=dummy_inputs["input_ids"],
         attention_mask=dummy_inputs["attention_mask"],
     )
-    assert outputs.shape == (BATCH_SIZE, 3), (
-        "Incorrect output shape for text-only mode"
-    )
+    assert outputs.shape == (BATCH_SIZE, 3), "Incorrect output shape for text-only mode"
 
 
 def test_image_only(dummy_inputs):

tests/test_triage_dataset.py

@@ -24,9 +24,7 @@ def test_dataset_loading(mode):
     sample = dataset[0]
 
     if mode in ["text", "multimodal"]:
-        assert "input_ids" in sample, (
-            "Missing input_ids in text/multimodal mode"
-        )
+        assert "input_ids" in sample, "Missing input_ids in text/multimodal mode"
         assert (
             "attention_mask" in sample
         ), "Missing attention_mask in text/multimodal mode"