chore(logging): add per-request timestamped logs and exception tracebacks for analyze paths

app.py

@@ -5,6 +5,12 @@ from PIL import Image
 import torch
 import numpy as np
 import cv2
+import time
+import traceback
+
+# Simple timestamped logger
+def log(msg: str) -> None:
+    print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
 
 # Writable cache directory for HF downloads
 HF_CACHE_DIR = os.getenv("HF_CACHE_DIR", "/data/hf-cache")
@@ -533,153 +539,118 @@ print(f"🔍 datapoint_model: {datapoint_model is not None}")
 
 # === Main prediction function ===
 def analyze(image):
-    """
-    Analyze a chart image and return comprehensive results.
-
-    Args:
-        image: Input chart image (filepath string or PIL.Image)
-
-    Returns:
-        dict: Analysis results containing:
-            - chart_type_id (int): Numeric chart type identifier (0-27)
-            - chart_type_label (str): Human-readable chart type name
-            - element_result (str): Detected chart elements (titles, axes, legends, etc.)
-            - datapoint_result (str): Segmented data points and regions
-            - status (str): Processing status message
-            - processing_time (float): Time taken for analysis in seconds
-    """
-    import time
-    from PIL import Image
-
-    start_time = time.time()
-
-    # Handle filepath input (convert to PIL Image)
-    if isinstance(image, str):
-        # It's a filepath, load the image
-        image = Image.open(image).convert("RGB")
-    elif image is None:
-        return {"error": "No image provided"}
-
-    # Ensure we have a PIL Image
-    if not isinstance(image, Image.Image):
-        return {"error": "Invalid image format"}
-
-    result = {
-        "chart_type_id": "Model not available",
-        "chart_type_label": "Model not available",
-        "element_result": "MMDetection models not available",
-        "datapoint_result": "MMDetection models not available",
-        "status": "Basic chart classification only",
-        "processing_time": 0.0,
-        "medsam": {"available": False}
-    }
+    try:
+        log("analyze: start")
+        start_time = time.time()
+        # Handle filepath input
+        if isinstance(image, str):
+            image = Image.open(image).convert("RGB")
+        elif image is None:
+            return {"error": "No image provided"}
+        if not isinstance(image, Image.Image):
+            return {"error": "Invalid image format"}
+
+        result = {
+            "chart_type_id": "Model not available",
+            "chart_type_label": "Model not available",
+            "element_result": "MMDetection models not available",
+            "datapoint_result": "MMDetection models not available",
+            "status": "Basic chart classification only",
+            "processing_time": 0.0,
+            "medsam": {"available": False}
+        }
 
-    # Chart Type Classification
-    if CHART_TYPE_AVAILABLE:
-        try:
-            # Preprocess image for PyTorch model
-            processed_image = chart_type_processor(image).unsqueeze(0)  # Add batch dimension
-
-            # Get prediction
-            with torch.no_grad():
-                outputs = chart_type_model(processed_image)
-                # Handle different output formats
-                if isinstance(outputs, torch.Tensor):
-                    logits = outputs
-                elif hasattr(outputs, 'logits'):
-                    logits = outputs.logits
+        # Chart Type Classification
+        if CHART_TYPE_AVAILABLE:
+            try:
+                processed_image = chart_type_processor(image).unsqueeze(0)
+                with torch.no_grad():
+                    outputs = chart_type_model(processed_image)
+                    logits = outputs if isinstance(outputs, torch.Tensor) else getattr(outputs, 'logits', outputs)
+                    predicted_class = logits.argmax(dim=-1).item()
+                result["chart_type_id"] = predicted_class
+                result["chart_type_label"] = CHART_TYPE_LABELS[predicted_class] if 0 <= predicted_class < len(CHART_TYPE_LABELS) else f"Unknown ({predicted_class})"
+                result["status"] = "Chart classification completed"
+                log(f"analyze: chart_type={result['chart_type_label']} ({result['chart_type_id']})")
+            except Exception:
+                log("analyze: chart classification error")
+                traceback.print_exc()
+
+        # Element Detection
+        if element_model is not None:
+            try:
+                np_img = np.array(image.convert("RGB"))[:, :, ::-1]
+                element_result = inference_detector(element_model, np_img)
+                if isinstance(element_result, tuple):
+                    bbox_result, segm_result = element_result
+                    element_data = {
+                        "bboxes": bbox_result.tolist() if hasattr(bbox_result, 'tolist') else str(bbox_result),
+                        "segments": segm_result.tolist() if hasattr(segm_result, 'tolist') else str(segm_result)
+                    }
                 else:
-                    logits = outputs
-
-                predicted_class = logits.argmax(dim=-1).item()
-
-            result["chart_type_id"] = predicted_class
-            result["chart_type_label"] = CHART_TYPE_LABELS[predicted_class] if 0 <= predicted_class < len(CHART_TYPE_LABELS) else f"Unknown ({predicted_class})"
-            result["status"] = "Chart classification completed"
-
-        except Exception as e:
-            result["chart_type_id"] = f"Error: {str(e)}"
-            result["chart_type_label"] = f"Error: {str(e)}"
-            result["status"] = "Error in chart classification"
-
-    # Chart Element Detection (Cascade R-CNN)
-    if element_model is not None:
-        try:
-            # Convert PIL image to numpy array for MMDetection
-            np_img = np.array(image.convert("RGB"))[:, :, ::-1]  # PIL → BGR
-
-            element_result = inference_detector(element_model, np_img)
-
-            # Convert result to more API-friendly format
-            if isinstance(element_result, tuple):
-                bbox_result, segm_result = element_result
-                element_data = {
-                    "bboxes": bbox_result.tolist() if hasattr(bbox_result, 'tolist') else str(bbox_result),
-                    "segments": segm_result.tolist() if hasattr(segm_result, 'tolist') else str(segm_result)
-                }
-            else:
-                element_data = str(element_result)
-
-            result["element_result"] = element_data
-            result["status"] = "Chart classification + element detection completed"
-        except Exception as e:
-            result["element_result"] = f"Error: {str(e)}"
-
-    # Chart Data Point Segmentation (Mask R-CNN)
-    if datapoint_model is not None:
+                    element_data = str(element_result)
+                result["element_result"] = element_data
+                result["status"] = "Chart classification + element detection completed"
+                log("analyze: element detection done")
+            except Exception:
+                log("analyze: element detection error")
+                traceback.print_exc()
+
+        # Datapoint Segmentation
+        if datapoint_model is not None:
+            try:
+                np_img = np.array(image.convert("RGB"))[:, :, ::-1]
+                datapoint_result = inference_detector(datapoint_model, np_img)
+                if isinstance(datapoint_result, tuple):
+                    bbox_result, segm_result = datapoint_result
+                    datapoint_data = {
+                        "bboxes": bbox_result.tolist() if hasattr(bbox_result, 'tolist') else str(bbox_result),
+                        "segments": segm_result.tolist() if hasattr(segm_result, 'tolist') else str(segm_result)
+                    }
+                else:
+                    datapoint_data = str(datapoint_result)
+                result["datapoint_result"] = datapoint_data
+                result["status"] = "Full analysis completed"
+                log("analyze: datapoint segmentation done")
+            except Exception:
+                log("analyze: datapoint segmentation error")
+                traceback.print_exc()
+
+        # MedSAM availability info
         try:
-            # Convert PIL image to numpy array for MMDetection
-            np_img = np.array(image.convert("RGB"))[:, :, ::-1]  # PIL → BGR
-
-            datapoint_result = inference_detector(datapoint_model, np_img)
-
-            # Convert result to more API-friendly format
-            if isinstance(datapoint_result, tuple):
-                bbox_result, segm_result = datapoint_result
-                datapoint_data = {
-                    "bboxes": bbox_result.tolist() if hasattr(bbox_result, 'tolist') else str(bbox_result),
-                    "segments": segm_result.tolist() if hasattr(segm_result, 'tolist') else str(segm_result)
-                }
-            else:
-                datapoint_data = str(datapoint_result)
-
-            result["datapoint_result"] = datapoint_data
-            result["status"] = "Full analysis completed"
-        except Exception as e:
-            result["datapoint_result"] = f"Error: {str(e)}"
-
-    # If predicted as medical image and MedSAM is available, include mask data (polygons)
-    try:
-        label_lower = str(result.get("chart_type_label", "")).strip().lower()
-        if label_lower == "medical image":
-            if _medsam.is_available():
-                # Indicate availability; masks are generated in then-chain
-                result["medsam"] = {"available": True}
-            else:
-                # Not available; include reason
-                result["medsam"] = {"available": False, "reason": "segment_anything or checkpoint missing"}
-    except Exception as e:
-        print(f"MedSAM JSON augmentation failed: {e}")
-
-    result["processing_time"] = round(time.time() - start_time, 3)
-    return result
+            label_lower = str(result.get("chart_type_label", "")).strip().lower()
+            if label_lower == "medical image":
+                if _medsam.is_available():
+                    result["medsam"] = {"available": True}
+                else:
+                    result["medsam"] = {"available": False, "reason": "segment_anything or checkpoint missing"}
+        except Exception:
+            log("analyze: medsam availability annotation error")
+            traceback.print_exc()
+
+        result["processing_time"] = round(time.time() - start_time, 3)
+        log(f"analyze: end in {result['processing_time']}s")
+        return result
+    except Exception:
+        log("analyze: fatal error")
+        traceback.print_exc()
+        return {"error": "Internal error in analyze"}
 
 
 def analyze_with_medsam(base_result, image):
-    """Auto-generate segmentations for medical images using SAM ViT-H if available,
-    otherwise fallback to MedSAM over top-K foreground boxes. Returns updated JSON and overlay image."""
     try:
+        log("analyze_with_medsam: start")
         if not isinstance(base_result, dict):
             return base_result, None
         label = str(base_result.get("chart_type_label", "")).strip().lower()
         if label != "medical image" or not _medsam.is_available():
+            log("analyze_with_medsam: skip (non-medical or MedSAM unavailable)")
             return base_result, None
 
         pil_img = Image.open(image).convert("RGB") if isinstance(image, str) else image
         if pil_img is None:
             return base_result, None
 
-        # Prepare embedding
         img_path = image if isinstance(image, str) else None
         if img_path is None:
             tmp_path = "./_tmp_input_image.png"
@@ -690,22 +661,19 @@ def analyze_with_medsam(base_result, image):
         segmentations = []
         masks_for_overlay = []
 
-        # AUTO segmentation path
-        try:
-            # Follow original behavior: use MedSAM with box prompts; no SAM auto in main path
-            cand_bboxes = _find_topk_foreground_bboxes(pil_img, max_regions=8, min_area=200)
-            for bbox in cand_bboxes:
-                m = _medsam.segment_with_box(bbox)
-                if m is None or not isinstance(m.get('mask'), np.ndarray):
-                    continue
-                segmentations.append({
-                    "mask": m['mask'].astype(np.uint8).tolist(),
-                    "confidence": float(m.get('confidence', 1.0)),
-                    "method": m.get("method", "medsam_box_auto")
-                })
-                masks_for_overlay.append(m)
-        except Exception as auto_e:
-            print(f"Automatic MedSAM segmentation failed: {auto_e}")
+        # MedSAM over candidate boxes (original behavior)
+        cand_bboxes = _find_topk_foreground_bboxes(pil_img, max_regions=8, min_area=200)
+        log(f"analyze_with_medsam: candidate boxes={len(cand_bboxes)}")
+        for bbox in cand_bboxes:
+            m = _medsam.segment_with_box(bbox)
+            if m is None or not isinstance(m.get('mask'), np.ndarray):
+                continue
+            segmentations.append({
+                "mask": m['mask'].astype(np.uint8).tolist(),
+                "confidence": float(m.get('confidence', 1.0)),
+                "method": m.get("method", "medsam_box_auto")
+            })
+            masks_for_overlay.append(m)
 
         W, H = pil_img.size
         base_result["medsam"] = {
@@ -715,11 +683,14 @@ def analyze_with_medsam(base_result, image):
             "segmentations": segmentations,
             "num_segments": len(segmentations)
         }
+        log(f"analyze_with_medsam: segments={len(segmentations)}")
 
         overlay_img = _overlay_masks_on_image(pil_img, masks_for_overlay) if masks_for_overlay else None
+        log("analyze_with_medsam: end")
         return base_result, overlay_img
-    except Exception as e:
-        print(f"analyze_with_medsam failed: {e}")
+    except Exception:
+        log("analyze_with_medsam: fatal error")
+        traceback.print_exc()
         return base_result, None
 
 # === Gradio UI with API enhancements ===