""" Simulador de Ruta — Módulo B Avanza `truck_status.current_position_id` cada tick. Antes de cada push verifica `notification_preferences` del usuario. Nunca envía coordenadas al cliente — solo ETA + mensaje. """ import asyncio import logging from datetime import datetime from typing import Optional from app.core.config import settings from app.data.repositories.ruta_repository import SQLiteRutaRepository from app.domain.entities.ruta import EstadoCamion, TruckStatus, TipoNotificacion from app.services.ws_manager import ws_manager logger = logging.getLogger(__name__) class SimuladorRuta: def __init__(self, route_id: str, tick_segundos: int = None): self.route_id = route_id self.tick = tick_segundos or settings.sim_tick_seconds self.repo = SQLiteRutaRepository() self._tarea: Optional[asyncio.Task] = None self._corriendo = False # ── Control ──────────────────────────────────────────────────────── def iniciar(self) -> None: if self._corriendo: return self._corriendo = True self._tarea = asyncio.create_task(self._loop()) logger.info(f"[SIM] {self.route_id} iniciada") def detener(self) -> None: self._corriendo = False if self._tarea: self._tarea.cancel() async def forzar_averia(self, mensaje: str = "Falla mecánica reportada.") -> None: """Endpoint /alerts/breakdown llama esto.""" ts = self.repo.obtener_truck_status(self.route_id) if not ts: return ts.status = EstadoCamion.AVERIADA ts.last_update = datetime.utcnow() self.repo.guardar_truck_status(ts) self.detener() await self._broadcast_a_usuarios( tipo=TipoNotificacion.FALLA_MECANICA, eta_minutos=None, mensaje=mensaje, preferencia_key="notify_breakdown", ) logger.warning(f"[SIM] Avería registrada en {self.route_id}") async def forzar_retraso(self, mensaje: str = "El camión reportó un retraso.") -> None: ts = self.repo.obtener_truck_status(self.route_id) if not ts: return ts.status = EstadoCamion.RETRASADA ts.last_update = datetime.utcnow() self.repo.guardar_truck_status(ts) await self._broadcast_a_usuarios( tipo=TipoNotificacion.RUTA_TARDE, eta_minutos=None, mensaje=mensaje, preferencia_key="notify_delay", ) # ── Loop principal ───────────────────────────────────────────────── async def _loop(self) -> None: ruta = self.repo.obtener_ruta(self.route_id) if not ruta or not ruta.puntos: logger.error(f"[SIM] Ruta {self.route_id} sin puntos") return # Inicializar truck_status en posición 1 ts = TruckStatus( route_id=self.route_id, current_position_id=1, last_update=datetime.utcnow(), status=EstadoCamion.EN_RUTA, ) self.repo.guardar_truck_status(ts) await self._broadcast_a_usuarios( tipo=TipoNotificacion.RUTA_INICIADA, eta_minutos=ruta.puntos[-1].tiempo_estimado_min, mensaje="El camión ha iniciado su ruta. Prepárate.", preferencia_key="notify_route_start", ) umbral = settings.sim_eta_alert_minutes ultimo_punto = ruta.puntos[-1] for punto in ruta.puntos[1:]: if not self._corriendo: break await asyncio.sleep(self.tick) eta = max(0, ultimo_punto.tiempo_estimado_min - punto.tiempo_estimado_min) # Detectar umbral de proximidad if eta <= umbral and ts.status == EstadoCamion.EN_RUTA: ts.status = EstadoCamion.APROXIMANDOSE tipo = TipoNotificacion.APROXIMANDOSE pref_key = "notify_proximity" msg = ( f"El camión llega en ~{eta} minutos. " "Saca tu basura ahora." ) else: tipo = TipoNotificacion.RUTA_INICIADA pref_key = "notify_route_start" msg = f"El camión está en camino. Llegada estimada en ~{eta} min." ts.current_position_id = punto.orden ts.last_update = datetime.utcnow() self.repo.guardar_truck_status(ts) await self._broadcast_a_usuarios( tipo=tipo, eta_minutos=eta, mensaje=msg, preferencia_key=pref_key, ) logger.info(f"[SIM] Pos {punto.orden} | ETA {eta} min | {ts.status}") if self._corriendo: ts.status = EstadoCamion.COMPLETADO ts.last_update = datetime.utcnow() self.repo.guardar_truck_status(ts) await self._broadcast_a_usuarios( tipo=TipoNotificacion.COMPLETADO, eta_minutos=0, mensaje="El camión completó su paso. ¡Hasta mañana!", preferencia_key=None, # completado siempre se notifica ) self._corriendo = False logger.info(f"[SIM] {self.route_id} completada") # ── Broadcast respetando preferencias ───────────────────────────── async def _broadcast_a_usuarios( self, tipo: TipoNotificacion, eta_minutos: Optional[int], mensaje: str, preferencia_key: Optional[str], ) -> None: """ Por cada domicilio en la ruta: 1. Consulta las preferencias del usuario. 2. Solo envía si la preferencia está activa. 3. Persiste la notificación en BD. 4. Empuja por WebSocket al address_id correspondiente. """ template = self.repo.obtener_template(tipo.value) if template and "{eta}" in template["body"]: mensaje = template["body"].replace("{eta}", str(eta_minutos or "?")) usuarios = self.repo.obtener_usuarios_por_ruta(self.route_id) for u in usuarios: user_id = u["user_id"] address_id = u["address_id"] # Verificar preferencia if preferencia_key: prefs = self.repo.obtener_preferencias(user_id) if not getattr(prefs, preferencia_key, True): logger.debug( f"[SIM] Usuario {user_id} desactivó {preferencia_key}, skip" ) continue payload = { "tipo": tipo.value, "address_id": address_id, "eta_minutos": eta_minutos, "mensaje": mensaje, "hora_utc": datetime.utcnow().isoformat(), } # Push WebSocket — el cliente escucha en /ws/{address_id} await ws_manager.broadcast_zona(str(address_id), payload) # Persistir self.repo.guardar_notificacion( tipo=tipo.value, route_id=self.route_id, address_id=address_id, mensaje=mensaje, eta_minutos=eta_minutos, ) # ── Registro global ──────────────────────────────────────────────────── _simuladores: dict[str, SimuladorRuta] = {} def obtener_simulador(route_id: str) -> SimuladorRuta: if route_id not in _simuladores: _simuladores[route_id] = SimuladorRuta(route_id) return _simuladores[route_id]