object_group_processor.py

import logging
from typing import Dict, List, Tuple, Optional, Any

class ObjectGroupProcessor:
    """
    物件組處理器 - 專門處理物件分組、排序和子句生成的邏輯
    負責物件按類別分組、重複物件檢測移除、物件組優先級排序以及描述子句的生成
    """

    def __init__(self, confidence_threshold_for_description: float = 0.25,
                 spatial_handler: Optional[Any] = None,
                 text_optimizer: Optional[Any] = None):
        """
        初始化物件組處理器

        Args:
            confidence_threshold_for_description: 用於描述的置信度閾值
            spatial_handler: 空間位置處理器實例
            text_optimizer: 文本優化器實例
        """
        self.logger = logging.getLogger(self.__class__.__name__)
        self.confidence_threshold_for_description = confidence_threshold_for_description
        self.spatial_handler = spatial_handler
        self.text_optimizer = text_optimizer

    def group_objects_by_class(self, confident_objects: List[Dict],
                              object_statistics: Optional[Dict]) -> Dict[str, List[Dict]]:
        """
        按類別分組物件

        Args:
            confident_objects: 置信度過濾後的物件
            object_statistics: 物件統計信息

        Returns:
            Dict[str, List[Dict]]: 按類別分組的物件
        """
        objects_by_class = {}

        if object_statistics:
            # 使用預計算的統計信息，採用動態的信心度
            for class_name, stats in object_statistics.items():
                count = stats.get("count", 0)
                avg_confidence = stats.get("avg_confidence", 0)

                # 動態調整置信度閾值
                dynamic_threshold = self.confidence_threshold_for_description
                if class_name in ["potted plant", "vase", "clock", "book"]:
                    dynamic_threshold = max(0.15, self.confidence_threshold_for_description * 0.6)
                elif count >= 3:
                    dynamic_threshold = max(0.2, self.confidence_threshold_for_description * 0.8)

                if count > 0 and avg_confidence >= dynamic_threshold:
                    matching_objects = [obj for obj in confident_objects if obj.get("class_name") == class_name]
                    if not matching_objects:
                        matching_objects = [obj for obj in confident_objects
                                          if obj.get("class_name") == class_name and obj.get("confidence", 0) >= dynamic_threshold]

                    if matching_objects:
                        actual_count = min(stats["count"], len(matching_objects))
                        objects_by_class[class_name] = matching_objects[:actual_count]

                        # Debug logging for specific classes
                        if class_name in ["car", "traffic light", "person", "handbag"]:
                            print(f"DEBUG: Before spatial deduplication:")
                            print(f"DEBUG: {class_name}: {len(objects_by_class[class_name])} objects before dedup")
        else:
            # 備用邏輯，同樣使用動態閾值
            for obj in confident_objects:
                name = obj.get("class_name", "unknown object")
                if name == "unknown object" or not name:
                    continue
                if name not in objects_by_class:
                    objects_by_class[name] = []
                objects_by_class[name].append(obj)

        return objects_by_class

    def remove_duplicate_objects(self, objects_by_class: Dict[str, List[Dict]]) -> Dict[str, List[Dict]]:
        """
        移除重複物件

        Args:
            objects_by_class: 按類別分組的物件

        Returns:
            Dict[str, List[Dict]]: 去重後的物件
        """
        deduplicated_objects_by_class = {}
        processed_positions = []

        for class_name, group_of_objects in objects_by_class.items():
            unique_objects = []

            for obj in group_of_objects:
                obj_position = obj.get("normalized_center", [0.5, 0.5])
                is_duplicate = False

                for processed_pos in processed_positions:
                    position_distance = abs(obj_position[0] - processed_pos[0]) + abs(obj_position[1] - processed_pos[1])
                    if position_distance < 0.15:
                        is_duplicate = True
                        break

                if not is_duplicate:
                    unique_objects.append(obj)
                    processed_positions.append(obj_position)

            if unique_objects:
                deduplicated_objects_by_class[class_name] = unique_objects

        # Debug logging after deduplication
        for class_name in ["car", "traffic light", "person", "handbag"]:
            if class_name in deduplicated_objects_by_class:
                print(f"DEBUG: After spatial deduplication:")
                print(f"DEBUG: {class_name}: {len(deduplicated_objects_by_class[class_name])} objects after dedup")

        return deduplicated_objects_by_class

    def sort_object_groups(self, objects_by_class: Dict[str, List[Dict]]) -> List[Tuple[str, List[Dict]]]:
        """
        排序物件組

        Args:
            objects_by_class: 按類別分組的物件

        Returns:
            List[Tuple[str, List[Dict]]]: 排序後的物件組
        """
        def sort_key_object_groups(item_tuple: Tuple[str, List[Dict]]):
            class_name_key, obj_group_list = item_tuple
            priority = 3
            count = len(obj_group_list)

            # 確保類別名稱已標準化
            normalized_class_name = self._normalize_object_class_name(class_name_key)

            # 動態優先級
            if normalized_class_name == "person":
                priority = 0
            elif normalized_class_name in ["dining table", "chair", "sofa", "bed"]:
                priority = 1
            elif normalized_class_name in ["car", "bus", "truck", "traffic light"]:
                priority = 2
            elif count >= 3:
                priority = max(1, priority - 1)
            elif normalized_class_name in ["potted plant", "vase", "clock", "book"] and count >= 2:
                priority = 2

            avg_area = sum(o.get("normalized_area", 0.0) for o in obj_group_list) / len(obj_group_list) if obj_group_list else 0
            quantity_bonus = min(count / 5.0, 1.0)

            return (priority, -len(obj_group_list), -avg_area, -quantity_bonus)

        return sorted(objects_by_class.items(), key=sort_key_object_groups)

    def generate_object_clauses(self, sorted_object_groups: List[Tuple[str, List[Dict]]],
                               object_statistics: Optional[Dict],
                               scene_type: str,
                               image_width: Optional[int],
                               image_height: Optional[int],
                               region_analyzer: Optional[Any] = None) -> List[str]:
        """
        生成物件描述子句

        Args:
            sorted_object_groups: 排序後的物件組
            object_statistics: 物件統計信息
            scene_type: 場景類型
            image_width: 圖像寬度
            image_height: 圖像高度
            region_analyzer: 區域分析器實例

        Returns:
            List[str]: 物件描述子句列表
        """
        object_clauses = []

        for class_name, group_of_objects in sorted_object_groups:
            count = len(group_of_objects)

            # Debug logging for final count
            if class_name in ["car", "traffic light", "person", "handbag"]:
                print(f"DEBUG: Final count for {class_name}: {count}")

            if count == 0:
                continue

            # 標準化class name
            normalized_class_name = self._normalize_object_class_name(class_name)

            # 使用統計信息確保準確的數量描述
            if object_statistics and class_name in object_statistics:
                actual_count = object_statistics[class_name]["count"]
                formatted_name_with_exact_count = self._format_object_count_description(
                    normalized_class_name,
                    actual_count,
                    scene_type=scene_type
                )
            else:
                formatted_name_with_exact_count = self._format_object_count_description(
                    normalized_class_name,
                    count,
                    scene_type=scene_type
                )

            if formatted_name_with_exact_count == "no specific objects clearly identified" or not formatted_name_with_exact_count:
                continue

            # 確定群組的集體位置
            location_description_suffix = self._generate_location_description(
                group_of_objects, count, image_width, image_height, region_analyzer
            )

            # 首字母大寫
            formatted_name_capitalized = formatted_name_with_exact_count[0].upper() + formatted_name_with_exact_count[1:]
            object_clauses.append(f"{formatted_name_capitalized} {location_description_suffix}")

        return object_clauses

    def format_object_clauses(self, object_clauses: List[str]) -> str:
        """
        格式化物件描述子句

        Args:
            object_clauses: 物件描述子句列表

        Returns:
            str: 格式化後的描述
        """
        if not object_clauses:
            return "No common objects were confidently identified for detailed description."

        # 處理第一個子句
        first_clause = object_clauses.pop(0)
        result = first_clause + "."

        # 處理剩餘子句
        if object_clauses:
            result += " The scene features:"
            joined_object_clauses = ". ".join(object_clauses)
            if joined_object_clauses and not joined_object_clauses.endswith("."):
                joined_object_clauses += "."
            result += " " + joined_object_clauses

        return result

    def _generate_location_description(self, group_of_objects: List[Dict], count: int,
                                     image_width: Optional[int], image_height: Optional[int],
                                     region_analyzer: Optional[Any] = None) -> str:
        """
        生成位置描述

        Args:
            group_of_objects: 物件組
            count: 物件數量
            image_width: 圖像寬度
            image_height: 圖像高度
            region_analyzer: 區域分析器實例

        Returns:
            str: 位置描述
        """
        if count == 1:
            if self.spatial_handler:
                spatial_desc = self.spatial_handler.generate_spatial_description(
                    group_of_objects[0], image_width, image_height, region_analyzer
                )
            else:
                spatial_desc = self._get_spatial_description_phrase(group_of_objects[0].get("region", ""))

            if spatial_desc:
                return f"is {spatial_desc}"
            else:
                distinct_regions = sorted(list(set(obj.get("region", "") for obj in group_of_objects if obj.get("region"))))
                valid_regions = [r for r in distinct_regions if r and r != "unknown" and r.strip()]
                if not valid_regions:
                    return "is positioned in the scene"
                elif len(valid_regions) == 1:
                    spatial_desc = self._get_spatial_description_phrase(valid_regions[0])
                    return f"is primarily {spatial_desc}" if spatial_desc else "is positioned in the scene"
                elif len(valid_regions) == 2:
                    clean_region1 = valid_regions[0].replace('_', ' ')
                    clean_region2 = valid_regions[1].replace('_', ' ')
                    return f"is mainly across the {clean_region1} and {clean_region2} areas"
                else:
                    return "is distributed in various parts of the scene"
        else:
            distinct_regions = sorted(list(set(obj.get("region", "") for obj in group_of_objects if obj.get("region"))))
            valid_regions = [r for r in distinct_regions if r and r != "unknown" and r.strip()]
            if not valid_regions:
                return "are visible in the scene"
            elif len(valid_regions) == 1:
                clean_region = valid_regions[0].replace('_', ' ')
                return f"are primarily in the {clean_region} area"
            elif len(valid_regions) == 2:
                clean_region1 = valid_regions[0].replace('_', ' ')
                clean_region2 = valid_regions[1].replace('_', ' ')
                return f"are mainly across the {clean_region1} and {clean_region2} areas"
            else:
                return "are distributed in various parts of the scene"

    def _get_spatial_description_phrase(self, region: str) -> str:
        """
        獲取空間描述短語的備用方法

        Args:
            region: 區域字符串

        Returns:
            str: 空間描述短語
        """
        if not region or region == "unknown":
            return ""

        clean_region = region.replace('_', ' ').strip().lower()

        region_map = {
            "top left": "in the upper left area",
            "top center": "in the upper area",
            "top right": "in the upper right area",
            "middle left": "on the left side",
            "middle center": "in the center",
            "center": "in the center",
            "middle right": "on the right side",
            "bottom left": "in the lower left area",
            "bottom center": "in the lower area",
            "bottom right": "in the lower right area"
        }

        return region_map.get(clean_region, "")

    def _normalize_object_class_name(self, class_name: str) -> str:
        """
        標準化物件類別名稱

        Args:
            class_name: 原始類別名稱

        Returns:
            str: 標準化後的類別名稱
        """
        if self.text_optimizer:
            return self.text_optimizer.normalize_object_class_name(class_name)
        else:
            # 備用標準化邏輯
            if not class_name or not isinstance(class_name, str):
                return "object"

            # 簡單的標準化處理
            normalized = class_name.replace('_', ' ').strip().lower()
            return normalized

    def _format_object_count_description(self, class_name: str, count: int,
                                       scene_type: Optional[str] = None,
                                       detected_objects: Optional[List[Dict]] = None,
                                       avg_confidence: float = 0.0) -> str:
        """
        格式化物件數量描述

        Args:
            class_name: 標準化後的類別名稱
            count: 物件數量
            scene_type: 場景類型
            detected_objects: 該類型的所有檢測物件
            avg_confidence: 平均檢測置信度

        Returns:
            str: 完整的格式化數量描述
        """
        if self.text_optimizer:
            return self.text_optimizer.format_object_count_description(
                class_name, count, scene_type, detected_objects, avg_confidence
            )
        else:
            # 備用格式化邏輯
            if count <= 0:
                return ""
            elif count == 1:
                article = "an" if class_name[0].lower() in 'aeiou' else "a"
                return f"{article} {class_name}"
            else:
                # 簡單的複數處理
                plural_form = class_name + "s" if not class_name.endswith("s") else class_name

                number_words = {
                    2: "two", 3: "three", 4: "four", 5: "five", 6: "six",
                    7: "seven", 8: "eight", 9: "nine", 10: "ten",
                    11: "eleven", 12: "twelve"
                }

                if count in number_words:
                    return f"{number_words[count]} {plural_form}"
                elif count <= 20:
                    return f"several {plural_form}"
                else:
                    return f"numerous {plural_form}"