bbox update

Author: kazmer97

lib/idp_common_pkg/idp_common/assessment/geometry_utils.py

@@ -0,0 +1,116 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0
+
+"""
+Shared utilities for geometry data conversion.
+
+Consolidates duplicate geometry conversion logic from service.py and granular_service.py.
+"""
+
+import os
+from typing import Any
+
+from aws_lambda_powertools import Logger
+
+from idp_common.assessment.models import Geometry
+
+logger = Logger(service="assessment", level=os.getenv("LOG_LEVEL", "INFO"))
+
+
+def process_assessment_geometry(
+    attr_assessment: dict[str, Any], attr_name: str = ""
+) -> dict[str, Any]:
+    """
+    Process and standardize geometry data in assessment response.
+
+    Args:
+        attr_assessment: Assessment dict with potential bbox/page fields
+        attr_name: Field name for logging
+
+    Returns:
+        Enhanced assessment with standardized geometry
+    """
+    enhanced = attr_assessment.copy()
+
+    # Check for bbox data
+    if "bbox" in attr_assessment and "page" in attr_assessment:
+        try:
+            bbox_coords = attr_assessment["bbox"]
+            page_num = attr_assessment["page"]
+
+            if isinstance(bbox_coords, list) and len(bbox_coords) == 4:
+                # Create Geometry object and convert to UI format
+                geometry = Geometry.from_bbox_list(bbox_coords, page_num)
+                enhanced["geometry"] = [geometry.to_ui_format()]
+
+                logger.debug(
+                    f"Converted bbox for {attr_name}: {bbox_coords} -> geometry"
+                )
+            else:
+                logger.warning(f"Invalid bbox format for {attr_name}: {bbox_coords}")
+        except Exception as e:
+            logger.warning(f"Failed to process bbox for {attr_name}: {e}")
+    elif "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"
+        )
+
+    # Remove raw bbox/page data
+    enhanced.pop("bbox", None)
+    enhanced.pop("page", None)
+
+    return enhanced
+
+
+def extract_geometry_from_nested_dict(
+    data: dict[str, Any], path: list[str] | None = None
+) -> dict[str, Any]:
+    """
+    Recursively process geometry data in nested assessment structures.
+
+    Args:
+        data: Assessment data dictionary (may contain nested dicts/lists)
+        path: Current path for logging
+
+    Returns:
+        Enhanced data with processed geometry
+    """
+    if path is None:
+        path = []
+
+    if not isinstance(data, dict):
+        return data
+
+    result = {}
+
+    for key, value in data.items():
+        current_path = path + [key]
+
+        if isinstance(value, dict):
+            # Check if this looks like an assessment entry
+            if "confidence" in value or "bbox" in value:
+                # Process this assessment
+                result[key] = process_assessment_geometry(value, ".".join(current_path))
+            else:
+                # Recurse into nested dict
+                result[key] = extract_geometry_from_nested_dict(value, current_path)
+
+        elif isinstance(value, list):
+            # Process each item in list
+            processed_list = []
+            for i, item in enumerate(value):
+                if isinstance(item, dict):
+                    processed_list.append(
+                        extract_geometry_from_nested_dict(item, current_path + [str(i)])
+                    )
+                else:
+                    processed_list.append(item)
+            result[key] = processed_list
+        else:
+            result[key] = value
+
+    return result

lib/idp_common_pkg/idp_common/assessment/granular_service.py

@@ -646,148 +646,7 @@ def _get_text_confidence_data(self, page) -> str:
         )
         return "Text Confidence Data Unavailable"
 
-    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)}"
-                    )
-                    raise
-            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
+    # Geometry processing uses shared utilities from geometry_utils module
 
     def process_document_section(self, document: Document, section_id: str) -> Document:
         """