"""Periodic trigger system for actions during navigation. This module provides a mechanism to execute actions periodically during long-running operations like navigation, with support for progress-based and conditional triggering. """ from __future__ import annotations import asyncio from dataclasses import dataclass, field import logging from typing import Any, Callable, Dict, List, Optional, Set # ============================================================================ # TriggerCondition # ============================================================================ class TriggerCondition: """Condition for when a trigger should fire. Attributes: interval_sec: Time-based trigger interval in seconds progress_milestones: Progress-based triggers (0.0-1.0) e.g., [0.25, 0.5, 0.75] while_navigating: If True, only trigger during active navigation """ def __init__( self, interval_sec: Optional[float] = None, progress_milestones: Optional[List[float]] = None, while_navigating: bool = True, ) -> None: """Initialize TriggerCondition. Args: interval_sec: Time interval in seconds (e.g., 2.0 for every 2 seconds) progress_milestones: List of progress points (0.0-1.0) to trigger at while_navigating: Only fire during active navigation state """ self.interval_sec = interval_sec self.progress_milestones = set(progress_milestones or []) self.while_navigating = while_navigating # Track which milestones have been fired self._fired_milestones: Set[float] = set() def should_fire_time(self, elapsed_sec: float) -> bool: """Check if trigger should fire based on elapsed time. Args: elapsed_sec: Time elapsed since start Returns: True if interval_sec is set and enough time has passed """ if self.interval_sec is None: return False return elapsed_sec >= self.interval_sec def should_fire_progress(self, current_progress: float) -> bool: """Check if trigger should fire based on progress milestone. Args: current_progress: Current progress (0.0-1.0) Returns: True if a new milestone has been reached (not yet fired) """ for milestone in self.progress_milestones: if current_progress >= milestone and milestone not in self._fired_milestones: return True return False def mark_milestone_fired(self, progress: float) -> None: """Mark a milestone as fired. Args: progress: Progress value to mark """ for milestone in self.progress_milestones: if progress >= milestone: self._fired_milestones.add(milestone) def reset(self) -> None: """Reset trigger state (clear fired milestones).""" self._fired_milestones.clear() # ============================================================================ # PeriodicTrigger # ============================================================================ @dataclass class PeriodicTrigger: """A periodic trigger that executes an action based on conditions. Attributes: action_id: The action to execute (e.g., "SCAN_AREA", "SPEAK") condition: TriggerCondition defining when to fire on_fire: Callback function to execute when trigger fires logger: Logger instance for debugging """ action_id: str condition: TriggerCondition on_fire: Callable[[], Any] logger: Optional[logging.Logger] = None # Internal state _running: bool = field(default=False, init=False) _task: Optional[asyncio.Task] = field(default=None, init=False) _start_time: float = field(default=0.0, init=False) _current_progress: float = field(default=0.0, init=False) _stop_event: asyncio.Event = field(default_factory=asyncio.Event, init=False) def start(self) -> None: """Start the periodic trigger. Creates a background task that monitors conditions and fires the trigger when conditions are met. """ if self._running: self.logger.warning( "Trigger '%s' is already running", self.action_id ) if self.logger else None return self._running = True self._stop_event.clear() self._start_time = asyncio.get_event_loop().time() self._current_progress = 0.0 self.condition.reset() # Start background task self._task = asyncio.create_task(self._run_loop()) if self.logger: self.logger.debug("Trigger '%s' started", self.action_id) def stop(self) -> None: """Stop the periodic trigger. Cancels the background task and resets state. """ if not self._running: return self._running = False self._stop_event.set() if self._task and not self._task.done(): self._task.cancel() if self.logger: self.logger.debug("Trigger '%s' stopped", self.action_id) def on_navigation_complete(self) -> None: """Called when navigation completes. Stops the trigger and performs cleanup. """ self.stop() self._current_progress = 1.0 if self.logger: self.logger.debug( "Trigger '%s': navigation complete", self.action_id ) def update_progress(self, progress: float) -> None: """Update navigation progress. Args: progress: Progress value from 0.0 to 1.0 """ self._current_progress = max(0.0, min(1.0, progress)) if self.logger: self.logger.debug( "Trigger '%s': progress updated to %.2f", self.action_id, self._current_progress, ) async def _run_loop(self) -> None: """Main trigger loop running in background task.""" check_interval = 0.1 # Check every 100ms try: while self._running and not self._stop_event.is_set(): loop_time = asyncio.get_event_loop().time() elapsed = loop_time - self._start_time # Check time-based condition if self.condition.should_fire_time(elapsed): # Reset start time BEFORE firing to prevent double-fire during await self._start_time = loop_time await self._fire() # Check progress-based condition elif self.condition.should_fire_progress(self._current_progress): await self._fire() self.condition.mark_milestone_fired(self._current_progress) # Wait for next check or stop signal try: await asyncio.wait_for( self._stop_event.wait(), timeout=check_interval ) break # Stop event was set except asyncio.TimeoutError: continue # Check interval elapsed, continue loop except asyncio.CancelledError: if self.logger: self.logger.debug("Trigger '%s' cancelled", self.action_id) except Exception as exc: if self.logger: self.logger.error( "Trigger '%s' error: %s", self.action_id, exc ) async def _fire(self) -> None: """Execute the trigger callback.""" try: if self.logger: self.logger.info( "Trigger '%s' firing (progress: %.2f)", self.action_id, self._current_progress, ) result = self.on_fire() if asyncio.iscoroutine(result): await result except Exception as exc: if self.logger: self.logger.error( "Trigger '%s' callback failed: %s", self.action_id, exc, )