Update src/ai_processor.py

src/ai_processor.py

@@ -771,7 +771,89 @@ class AIProcessor:
         os.makedirs(out_dir, exist_ok=True)
         return out_dir
 
-    def perform_visual_analysis(self, image_pil: Image.Image) -> Dict:
+    def _compute_skin_tone(self, image_cv: np.ndarray, mask01: np.ndarray) -> Tuple[str, float]:
+        """
+        Estimate Fitzpatrick skin tone based on the Individual Typology Angle (ITA) computed
+        from the non-wound region of the provided ROI. If no background region exists,
+        use the entire image. Returns a tuple (label, ita_degrees).
+        """
+        try:
+            # Convert BGR to LAB color space
+            lab = cv2.cvtColor(image_cv, cv2.COLOR_BGR2LAB).astype(np.float32)
+            # Split channels (L in [0,255], A and B in [0,255])
+            L_channel = lab[:, :, 0]
+            b_channel = lab[:, :, 2]
+
+            # Determine background (skin) pixels: where mask==0
+            if mask01 is not None and mask01.size == L_channel.shape[0] * L_channel.shape[1]:
+                bg_mask = (mask01 == 0)
+            else:
+                bg_mask = np.ones_like(L_channel, dtype=bool)
+
+            # If background region is too small, use entire image
+            if bg_mask.sum() < max(100, 0.05 * bg_mask.size):
+                bg_mask = np.ones_like(L_channel, dtype=bool)
+
+            L_vals = L_channel[bg_mask]
+            b_vals = b_channel[bg_mask]
+
+            # Convert to CIELAB ranges: L* ∈ [0,100]; b* ∈ [-128,127]
+            L_star = L_vals * (100.0 / 255.0)
+            b_star = (b_vals - 128.0) * (200.0 / 255.0)
+            # Mean values
+            L_mean = float(np.mean(L_star)) if L_star.size > 0 else 50.0
+            b_mean = float(np.mean(b_star)) if b_star.size > 0 else 0.0
+
+            # ITA calculation
+            ita = np.degrees(np.arctan2((L_mean - 50.0), b_mean))
+            ita = float(ita)
+
+            # Classification based on Del Bino ranges
+            if ita > 55:
+                label = "Type I (Very Light)"
+            elif ita > 41:
+                label = "Type II (Light)"
+            elif ita > 28:
+                label = "Type III (Intermediate)"
+            elif ita > 10:
+                label = "Type IV (Tan)"
+            elif ita > -30:
+                label = "Type V (Brown)"
+            else:
+                label = "Type VI (Dark)"
+            return label, round(ita, 2)
+        except Exception:
+            return "Unknown", 0.0
+
+    def _compute_tissue_type(self, image_cv: np.ndarray, mask01: np.ndarray) -> str:
+        """
+        Classify wound tissue based on the average color of the wound region.
+        Returns one of ["Granulation", "Slough", "Necrotic", "Unknown"].
+        """
+        try:
+            if mask01 is None or not mask01.any():
+                return "Unknown"
+            # convert BGR to HSV; OpenCV hue in [0,179], saturation/value in [0,255]
+            hsv = cv2.cvtColor(image_cv, cv2.COLOR_BGR2HSV).astype(np.float32)
+            h = hsv[:, :, 0][mask01 == 1]
+            s = hsv[:, :, 1][mask01 == 1] / 255.0
+            v = hsv[:, :, 2][mask01 == 1] / 255.0
+            if v.size == 0:
+                return "Unknown"
+            h_mean = float(np.mean(h)) * 2.0  # convert to degrees 0-360
+            v_mean = float(np.mean(v))
+            # Dark = necrotic
+            if v_mean < 0.2:
+                return "Necrotic"
+            # Slough: yellowish hues ~15-40°
+            if 15.0 <= h_mean <= 40.0:
+                return "Slough"
+            # Otherwise granulation (reddish)
+            return "Granulation"
+        except Exception:
+            return "Unknown"
+
+    def perform_visual_analysis(self, image_pil: Image.Image, manual_mask_data: Optional[dict] = None) -> Dict:
         """
         YOLO detect → crop ROI → segment_wound(ROI) → clean mask →
         minAreaRect measurement (cm) using EXIF px/cm → save outputs.
@@ -783,8 +865,158 @@ class AIProcessor:
             if (exif_meta or {}).get("used") != "exif":
                 logging.warning(f"Calibration fallback used: px_per_cm={px_per_cm:.2f} (default). Prefer ruler/Aruco for accuracy.")
 
+            # Convert input PIL image to OpenCV BGR once here; manual mode may reuse it
             image_cv = cv2.cvtColor(np.array(image_pil.convert("RGB")), cv2.COLOR_RGB2BGR)
 
+            # ----------------------------------------------------------------------
+            # Manual annotation branch
+            # If a manual_mask_data dictionary/image is provided, bypass model detection
+            if manual_mask_data:
+                try:
+                    # Extract mask image from various possible structures
+                    mask_source = manual_mask_data
+                    if isinstance(mask_source, dict):
+                        mask_source = mask_source.get("mask") or mask_source.get("image") or mask_source
+                    # Load mask into numpy grayscale
+                    if isinstance(mask_source, Image.Image):
+                        mask_np = np.array(mask_source.convert("L"))
+                    elif isinstance(mask_source, np.ndarray):
+                        mask_np = mask_source.copy()
+                        if mask_np.ndim == 3:
+                            # If mask is RGB/A, convert to grayscale
+                            mask_np = cv2.cvtColor(mask_np, cv2.COLOR_BGR2GRAY)
+                    elif isinstance(mask_source, str) and os.path.exists(mask_source):
+                        mask_np = np.array(Image.open(mask_source).convert("L"))
+                    else:
+                        mask_np = None
+
+                    if mask_np is None:
+                        raise ValueError("Invalid manual mask")
+
+                    # Binary mask (1 for wound, 0 for background)
+                    mask01_full = (mask_np > 0).astype(np.uint8)
+                    h_full, w_full = image_cv.shape[:2]
+
+                    # Measurement using full-size mask
+                    segmentation_empty = not mask01_full.any()
+                    if not segmentation_empty:
+                        length_cm, breadth_cm, (box_pts, _) = measure_min_area_rect(mask01_full, px_per_cm)
+                        area_poly_cm2, largest_cnt = area_cm2_from_contour(mask01_full, px_per_cm)
+                        if largest_cnt is not None:
+                            surface_area_cm2 = clamp_area_with_minrect(largest_cnt, px_per_cm, area_poly_cm2)
+                        else:
+                            surface_area_cm2 = area_poly_cm2
+                        anno_roi = draw_measurement_overlay(image_cv.copy(), mask01_full, box_pts, length_cm, breadth_cm)
+                    else:
+                        # fallback: use full image dims
+                        h_px = h_full; w_px = w_full
+                        length_cm = round(max(h_px, w_px) / px_per_cm, 2)
+                        breadth_cm = round(min(h_px, w_px) / px_per_cm, 2)
+                        surface_area_cm2 = round((h_px * w_px) / (px_per_cm ** 2), 2)
+                        anno_roi = image_cv.copy()
+                        cv2.rectangle(anno_roi, (2, 2), (anno_roi.shape[1]-3, anno_roi.shape[0]-3), (0, 0, 255), 3)
+                        cv2.line(anno_roi, (0, 0), (anno_roi.shape[1]-1, anno_roi.shape[0]-1), (0, 0, 255), 2)
+                        cv2.line(anno_roi, (anno_roi.shape[1]-1, 0), (0, anno_roi.shape[0]-1), (0, 0, 255), 2)
+                        box_pts = None
+
+                    # Prepare output directory
+                    out_dir = self._ensure_analysis_dir()
+                    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+                    original_path = os.path.join(out_dir, f"original_{ts}.png")
+                    cv2.imwrite(original_path, image_cv)
+
+                    # Detection visualization: draw bounding box of mask
+                    det_vis = image_cv.copy()
+                    if mask01_full.any():
+                        ys, xs = np.where(mask01_full == 1)
+                        y_min, y_max = int(ys.min()), int(ys.max())
+                        x_min, x_max = int(xs.min()), int(xs.max())
+                        cv2.rectangle(det_vis, (x_min, y_min), (x_max, y_max), (0, 255, 0), 2)
+                    detection_path = os.path.join(out_dir, f"detection_{ts}.png")
+                    cv2.imwrite(detection_path, det_vis)
+
+                    # Save mask and overlays
+                    roi_mask_path = os.path.join(out_dir, f"roi_mask_{ts}.png")
+                    cv2.imwrite(roi_mask_path, (mask01_full * 255).astype(np.uint8))
+
+                    # ROI overlay: tinted mask
+                    mask255 = (mask01_full * 255).astype(np.uint8)
+                    mask3   = cv2.merge([mask255, mask255, mask255])
+                    red     = np.zeros_like(image_cv); red[:] = (0, 0, 255)
+                    alpha   = 0.55
+                    tinted  = cv2.addWeighted(image_cv, 1 - alpha, red, alpha, 0)
+                    if mask255.any():
+                        roi_overlay = np.where(mask3 > 0, tinted, image_cv)
+                        cnts, _ = cv2.findContours(mask255, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+                        cv2.drawContours(roi_overlay, cnts, -1, (255, 255, 255), 2)
+                    else:
+                        roi_overlay = anno_roi
+
+                    seg_full = image_cv.copy()
+                    seg_full[:, :] = roi_overlay
+                    segmentation_path = os.path.join(out_dir, f"segmentation_{ts}.png")
+                    cv2.imwrite(segmentation_path, seg_full)
+
+                    segmentation_roi_path = os.path.join(out_dir, f"segmentation_roi_{ts}.png")
+                    cv2.imwrite(segmentation_roi_path, roi_overlay)
+
+                    anno_full = image_cv.copy()
+                    anno_full[:, :] = anno_roi
+                    annotated_seg_path = os.path.join(out_dir, f"segmentation_annotated_{ts}.png")
+                    cv2.imwrite(annotated_seg_path, anno_full)
+
+                    # Classification: crop bounding box region for classification
+                    wound_type = "Unknown"
+                    cls_pipe = self.models_cache.get("cls")
+                    if cls_pipe is not None and mask01_full.any():
+                        try:
+                            roi_bbox = image_cv[y_min:y_max+1, x_min:x_max+1]
+                            preds = cls_pipe(Image.fromarray(cv2.cvtColor(roi_bbox, cv2.COLOR_BGR2RGB)))
+                            if preds:
+                                wound_type = max(preds, key=lambda x: x.get("score", 0)).get("label", "Unknown")
+                        except Exception as e:
+                            logging.warning(f"Classification failed: {e}")
+
+                    # Compute skin tone and tissue classification using full image and mask
+                    skin_tone_label, ita_degrees = self._compute_skin_tone(image_cv, mask01_full)
+                    tissue_type = self._compute_tissue_type(image_cv, mask01_full)
+
+                    seg_debug = {
+                        "used": "manual",
+                        "reason": "Manual annotation provided",
+                        "positive_fraction": float(mask01_full.mean()) if mask01_full.size > 0 else 0.0,
+                        "thr": SEG_THRESH,
+                    }
+
+                    return {
+                        "wound_type": wound_type,
+                        "length_cm": length_cm,
+                        "breadth_cm": breadth_cm,
+                        "surface_area_cm2": surface_area_cm2,
+                        "px_per_cm": round(px_per_cm, 2),
+                        "calibration_meta": exif_meta,
+                        "detection_confidence": 1.0 if mask01_full.any() else 0.0,
+                        "detection_image_path": detection_path,
+                        "segmentation_image_path": annotated_seg_path,
+                        "segmentation_annotated_path": annotated_seg_path,
+                        "segmentation_roi_path": segmentation_roi_path,
+                        "roi_mask_path": roi_mask_path,
+                        "segmentation_empty": segmentation_empty,
+                        "segmentation_debug": seg_debug,
+                        "original_image_path": original_path,
+                        "skin_tone_label": skin_tone_label,
+                        "ita_degrees": ita_degrees,
+                        "tissue_type": tissue_type,
+                        "segmentation_used": "manual",
+                    }
+                except Exception as e_manual:
+                    logging.error(f"Manual analysis failed: {e_manual}", exc_info=True)
+                    # If manual branch fails, fall back to automatic detection
+                    pass
+
+            # ----------------------------------------------------------------------
+            # Automatic model-based pipeline below
+
             # --- Detection ---
             det_model = self.models_cache.get("det")
             if det_model is None:
@@ -906,6 +1138,9 @@ class AIProcessor:
             }
             _log_kv("SEG_SUMMARY", seg_summary)
 
+            # Compute skin tone and tissue classification on ROI for automatic segmentation
+            skin_tone_label, ita_degrees = self._compute_skin_tone(roi, mask01)
+            tissue_type = self._compute_tissue_type(roi, mask01)
             return {
                 "wound_type": wound_type,
                 "length_cm": length_cm,
@@ -923,6 +1158,10 @@ class AIProcessor:
                 "segmentation_empty": segmentation_empty,
                 "segmentation_debug": seg_debug,
                 "original_image_path": original_path,
+                "skin_tone_label": skin_tone_label,
+                "ita_degrees": ita_degrees,
+                "tissue_type": tissue_type,
+                "segmentation_used": "automatic",
             }
         except Exception as e:
             logging.error(f"Visual analysis failed: {e}", exc_info=True)
@@ -1029,7 +1268,12 @@ Automated analysis provides quantitative measurements; verify via clinical exami
     def full_analysis_pipeline(self, image_pil: Image.Image, questionnaire_data: Dict) -> Dict:
         try:
             saved_path = self.save_and_commit_image(image_pil)
-            visual_results = self.perform_visual_analysis(image_pil)
+            # Extract any manual annotation mask from questionnaire data; remove it from the dict so it doesn't
+            # get forwarded to the text generation pipeline.
+            manual_mask_data = None
+            if isinstance(questionnaire_data, dict) and 'manual_mask' in questionnaire_data:
+                manual_mask_data = questionnaire_data.pop('manual_mask')
+            visual_results = self.perform_visual_analysis(image_pil, manual_mask_data)
 
             pi = questionnaire_data or {}
             patient_info = (