VisionScout / spatial_analyzer.py
DawnC's picture
Upload spatial_analyzer.py
d823aa6 verified
import os
import numpy as np
from typing import Dict, List, Tuple, Any, Optional
from scene_type import SCENE_TYPES
from enhance_scene_describer import EnhancedSceneDescriber
class SpatialAnalyzer:
"""
Analyzes spatial relationships between objects in an image.
Handles region assignment, object positioning, and functional zone identification.
"""
def __init__(self, class_names: Dict[int, str] = None, object_categories=None):
"""Initialize the spatial analyzer with image regions"""
# Define regions of the image (3x3 grid)
self.regions = {
"top_left": (0, 0, 1/3, 1/3),
"top_center": (1/3, 0, 2/3, 1/3),
"top_right": (2/3, 0, 1, 1/3),
"middle_left": (0, 1/3, 1/3, 2/3),
"middle_center": (1/3, 1/3, 2/3, 2/3),
"middle_right": (2/3, 1/3, 1, 2/3),
"bottom_left": (0, 2/3, 1/3, 1),
"bottom_center": (1/3, 2/3, 2/3, 1),
"bottom_right": (2/3, 2/3, 1, 1)
}
self.class_names = class_names
self.OBJECT_CATEGORIES = object_categories or {}
self.enhance_descriptor = EnhancedSceneDescriber(scene_types=SCENE_TYPES)
# Distances thresholds for proximity analysis (normalized)
self.proximity_threshold = 0.2
def _determine_region(self, x: float, y: float) -> str:
"""
Determine which region a point falls into.
Args:
x: Normalized x-coordinate (0-1)
y: Normalized y-coordinate (0-1)
Returns:
Region name
"""
for region_name, (x1, y1, x2, y2) in self.regions.items():
if x1 <= x < x2 and y1 <= y < y2:
return region_name
return "unknown"
def _analyze_regions(self, detected_objects: List[Dict]) -> Dict:
"""
Analyze object distribution across image regions.
Args:
detected_objects: List of detected objects with position information
Returns:
Dictionary with region analysis
"""
# Count objects in each region
region_counts = {region: 0 for region in self.regions.keys()}
region_objects = {region: [] for region in self.regions.keys()}
for obj in detected_objects:
region = obj["region"]
if region in region_counts:
region_counts[region] += 1
region_objects[region].append({
"class_id": obj["class_id"],
"class_name": obj["class_name"]
})
# Determine main focus regions (top 1-2 regions by object count)
sorted_regions = sorted(region_counts.items(), key=lambda x: x[1], reverse=True)
main_regions = [region for region, count in sorted_regions if count > 0][:2]
return {
"counts": region_counts,
"main_focus": main_regions,
"objects_by_region": region_objects
}
def _extract_detected_objects(self, detection_result: Any, confidence_threshold: float = 0.25) -> List[Dict]:
"""
Extract detected objects from detection result with position information.
Args:
detection_result: Detection result from YOLOv8
confidence_threshold: Minimum confidence threshold
Returns:
List of dictionaries with detected object information
"""
boxes = detection_result.boxes.xyxy.cpu().numpy()
classes = detection_result.boxes.cls.cpu().numpy().astype(int)
confidences = detection_result.boxes.conf.cpu().numpy()
# Image dimensions
img_height, img_width = detection_result.orig_shape[:2]
detected_objects = []
for box, class_id, confidence in zip(boxes, classes, confidences):
# Skip objects with confidence below threshold
if confidence < confidence_threshold:
continue
x1, y1, x2, y2 = box
width = x2 - x1
height = y2 - y1
# Center point
center_x = (x1 + x2) / 2
center_y = (y1 + y2) / 2
# Normalized positions (0-1)
norm_x = center_x / img_width
norm_y = center_y / img_height
norm_width = width / img_width
norm_height = height / img_height
# Area calculation
area = width * height
norm_area = area / (img_width * img_height)
# Region determination
object_region = self._determine_region(norm_x, norm_y)
detected_objects.append({
"class_id": int(class_id),
"class_name": self.class_names[int(class_id)],
"confidence": float(confidence),
"box": [float(x1), float(y1), float(x2), float(y2)],
"center": [float(center_x), float(center_y)],
"normalized_center": [float(norm_x), float(norm_y)],
"size": [float(width), float(height)],
"normalized_size": [float(norm_width), float(norm_height)],
"area": float(area),
"normalized_area": float(norm_area),
"region": object_region
})
return detected_objects
def _detect_scene_viewpoint(self, detected_objects: List[Dict]) -> Dict:
"""
檢測場景視角並識別特殊場景模式。
Args:
detected_objects: 檢測到的物體列表
Returns:
Dict: 包含視角和場景模式信息的字典
"""
if not detected_objects:
return {"viewpoint": "eye_level", "patterns": []}
# 從物體位置中提取信息
patterns = []
# 檢測行人位置模式
pedestrian_objs = [obj for obj in detected_objects if obj["class_id"] == 0]
# 檢查是否有足夠的行人來識別模式
if len(pedestrian_objs) >= 4:
pedestrian_positions = [obj["normalized_center"] for obj in pedestrian_objs]
# 檢測十字交叉模式
if self._detect_cross_pattern(pedestrian_positions):
patterns.append("crosswalk_intersection")
# 檢測多方向行人流
directions = self._analyze_movement_directions(pedestrian_positions)
if len(directions) >= 2:
patterns.append("multi_directional_movement")
# 檢查物體的大小一致性 - 在空中俯視圖中,物體大小通常更一致
if len(detected_objects) >= 5:
sizes = [obj.get("normalized_area", 0) for obj in detected_objects]
size_variance = np.var(sizes) / (np.mean(sizes) ** 2) # 標準化變異數,不會受到平均值影響
if size_variance < 0.3: # 低變異表示大小一致
patterns.append("consistent_object_size")
# 基本視角檢測
viewpoint = self.enhance_descriptor._detect_viewpoint(detected_objects)
# 根據檢測到的模式增強視角判斷
if "crosswalk_intersection" in patterns and viewpoint != "aerial":
# 如果檢測到斑馬線交叉但視角判斷不是空中視角,優先採用模式判斷
viewpoint = "aerial"
return {
"viewpoint": viewpoint,
"patterns": patterns
}
def _detect_cross_pattern(self, positions):
"""
檢測位置中的十字交叉模式
Args:
positions: 位置列表 [[x1, y1], [x2, y2], ...]
Returns:
bool: 是否檢測到十字交叉模式
"""
if len(positions) < 8: # 需要足夠多的點
return False
# 提取 x 和 y 坐標
x_coords = [pos[0] for pos in positions]
y_coords = [pos[1] for pos in positions]
# 檢測 x 和 y 方向的聚類
x_clusters = []
y_clusters = []
# 簡化的聚類分析
x_mean = np.mean(x_coords)
y_mean = np.mean(y_coords)
# 計算在中心線附近的點
near_x_center = sum(1 for x in x_coords if abs(x - x_mean) < 0.1)
near_y_center = sum(1 for y in y_coords if abs(y - y_mean) < 0.1)
# 如果有足夠的點在中心線附近,可能是十字交叉
return near_x_center >= 3 and near_y_center >= 3
def _analyze_movement_directions(self, positions):
"""
分析位置中的移動方向
Args:
positions: 位置列表 [[x1, y1], [x2, y2], ...]
Returns:
list: 檢測到的主要方向
"""
if len(positions) < 6:
return []
# extract x 和 y 坐標
x_coords = [pos[0] for pos in positions]
y_coords = [pos[1] for pos in positions]
directions = []
# horizontal move (left --> right)
x_std = np.std(x_coords)
x_range = max(x_coords) - min(x_coords)
# vertical move(up --> down)
y_std = np.std(y_coords)
y_range = max(y_coords) - min(y_coords)
# 足夠大的範圍表示該方向有運動
if x_range > 0.4:
directions.append("horizontal")
if y_range > 0.4:
directions.append("vertical")
return directions
def _identify_functional_zones(self, detected_objects: List[Dict], scene_type: str) -> Dict:
"""
Identify functional zones within the scene with improved detection for different viewpoints
and cultural contexts.
Args:
detected_objects: List of detected objects
scene_type: Identified scene type
Returns:
Dictionary of functional zones with their descriptions
"""
# Group objects by category and region
category_regions = {}
for obj in detected_objects:
# Find object category
category = "other"
for cat_name, cat_ids in self.OBJECT_CATEGORIES.items():
if obj["class_id"] in cat_ids:
category = cat_name
break
# Add to category-region mapping
if category not in category_regions:
category_regions[category] = {}
region = obj["region"]
if region not in category_regions[category]:
category_regions[category][region] = []
category_regions[category][region].append(obj)
# Identify zones based on object groupings
zones = {}
# Detect viewpoint to adjust zone identification strategy
viewpoint = self._detect_scene_viewpoint(detected_objects)
# Choose appropriate zone identification strategy based on scene type and viewpoint
if scene_type in ["living_room", "bedroom", "dining_area", "kitchen", "office_workspace", "meeting_room"]:
# Indoor scenes
zones.update(self._identify_indoor_zones(category_regions, detected_objects, scene_type))
elif scene_type in ["city_street", "parking_lot", "park_area"]:
# Outdoor general scenes
zones.update(self._identify_outdoor_general_zones(category_regions, detected_objects, scene_type))
elif "aerial" in scene_type or viewpoint == "aerial":
# Aerial viewpoint scenes
zones.update(self._identify_aerial_view_zones(category_regions, detected_objects, scene_type))
elif "asian" in scene_type:
# Asian cultural context scenes
zones.update(self._identify_asian_cultural_zones(category_regions, detected_objects, scene_type))
elif scene_type == "urban_intersection":
# Specific urban intersection logic
zones.update(self._identify_intersection_zones(category_regions, detected_objects, viewpoint))
elif scene_type == "financial_district":
# Financial district specific logic
zones.update(self._identify_financial_district_zones(category_regions, detected_objects))
elif scene_type == "upscale_dining":
# Upscale dining specific logic
zones.update(self._identify_upscale_dining_zones(category_regions, detected_objects))
else:
# Default zone identification for other scene types
zones.update(self._identify_default_zones(category_regions, detected_objects))
# If no zones were identified, try the default approach
if not zones:
zones.update(self._identify_default_zones(category_regions, detected_objects))
return zones
def _identify_indoor_zones(self, category_regions: Dict, detected_objects: List[Dict], scene_type: str) -> Dict:
"""
Identify functional zones for indoor scenes.
Args:
category_regions: Objects grouped by category and region
detected_objects: List of detected objects
scene_type: Specific indoor scene type
Returns:
Dict: Indoor functional zones
"""
zones = {}
# Seating/social zone
if "furniture" in category_regions:
furniture_regions = category_regions["furniture"]
main_furniture_region = max(furniture_regions.items(),
key=lambda x: len(x[1]),
default=(None, []))
if main_furniture_region[0] is not None and len(main_furniture_region[1]) >= 2:
zone_objects = [obj["class_name"] for obj in main_furniture_region[1]]
zones["social_zone"] = {
"region": main_furniture_region[0],
"objects": zone_objects,
"description": f"Social or seating area with {', '.join(zone_objects)}"
}
# Entertainment zone
if "electronics" in category_regions:
electronics_items = []
for region_objects in category_regions["electronics"].values():
electronics_items.extend([obj["class_name"] for obj in region_objects])
if electronics_items:
zones["entertainment_zone"] = {
"region": self._find_main_region(category_regions.get("electronics", {})),
"objects": electronics_items,
"description": f"Entertainment or media area with {', '.join(electronics_items)}"
}
# Dining/food zone
food_zone_categories = ["kitchen_items", "food"]
food_items = []
food_regions = {}
for category in food_zone_categories:
if category in category_regions:
for region, objects in category_regions[category].items():
if region not in food_regions:
food_regions[region] = []
food_regions[region].extend(objects)
food_items.extend([obj["class_name"] for obj in objects])
if food_items:
main_food_region = max(food_regions.items(),
key=lambda x: len(x[1]),
default=(None, []))
if main_food_region[0] is not None:
zones["dining_zone"] = {
"region": main_food_region[0],
"objects": list(set(food_items)),
"description": f"Dining or food preparation area with {', '.join(list(set(food_items))[:3])}"
}
# Work/study zone - enhanced to detect even when scene_type is not explicitly office
work_items = []
work_regions = {}
for obj in detected_objects:
if obj["class_id"] in [56, 60, 63, 64, 66, 73]: # chair, table, laptop, mouse, keyboard, book
region = obj["region"]
if region not in work_regions:
work_regions[region] = []
work_regions[region].append(obj)
work_items.append(obj["class_name"])
# Check for laptop and table/chair combinations that suggest a workspace
has_laptop = any(obj["class_id"] == 63 for obj in detected_objects)
has_keyboard = any(obj["class_id"] == 66 for obj in detected_objects)
has_table = any(obj["class_id"] == 60 for obj in detected_objects)
has_chair = any(obj["class_id"] == 56 for obj in detected_objects)
# If we have electronics with furniture in the same region, likely a workspace
workspace_detected = (has_laptop or has_keyboard) and (has_table or has_chair)
if (workspace_detected or scene_type in ["office_workspace", "meeting_room"]) and work_items:
main_work_region = max(work_regions.items(),
key=lambda x: len(x[1]),
default=(None, []))
if main_work_region[0] is not None:
zones["workspace_zone"] = {
"region": main_work_region[0],
"objects": list(set(work_items)),
"description": f"Work or study area with {', '.join(list(set(work_items))[:3])}"
}
# Bedroom-specific zones
if scene_type == "bedroom":
bed_objects = [obj for obj in detected_objects if obj["class_id"] == 59] # Bed
if bed_objects:
bed_region = bed_objects[0]["region"]
zones["sleeping_zone"] = {
"region": bed_region,
"objects": ["bed"],
"description": "Sleeping area with bed"
}
# Kitchen-specific zones
if scene_type == "kitchen":
# Look for appliances (refrigerator, oven, microwave, sink)
appliance_ids = [68, 69, 71, 72] # microwave, oven, sink, refrigerator
appliance_objects = [obj for obj in detected_objects if obj["class_id"] in appliance_ids]
if appliance_objects:
appliance_regions = {}
for obj in appliance_objects:
region = obj["region"]
if region not in appliance_regions:
appliance_regions[region] = []
appliance_regions[region].append(obj)
if appliance_regions:
main_appliance_region = max(appliance_regions.items(),
key=lambda x: len(x[1]),
default=(None, []))
if main_appliance_region[0] is not None:
appliance_names = [obj["class_name"] for obj in main_appliance_region[1]]
zones["kitchen_appliance_zone"] = {
"region": main_appliance_region[0],
"objects": appliance_names,
"description": f"Kitchen appliance area with {', '.join(appliance_names)}"
}
return zones
def _identify_intersection_zones(self, category_regions: Dict, detected_objects: List[Dict], viewpoint: str) -> Dict:
"""
Identify functional zones for urban intersections with enhanced spatial awareness.
Args:
category_regions: Objects grouped by category and region
detected_objects: List of detected objects
viewpoint: Detected viewpoint
Returns:
Dict: Refined intersection functional zones
"""
zones = {}
# Get pedestrians, vehicles and traffic signals
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]] # bicycle, car, motorcycle, bus, truck
traffic_light_objs = [obj for obj in detected_objects if obj["class_id"] == 9]
# Create distribution maps for better spatial understanding
regions_distribution = self._create_distribution_map(detected_objects)
# Analyze pedestrian crossing patterns
crossing_zones = self._analyze_crossing_patterns(pedestrian_objs, traffic_light_objs, regions_distribution)
zones.update(crossing_zones)
# Analyze vehicle traffic zones with directional awareness
traffic_zones = self._analyze_traffic_zones(vehicle_objs, regions_distribution)
zones.update(traffic_zones)
# Identify traffic control zones based on signal placement
if traffic_light_objs:
# Group traffic lights by region for better organization
signal_regions = {}
for obj in traffic_light_objs:
region = obj["region"]
if region not in signal_regions:
signal_regions[region] = []
signal_regions[region].append(obj)
# Create traffic control zones for each region with signals
for idx, (region, signals) in enumerate(signal_regions.items()):
# Check if this region has a directional name
direction = self._get_directional_description(region)
zones[f"traffic_control_zone_{idx+1}"] = {
"region": region,
"objects": ["traffic light"] * len(signals),
"description": f"Traffic control area with {len(signals)} traffic signals" +
(f" in {direction} area" if direction else "")
}
return zones
def _analyze_crossing_patterns(self, pedestrians: List[Dict], traffic_lights: List[Dict],
region_distribution: Dict) -> Dict:
"""
Analyze pedestrian crossing patterns to identify crosswalk zones.
Args:
pedestrians: List of pedestrian objects
traffic_lights: List of traffic light objects
region_distribution: Distribution of objects by region
Returns:
Dict: Identified crossing zones
"""
crossing_zones = {}
if not pedestrians:
return crossing_zones
# Group pedestrians by region
pedestrian_regions = {}
for p in pedestrians:
region = p["region"]
if region not in pedestrian_regions:
pedestrian_regions[region] = []
pedestrian_regions[region].append(p)
# Sort regions by pedestrian count to find main crossing areas
sorted_regions = sorted(pedestrian_regions.items(), key=lambda x: len(x[1]), reverse=True)
# Create crossing zones for regions with pedestrians
for idx, (region, peds) in enumerate(sorted_regions[:2]): # Focus on top 2 regions
# Check if there are traffic lights nearby to indicate a crosswalk
has_nearby_signals = any(t["region"] == region for t in traffic_lights)
# Create crossing zone with descriptive naming
zone_name = f"crossing_zone_{idx+1}"
direction = self._get_directional_description(region)
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"
crossing_zones[zone_name] = {
"region": region,
"objects": ["pedestrian"] * len(peds),
"description": description
}
return crossing_zones
def _analyze_traffic_zones(self, vehicles: List[Dict], region_distribution: Dict) -> Dict:
"""
Analyze vehicle distribution to identify traffic zones with directional awareness.
Args:
vehicles: List of vehicle objects
region_distribution: Distribution of objects by region
Returns:
Dict: Identified traffic zones
"""
traffic_zones = {}
if not vehicles:
return traffic_zones
# Group vehicles by region
vehicle_regions = {}
for v in vehicles:
region = v["region"]
if region not in vehicle_regions:
vehicle_regions[region] = []
vehicle_regions[region].append(v)
# Create traffic zones for regions with vehicles
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]
# Get a list of vehicle types for description
vehicle_types = [v["class_name"] for v in vehicles_in_region]
unique_types = list(set(vehicle_types))
# Get directional description
direction = self._get_directional_description(region)
# Create descriptive zone
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 vehicles are distributed across multiple regions, create secondary zones
if len(vehicle_regions) > 1:
# Get second most populated region
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
def _get_directional_description(self, region: str) -> str:
"""
Convert region name to a directional description.
Args:
region: Region name from the grid
Returns:
str: Directional description
"""
if "top" in region and "left" in region:
return "northwest"
elif "top" in region and "right" in region:
return "northeast"
elif "bottom" in region and "left" in region:
return "southwest"
elif "bottom" in region and "right" in region:
return "southeast"
elif "top" in region:
return "north"
elif "bottom" in region:
return "south"
elif "left" in region:
return "west"
elif "right" in region:
return "east"
else:
return "central"
def _create_distribution_map(self, detected_objects: List[Dict]) -> Dict:
"""
Create a distribution map of objects across regions for spatial analysis.
Args:
detected_objects: List of detected objects
Returns:
Dict: Distribution map of objects by region and class
"""
distribution = {}
# Initialize all regions
for region in self.regions.keys():
distribution[region] = {
"total": 0,
"objects": {},
"density": 0
}
# Populate the distribution
for obj in detected_objects:
region = obj["region"]
class_id = obj["class_id"]
class_name = obj["class_name"]
distribution[region]["total"] += 1
if class_id not in distribution[region]["objects"]:
distribution[region]["objects"][class_id] = {
"name": class_name,
"count": 0,
"positions": []
}
distribution[region]["objects"][class_id]["count"] += 1
# Store position for spatial relationship analysis
if "normalized_center" in obj:
distribution[region]["objects"][class_id]["positions"].append(obj["normalized_center"])
# Calculate object density for each region
for region, data in distribution.items():
# Assuming all regions are equal size in the grid
data["density"] = data["total"] / 1
return distribution
def _identify_asian_cultural_zones(self, category_regions: Dict, detected_objects: List[Dict], scene_type: str) -> Dict:
"""
Identify functional zones for scenes with Asian cultural context.
Args:
category_regions: Objects grouped by category and region
detected_objects: List of detected objects
scene_type: Specific scene type
Returns:
Dict: Asian cultural functional zones
"""
zones = {}
# Identify storefront zone
storefront_items = []
storefront_regions = {}
# Since storefronts aren't directly detectable, infer from context
# For example, look for regions with signs, people, and smaller objects
sign_regions = set()
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)
# Add regions with people as potential storefront areas
sign_regions.add(region)
# Use the areas with most people as storefront zones
if storefront_regions:
main_storefront_regions = sorted(storefront_regions.items(),
key=lambda x: len(x[1]),
reverse=True)[:2] # Top 2 regions
for idx, (region, objs) in enumerate(main_storefront_regions):
zones[f"commercial_zone_{idx+1}"] = {
"region": region,
"objects": [obj["class_name"] for obj in objs],
"description": f"Asian commercial storefront with pedestrian activity"
}
# Identify pedestrian pathway - enhanced to better detect linear pathways
pathway_items = []
pathway_regions = {}
# Extract people for pathway analysis
people_objs = [obj for obj in detected_objects if obj["class_id"] == 0]
# Analyze if people form a line (typical of shopping streets)
people_positions = [obj["normalized_center"] for obj in people_objs]
structured_path = False
if len(people_positions) >= 3:
# Check if people are arranged along a similar y-coordinate (horizontal path)
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 # Low variance indicates horizontal alignment
# Check if people are arranged along a similar x-coordinate (vertical path)
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 # Low variance indicates vertical alignment
structured_path = horizontal_path or vertical_path
path_direction = "horizontal" if horizontal_path else "vertical" if vertical_path else "meandering"
# Collect pathway objects (people, bicycles, motorcycles in middle area)
for obj in detected_objects:
if obj["class_id"] in [0, 1, 3]: # Person, bicycle, motorcycle
y_pos = obj["normalized_center"][1]
# Group by vertical position (middle of image likely pathway)
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", # Assumption: pathway often in middle
"objects": list(set(pathway_items)),
"description": path_desc
}
# Identify vendor zone (small stalls/shops - inferred from context)
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:
# Likely vendor areas are where people and small objects cluster
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"
}
# For night markets, identify illuminated zones
if scene_type == "asian_night_market":
# Night markets typically have bright spots for food stalls
# This would be enhanced with lighting analysis integration
zones["food_stall_zone"] = {
"region": "middle_center",
"objects": ["inferred food stalls"],
"description": "Food stall area typical of Asian night markets"
}
return zones
def _identify_upscale_dining_zones(self, category_regions: Dict, detected_objects: List[Dict]) -> Dict:
"""
Identify functional zones for upscale dining settings.
Args:
category_regions: Objects grouped by category and region
detected_objects: List of detected objects
Returns:
Dict: Upscale dining functional zones
"""
zones = {}
# Identify dining table zone
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])}"
}
# Identify decorative zone with enhanced detection
decor_items = []
decor_regions = {}
# Look for decorative elements (vases, wine glasses, unused dishes)
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)))}"
}
# Identify seating arrangement zone
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"
}
# Identify serving area (if different from dining area)
serving_items = []
serving_regions = {}
# Serving areas might have bottles, bowls, containers
for obj in detected_objects:
if obj["class_id"] in [39, 45]: # Bottle, bowl
# Check if it's in a different region from the main dining table
if "formal_dining_zone" in zones and obj["region"] != 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
def _identify_financial_district_zones(self, category_regions: Dict, detected_objects: List[Dict]) -> Dict:
"""
Identify functional zones for financial district scenes.
Args:
category_regions: Objects grouped by category and region
detected_objects: List of detected objects
Returns:
Dict: Financial district functional zones
"""
zones = {}
# Identify traffic zone
traffic_items = []
traffic_regions = {}
for obj in detected_objects:
if obj["class_id"] in [1, 2, 3, 5, 6, 7, 9]: # Various vehicles and traffic lights
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])}"
}
# Building zones on the sides (inferred from scene context)
# Enhanced to check if there are actual regions that might contain buildings
# Check for regions without vehicles or pedestrians - likely building areas
left_side_regions = ["top_left", "middle_left", "bottom_left"]
right_side_regions = ["top_right", "middle_right", "bottom_right"]
# Check left side
left_building_evidence = True
for region in left_side_regions:
# If many vehicles or people in this region, less likely to be buildings
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
# Check right side
right_building_evidence = True
for region in right_side_regions:
# If many vehicles or people in this region, less likely to be buildings
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
# Add building zones if evidence supports them
if left_building_evidence:
zones["building_zone_left"] = {
"region": "middle_left",
"objects": ["building"], # Inferred
"description": "Tall buildings line the left side of the street"
}
if right_building_evidence:
zones["building_zone_right"] = {
"region": "middle_right",
"objects": ["building"], # Inferred
"description": "Tall buildings line the right side of the street"
}
# Identify pedestrian zone if people are present
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
def _identify_aerial_view_zones(self, category_regions: Dict, detected_objects: List[Dict], scene_type: str) -> Dict:
"""
Identify functional zones for scenes viewed from an aerial perspective.
Args:
category_regions: Objects grouped by category and region
detected_objects: List of detected objects
scene_type: Specific scene type
Returns:
Dict: Aerial view functional zones
"""
zones = {}
# For aerial views, we focus on patterns and flows rather than specific zones
# Identify pedestrian patterns
people_objs = [obj for obj in detected_objects if obj["class_id"] == 0]
if people_objs:
# Convert positions to arrays for pattern analysis
positions = np.array([obj["normalized_center"] for obj in people_objs])
if len(positions) >= 3:
# Calculate distribution metrics
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)
# Determine if people are organized in a linear pattern
if x_std < 0.1 or y_std < 0.1:
# Linear distribution along one axis
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:
# More dispersed pattern
zones["pedestrian_distribution"] = {
"region": "wide",
"objects": ["person"] * len(people_objs),
"description": f"Aerial view shows pedestrians distributed across the area"
}
# Identify vehicle patterns for traffic analysis
vehicle_objs = [obj for obj in detected_objects if obj["class_id"] in [1, 2, 3, 5, 6, 7]]
if vehicle_objs:
# Convert positions to arrays for pattern analysis
positions = np.array([obj["normalized_center"] for obj in vehicle_objs])
if len(positions) >= 2:
# Calculate distribution metrics
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)
# Determine if vehicles are organized in lanes
if x_std < y_std * 0.5:
# Vehicles aligned vertically - indicates north-south traffic
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:
# Vehicles aligned horizontally - indicates east-west traffic
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:
# Vehicles in multiple directions - indicates intersection
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"
}
# For intersection specific aerial views, identify crossing patterns
if "intersection" in scene_type:
# Check for traffic signals
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"
}
# Crosswalks are inferred from context in aerial views
zones["crossing_pattern"] = {
"region": "central",
"objects": ["inferred crosswalk"],
"description": "Crossing pattern visible from aerial perspective"
}
# For plaza aerial views, identify gathering patterns
if "plaza" in scene_type:
# Plazas typically have central open area with people
if people_objs:
# Check if people are clustered in central region
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
def _identify_outdoor_general_zones(self, category_regions: Dict, detected_objects: List[Dict], scene_type: str) -> Dict:
"""
Identify functional zones for general outdoor scenes.
Args:
category_regions: Objects grouped by category and region
detected_objects: List of detected objects
scene_type: Specific outdoor scene type
Returns:
Dict: Outdoor functional zones
"""
zones = {}
# Identify pedestrian 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:
# Find main pedestrian areas
main_people_regions = sorted(people_regions.items(),
key=lambda x: len(x[1]),
reverse=True)[:2] # Top 2 regions
for idx, (region, objs) in enumerate(main_people_regions):
if len(objs) > 0:
zones[f"pedestrian_zone_{idx+1}"] = {
"region": region,
"objects": ["person"] * len(objs),
"description": f"Pedestrian area with {len(objs)} {'people' if len(objs) > 1 else 'person'}"
}
# Identify vehicle zones for streets and parking lots
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])}"
}
# For park areas, identify recreational zones
if scene_type == "park_area":
# Look for recreational objects (sports balls, kites, etc.)
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)))}"
}
# For parking lots, identify parking zones
if scene_type == "parking_lot":
# Look for parked cars with consistent spacing
car_objs = [obj for obj in detected_objects if obj["class_id"] == 2] # cars
if len(car_objs) >= 3:
# Check if cars are arranged in patterns (simplified)
car_positions = [obj["normalized_center"] for obj in car_objs]
# Check for row patterns by analyzing vertical positions
y_coords = [pos[1] for pos in car_positions]
y_clusters = {}
# Simplified clustering - group cars by similar y-coordinates
for i, y in enumerate(y_coords):
assigned = False
for cluster_y in y_clusters.keys():
if abs(y - cluster_y) < 0.1: # Within 10% of image height
y_clusters[cluster_y].append(i)
assigned = True
break
if not assigned:
y_clusters[y] = [i]
# If we have row patterns
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
def _identify_default_zones(self, category_regions: Dict, detected_objects: List[Dict]) -> Dict:
"""
Identify general functional zones when no specific scene type is matched.
Args:
category_regions: Objects grouped by category and region
detected_objects: List of detected objects
Returns:
Dict: Default functional zones
"""
zones = {}
# Group objects by category and find main concentrations
for category, regions in category_regions.items():
if not regions:
continue
# Find region with most objects in this category
main_region = max(regions.items(),
key=lambda x: len(x[1]),
default=(None, []))
if main_region[0] is None or len(main_region[1]) < 2:
continue
# Create zone based on object category
zone_objects = [obj["class_name"] for obj in main_region[1]]
# Skip if too few objects
if len(zone_objects) < 2:
continue
# Create appropriate zone name and description based on category
if category == "furniture":
zones["furniture_zone"] = {
"region": main_region[0],
"objects": zone_objects,
"description": f"Area with furniture including {', '.join(zone_objects[:3])}"
}
elif category == "electronics":
zones["electronics_zone"] = {
"region": main_region[0],
"objects": zone_objects,
"description": f"Area with electronic devices including {', '.join(zone_objects[:3])}"
}
elif category == "kitchen_items":
zones["dining_zone"] = {
"region": main_region[0],
"objects": zone_objects,
"description": f"Dining or food area with {', '.join(zone_objects[:3])}"
}
elif category == "vehicles":
zones["vehicle_zone"] = {
"region": main_region[0],
"objects": zone_objects,
"description": f"Area with vehicles including {', '.join(zone_objects[:3])}"
}
elif category == "personal_items":
zones["personal_items_zone"] = {
"region": main_region[0],
"objects": zone_objects,
"description": f"Area with personal items including {', '.join(zone_objects[:3])}"
}
# Check for people groups
people_objs = [obj for obj in detected_objects if obj["class_id"] == 0]
if len(people_objs) >= 2:
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_region = max(people_regions.items(),
key=lambda x: len(x[1]),
default=(None, []))
if main_people_region[0] is not None:
zones["people_zone"] = {
"region": main_people_region[0],
"objects": ["person"] * len(main_people_region[1]),
"description": f"Area with {len(main_people_region[1])} people"
}
return zones
def _find_main_region(self, region_objects_dict: Dict) -> str:
"""Find the main region with the most objects"""
if not region_objects_dict:
return "unknown"
return max(region_objects_dict.items(),
key=lambda x: len(x[1]),
default=("unknown", []))[0]
def _find_main_region(self, region_objects_dict: Dict) -> str:
"""Find the main region with the most objects"""
if not region_objects_dict:
return "unknown"
return max(region_objects_dict.items(),
key=lambda x: len(x[1]),
default=("unknown", []))[0]