# -*- coding: utf-8 -*- """ 斑马线感知监控器 基于面积变化的斑马线检测和语音提示 不涉及状态切换，只提供语音引导 """ import time import numpy as np from collections import deque from typing import Optional, Dict, Any import logging logger = logging.getLogger(__name__) class CrosswalkAwarenessMonitor: """斑马线感知监控器 - 纯语音提示模块""" def __init__(self): # 面积阈值（固定锚点） self.THRESHOLDS = { 'discover': 0.01, # 1% - 发现 'approaching': 0.08, # 8% - 靠近 'near': 0.18, # 18% - 很近 'arrival': 0.25, # 25% - 到达（可以过马路） } # 已播报的阈值（避免重复） self.broadcasted_thresholds = set() # 面积历史记录 self.area_history = deque(maxlen=30) # 保存最近30帧 # 时间记录 self.last_broadcast_time = 0 self.arrival_first_broadcast_time = 0 # 状态标志 self.in_arrival_state = False # 是否在"可以过马路"状态 self.last_position_zone = None # 上次播报的方位 # 播报间隔配置（可调整参数 - 数值越小播报越频繁） # 【参数调整】将所有间隔除以1.5，提高播报频率1.5倍 self.REPEAT_INTERVALS = { 'approaching': 6.7, # 靠近阶段：每6.7秒重复（原10秒÷1.5） 'near': 3.3, # 很近阶段：每3.3秒重复（原5秒÷1.5） 'arrival': 5.3, # 到达阶段：每5.3秒重复（原8秒÷1.5） } # 提示：如需调整频率，修改这些数值即可 # - 数值越小 = 播报越频繁 # - 数值越大 = 播报越稀疏 # 无遮挡判断阈值 self.OCCLUSION_THRESHOLD = 0.30 # 重叠>30%认为有遮挡 def process_frame(self, crosswalk_mask, blind_path_mask=None) -> Optional[Dict[str, Any]]: """ 处理每帧的斑马线检测 返回： { 'voice_text': 语音文本, 'priority': 优先级, 'should_broadcast': 是否应该播报, 'area': 当前面积, 'position': 方位描述, 'visualization': 可视化信息（用于外部绘制） } 或 None（无需播报） """ # 如果没有斑马线，重置状态 if crosswalk_mask is None: self._reset_if_needed() return None # 1. 计算面积 total_pixels = crosswalk_mask.size crosswalk_pixels = np.sum(crosswalk_mask > 0) area_ratio = crosswalk_pixels / total_pixels # 2. 计算中心位置 y_coords, x_coords = np.where(crosswalk_mask > 0) if len(y_coords) == 0: return None center_x_ratio = np.mean(x_coords) / crosswalk_mask.shape[1] center_y_ratio = np.mean(y_coords) / crosswalk_mask.shape[0] # 3. 记录历史 current_time = time.time() self.area_history.append({ 'area': area_ratio, 'center_x': center_x_ratio, 'center_y': center_y_ratio, 'time': current_time }) # 4. 检查遮挡 has_occlusion = self._check_occlusion(crosswalk_mask, blind_path_mask) # 5. 判断当前阶段和生成语音 return self._generate_guidance(area_ratio, center_x_ratio, center_y_ratio, has_occlusion, current_time) def _check_occlusion(self, crosswalk_mask, blind_path_mask) -> bool: """检查斑马线是否被盲道遮挡""" if blind_path_mask is None: return False crosswalk_area = crosswalk_mask > 0 blind_path_area = blind_path_mask > 0 # 计算重叠 overlap = np.logical_and(crosswalk_area, blind_path_area) overlap_ratio = np.sum(overlap) / max(np.sum(crosswalk_area), 1) # 重叠超过阈值认为有遮挡 return overlap_ratio > self.OCCLUSION_THRESHOLD def _get_position_description(self, center_x_ratio) -> str: """获取方位描述（3分法）""" if center_x_ratio < 0.40: return "在画面左侧" elif center_x_ratio < 0.60: return "在画面中间" else: return "在画面右侧" def _generate_guidance(self, area_ratio, center_x_ratio, center_y_ratio, has_occlusion, current_time) -> Optional[Dict[str, Any]]: """生成引导语音""" # 检查面积是否稳定（避免抖动） if not self._is_area_stable(area_ratio): return None position_desc = self._get_position_description(center_x_ratio) # 阶段1：发现阶段（0.01-0.08） if area_ratio >= self.THRESHOLDS['discover'] and area_ratio < self.THRESHOLDS['approaching']: if self.THRESHOLDS['discover'] not in self.broadcasted_thresholds: self.broadcasted_thresholds.add(self.THRESHOLDS['discover']) return { 'voice_text': f"远处发现斑马线,{position_desc}", 'priority': 55, # 提高到55，超过盲道方向指令(50) 'should_broadcast': True, 'area': area_ratio, 'position': position_desc } # 阶段2：靠近阶段（0.08-0.18） elif area_ratio >= self.THRESHOLDS['approaching'] and area_ratio < self.THRESHOLDS['near']: # 首次播报 if self.THRESHOLDS['approaching'] not in self.broadcasted_thresholds: self.broadcasted_thresholds.add(self.THRESHOLDS['approaching']) self.last_broadcast_time = current_time self.last_position_zone = position_desc return { 'voice_text': f"正在靠近斑马线,{position_desc}", 'priority': 55, # 提高到55 'should_broadcast': True, 'area': area_ratio, 'position': position_desc } # 重复播报（每10秒或方位变化） elif (current_time - self.last_broadcast_time >= self.REPEAT_INTERVALS['approaching'] or position_desc != self.last_position_zone): self.last_broadcast_time = current_time self.last_position_zone = position_desc return { 'voice_text': f"正在靠近斑马线,{position_desc}", 'priority': 55, # 提高到55 'should_broadcast': True, 'area': area_ratio, 'position': position_desc } # 阶段3：很近阶段（0.18-0.25） elif area_ratio >= self.THRESHOLDS['near'] and area_ratio < self.THRESHOLDS['arrival']: # 首次播报 if self.THRESHOLDS['near'] not in self.broadcasted_thresholds: self.broadcasted_thresholds.add(self.THRESHOLDS['near']) self.last_broadcast_time = current_time self.last_position_zone = position_desc return { 'voice_text': f"接近斑马线,{position_desc}", 'priority': 60, 'should_broadcast': True, 'area': area_ratio, 'position': position_desc } # 重复播报（每5秒或方位变化） elif (current_time - self.last_broadcast_time >= self.REPEAT_INTERVALS['near'] or position_desc != self.last_position_zone): self.last_broadcast_time = current_time self.last_position_zone = position_desc return { 'voice_text': f"接近斑马线,{position_desc}", 'priority': 60, 'should_broadcast': True, 'area': area_ratio, 'position': position_desc } # 阶段4：到达阶段（area ≥ 0.25，无遮挡） elif area_ratio >= self.THRESHOLDS['arrival']: # 必须无遮挡才能提示过马路 if has_occlusion: # 有遮挡，暂不提示过马路，停留在阶段3 logger.info(f"[斑马线] 面积达到{area_ratio:.2f}但被遮挡，暂不提示过马路") return None # 首次到达 if not self.in_arrival_state: self.in_arrival_state = True self.arrival_first_broadcast_time = current_time self.last_broadcast_time = current_time logger.info(f"[斑马线] 到达状态：area={area_ratio:.2f}, 无遮挡") return { 'voice_text': "斑马线到了可以过马路", 'priority': 80, 'should_broadcast': True, 'area': area_ratio, 'position': '到达' } # 重复播报（每8秒） elif current_time - self.last_broadcast_time >= self.REPEAT_INTERVALS['arrival']: self.last_broadcast_time = current_time return { 'voice_text': "斑马线到了可以过马路", 'priority': 80, 'should_broadcast': True, 'area': area_ratio, 'position': '到达' } # 超时处理（30秒后自动退出到达状态） elif current_time - self.arrival_first_broadcast_time > 30.0: logger.info("[斑马线] 到达状态超时30秒，自动退出") self.in_arrival_state = False return None # 降级处理：如果从到达状态面积减小 if self.in_arrival_state and area_ratio < 0.20: logger.info(f"[斑马线] 面积降至{area_ratio:.2f}，退出到达状态") self.in_arrival_state = False # 清除部分已播报标记，允许重新播报 self.broadcasted_thresholds.discard(self.THRESHOLDS['arrival']) return None def _is_area_stable(self, area_ratio, stability_frames=5) -> bool: """检查面积是否稳定（避免抖动触发）""" if len(self.area_history) < stability_frames: return True # 初始阶段，认为稳定 recent_areas = [h['area'] for h in list(self.area_history)[-stability_frames:]] # 检查最近N帧是否都在当前面积附近（±20%） for recent_area in recent_areas: if abs(recent_area - area_ratio) / max(area_ratio, 0.001) > 0.20: return False return True def _reset_if_needed(self): """重置状态（斑马线消失时）""" if len(self.area_history) > 0: logger.info("[斑马线] 斑马线消失，重置状态") self.broadcasted_thresholds.clear() self.area_history.clear() self.in_arrival_state = False self.last_position_zone = None def reset(self): """完全重置""" self.broadcasted_thresholds.clear() self.area_history.clear() self.in_arrival_state = False self.last_broadcast_time = 0 self.arrival_first_broadcast_time = 0 self.last_position_zone = None logger.info("[斑马线] 感知监控器已重置") def is_in_arrival_state(self) -> bool: """是否在到达状态（用于外部判断是否暂停盲道语音）""" return self.in_arrival_state def get_current_area(self) -> float: """获取当前面积""" if len(self.area_history) > 0: return self.area_history[-1]['area'] return 0.0 def get_visualization_data(self, crosswalk_mask, area_ratio, center_x_ratio, center_y_ratio, has_occlusion) -> Dict[str, Any]: """ 获取可视化数据 返回包含所有可视化元素的字典 """ if crosswalk_mask is None: return {} # 确定当前阶段（统一使用橙色） if area_ratio >= self.THRESHOLDS['arrival']: stage = "到达" stage_color = "rgba(255, 165, 0, 0.5)" # 橙色 elif area_ratio >= self.THRESHOLDS['near']: stage = "接近" stage_color = "rgba(255, 165, 0, 0.45)" # 橙色 elif area_ratio >= self.THRESHOLDS['approaching']: stage = "靠近" stage_color = "rgba(255, 165, 0, 0.40)" # 橙色 else: stage = "发现" stage_color = "rgba(255, 165, 0, 0.35)" # 橙色 # 方位描述 position = self._get_position_description(center_x_ratio) return { 'area_ratio': area_ratio, 'stage': stage, 'stage_color': stage_color, 'position': position.replace("在画面", ""), # 去掉"在画面"前缀 'center_x_ratio': center_x_ratio, 'center_y_ratio': center_y_ratio, 'has_occlusion': has_occlusion, 'in_arrival': self.in_arrival_state } # 辅助函数 def split_combined_voice(combined_text: str) -> list: """ 将组合语音拆分为多个独立语音 例如："远处发现斑马线,在画面左侧" → ["远处发现斑马线", "在画面左侧"] """ if ',' in combined_text: parts = combined_text.split(',') return [p.strip() for p in parts if p.strip()] return [combined_text]