""" Sentinel Prediction Engine — Proactive risk forecasting. Analyzes the current ContextState, historical MemoryEntries, and event patterns to predict risks BEFORE they become critical. Each prediction carries a type, urgency level, confidence score, and recommended action. Prediction rules: 1. Person Contact: Approaching person will reach user in N seconds 2. Fall Risk: Dark + walking + elevated risk = fall probability 3. Situation Escalation: Multiple recent warnings = deteriorating trend 4. Battery Depletion: Low battery + outdoors = stranding risk 5. Prolonged Inactivity: Stationary too long = possible medical emergency Usage: predictor = PredictionEngine() predictions = predictor.predict(context_engine.state, memory_engine) for p in predictions: print(f"[{p.urgency}] {p.message} (conf: {p.confidence:.2f})") prompt_text = predictor.get_prediction_prompt(predictions) """ import time from dataclasses import dataclass, field try: import structlog logger = structlog.get_logger() except ImportError: import logging logger = logging.getLogger("sentinel.prediction") from context_engine import ContextState from memory_engine import MemoryEngine @dataclass class Prediction: """ A single predicted outcome with confidence and recommended action. Attributes: type: Machine-readable category for programmatic handling. Examples: "person_contact", "fall_risk", "escalation", "battery_risk", "inactivity_alert" urgency: Severity level: "critical", "warning", "info" message: Human-readable description for display or TTS. confidence: Probability estimate 0.0 to 1.0. action: Optional recommended response (e.g., "send_sos", "alert_user"). """ type: str urgency: str message: str confidence: float action: str = "" def to_dict(self) -> dict: return { "type": self.type, "urgency": self.urgency, "message": self.message, "confidence": round(self.confidence, 2), "action": self.action, } class PredictionEngine: """ Rule-based prediction system that forecasts risks from context and memory. All rules are deterministic and interpretable — no black-box ML. Each rule evaluates independently and returns zero or one Prediction. Rules can be individually tuned via sensitivity parameters. Sensitivity parameters: memory_window: Seconds of history to analyze (default 60) escalation_threshold: Number of warnings in window to trigger escalation (default 3) risk_threshold: Minimum risk_level to consider for predictions (default 0.2) """ def __init__( self, memory_window: float = 60.0, escalation_threshold: int = 3, risk_threshold: float = 0.2, ): self.memory_window = memory_window self.escalation_threshold = escalation_threshold self.risk_threshold = risk_threshold self.last_predictions: list[Prediction] = [] self._last_activity_time: float = time.time() self._last_known_activity: str = "idle" def predict( self, context: ContextState, memory: MemoryEngine | None = None ) -> list[Prediction]: """ Run all prediction rules against current context and memory. Returns a list of active predictions (may be empty). Results are also stored in self.last_predictions. """ predictions: list[Prediction] = [] recent_entries = memory.recent(self.memory_window) if memory else [] trend = memory.analyze_trend(self.memory_window) if memory else {} self._check_person_contact(context, predictions) self._check_fall_risk(context, predictions) self._check_situation_escalation(context, recent_entries, trend, predictions) self._check_battery_risk(context, predictions) self._check_inactivity(context, predictions) self.last_predictions = predictions return predictions # ── Prediction Rules ──────────────────────────────────────────────────── def _check_person_contact( self, ctx: ContextState, out: list[Prediction] ) -> None: """ Rule: Person approaching + close distance = imminent contact. Estimates seconds to contact from distance and approach speed. """ if ctx.closest_person_trend != "approaching": return if ctx.closest_person_distance > 5.0: return speed = max(0.2, ctx.speed) seconds_to_contact = ctx.closest_person_distance / speed if seconds_to_contact < 3.0: urgency = "critical" confidence = 0.92 elif seconds_to_contact < 8.0: urgency = "warning" confidence = 0.80 else: urgency = "info" confidence = 0.60 out.append(Prediction( type="person_contact", urgency=urgency, message=( f"Person approaching, ~{seconds_to_contact:.0f}s to contact " f"(currently {ctx.closest_person_distance:.1f}m)" ), confidence=confidence, action="alert_user" if urgency in ("critical", "warning") else "", )) def _check_fall_risk( self, ctx: ContextState, out: list[Prediction] ) -> None: """ Rule: Walking in dark + elevated risk = high fall probability. Factors: light level, activity, current risk score, noise. """ if ctx.activity not in ("walking", "running"): return fall_score = 0.0 if ctx.light_condition == "dark": fall_score += 0.35 elif ctx.light_condition == "dim": fall_score += 0.15 if ctx.risk_level > 0.3: fall_score += 0.2 if ctx.noise_condition == "loud": fall_score += 0.1 if ctx.battery < 20: fall_score += 0.1 if ctx.movement_speed > 2.0: fall_score += 0.1 if fall_score < 0.3: return urgency = "critical" if fall_score > 0.6 else "warning" out.append(Prediction( type="fall_risk", urgency=urgency, message=f"Elevated fall risk: {ctx.activity} in {ctx.light_condition} conditions (score: {fall_score:.2f})", confidence=min(0.95, fall_score), action="alert_user", )) def _check_situation_escalation( self, ctx: ContextState, recent: list, trend: dict, out: list[Prediction] ) -> None: """ Rule: Multiple warnings in short time = situation deteriorating. Escalates from warning to critical when threshold is exceeded. """ warnings = sum(1 for e in recent if e.alert_level == "warning") criticals = sum(1 for e in recent if e.alert_level == "critical") direction = trend.get("direction", "stable") if warnings >= self.escalation_threshold and direction == "deteriorating": out.append(Prediction( type="escalation", urgency="critical", message=( f"Situation deteriorating: {warnings} warnings in " f"{self.memory_window:.0f}s, risk trending upward" ), confidence=0.88, action="send_sos", )) elif warnings >= 2 and direction == "deteriorating": confidence = 0.72 if len(recent) >= 5 else 0.55 out.append(Prediction( type="escalation", urgency="warning", message=f"Multiple warnings detected ({warnings}), situation may worsen", confidence=confidence, action="alert_user", )) elif criticals >= 2: out.append(Prediction( type="repeated_critical", urgency="critical", message=f"Repeated critical alerts ({criticals}), emergency situation likely", confidence=0.90, action="send_sos", )) def _check_battery_risk( self, ctx: ContextState, out: list[Prediction] ) -> None: """ Rule: Low battery + outdoor/mobile = stranding risk. Warns user to head home or charge before losing connectivity. """ if ctx.battery >= 20: return is_mobile = ctx.activity in ("walking", "running") is_outdoors = ctx.location_type == "outdoors" if ctx.battery < 10 and (is_outdoors or is_mobile): out.append(Prediction( type="battery_critical", urgency="warning", message=( f"Critical battery ({ctx.battery:.0f}%) while " f"{ctx.activity} {ctx.location_type}. " f"Sentinel may lose connectivity." ), confidence=0.95, action="alert_user", )) elif is_outdoors: out.append(Prediction( type="battery_risk", urgency="info", message=f"Battery at {ctx.battery:.0f}%. Consider heading to a charging point.", confidence=0.80, )) def _check_inactivity( self, ctx: ContextState, out: list[Prediction] ) -> None: """ Rule: Stationary for extended time = possible medical emergency. Especially relevant for elderly users or after a detected fall. """ if ctx.activity not in ("standing", "idle", "sitting"): self._last_activity_time = time.time() self._last_known_activity = ctx.activity return stationary_seconds = time.time() - self._last_activity_time if ctx.activity == "sitting" and stationary_seconds < 1800: return if stationary_seconds < 600: return minutes = stationary_seconds / 60.0 if "fall_detected" in ctx.risk_factors: out.append(Prediction( type="post_fall_inactivity", urgency="critical", message=( f"No movement for {minutes:.0f} minutes after fall detection. " f"User may be unconscious or unable to get up." ), confidence=0.90, action="send_sos", )) elif stationary_seconds > 1800: out.append(Prediction( type="prolonged_inactivity", urgency="warning", message=f"No significant movement for {minutes:.0f} minutes. Checking on user.", confidence=0.55, action="alert_user", )) else: out.append(Prediction( type="inactivity", urgency="info", message=f"Stationary for {minutes:.0f} minutes.", confidence=0.40, )) # ── Risk Decay ────────────────────────────────────────────────────────── def decay_context_risk(self, context: ContextState, seconds_since_trigger: float) -> None: """ Apply time-based risk decay to a ContextState. Call this each frame to prevent stale risk accumulation. Risk decays faster the longer it has been since the last trigger. """ if seconds_since_trigger > 10.0: decay_rate = min(0.05, seconds_since_trigger * 0.002) context.risk_level = max(0.0, context.risk_level - decay_rate) if context.risk_level < 0.05: context.risk_factors.clear() # ── Formatted Output ──────────────────────────────────────────────────── @staticmethod def get_prediction_prompt(predictions: list[Prediction]) -> str: """ Format active predictions for VLM prompt enrichment. Returns empty string if no predictions. """ if not predictions: return "" lines = ["Active predictions (consider these in your response):"] for p in sorted(predictions, key=lambda x: {"critical": 0, "warning": 1, "info": 2}.get(x.urgency, 3)): lines.append(f" [{p.urgency.upper()}] {p.message} (confidence: {p.confidence:.0%})") return "\n".join(lines) @staticmethod def get_ui_summary(predictions: list[Prediction]) -> str: """Compact HTML-ready summary for dashboard display.""" if not predictions: return "No active predictions" critical = [p for p in predictions if p.urgency == "critical"] warning = [p for p in predictions if p.urgency == "warning"] info = [p for p in predictions if p.urgency == "info"] parts = [] if critical: parts.append(f"{len(critical)} critical") if warning: parts.append(f"{len(warning)} warnings") if info: parts.append(f"{len(info)} info") return " | ".join(parts) if parts else "All clear" def reset(self) -> None: """Reset prediction state.""" self.last_predictions.clear() self._last_activity_time = time.time() self._last_known_activity = "idle"