127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
import json
|
|
import os
|
|
import time
|
|
from typing import Dict, List, Optional
|
|
from app.services import notifications
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(__file__)) # backend/app
|
|
DATA_DIR = os.path.join(ROOT, "data")
|
|
|
|
ROUTES: List[Dict] = []
|
|
NOTIFS: List[Dict] = []
|
|
COLONIAS: List[Dict] = []
|
|
ESTADO: Dict[str, int] = {}
|
|
STATUS: Dict[str, str] = {}
|
|
LAST_EVENTS: List[Dict] = []
|
|
|
|
|
|
def _load_json(filename: str):
|
|
path = os.path.join(DATA_DIR, filename)
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def load_data():
|
|
global ROUTES, NOTIFS, COLONIAS
|
|
ROUTES = _load_json("rutas.json")
|
|
NOTIFS = _load_json("notificaciones.json")
|
|
COLONIAS = _load_json("colonias-rutas.json")
|
|
|
|
|
|
def start_simulation_state():
|
|
"""Inicializa el estado (positionId) para cada ruta presente en `rutas.json`."""
|
|
global ESTADO, STATUS
|
|
ESTADO = {}
|
|
STATUS = {}
|
|
for r in ROUTES:
|
|
rid = r.get("routeId")
|
|
ESTADO[rid] = 1
|
|
STATUS[rid] = r.get("status", "PENDIENTE")
|
|
|
|
|
|
def get_colonias():
|
|
return COLONIAS
|
|
|
|
|
|
def get_route_position(routeId: str) -> Optional[int]:
|
|
return ESTADO.get(routeId)
|
|
|
|
|
|
def get_route_status(routeId: str) -> Optional[str]:
|
|
return STATUS.get(routeId)
|
|
|
|
|
|
def _find_notif(event_name: str) -> Optional[Dict]:
|
|
for n in NOTIFS:
|
|
if n.get("triggerEvent") == event_name:
|
|
return n
|
|
return None
|
|
|
|
|
|
def tick() -> List[Dict]:
|
|
"""Avanza todas las rutas en memoria (pos 1→8) y devuelve eventos disparados.
|
|
Al llegar a pos 8, reinicia la ruta para que la demo sea un ciclo continuo."""
|
|
global ESTADO, STATUS, LAST_EVENTS
|
|
events = []
|
|
for route_id, pos in list(ESTADO.items()):
|
|
# Ciclo: cuando la ruta completó, reiniciar en el siguiente tick
|
|
if pos >= 8:
|
|
ESTADO[route_id] = 1
|
|
STATUS[route_id] = "PENDIENTE"
|
|
print(f"[SIM RESET] {route_id} reiniciado para nuevo ciclo.")
|
|
continue
|
|
|
|
antes = pos
|
|
ahora = pos + 1
|
|
ESTADO[route_id] = ahora
|
|
|
|
# Actualizar STATUS según la nueva posición
|
|
if ahora == 2:
|
|
STATUS[route_id] = "en_ruta"
|
|
elif ahora == 8:
|
|
STATUS[route_id] = "completada"
|
|
|
|
evt = None
|
|
if antes == 1 and ahora == 2:
|
|
evt = "ROUTE_START"
|
|
elif ahora == 4:
|
|
evt = "TRUCK_PROXIMITY"
|
|
elif ahora == 8:
|
|
evt = "ROUTE_COMPLETED"
|
|
|
|
if evt:
|
|
notif = _find_notif(evt)
|
|
payload = notif.get("pushPayload") if notif else {"title": evt, "body": ""}
|
|
# Adjuntar event + routeId en `data` para que el cliente Flutter
|
|
# los clasifique (notifications_screen.dart usa msg.data['event']).
|
|
payload = {
|
|
**payload,
|
|
"data": {
|
|
**(payload.get("data") or {}),
|
|
"event": evt,
|
|
"routeId": route_id,
|
|
},
|
|
}
|
|
simulated = {"routeId": route_id, "event": evt, "payload": payload}
|
|
events.append(simulated)
|
|
LAST_EVENTS.append(simulated)
|
|
topic = f"topic_{route_id}"
|
|
try:
|
|
notifications.send_to_topic(topic, payload)
|
|
|
|
# ── WHATSAPP: Se dispara cuando el camión está cerca (posición 4) ──
|
|
if evt == "TRUCK_PROXIMITY":
|
|
# Para evitar saturar el bot y bloquear el servidor,
|
|
# enviamos el WhatsApp de demostración solo para la primera ruta.
|
|
if route_id == "RUTA-01":
|
|
msg = f"¡Hola! Soy Eco 🍃. El camión recolector de tu ruta ({route_id}) está a menos de 15 minutos. ¡Saca la basura! ♻️"
|
|
notifications.send_whatsapp_alert("5214131060699", msg)
|
|
except Exception:
|
|
print(f"[SIM PUSH FAIL] {route_id} -> {evt}: {payload.get('title')} - {payload.get('body')}")
|
|
|
|
return events
|
|
|
|
|
|
def get_last_events() -> List[Dict]:
|
|
return LAST_EVENTS[-20:]
|