"""Risk fusion engine — contextual, trend-aware, with user profiles. Improvements over POC: 1. Contextual: bent knees on flat ground ≠ bent knees on a slope 2. Trend-based: steepening slope is riskier than steady slope 3. Obstacle velocity: closing obstacles score higher than static ones 4. User profiles: elderly/athletic/default adjust sensitivity 5. Temporal: risk decays slowly (no flicker between HIGH/LOW) """ from config import RISK_HIGH, RISK_MEDIUM, RISK_LOW, RISK_PROFILE # ─── User profiles: multipliers on base scores ─── _PROFILES = { "default": {"slope": 1.0, "gait": 1.0, "obstacle": 1.0, "compound": 1.0}, "elderly": {"slope": 1.5, "gait": 1.5, "obstacle": 1.3, "compound": 1.5}, "athletic": {"slope": 0.6, "gait": 0.5, "obstacle": 1.0, "compound": 0.5}, } def _get_profile(): return _PROFILES.get(RISK_PROFILE, _PROFILES["default"]) class RiskEngine: """Stateful risk engine with temporal smoothing and context awareness.""" def __init__(self, profile=None): self.profile = _PROFILES.get(profile or RISK_PROFILE, _PROFILES["default"]) self._prev_score = 0 self._prev_level = "SAFE" self._decay = 0.7 # risk decays slowly to prevent flicker def assess(self, slope_angle, slope_dir, gait, obstacles, slope_trend="STABLE", new_obstacles=None, closing_obstacles=None): """Compute contextual risk from all signals. Returns dict with: risk, score, reasons, gait_summary, components. """ sa = abs(slope_angle) p = self.profile score = 0 reasons = [] components = {} # ── 1. Terrain slope ── s_score = 0 if sa > 20: s_score = 40 reasons.append(f"steep slope ({slope_angle:.0f}°)") elif sa > 10: s_score = 20 reasons.append(f"moderate slope ({slope_angle:.0f}°)") elif sa > 3: s_score = 5 reasons.append(f"mild slope ({slope_angle:.0f}°)") # Trend bonus: steepening is riskier than steady if slope_trend == "STEEPENING": s_score = int(s_score * 1.4) reasons.append("slope steepening") elif slope_trend == "FLATTENING" and s_score > 0: s_score = int(s_score * 0.7) s_score = int(s_score * p["slope"]) components['slope'] = s_score score += s_score # ── 2. Gait analysis (contextual) ── g_score = 0 gait_summary = {} if gait: kn = gait['avg_knee'] sh = abs(gait['avg_shin']) ln = abs(gait['avg_lean']) sy = gait['symmetry'] gait_summary = {'knee': kn, 'shin': sh, 'lean': ln, 'symmetry': sy} # Bent knees: only risky if NOT on a slope (on slopes it's expected adaptation) if kn < 130: if sa < 5: # Bent knees on flat = potential instability g_score += 25 reasons.append(f"heavily bent knees on flat ({kn:.0f}°)") else: # Bent knees on slope = expected, mild concern only if extreme if kn < 110: g_score += 15 reasons.append(f"extreme knee bend ({kn:.0f}°)") elif kn < 150: if sa < 5: g_score += 10 reasons.append(f"bent knees on flat ({kn:.0f}°)") # Shin tilt: contextual — expected to tilt on slopes expected_shin = sa * 0.4 # rough expected shin tilt for slope excess_shin = max(0, sh - expected_shin) if excess_shin > 15: g_score += 15 reasons.append(f"excess shin tilt ({sh:.0f}° vs expected {expected_shin:.0f}°)") elif excess_shin > 8: g_score += 8 # Body lean: expected on slopes, risky if opposite direction if slope_dir == "UPHILL" and ln < -10: g_score += 20 reasons.append(f"leaning backward on uphill ({ln:.0f}°)") elif slope_dir == "DOWNHILL" and ln > 10: g_score += 20 reasons.append(f"leaning forward on downhill ({ln:.0f}°)") elif abs(ln) > 20: g_score += 10 reasons.append(f"excessive lean ({ln:.0f}°)") # Asymmetry: always concerning if sy > 25: g_score += 25 reasons.append(f"severe gait asymmetry ({sy:.0f}°)") elif sy > 15: g_score += 12 reasons.append(f"gait asymmetry ({sy:.0f}°)") g_score = int(g_score * p["gait"]) components['gait'] = g_score score += g_score # ── 3. Obstacles ── o_score = 0 num_obs = len(obstacles) if isinstance(obstacles, list) else obstacles if isinstance(obstacles, list): near = [o for o in obstacles if o.get('proximity', 0) > 0.6] o_score += min(20, len(near) * 10) if len(near) >= 2: reasons.append(f"{len(near)} obstacles nearby") # Closing obstacles are more dangerous if closing_obstacles: o_score += min(20, len(closing_obstacles) * 12) for ob in closing_obstacles[:2]: rate = ob.get('closing_rate', 0) reasons.append(f"{ob['label']} closing ({rate:.0%}/frame)") # New obstacles: brief awareness bump if new_obstacles: o_score += min(10, len(new_obstacles) * 5) else: o_score += min(30, num_obs * 10) o_score = int(o_score * p["obstacle"]) components['obstacles'] = o_score score += o_score # ── 4. Compound risks ── c_score = 0 if gait and sa > 10: kn = gait['avg_knee'] sy = gait['symmetry'] if kn < 150 and sy > 15: c_score += 20 reasons.append("slope + bent knees + asymmetry") elif kn < 150: c_score += 12 reasons.append("slope + bent knees") elif sy > 15: c_score += 12 reasons.append("slope + asymmetry") if isinstance(obstacles, list): near = [o for o in obstacles if o.get('proximity', 0) > 0.7] if near and sa > 10: c_score += 15 reasons.append("slope + near obstacle") c_score = int(c_score * p["compound"]) components['compound'] = c_score score += c_score # ── 5. Temporal smoothing (prevent flicker) ── raw_score = min(100, score) smoothed = self._decay * self._prev_score + (1 - self._decay) * raw_score # Snap up fast (danger), decay down slowly (safety) if raw_score > self._prev_score: smoothed = max(smoothed, raw_score * 0.85) # jump up quickly self._prev_score = smoothed final_score = int(smoothed) # Level if final_score >= RISK_HIGH: level = "HIGH" elif final_score >= RISK_MEDIUM: level = "MEDIUM" elif final_score >= RISK_LOW: level = "LOW" else: level = "SAFE" self._prev_level = level return { 'risk': level, 'score': final_score, 'raw_score': raw_score, 'terrain': slope_dir, 'terrain_slope': round(slope_angle, 1), 'slope_trend': slope_trend, 'reasons': reasons, 'gait_summary': gait_summary, 'components': components, 'profile': RISK_PROFILE, } # ─── Backward-compatible module-level function ─── _default_engine = None def assess(slope_angle, slope_dir, gait, num_obstacles, slope_trend="STABLE", new_obstacles=None, closing_obstacles=None): """Stateless convenience wrapper. For video, use RiskEngine class directly.""" global _default_engine if _default_engine is None: _default_engine = RiskEngine() return _default_engine.assess( slope_angle, slope_dir, gait, num_obstacles, slope_trend, new_obstacles, closing_obstacles)