Add granular assessment bounding box conversion support with comprehensive testing

Author: Bob Strahan

lib/idp_common_pkg/idp_common/assessment/granular_service.py

@@ -720,6 +720,20 @@ def _process_assessment_task(
                             "confidence_reason": f"Unable to parse assessment response for {attr_name} - default score assigned",
                         }
 
+            # Process bounding boxes automatically if bbox data is present
+            try:
+                logger.debug(
+                    f"Checking for bounding box data in granular assessment task {task.task_id}"
+                )
+                assessment_data = self._extract_geometry_from_assessment(
+                    assessment_data
+                )
+            except Exception as e:
+                logger.warning(
+                    f"Failed to extract geometry data for task {task.task_id}: {str(e)}"
+                )
+                # Continue with assessment even if geometry extraction fails
+
             # Check for confidence threshold alerts
             confidence_alerts = []
             self._check_confidence_alerts_for_task(
@@ -987,6 +1001,149 @@ def _get_text_confidence_data(self, page) -> str:
 
         return ""
 
+    def _convert_bbox_to_geometry(
+        self, bbox_coords: List[float], page_num: int
+    ) -> Dict[str, Any]:
+        """
+        Convert [x1,y1,x2,y2] coordinates to geometry format.
+
+        Args:
+            bbox_coords: List of 4 coordinates [x1, y1, x2, y2] in 0-1000 scale
+            page_num: Page number where the bounding box appears
+
+        Returns:
+            Dictionary in geometry format compatible with pattern-1 UI
+        """
+        if len(bbox_coords) != 4:
+            raise ValueError(f"Expected 4 coordinates, got {len(bbox_coords)}")
+
+        x1, y1, x2, y2 = bbox_coords
+
+        # Ensure coordinates are in correct order
+        x1, x2 = min(x1, x2), max(x1, x2)
+        y1, y2 = min(y1, y2), max(y1, y2)
+
+        # Convert from normalized 0-1000 scale to 0-1
+        left = x1 / 1000.0
+        top = y1 / 1000.0
+        width = (x2 - x1) / 1000.0
+        height = (y2 - y1) / 1000.0
+
+        return {
+            "boundingBox": {"top": top, "left": left, "width": width, "height": height},
+            "page": page_num,
+        }
+
+    def _process_single_assessment_geometry(
+        self, attr_assessment: Dict[str, Any], attr_name: str = ""
+    ) -> Dict[str, Any]:
+        """
+        Process geometry data for a single assessment (with confidence key).
+
+        Args:
+            attr_assessment: Single assessment dictionary with confidence data
+            attr_name: Name of attribute for logging
+
+        Returns:
+            Enhanced assessment with geometry converted to proper format
+        """
+        enhanced_attr = attr_assessment.copy()
+
+        # Check if this assessment includes bbox data
+        if "bbox" in attr_assessment or "page" in attr_assessment:
+            # Both bbox and page are required for valid geometry
+            if "bbox" in attr_assessment and "page" in attr_assessment:
+                try:
+                    bbox_coords = attr_assessment["bbox"]
+                    page_num = attr_assessment["page"]
+
+                    # Validate bbox coordinates
+                    if isinstance(bbox_coords, list) and len(bbox_coords) == 4:
+                        # Convert to geometry format
+                        geometry = self._convert_bbox_to_geometry(bbox_coords, page_num)
+                        enhanced_attr["geometry"] = [geometry]
+
+                        logger.debug(
+                            f"Converted bounding box for {attr_name}: {bbox_coords} -> geometry format"
+                        )
+                    else:
+                        logger.warning(
+                            f"Invalid bounding box format for {attr_name}: {bbox_coords}"
+                        )
+
+                except Exception as e:
+                    logger.warning(
+                        f"Failed to process bounding box for {attr_name}: {str(e)}"
+                    )
+            else:
+                # If only one of bbox/page exists, log a warning about incomplete data
+                if "bbox" in attr_assessment and "page" not in attr_assessment:
+                    logger.warning(
+                        f"Found bbox without page for {attr_name} - removing incomplete bbox data"
+                    )
+                elif "page" in attr_assessment and "bbox" not in attr_assessment:
+                    logger.warning(
+                        f"Found page without bbox for {attr_name} - removing incomplete page data"
+                    )
+
+            # Always remove raw bbox/page data from output (whether processed or incomplete)
+            enhanced_attr.pop("bbox", None)
+            enhanced_attr.pop("page", None)
+
+        return enhanced_attr
+
+    def _extract_geometry_from_assessment(
+        self, assessment_data: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """
+        Extract geometry data from assessment response and convert to proper format.
+        Now supports recursive processing of nested group attributes.
+
+        Args:
+            assessment_data: Dictionary containing assessment results from LLM
+
+        Returns:
+            Enhanced assessment data with geometry information converted to proper format
+        """
+        enhanced_assessment = {}
+
+        for attr_name, attr_assessment in assessment_data.items():
+            if isinstance(attr_assessment, dict):
+                # Check if this is a direct confidence assessment
+                if "confidence" in attr_assessment:
+                    # This is a direct assessment - process its geometry
+                    enhanced_assessment[attr_name] = (
+                        self._process_single_assessment_geometry(
+                            attr_assessment, attr_name
+                        )
+                    )
+                else:
+                    # This is a group attribute (no direct confidence) - recursively process nested attributes
+                    logger.debug(f"Processing group attribute: {attr_name}")
+                    enhanced_assessment[attr_name] = (
+                        self._extract_geometry_from_assessment(attr_assessment)
+                    )
+
+            elif isinstance(attr_assessment, list):
+                # Handle list attributes - process each item recursively
+                enhanced_list = []
+                for i, item_assessment in enumerate(attr_assessment):
+                    if isinstance(item_assessment, dict):
+                        # Recursively process each list item
+                        enhanced_item = self._extract_geometry_from_assessment(
+                            item_assessment
+                        )
+                        enhanced_list.append(enhanced_item)
+                    else:
+                        # Non-dict items pass through unchanged
+                        enhanced_list.append(item_assessment)
+                enhanced_assessment[attr_name] = enhanced_list
+            else:
+                # Other types pass through unchanged
+                enhanced_assessment[attr_name] = attr_assessment
+
+        return enhanced_assessment
+
     def process_document_section(self, document: Document, section_id: str) -> Document:
         """
         Process a single section from a Document object to assess extraction confidence using granular approach.

lib/idp_common_pkg/tests/unit/assessment/test_granular_bbox_conversion.py

@@ -0,0 +1,144 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0
+
+"""
+Test to verify that both regular and granular assessment services
+handle bounding box conversion correctly.
+"""
+
+from idp_common.assessment.granular_service import GranularAssessmentService
+from idp_common.assessment.service import AssessmentService
+
+
+def test_both_services_convert_bbox_to_geometry():
+    """Test that both regular and granular services convert bbox to geometry."""
+
+    # Test data with bbox coordinates
+    mock_assessment_data = {
+        "YTDNetPay": {
+            "confidence": 1.0,
+            "confidence_reason": "Clear text with high OCR confidence",
+            "bbox": [443, 333, 507, 345],
+            "page": 1,
+        },
+        "CompanyAddress": {
+            "State": {
+                "confidence": 0.99,
+                "confidence_reason": "Clear text",
+                "bbox": [230, 116, 259, 126],
+                "page": 1,
+            },
+            "ZipCode": {
+                "confidence": 0.99,
+                "confidence_reason": "Clear text",
+                "bbox": [261, 116, 298, 126],
+                "page": 1,
+            },
+        },
+    }
+
+    print("=== Testing Bounding Box Conversion in Both Services ===")
+
+    # Test regular assessment service
+    print("\n📝 Testing Regular AssessmentService")
+    regular_service = AssessmentService()
+    regular_result = regular_service._extract_geometry_from_assessment(
+        mock_assessment_data
+    )
+
+    # Check YTDNetPay conversion
+    regular_ytd = regular_result["YTDNetPay"]
+    regular_ytd_has_geometry = "geometry" in regular_ytd
+    regular_ytd_has_bbox = "bbox" in regular_ytd
+
+    print(
+        f"Regular Service - YTDNetPay: geometry={regular_ytd_has_geometry}, bbox={regular_ytd_has_bbox}"
+    )
+
+    # Check CompanyAddress.State conversion
+    regular_state = regular_result["CompanyAddress"]["State"]
+    regular_state_has_geometry = "geometry" in regular_state
+    regular_state_has_bbox = "bbox" in regular_state
+
+    print(
+        f"Regular Service - CompanyAddress.State: geometry={regular_state_has_geometry}, bbox={regular_state_has_bbox}"
+    )
+
+    # Test granular assessment service
+    print("\n📝 Testing GranularAssessmentService")
+    granular_service = GranularAssessmentService()
+    granular_result = granular_service._extract_geometry_from_assessment(
+        mock_assessment_data
+    )
+
+    # Check YTDNetPay conversion
+    granular_ytd = granular_result["YTDNetPay"]
+    granular_ytd_has_geometry = "geometry" in granular_ytd
+    granular_ytd_has_bbox = "bbox" in granular_ytd
+
+    print(
+        f"Granular Service - YTDNetPay: geometry={granular_ytd_has_geometry}, bbox={granular_ytd_has_bbox}"
+    )
+
+    # Check CompanyAddress.State conversion
+    granular_state = granular_result["CompanyAddress"]["State"]
+    granular_state_has_geometry = "geometry" in granular_state
+    granular_state_has_bbox = "bbox" in granular_state
+
+    print(
+        f"Granular Service - CompanyAddress.State: geometry={granular_state_has_geometry}, bbox={granular_state_has_bbox}"
+    )
+
+    # Verify both services work identically
+    print("\n🔍 Verification:")
+
+    # Both should convert bbox to geometry
+    assert regular_ytd_has_geometry, (
+        "Regular service should convert YTDNetPay bbox to geometry"
+    )
+    assert not regular_ytd_has_bbox, (
+        "Regular service should remove YTDNetPay bbox after conversion"
+    )
+    assert granular_ytd_has_geometry, (
+        "Granular service should convert YTDNetPay bbox to geometry"
+    )
+    assert not granular_ytd_has_bbox, (
+        "Granular service should remove YTDNetPay bbox after conversion"
+    )
+
+    # Both should handle nested attributes
+    assert regular_state_has_geometry, (
+        "Regular service should convert nested State bbox to geometry"
+    )
+    assert not regular_state_has_bbox, (
+        "Regular service should remove nested State bbox after conversion"
+    )
+    assert granular_state_has_geometry, (
+        "Granular service should convert nested State bbox to geometry"
+    )
+    assert not granular_state_has_bbox, (
+        "Granular service should remove nested State bbox after conversion"
+    )
+
+    # Check geometry values are equivalent
+    regular_ytd_geometry = regular_ytd["geometry"][0]["boundingBox"]
+    granular_ytd_geometry = granular_ytd["geometry"][0]["boundingBox"]
+
+    assert regular_ytd_geometry == granular_ytd_geometry, (
+        "Both services should produce identical geometry"
+    )
+
+    print("✅ Regular AssessmentService: Converts bbox → geometry correctly")
+    print("✅ GranularAssessmentService: Converts bbox → geometry correctly")
+    print("✅ Both services handle nested attributes (CompanyAddress.State)")
+    print("✅ Both services produce identical geometry output")
+    print("✅ Both services remove raw bbox data after conversion")
+
+    print("\n🎉 Both services now support automatic bounding box conversion!")
+    print("Your deployed stack with granular assessment will now work correctly!")
+
+    return True
+
+
+if __name__ == "__main__":
+    test_both_services_convert_bbox_to_geometry()