Spaces:
Running
on
Zero
Running
on
Zero
import logging | |
import traceback | |
import numpy as np | |
from typing import Dict, List, Any, Optional | |
logger = logging.getLogger(__name__) | |
class SceneZoneIdentifier: | |
""" | |
負責不同場景類型的區域識別邏輯 | |
專注於根據場景類型執行相應的功能區域識別策略 | |
""" | |
def __init__(self): | |
"""初始化場景區域辨識器""" | |
try: | |
logger.info("SceneZoneIdentifier initialized successfully") | |
except Exception as e: | |
logger.error(f"Failed to initialize SceneZoneIdentifier: {str(e)}") | |
logger.error(traceback.format_exc()) | |
raise | |
def identify_indoor_zones(self, category_regions: Dict, detected_objects: List[Dict], scene_type: str) -> Dict: | |
""" | |
平衡化的室內功能區域識別並標準化命名 | |
採用通用的物件關聯性分析,避免只針對特定場景 | |
Args: | |
category_regions: 按類別和區域分組的物件字典 | |
detected_objects: 檢測到的物件列表 | |
scene_type: 場景類型 | |
Returns: | |
識別出的室內功能區域字典,使用描述性鍵名 | |
""" | |
try: | |
zones = {} | |
# 主要功能區域(基於物件關聯性而非場景類型) | |
primary_zone = self._identify_primary_functional_area(detected_objects) | |
if primary_zone: | |
# 基於區域內容生成描述性鍵名 | |
descriptive_key = self._generate_descriptive_zone_key_from_data(primary_zone, "primary") | |
zones[descriptive_key] = primary_zone | |
# 只有明確證據且物件數量足夠時創建次要功能區域 | |
if len(zones) >= 1 and len(detected_objects) >= 6: | |
secondary_zone = self._identify_secondary_functional_area(detected_objects, zones) | |
if secondary_zone: | |
# 基於區域內容生成描述性鍵名 | |
descriptive_key = self._generate_descriptive_zone_key_from_data(secondary_zone, "secondary") | |
zones[descriptive_key] = secondary_zone | |
logger.info(f"Identified {len(zones)} indoor zones for scene type '{scene_type}'") | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying indoor zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _generate_descriptive_zone_key_from_data(self, zone_data: Dict, priority_level: str) -> str: | |
""" | |
基於區域數據生成描述性鍵名 | |
Args: | |
zone_data: 區域數據字典 | |
priority_level: 優先級別(primary/secondary) | |
Returns: | |
str: 描述性區域鍵名 | |
""" | |
try: | |
objects = zone_data.get("objects", []) | |
region = zone_data.get("region", "") | |
description = zone_data.get("description", "") | |
# 基於物件內容確定功能類型 | |
if any("dining" in obj.lower() or "table" in obj.lower() for obj in objects): | |
base_name = "dining area" | |
elif any("chair" in obj.lower() or "sofa" in obj.lower() for obj in objects): | |
base_name = "seating area" | |
elif any("bed" in obj.lower() for obj in objects): | |
base_name = "sleeping area" | |
elif any("laptop" in obj.lower() or "keyboard" in obj.lower() for obj in objects): | |
base_name = "workspace area" | |
elif any("plant" in obj.lower() or "vase" in obj.lower() for obj in objects): | |
base_name = "decorative area" | |
elif any("refrigerator" in obj.lower() or "microwave" in obj.lower() for obj in objects): | |
base_name = "kitchen area" | |
else: | |
# 基於描述內容推斷 | |
if "dining" in description.lower(): | |
base_name = "dining area" | |
elif "seating" in description.lower() or "relaxation" in description.lower(): | |
base_name = "seating area" | |
elif "work" in description.lower(): | |
base_name = "workspace area" | |
elif "decorative" in description.lower(): | |
base_name = "decorative area" | |
else: | |
base_name = "functional area" | |
# 為次要區域添加位置標識以區分 | |
if priority_level == "secondary" and region: | |
spatial_context = self._get_spatial_context_description(region) | |
if spatial_context: | |
return f"{spatial_context} {base_name}" | |
return base_name | |
except Exception as e: | |
logger.warning(f"Error generating descriptive zone key: {str(e)}") | |
return "activity area" | |
def _get_spatial_context_description(self, region: str) -> str: | |
""" | |
獲取空間上下文描述 | |
Args: | |
region: 區域位置標識 | |
Returns: | |
str: 空間上下文描述 | |
""" | |
try: | |
spatial_mapping = { | |
"top_left": "upper left", | |
"top_center": "upper", | |
"top_right": "upper right", | |
"middle_left": "left side", | |
"middle_center": "central", | |
"middle_right": "right side", | |
"bottom_left": "lower left", | |
"bottom_center": "lower", | |
"bottom_right": "lower right" | |
} | |
return spatial_mapping.get(region, "") | |
except Exception as e: | |
logger.warning(f"Error getting spatial context for region '{region}': {str(e)}") | |
return "" | |
def identify_outdoor_general_zones(self, category_regions: Dict, detected_objects: List[Dict], scene_type: str) -> Dict: | |
""" | |
識別一般戶外場景的功能區域 | |
Args: | |
category_regions: 按類別和區域分組的物件字典 | |
detected_objects: 檢測到的物件列表 | |
scene_type: 特定戶外場景類型 | |
Returns: | |
戶外功能區域字典 | |
""" | |
try: | |
zones = {} | |
# 識別行人區域 | |
people_objs = [obj for obj in detected_objects if obj["class_id"] == 0] | |
if people_objs: | |
people_regions = {} | |
for obj in people_objs: | |
region = obj["region"] | |
if region not in people_regions: | |
people_regions[region] = [] | |
people_regions[region].append(obj) | |
if people_regions: | |
# 找到主要的行人活動區域 | |
main_people_regions = sorted(people_regions.items(), | |
key=lambda x: len(x[1]), | |
reverse=True)[:2] # 取前2個區域 | |
for idx, (region, objs) in enumerate(main_people_regions): | |
if len(objs) > 0: | |
# 生成基於位置的描述性鍵名 | |
spatial_desc = self._get_directional_description(region) | |
if spatial_desc and spatial_desc != "central": | |
zone_key = f"{spatial_desc} pedestrian area" | |
else: | |
zone_key = "main pedestrian area" if idx == 0 else "secondary pedestrian area" | |
zones[zone_key] = { | |
"region": region, | |
"objects": ["person"] * len(objs), | |
"description": f"Pedestrian area with {len(objs)} {'people' if len(objs) > 1 else 'person'}" | |
} | |
# 識別車輛區域,適用於街道和停車場 | |
vehicle_objs = [obj for obj in detected_objects if obj["class_id"] in [1, 2, 3, 5, 6, 7]] | |
if vehicle_objs: | |
vehicle_regions = {} | |
for obj in vehicle_objs: | |
region = obj["region"] | |
if region not in vehicle_regions: | |
vehicle_regions[region] = [] | |
vehicle_regions[region].append(obj) | |
if vehicle_regions: | |
main_vehicle_region = max(vehicle_regions.items(), | |
key=lambda x: len(x[1]), | |
default=(None, [])) | |
if main_vehicle_region[0] is not None: | |
vehicle_types = [obj["class_name"] for obj in main_vehicle_region[1]] | |
zones["vehicle_zone"] = { | |
"region": main_vehicle_region[0], | |
"objects": vehicle_types, | |
"description": f"Traffic area with {', '.join(list(set(vehicle_types))[:3])}" | |
} | |
# 針對公園區域的特殊處理 | |
if scene_type == "park_area": | |
zones.update(self._identify_park_recreational_zones(detected_objects)) | |
# 針對停車場的特殊處理 | |
if scene_type == "parking_lot": | |
zones.update(self._identify_parking_zones(detected_objects)) | |
logger.info(f"Identified {len(zones)} outdoor zones for scene type '{scene_type}'") | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying outdoor general zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def identify_intersection_zones(self, category_regions: Dict, detected_objects: List[Dict], viewpoint: str) -> Dict: | |
""" | |
辨識城市十字路口的功能區域,無論是否有行人,只要偵測到紅綠燈就一定顯示 Traffic Control Area; | |
若有行人,則額外建立 Crossing Zone 並把行人 + 同 region 的紅綠燈歸在一起。 | |
Args: | |
category_regions: 按類別和 region 分組的物件字典 | |
detected_objects: YOLO 檢測到的所有物件列表 | |
viewpoint: 偵測到的視角字串 | |
Returns: | |
zones: 最終的十字路口功能區域字典 | |
""" | |
try: | |
zones = {} | |
# 1. 按 class_id 分出行人、車輛、紅綠燈 | |
pedestrian_objs = [obj for obj in detected_objects if obj["class_id"] == 0] | |
vehicle_objs = [obj for obj in detected_objects if obj["class_id"] in [1, 2, 3, 5, 7]] | |
traffic_light_objs = [obj for obj in detected_objects if obj["class_id"] == 9] | |
# 2. Step A: 無條件建立 Traffic Control Area | |
# 把每個 region 下的紅綠燈都先分群,生成對應 zone,確保「只要偵測到紅綠燈就一定顯示」 | |
signal_regions_all = {} | |
for t in traffic_light_objs: | |
region = t["region"] | |
signal_regions_all.setdefault(region, []).append(t) | |
for idx, (region, signals) in enumerate(signal_regions_all.items()): | |
# 先決定 zone_key (依 direction 或 primary/auxiliary) | |
direction = self._get_directional_description(region) | |
if direction and direction != "central": | |
zone_key = f"{direction} traffic control area" | |
else: | |
zone_key = "primary traffic control area" if idx == 0 else "auxiliary traffic control area" | |
# 確保命名不衝突 | |
if zone_key in zones: | |
suffix = 1 | |
new_key = f"{zone_key} ({suffix})" | |
while new_key in zones: | |
suffix += 1 | |
new_key = f"{zone_key} ({suffix})" | |
zone_key = new_key | |
zones[zone_key] = { | |
"region": region, | |
"objects": ["traffic light"] * len(signals), | |
"description": f"Traffic control area with {len(signals)} traffic lights in {region}" | |
} | |
# (用於後面計算 Crossing 使用掉的 traffic light) | |
used_tl_count_per_region = dict.fromkeys(signal_regions_all.keys(), 0) | |
# 3. Step B: 如果有行人,就建立 Crossing Zone,並移除已被打包的紅綠燈 | |
if pedestrian_objs: | |
# 先呼叫 _analyze_crossing_patterns,讓它回傳「行人 + 同 region 的紅綠燈」區 | |
crossing_zones = self._analyze_crossing_patterns(pedestrian_objs, traffic_light_objs) | |
# 把 Crossing Zone 加到最終 zones,並同時記錄已使用掉的紅綠燈數量 | |
for zone_key, zone_info in crossing_zones.items(): | |
region = zone_info.get("region", "") | |
obj_list = zone_info.get("objects", []) | |
# 如果該 zone_info["objects"] 裡含有紅綠燈,就累加到 used_tl_count_per_region | |
count_in_zone = obj_list.count("traffic light") | |
if count_in_zone > 0: | |
used_tl_count_per_region[region] = used_tl_count_per_region.get(region, 0) + count_in_zone | |
# 加入最終結果 | |
# 如果 key 重複,也可以在此加上 index,或直接覆蓋 | |
if zone_key in zones: | |
suffix = 1 | |
new_key = f"{zone_key} ({suffix})" | |
while new_key in zones: | |
suffix += 1 | |
new_key = f"{zone_key} ({suffix})" | |
zone_key = new_key | |
zones[zone_key] = { | |
"region": region, | |
"objects": obj_list, | |
"description": zone_info.get("description", "") | |
} | |
# 4. Step C: 計算並顯示 debug 資訊 (Total / Used / Remaining) | |
for region, signals in signal_regions_all.items(): | |
total = len(signals) | |
used = used_tl_count_per_region.get(region, 0) | |
remaining = total - used | |
# print(f"[DEBUG] Region '{region}': Total TL = {total}, Used in crossing = {used}, Remaining = {remaining}") | |
# 5. Step D: 分析車輛交通區域(Vehicle Zones) | |
if vehicle_objs: | |
traffic_zones = self._analyze_traffic_zones(vehicle_objs) | |
# _analyze_traffic_zones 內部已用英文 debug,直接更新 | |
for zone_key, zone_info in traffic_zones.items(): | |
if zone_key in zones: | |
suffix = 1 | |
new_key = f"{zone_key} ({suffix})" | |
while new_key in zones: | |
suffix += 1 | |
new_key = f"{zone_key} ({suffix})" | |
zone_key = new_key | |
zones[zone_key] = zone_info | |
logger.info(f"Identified {len(zones)} intersection zones") | |
return zones | |
except Exception as e: | |
logger.error(f"Error in identify_intersection_zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def identify_aerial_view_zones(self, category_regions: Dict, detected_objects: List[Dict], scene_type: str) -> Dict: | |
""" | |
辨識空中視角場景的功能區域 | |
專注於模式和流動而非特定區域 | |
Args: | |
category_regions: 按類別和區域分組的物件字典 | |
detected_objects: 檢測到的物件列表 | |
scene_type: 特定場景類型 | |
Returns: | |
空中視角功能區域字典 | |
""" | |
try: | |
zones = {} | |
# 識別行人模式 | |
people_objs = [obj for obj in detected_objects if obj["class_id"] == 0] | |
if people_objs: | |
# 將位置轉換為數組進行模式分析 | |
positions = np.array([obj["normalized_center"] for obj in people_objs]) | |
if len(positions) >= 3: | |
# 計算分布指標 | |
x_coords = positions[:, 0] | |
y_coords = positions[:, 1] | |
x_mean = np.mean(x_coords) | |
y_mean = np.mean(y_coords) | |
x_std = np.std(x_coords) | |
y_std = np.std(y_coords) | |
# 判斷人群是否組織成線性模式 | |
if x_std < 0.1 or y_std < 0.1: | |
# 沿一個軸的線性分布 | |
pattern_direction = "vertical" if x_std < y_std else "horizontal" | |
zones["pedestrian_pattern"] = { | |
"region": "central", | |
"objects": ["person"] * len(people_objs), | |
"description": f"Aerial view shows a {pattern_direction} pedestrian movement pattern" | |
} | |
else: | |
# 更分散的模式 | |
zones["pedestrian_distribution"] = { | |
"region": "wide", | |
"objects": ["person"] * len(people_objs), | |
"description": f"Aerial view shows pedestrians distributed across the area" | |
} | |
# 識別車輛模式進行交通分析 | |
vehicle_objs = [obj for obj in detected_objects if obj["class_id"] in [1, 2, 3, 5, 6, 7]] | |
if vehicle_objs: | |
zones.update(self._analyze_aerial_traffic_patterns(vehicle_objs)) | |
# 針對十字路口特定空中視角的處理 | |
if "intersection" in scene_type: | |
zones.update(self._identify_aerial_intersection_features(detected_objects)) | |
# 針對廣場空中視角的處理 | |
if "plaza" in scene_type: | |
zones.update(self._identify_aerial_plaza_features(people_objs)) | |
logger.info(f"Identified {len(zones)} aerial view zones") | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying aerial view zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def identify_asian_cultural_zones(self, category_regions: Dict, detected_objects: List[Dict], scene_type: str) -> Dict: | |
""" | |
辨識有亞洲文化背景的場景功能區域 | |
Args: | |
category_regions: 按類別和區域分組的物件字典 | |
detected_objects: 檢測到的物件列表 | |
scene_type: 特定場景類型 | |
Returns: | |
亞洲文化功能區域字典 | |
""" | |
try: | |
zones = {} | |
# 識別店面區域 | |
# 由於店面不能直接檢測,從情境推斷 | |
# 例如,尋找有標誌、行人和小物件的區域 | |
storefront_regions = {} | |
for obj in detected_objects: | |
if obj["class_id"] == 0: # Person | |
region = obj["region"] | |
if region not in storefront_regions: | |
storefront_regions[region] = [] | |
storefront_regions[region].append(obj) | |
# 將人最多的區域作為店面區域 | |
if storefront_regions: | |
main_storefront_regions = sorted(storefront_regions.items(), | |
key=lambda x: len(x[1]), | |
reverse=True)[:2] # 前2個區域 | |
for idx, (region, objs) in enumerate(main_storefront_regions): | |
# 生成基於位置的描述性鍵名 | |
spatial_desc = self._get_directional_description(region) | |
if spatial_desc and spatial_desc != "central": | |
zone_key = f"{spatial_desc} commercial area" | |
else: | |
zone_key = "main commercial area" if idx == 0 else "secondary commercial area" | |
zones[zone_key] = { | |
"region": region, | |
"objects": [obj["class_name"] for obj in objs], | |
"description": f"Asian commercial storefront with pedestrian activity" | |
} | |
# 辨識行人通道 | |
zones.update(self._identify_asian_pedestrian_pathway(detected_objects)) | |
# 辨識攤販區域(小攤/商店 - 從情境推斷) | |
zones.update(self._identify_vendor_zones(detected_objects)) | |
# 針對夜市的特殊處理 | |
if scene_type == "asian_night_market": | |
zones["food_stall_zone"] = { | |
"region": "middle_center", | |
"objects": ["inferred food stalls"], | |
"description": "Food stall area typical of Asian night markets" | |
} | |
logger.info(f"Identified {len(zones)} Asian cultural zones") | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying Asian cultural zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def identify_upscale_dining_zones(self, category_regions: Dict, detected_objects: List[Dict]) -> Dict: | |
""" | |
辨識高級餐飲設置的功能區域 | |
Args: | |
category_regions: 按類別和區域分組的物件字典 | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
高級餐飲功能區域字典 | |
""" | |
try: | |
zones = {} | |
# 辨識餐桌區域 | |
dining_items = [] | |
dining_regions = {} | |
for obj in detected_objects: | |
if obj["class_id"] in [40, 41, 42, 43, 44, 45, 60]: # Wine glass, cup, fork, knife, spoon, bowl, table | |
region = obj["region"] | |
if region not in dining_regions: | |
dining_regions[region] = [] | |
dining_regions[region].append(obj) | |
dining_items.append(obj["class_name"]) | |
if dining_items: | |
main_dining_region = max(dining_regions.items(), | |
key=lambda x: len(x[1]), | |
default=(None, [])) | |
if main_dining_region[0] is not None: | |
zones["formal_dining_zone"] = { | |
"region": main_dining_region[0], | |
"objects": list(set(dining_items)), | |
"description": f"Formal dining area with {', '.join(list(set(dining_items))[:3])}" | |
} | |
# 識別裝飾區域,增強檢測 | |
zones.update(self._identify_upscale_decorative_zones(detected_objects)) | |
# 識別座位安排區域 | |
zones.update(self._identify_dining_seating_zones(detected_objects)) | |
# 識別服務區域(如果與餐飲區域不同) | |
zones.update(self._identify_serving_zones(detected_objects, zones)) | |
logger.info(f"Identified {len(zones)} upscale dining zones") | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying upscale dining zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def identify_financial_district_zones(self, category_regions: Dict, detected_objects: List[Dict]) -> Dict: | |
""" | |
金融區場景的功能區域 | |
Args: | |
category_regions: 按類別和區域分組的物件字典 | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
金融區功能區域字典 | |
""" | |
try: | |
zones = {} | |
# 識別交通區域 | |
traffic_items = [] | |
traffic_regions = {} | |
for obj in detected_objects: | |
if obj["class_id"] in [1, 2, 3, 5, 6, 7, 9]: # 各種車輛和交通燈 | |
region = obj["region"] | |
if region not in traffic_regions: | |
traffic_regions[region] = [] | |
traffic_regions[region].append(obj) | |
traffic_items.append(obj["class_name"]) | |
if traffic_items: | |
main_traffic_region = max(traffic_regions.items(), | |
key=lambda x: len(x[1]), | |
default=(None, [])) | |
if main_traffic_region[0] is not None: | |
zones["traffic_zone"] = { | |
"region": main_traffic_region[0], | |
"objects": list(set(traffic_items)), | |
"description": f"Urban traffic area with {', '.join(list(set(traffic_items))[:3])}" | |
} | |
# 側邊建築區域(從場景情境推斷) | |
zones.update(self._identify_building_zones(detected_objects)) | |
# 行人區域 | |
zones.update(self._identify_financial_pedestrian_zones(detected_objects)) | |
logger.info(f"Identified {len(zones)} financial district zones") | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying financial district zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def identify_landmark_zones(self, landmark_objects: List[Dict]) -> Dict: | |
""" | |
辨識與地標相關的功能區域 | |
Args: | |
landmark_objects: 被辨識為地標的物體列表 | |
Returns: | |
地標相關的功能區域字典 | |
""" | |
try: | |
landmark_zones = {} | |
# 如果沒有任何地標,就直接回空字典 | |
if not landmark_objects: | |
logger.warning("No landmark objects provided to identify_landmark_zones") | |
return landmark_zones | |
# 只取第一個地標來示範:至少產生一個地標 | |
landmark = landmark_objects[0] | |
# 確保傳入的 landmark 是 dict | |
if not isinstance(landmark, dict): | |
logger.warning("First landmark object is not a dict") | |
return landmark_zones | |
# 從 landmark dict 拿出必要欄位 | |
landmark_id = landmark.get("landmark_id", "unknown_landmark") | |
landmark_name = landmark.get("class_name", "Landmark") | |
landmark_type = landmark.get("landmark_type", "architectural") | |
landmark_region = landmark.get("region", "middle_center") | |
# 如果 location 沒提供,就給預設 "this area" | |
location = landmark.get("location") | |
if not location: | |
location = "this area" | |
# 為地標創建主要觀景區 | |
zone_id = f"{landmark_name.lower().replace(' ', '_')}_viewing_area" | |
zone_name = f"{landmark_name} Viewing Area" | |
# 根據地標類型調整描述,並確保帶入地點 | |
if landmark_type == "natural": | |
zone_description = ( | |
f"Scenic viewpoint for observing {landmark_name}, " | |
f"a notable natural landmark in {location}." | |
) | |
primary_function = "Nature observation and photography" | |
elif landmark_type == "monument": | |
zone_description = ( | |
f"Viewing area around {landmark_name}, " | |
f"a significant monument in {location}." | |
) | |
primary_function = "Historical appreciation and cultural tourism" | |
else: # architectural | |
zone_description = ( | |
f"Area centered around {landmark_name}, " | |
f"where visitors can observe and appreciate this iconic structure in {location}." | |
) | |
primary_function = "Architectural tourism and photography" | |
# 確定與地標相關的物體(如果被偵測到) | |
related_objects = [] | |
for o in landmark_objects: | |
cn = o.get("class_name", "").lower() | |
if cn in ["person", "camera", "cell phone", "backpack"]: | |
related_objects.append(cn) | |
# 建立地標功能區 | |
landmark_zones[zone_id] = { | |
"name": zone_name, | |
"description": zone_description, | |
"objects": ["landmark"] + related_objects, | |
"region": landmark_region, | |
"primary_function": primary_function | |
} | |
# 創建相關輔助功能區,如攝影區、紀念品販賣區 | |
auxiliary_zones = self._create_landmark_auxiliary_zones(landmark, 0) | |
if auxiliary_zones: | |
landmark_zones.update(auxiliary_zones) | |
logger.info(f"Identified {len(landmark_zones)} landmark zones") | |
return landmark_zones | |
except Exception as e: | |
logger.error(f"Error in identify_landmark_zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_primary_functional_area(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
識別主要功能區域,基於最強的物件關聯性組合 | |
採用通用邏輯處理各種室內場景 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
主要功能區域字典或None | |
""" | |
try: | |
# 用餐區域檢測(桌椅組合) | |
dining_area = self._detect_functional_combination( | |
detected_objects, | |
primary_objects=[60], # dining table | |
supporting_objects=[56, 40, 41, 42, 43], # chair, wine glass, cup, fork, knife | |
min_supporting=2, | |
description_template="Dining area with table and seating arrangement" | |
) | |
if dining_area: | |
return dining_area | |
# 休息區域檢測(沙發電視組合或床) | |
seating_area = self._detect_functional_combination( | |
detected_objects, | |
primary_objects=[57, 59], # sofa, bed | |
supporting_objects=[62, 58, 56], # tv, potted plant, chair | |
min_supporting=1, | |
description_template="Seating and relaxation area" | |
) | |
if seating_area: | |
return seating_area | |
# 工作區域檢測(電子設備與家具組合) | |
work_area = self._detect_functional_combination( | |
detected_objects, | |
primary_objects=[63, 66], # laptop, keyboard | |
supporting_objects=[60, 56, 64], # dining table, chair, mouse | |
min_supporting=2, | |
description_template="Workspace area with electronics and furniture" | |
) | |
if work_area: | |
return work_area | |
return None | |
except Exception as e: | |
logger.error(f"Error identifying primary functional area: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return None | |
def _identify_secondary_functional_area(self, detected_objects: List[Dict], existing_zones: Dict) -> Dict: | |
""" | |
識別次要功能區域,避免與主要區域重疊 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
existing_zones: 已存在的功能區域 | |
Returns: | |
次要功能區域字典或None | |
""" | |
try: | |
# 獲取已使用的區域 | |
used_regions = set(zone.get("region") for zone in existing_zones.values()) | |
# 裝飾區域檢測(植物集中區域) | |
decorative_area = self._detect_functional_combination( | |
detected_objects, | |
primary_objects=[58], # potted plant | |
supporting_objects=[75], # vase | |
min_supporting=0, | |
min_primary=3, # 至少需要3個植物 | |
description_template="Decorative area with plants and ornamental items", | |
exclude_regions=used_regions | |
) | |
if decorative_area: | |
return decorative_area | |
# 儲存區域檢測(廚房電器組合) | |
storage_area = self._detect_functional_combination( | |
detected_objects, | |
primary_objects=[72, 68, 69], # refrigerator, microwave, oven | |
supporting_objects=[71], # sink | |
min_supporting=0, | |
min_primary=2, | |
description_template="Kitchen appliance and storage area", | |
exclude_regions=used_regions | |
) | |
if storage_area: | |
return storage_area | |
return None | |
except Exception as e: | |
logger.error(f"Error identifying secondary functional area: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return None | |
def _detect_functional_combination(self, detected_objects: List[Dict], primary_objects: List[int], | |
supporting_objects: List[int], min_supporting: int, | |
description_template: str, min_primary: int = 1, | |
exclude_regions: set = None) -> Dict: | |
""" | |
通用的功能組合檢測方法 | |
基於主要物件和支持物件的組合判斷功能區域 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
primary_objects: 主要物件的class_id列表 | |
supporting_objects: 支持物件的class_id列表 | |
min_supporting: 最少需要的支持物件數量 | |
description_template: 描述模板 | |
min_primary: 最少需要的主要物件數量 | |
exclude_regions: 需要排除的區域集合 | |
Returns: | |
功能區域資訊字典,如果不符合條件則返回None | |
""" | |
try: | |
if exclude_regions is None: | |
exclude_regions = set() | |
# 收集主要物件 | |
primary_objs = [obj for obj in detected_objects | |
if obj.get("class_id") in primary_objects and obj.get("confidence", 0) >= 0.4] | |
# 收集支持物件 | |
supporting_objs = [obj for obj in detected_objects | |
if obj.get("class_id") in supporting_objects and obj.get("confidence", 0) >= 0.4] | |
# 檢查是否滿足最少數量要求 | |
if len(primary_objs) < min_primary or len(supporting_objs) < min_supporting: | |
return None | |
# 按區域組織物件 | |
region_combinations = {} | |
all_relevant_objs = primary_objs + supporting_objs | |
for obj in all_relevant_objs: | |
region = obj.get("region") | |
# 排除指定區域 | |
if region in exclude_regions: | |
continue | |
if region not in region_combinations: | |
region_combinations[region] = {"primary": [], "supporting": [], "all": []} | |
region_combinations[region]["all"].append(obj) | |
if obj.get("class_id") in primary_objects: | |
region_combinations[region]["primary"].append(obj) | |
else: | |
region_combinations[region]["supporting"].append(obj) | |
# 找到最佳區域組合 | |
best_region = None | |
best_score = 0 | |
for region, objs in region_combinations.items(): | |
# 計算該區域的評分 | |
primary_count = len(objs["primary"]) | |
supporting_count = len(objs["supporting"]) | |
# 必須滿足最低要求 | |
if primary_count < min_primary or supporting_count < min_supporting: | |
continue | |
# 計算組合評分(主要物件權重較高) | |
score = primary_count * 2 + supporting_count | |
if score > best_score: | |
best_score = score | |
best_region = region | |
if best_region is None: | |
return None | |
best_combination = region_combinations[best_region] | |
all_objects = [obj["class_name"] for obj in best_combination["all"]] | |
return { | |
"region": best_region, | |
"objects": all_objects, | |
"description": description_template | |
} | |
except Exception as e: | |
logger.error(f"Error detecting functional combination: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return None | |
def _analyze_crossing_patterns(self, pedestrians: List[Dict], traffic_lights: List[Dict]) -> Dict: | |
""" | |
Analyze pedestrian crossing patterns to identify crossing zones. | |
若同一 region 中同時有行人與紅綠燈,則將兩者都放入該區域的 objects。 | |
Args: | |
pedestrians: 行人物件列表(每個 obj 應包含 'class_id', 'region', 'confidence' 等) | |
traffic_lights: 紅綠燈物件列表(每個 obj 應包含 'class_id', 'region', 'confidence' 等) | |
Returns: | |
crossing_zones: 字典,key 為 zone 名稱,value 包含 'region', 'objects', 'description' | |
""" | |
try: | |
crossing_zones = {} | |
# 如果沒有任何行人,就不辨識任何 crossing zone | |
if not pedestrians: | |
return crossing_zones | |
# (1) 按照 region 分組行人 | |
pedestrian_regions = {} | |
for p in pedestrians: | |
region = p["region"] | |
pedestrian_regions.setdefault(region, []).append(p) | |
# (2) 針對每個 region,看是否同時有紅綠燈 | |
# 建立一個 mapping: region -> { "pedestrians": [...], "traffic_lights": [...] } | |
combined_regions = {} | |
for region, peds in pedestrian_regions.items(): | |
# 取得該 region 下所有紅綠燈 | |
tls_in_region = [t for t in traffic_lights if t["region"] == region] | |
combined_regions[region] = { | |
"pedestrians": peds, | |
"traffic_lights": tls_in_region | |
} | |
# (3) 按照行人數量排序,找出前兩個需要建立 crossing zone 的 region | |
sorted_regions = sorted( | |
combined_regions.items(), | |
key=lambda x: len(x[1]["pedestrians"]), | |
reverse=True | |
) | |
# (4) 將前兩個 region 建立 Crossing Zone,objects 同時包含行人與紅綠燈 | |
for idx, (region, group) in enumerate(sorted_regions[:2]): | |
peds = group["pedestrians"] | |
tls = group["traffic_lights"] | |
has_nearby_signals = len(tls) > 0 | |
# 生成 zone_name(基於 region 方向 + idx 決定主/次 crossing) | |
direction = self._get_directional_description(region) | |
if direction and direction != "central": | |
zone_name = f"{direction} crossing area" | |
else: | |
zone_name = "main crossing area" if idx == 0 else "secondary crossing area" | |
# 組合 description | |
description = f"Pedestrian crossing area with {len(peds)} " | |
description += "person" if len(peds) == 1 else "people" | |
if direction: | |
description += f" in {direction} direction" | |
if has_nearby_signals: | |
description += " near traffic signals" | |
# ======= 將行人 + 同區紅綠燈一併放入 objects ======= | |
obj_list = ["pedestrian"] * len(peds) | |
if has_nearby_signals: | |
obj_list += ["traffic light"] * len(tls) | |
crossing_zones[zone_name] = { | |
"region": region, | |
"objects": obj_list, | |
"description": description | |
} | |
return crossing_zones | |
except Exception as e: | |
logger.error(f"Error in _analyze_crossing_patterns: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _analyze_traffic_zones(self, vehicles: List[Dict]) -> Dict: | |
""" | |
分析車輛分布以識別具有方向感知的交通區域 | |
Args: | |
vehicles: 車輛物件列表 | |
Returns: | |
識別出的交通區域字典 | |
""" | |
try: | |
traffic_zones = {} | |
if not vehicles: | |
return traffic_zones | |
# 按區域分組車輛 | |
vehicle_regions = {} | |
for v in vehicles: | |
region = v["region"] | |
if region not in vehicle_regions: | |
vehicle_regions[region] = [] | |
vehicle_regions[region].append(v) | |
# 為有車輛的區域創建交通區域 | |
main_traffic_region = max(vehicle_regions.items(), key=lambda x: len(x[1]), default=(None, [])) | |
if main_traffic_region[0] is not None: | |
region = main_traffic_region[0] | |
vehicles_in_region = main_traffic_region[1] | |
# 獲取車輛類型列表用於描述 | |
vehicle_types = [v["class_name"] for v in vehicles_in_region] | |
unique_types = list(set(vehicle_types)) | |
# 獲取方向描述 | |
direction = self._get_directional_description(region) | |
# 創建描述性區域 | |
traffic_zones["vehicle_zone"] = { | |
"region": region, | |
"objects": vehicle_types, | |
"description": f"Vehicle traffic area with {', '.join(unique_types[:3])}" + | |
(f" in {direction} area" if direction else "") | |
} | |
# 如果車輛分布在多個區域,創建次要區域 | |
if len(vehicle_regions) > 1: | |
# 獲取第二大車輛聚集區域 | |
sorted_regions = sorted(vehicle_regions.items(), key=lambda x: len(x[1]), reverse=True) | |
if len(sorted_regions) > 1: | |
second_region, second_vehicles = sorted_regions[1] | |
direction = self._get_directional_description(second_region) | |
vehicle_types = [v["class_name"] for v in second_vehicles] | |
unique_types = list(set(vehicle_types)) | |
traffic_zones["secondary_vehicle_zone"] = { | |
"region": second_region, | |
"objects": vehicle_types, | |
"description": f"Secondary traffic area with {', '.join(unique_types[:2])}" + | |
(f" in {direction} direction" if direction else "") | |
} | |
return traffic_zones | |
except Exception as e: | |
logger.error(f"Error analyzing traffic zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _get_directional_description(self, region: str) -> str: | |
""" | |
將區域名稱轉換為方位描述(東西南北) | |
Args: | |
region: 區域名稱 | |
Returns: | |
方位描述字串 | |
""" | |
try: | |
region_lower = region.lower() | |
if "top" in region_lower and "left" in region_lower: | |
return "northwest" | |
elif "top" in region_lower and "right" in region_lower: | |
return "northeast" | |
elif "bottom" in region_lower and "left" in region_lower: | |
return "southwest" | |
elif "bottom" in region_lower and "right" in region_lower: | |
return "southeast" | |
elif "top" in region_lower: | |
return "north" | |
elif "bottom" in region_lower: | |
return "south" | |
elif "left" in region_lower: | |
return "west" | |
elif "right" in region_lower: | |
return "east" | |
else: | |
return "central" | |
except Exception as e: | |
logger.error(f"Error getting directional description for region '{region}': {str(e)}") | |
return "central" | |
def _identify_park_recreational_zones(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
識別公園的休閒活動區域 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
休閒區域字典 | |
""" | |
try: | |
zones = {} | |
# 尋找休閒物件(運動球、風箏等) | |
rec_items = [] | |
rec_regions = {} | |
for obj in detected_objects: | |
if obj["class_id"] in [32, 33, 34, 35, 38]: # sports ball, kite, baseball bat, glove, tennis racket | |
region = obj["region"] | |
if region not in rec_regions: | |
rec_regions[region] = [] | |
rec_regions[region].append(obj) | |
rec_items.append(obj["class_name"]) | |
if rec_items: | |
main_rec_region = max(rec_regions.items(), | |
key=lambda x: len(x[1]), | |
default=(None, [])) | |
if main_rec_region[0] is not None: | |
zones["recreational_zone"] = { | |
"region": main_rec_region[0], | |
"objects": list(set(rec_items)), | |
"description": f"Recreational area with {', '.join(list(set(rec_items)))}" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying park recreational zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_parking_zones(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
停車場的停車區域 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
停車區域字典 | |
""" | |
try: | |
zones = {} | |
# 尋找停放的汽車 | |
car_objs = [obj for obj in detected_objects if obj["class_id"] == 2] # cars | |
if len(car_objs) >= 3: | |
# 檢查汽車是否按模式排列(簡化) | |
car_positions = [obj["normalized_center"] for obj in car_objs] | |
# 通過分析垂直位置檢查行模式 | |
y_coords = [pos[1] for pos in car_positions] | |
y_clusters = {} | |
# 簡化聚類 - 按相似y坐標分組汽車 | |
for i, y in enumerate(y_coords): | |
assigned = False | |
for cluster_y in y_clusters.keys(): | |
if abs(y - cluster_y) < 0.1: # 圖像高度的10%內 | |
y_clusters[cluster_y].append(i) | |
assigned = True | |
break | |
if not assigned: | |
y_clusters[y] = [i] | |
# 如果有行模式 | |
if max(len(indices) for indices in y_clusters.values()) >= 2: | |
zones["parking_row"] = { | |
"region": "central", | |
"objects": ["car"] * len(car_objs), | |
"description": f"Organized parking area with vehicles arranged in rows" | |
} | |
else: | |
zones["parking_area"] = { | |
"region": "wide", | |
"objects": ["car"] * len(car_objs), | |
"description": f"Parking area with {len(car_objs)} vehicles" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying parking zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _analyze_aerial_traffic_patterns(self, vehicle_objs: List[Dict]) -> Dict: | |
""" | |
分析空中視角的車輛交通模式 | |
Args: | |
vehicle_objs: 車輛物件列表 | |
Returns: | |
交通模式區域字典 | |
""" | |
try: | |
zones = {} | |
if not vehicle_objs: | |
return zones | |
# 將位置轉換為數組進行模式分析 | |
positions = np.array([obj["normalized_center"] for obj in vehicle_objs]) | |
if len(positions) >= 2: | |
# 計算分布指標 | |
x_coords = positions[:, 0] | |
y_coords = positions[:, 1] | |
x_mean = np.mean(x_coords) | |
y_mean = np.mean(y_coords) | |
x_std = np.std(x_coords) | |
y_std = np.std(y_coords) | |
# 判斷車輛是否組織成車道 | |
if x_std < y_std * 0.5: | |
# 車輛垂直對齊 - 表示南北交通 | |
zones["vertical_traffic_flow"] = { | |
"region": "central_vertical", | |
"objects": [obj["class_name"] for obj in vehicle_objs[:5]], | |
"description": "North-south traffic flow visible from aerial view" | |
} | |
elif y_std < x_std * 0.5: | |
# 車輛水平對齊 - 表示東西交通 | |
zones["horizontal_traffic_flow"] = { | |
"region": "central_horizontal", | |
"objects": [obj["class_name"] for obj in vehicle_objs[:5]], | |
"description": "East-west traffic flow visible from aerial view" | |
} | |
else: | |
# 車輛多方向 - 表示十字路口 | |
zones["intersection_traffic"] = { | |
"region": "central", | |
"objects": [obj["class_name"] for obj in vehicle_objs[:5]], | |
"description": "Multi-directional traffic at intersection visible from aerial view" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error analyzing aerial traffic patterns: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_aerial_intersection_features(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
空中視角十字路口特徵 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
十字路口特徵區域字典 | |
""" | |
try: | |
zones = {} | |
# 檢查交通信號 | |
traffic_light_objs = [obj for obj in detected_objects if obj["class_id"] == 9] | |
if traffic_light_objs: | |
zones["traffic_control_pattern"] = { | |
"region": "intersection", | |
"objects": ["traffic light"] * len(traffic_light_objs), | |
"description": f"Intersection traffic control with {len(traffic_light_objs)} signals visible from above" | |
} | |
# 人行道從空中視角的情境推斷 | |
zones["crossing_pattern"] = { | |
"region": "central", | |
"objects": ["inferred crosswalk"], | |
"description": "Crossing pattern visible from aerial perspective" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying aerial intersection features: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_aerial_plaza_features(self, people_objs: List[Dict]) -> Dict: | |
""" | |
識別空中視角廣場特徵 | |
Args: | |
people_objs: 行人物件列表 | |
Returns: | |
廣場特徵區域字典 | |
""" | |
try: | |
zones = {} | |
if people_objs: | |
# 檢查人群是否聚集在中央區域 | |
central_people = [obj for obj in people_objs | |
if "middle" in obj["region"]] | |
if central_people: | |
zones["central_gathering"] = { | |
"region": "middle_center", | |
"objects": ["person"] * len(central_people), | |
"description": f"Central plaza gathering area with {len(central_people)} people viewed from above" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying aerial plaza features: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_asian_pedestrian_pathway(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
亞洲文化場景中的行人通道 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
行人通道區域字典 | |
""" | |
try: | |
zones = {} | |
pathway_items = [] | |
pathway_regions = {} | |
# 提取人群用於通道分析 | |
people_objs = [obj for obj in detected_objects if obj["class_id"] == 0] | |
# 分析人群是否形成線形(商業街的特徵) | |
people_positions = [obj["normalized_center"] for obj in people_objs] | |
structured_path = False | |
path_direction = "meandering" | |
if len(people_positions) >= 3: | |
# 檢查人群是否沿相似y坐標排列(水平路徑) | |
y_coords = [pos[1] for pos in people_positions] | |
y_mean = sum(y_coords) / len(y_coords) | |
y_variance = sum((y - y_mean)**2 for y in y_coords) / len(y_coords) | |
horizontal_path = y_variance < 0.05 # 低變異表示水平對齊 | |
# 檢查人群是否沿相似x坐標排列(垂直路徑) | |
x_coords = [pos[0] for pos in people_positions] | |
x_mean = sum(x_coords) / len(x_coords) | |
x_variance = sum((x - x_mean)**2 for x in x_coords) / len(x_coords) | |
vertical_path = x_variance < 0.05 # 低變異表示垂直對齊 | |
structured_path = horizontal_path or vertical_path | |
path_direction = "horizontal" if horizontal_path else "vertical" if vertical_path else "meandering" | |
# 收集通道物件(人、自行車、摩托車在中間區域) | |
for obj in detected_objects: | |
if obj["class_id"] in [0, 1, 3]: # Person, bicycle, motorcycle | |
y_pos = obj["normalized_center"][1] | |
# 按垂直位置分組(圖像中間可能是通道) | |
if 0.25 <= y_pos <= 0.75: | |
region = obj["region"] | |
if region not in pathway_regions: | |
pathway_regions[region] = [] | |
pathway_regions[region].append(obj) | |
pathway_items.append(obj["class_name"]) | |
if pathway_items: | |
path_desc = "Pedestrian walkway with people moving through the commercial area" | |
if structured_path: | |
path_desc = f"{path_direction.capitalize()} pedestrian walkway with organized foot traffic" | |
zones["pedestrian_pathway"] = { | |
"region": "middle_center", # 假設:通道通常在中間 | |
"objects": list(set(pathway_items)), | |
"description": path_desc | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying Asian pedestrian pathway: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_vendor_zones(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
識別攤販區域 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
攤販區域字典 | |
""" | |
try: | |
zones = {} | |
# 識別攤販區域(小攤/商店 - 從情境推斷) | |
has_small_objects = any(obj["class_id"] in [24, 26, 39, 41] for obj in detected_objects) # bags, bottles, cups | |
has_people = any(obj["class_id"] == 0 for obj in detected_objects) | |
if has_small_objects and has_people: | |
# 可能的攤販區域是人群和小物件聚集的地方 | |
small_obj_regions = {} | |
for obj in detected_objects: | |
if obj["class_id"] in [24, 26, 39, 41, 67]: # bags, bottles, cups, phones | |
region = obj["region"] | |
if region not in small_obj_regions: | |
small_obj_regions[region] = [] | |
small_obj_regions[region].append(obj) | |
if small_obj_regions: | |
main_vendor_region = max(small_obj_regions.items(), | |
key=lambda x: len(x[1]), | |
default=(None, [])) | |
if main_vendor_region[0] is not None: | |
vendor_items = [obj["class_name"] for obj in main_vendor_region[1]] | |
zones["vendor_zone"] = { | |
"region": main_vendor_region[0], | |
"objects": list(set(vendor_items)), | |
"description": "Vendor or market stall area with small merchandise" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying vendor zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_upscale_decorative_zones(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
識別高級餐飲的裝飾區域 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
裝飾區域字典 | |
""" | |
try: | |
zones = {} | |
decor_items = [] | |
decor_regions = {} | |
# 尋找裝飾元素(花瓶、酒杯、未使用的餐具) | |
for obj in detected_objects: | |
if obj["class_id"] in [75, 40]: # Vase, wine glass | |
region = obj["region"] | |
if region not in decor_regions: | |
decor_regions[region] = [] | |
decor_regions[region].append(obj) | |
decor_items.append(obj["class_name"]) | |
if decor_items: | |
main_decor_region = max(decor_regions.items(), | |
key=lambda x: len(x[1]), | |
default=(None, [])) | |
if main_decor_region[0] is not None: | |
zones["decorative_zone"] = { | |
"region": main_decor_region[0], | |
"objects": list(set(decor_items)), | |
"description": f"Decorative area with {', '.join(list(set(decor_items)))}" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying upscale decorative zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_dining_seating_zones(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
識別餐廳座位安排區域 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
座位區域字典 | |
""" | |
try: | |
zones = {} | |
# 識別座位安排區域 | |
chairs = [obj for obj in detected_objects if obj["class_id"] == 56] # chairs | |
if len(chairs) >= 2: | |
chair_regions = {} | |
for obj in chairs: | |
region = obj["region"] | |
if region not in chair_regions: | |
chair_regions[region] = [] | |
chair_regions[region].append(obj) | |
if chair_regions: | |
main_seating_region = max(chair_regions.items(), | |
key=lambda x: len(x[1]), | |
default=(None, [])) | |
if main_seating_region[0] is not None: | |
zones["dining_seating_zone"] = { | |
"region": main_seating_region[0], | |
"objects": ["chair"] * len(main_seating_region[1]), | |
"description": f"Formal dining seating arrangement with {len(main_seating_region[1])} chairs" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying dining seating zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_serving_zones(self, detected_objects: List[Dict], existing_zones: Dict) -> Dict: | |
""" | |
識別服務區域 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
existing_zones: 已存在的功能區域 | |
Returns: | |
服務區域字典 | |
""" | |
try: | |
zones = {} | |
serving_items = [] | |
serving_regions = {} | |
# 服務區域可能有瓶子、碗、容器 | |
for obj in detected_objects: | |
if obj["class_id"] in [39, 45]: # Bottle, bowl | |
# 檢查是否在與主餐桌不同的區域 | |
if "formal_dining_zone" in existing_zones and obj["region"] != existing_zones["formal_dining_zone"]["region"]: | |
region = obj["region"] | |
if region not in serving_regions: | |
serving_regions[region] = [] | |
serving_regions[region].append(obj) | |
serving_items.append(obj["class_name"]) | |
if serving_items: | |
main_serving_region = max(serving_regions.items(), | |
key=lambda x: len(x[1]), | |
default=(None, [])) | |
if main_serving_region[0] is not None: | |
zones["serving_zone"] = { | |
"region": main_serving_region[0], | |
"objects": list(set(serving_items)), | |
"description": f"Serving or sideboard area with {', '.join(list(set(serving_items)))}" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying serving zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_building_zones(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
識別建築區域(從場景情境推斷) | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
建築區域字典 | |
""" | |
try: | |
zones = {} | |
# 側邊建築區域(從場景情境推斷) | |
# 檢查是否有實際可能包含建築物的區域 | |
left_side_regions = ["top_left", "middle_left", "bottom_left"] | |
right_side_regions = ["top_right", "middle_right", "bottom_right"] | |
# 檢查左側 | |
left_building_evidence = True | |
for region in left_side_regions: | |
# 如果此區域有很多車輛或人群,不太可能是建築物 | |
vehicle_in_region = any(obj["region"] == region and obj["class_id"] in [1, 2, 3, 5, 7] | |
for obj in detected_objects) | |
people_in_region = any(obj["region"] == region and obj["class_id"] == 0 | |
for obj in detected_objects) | |
if vehicle_in_region or people_in_region: | |
left_building_evidence = False | |
break | |
# 檢查右側 | |
right_building_evidence = True | |
for region in right_side_regions: | |
# 如果此區域有很多車輛或人群,不太可能是建築物 | |
vehicle_in_region = any(obj["region"] == region and obj["class_id"] in [1, 2, 3, 5, 7] | |
for obj in detected_objects) | |
people_in_region = any(obj["region"] == region and obj["class_id"] == 0 | |
for obj in detected_objects) | |
if vehicle_in_region or people_in_region: | |
right_building_evidence = False | |
break | |
# 如果證據支持,添加建築區域 | |
if left_building_evidence: | |
zones["building_zone_left"] = { | |
"region": "middle_left", | |
"objects": ["building"], # 推斷 | |
"description": "Tall buildings line the left side of the street" | |
} | |
if right_building_evidence: | |
zones["building_zone_right"] = { | |
"region": "middle_right", | |
"objects": ["building"], # 推斷 | |
"description": "Tall buildings line the right side of the street" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying building zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _identify_financial_pedestrian_zones(self, detected_objects: List[Dict]) -> Dict: | |
""" | |
識別金融區的行人區域 | |
Args: | |
detected_objects: 檢測到的物件列表 | |
Returns: | |
行人區域字典 | |
""" | |
try: | |
zones = {} | |
# 識別行人區域(如果有人群) | |
people_objs = [obj for obj in detected_objects if obj["class_id"] == 0] | |
if people_objs: | |
people_regions = {} | |
for obj in people_objs: | |
region = obj["region"] | |
if region not in people_regions: | |
people_regions[region] = [] | |
people_regions[region].append(obj) | |
if people_regions: | |
main_pedestrian_region = max(people_regions.items(), | |
key=lambda x: len(x[1]), | |
default=(None, [])) | |
if main_pedestrian_region[0] is not None: | |
zones["pedestrian_zone"] = { | |
"region": main_pedestrian_region[0], | |
"objects": ["person"] * len(main_pedestrian_region[1]), | |
"description": f"Pedestrian area with {len(main_pedestrian_region[1])} people navigating the financial district" | |
} | |
return zones | |
except Exception as e: | |
logger.error(f"Error identifying financial pedestrian zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |
def _create_landmark_auxiliary_zones(self, landmark: Dict, index: int) -> Dict: | |
""" | |
創建地標相關的輔助區域(攝影區、紀念品區等) | |
Args: | |
landmark: 地標物件字典 | |
index: 地標索引 | |
Returns: | |
輔助區域字典 | |
""" | |
try: | |
auxiliary_zones = {} | |
landmark_region = landmark.get("region", "middle_center") | |
landmark_name = landmark.get("class_name", "Landmark") | |
# 創建攝影區 | |
# 根據地標位置調整攝影區位置(地標前方通常是攝影區) | |
region_mapping = { | |
"top_left": "bottom_right", | |
"top_center": "bottom_center", | |
"top_right": "bottom_left", | |
"middle_left": "middle_right", | |
"middle_center": "bottom_center", | |
"middle_right": "middle_left", | |
"bottom_left": "top_right", | |
"bottom_center": "top_center", | |
"bottom_right": "top_left" | |
} | |
photo_region = region_mapping.get(landmark_region, landmark_region) | |
photo_key = f"{landmark_name.lower().replace(' ', '_')}_photography_spot" | |
auxiliary_zones[photo_key] = { | |
"name": f"{landmark_name} Photography Spot", | |
"description": f"Popular position for photographing {landmark_name} with optimal viewing angle.", | |
"objects": ["camera", "person", "cell phone"], | |
"region": photo_region, | |
"primary_function": "Tourist photography" | |
} | |
# 如果是著名地標,可能有紀念品販售區 | |
if landmark.get("confidence", 0) > 0.7: # 高置信度地標更可能有紀念品區 | |
# 根據地標位置找到適合的紀念品區位置(通常在地標附近但不直接在地標上) | |
adjacent_regions = { | |
"top_left": ["top_center", "middle_left"], | |
"top_center": ["top_left", "top_right"], | |
"top_right": ["top_center", "middle_right"], | |
"middle_left": ["top_left", "bottom_left"], | |
"middle_center": ["middle_left", "middle_right"], | |
"middle_right": ["top_right", "bottom_right"], | |
"bottom_left": ["middle_left", "bottom_center"], | |
"bottom_center": ["bottom_left", "bottom_right"], | |
"bottom_right": ["bottom_center", "middle_right"] | |
} | |
if landmark_region in adjacent_regions: | |
souvenir_region = adjacent_regions[landmark_region][0] # 選擇第一個相鄰區域 | |
souvenir_key = f"{landmark_name.lower().replace(' ', '_')}_souvenir_area" | |
auxiliary_zones[souvenir_key] = { | |
"name": f"{landmark_name} Souvenir Area", | |
"description": f"Area where visitors can purchase souvenirs and memorabilia related to {landmark_name}.", | |
"objects": ["person", "handbag", "backpack"], | |
"region": souvenir_region, | |
"primary_function": "Tourism commerce" | |
} | |
return auxiliary_zones | |
except Exception as e: | |
logger.error(f"Error creating landmark auxiliary zones: {str(e)}") | |
logger.error(traceback.format_exc()) | |
return {} | |