"""Voice and movement guidance with throttling and tracker-aware alerts.""" import time from config import VOICE_COOLDOWN_SEC, VOICE_SLOPE_DELTA class GuidanceEngine: """Generates movement guidance and throttled voice messages.""" def __init__(self): self._last_voice = "" self._last_voice_time = 0.0 self._last_slope_announced = 0.0 def compute(self, slope_angle, slope_dir, terrain, obstacles, slope_trend="STABLE", new_obstacles=None, closing_obstacles=None): """Return guidance dict with movement recommendations and voice string.""" sa = abs(slope_angle) knee_rec = 135 if sa > 20 else 145 if sa > 10 else 155 if sa > 5 else 165 if sa > 15 or terrain in ("ROCKY", "ROUGH"): step, step_len = "SHORT", "30cm" elif sa > 5: step, step_len = "MEDIUM", "50cm" else: step, step_len = "NORMAL", "70cm" if slope_dir == "UPHILL": foot_adj = round(min(25, sa * 0.7), 1) lean = "FORWARD" elif slope_dir == "DOWNHILL": foot_adj = round(-min(20, sa * 0.6), 1) lean = "BACKWARD" else: foot_adj, lean = 0.0, "UPRIGHT" near_obs = [o for o in obstacles if o['proximity'] > 0.6] risk_score = min(100, int(sa * 1.5 + len(near_obs) * 20 + (15 if terrain == "ROCKY" else 0))) voice = self._build_voice( slope_angle, slope_dir, terrain, obstacles, sa, step, near_obs, slope_trend, new_obstacles, closing_obstacles) return { 'knee_rec': knee_rec, 'step': step, 'step_len': step_len, 'foot_adj': foot_adj, 'lean': lean, 'risk_score': risk_score, 'voice': voice, 'obstacle_warning': self._obstacle_warning(near_obs), 'slope_trend': slope_trend, } def _build_voice(self, slope_angle, slope_dir, terrain, obstacles, sa, step, near_obs, slope_trend, new_obstacles, closing_obstacles): parts = [] # Priority 1: NEW obstacles entering scene if new_obstacles: for ob in new_obstacles[:2]: parts.append(f"New {ob['label']} on {ob['direction'].lower()}.") # Priority 2: CLOSING obstacles (approaching fast) if closing_obstacles: for ob in closing_obstacles[:2]: if ob['direction'] == "CENTER": parts.append(f"{ob['label']} approaching ahead!") else: opp = "right" if ob['direction'] == "LEFT" else "left" parts.append(f"{ob['label']} closing from {ob['direction'].lower()}. Move {opp}.") # Priority 3: Already-near obstacles if not closing_obstacles: for ob in near_obs[:2]: if ob['direction'] == "CENTER": parts.append(f"{ob['label']} ahead! Stop.") else: opp = "right" if ob['direction'] == "LEFT" else "left" parts.append(f"{ob['label']} on {ob['direction'].lower()}. Move {opp}.") # Priority 4: Slope trend changes if slope_trend == "STEEPENING" and sa > 5: parts.append("Slope increasing. Slow down.") elif slope_trend == "FLATTENING" and sa > 3: parts.append("Slope easing.") # Priority 5: Current slope guidance if sa > 3: parts.append(f"Slope {slope_angle:+.0f} degrees.") if slope_dir == "UPHILL": parts.append("Lean forward." if sa > 10 else "Slight forward lean.") else: parts.append("Lean back." if sa > 10 else "Slight backward lean.") if sa > 15 or terrain in ("ROCKY", "ROUGH"): parts.append("Short steps.") elif sa > 5: parts.append("Medium steps.") if terrain == "ROCKY": parts.append("Uneven ground.") elif terrain == "ROUGH": parts.append("Rough surface.") if not parts: parts.append("Path clear.") return " ".join(parts) def _obstacle_warning(self, near_obs): if not near_obs: return None ob = near_obs[0] if ob['direction'] == "CENTER": return f"{ob['label']} AHEAD — STOP" opp = "RIGHT" if ob['direction'] == "LEFT" else "LEFT" return f"{ob['label']} on {ob['direction']} — move {opp}" def should_speak(self, voice, slope_angle): """Throttle: returns True if this message should be spoken aloud.""" now = time.time() # Always speak obstacle warnings immediately if any(kw in voice.lower() for kw in ["ahead", "stop", "new ", "closing", "approaching"]): self._last_voice = voice self._last_voice_time = now return True # Slope changed significantly if abs(slope_angle - self._last_slope_announced) > VOICE_SLOPE_DELTA: self._last_slope_announced = slope_angle self._last_voice = voice self._last_voice_time = now return True # Cooldown elapsed and message changed if now - self._last_voice_time > VOICE_COOLDOWN_SEC and voice != self._last_voice: self._last_voice = voice self._last_voice_time = now return True return False