basura_backend/
├── app/
│   ├── domain/              # ← Reglas de negocio puras
│   │   ├── entities/
│   │   │   └── ruta.py      # Ruta, PuntoRuta, TruckStatus
│   │   └── interfaces/
│   │       └── i_ruta_repository.py
│   │
│   ├── data/                # ← Implementaciones concretas
│   │   └── repositories/
│   │       └── ruta_repository.py  # SQLite queries
│   │
│   ├── use_cases/           # ← Lógica de aplicación
│   │   └── obtener_eta.py   # Calcula ETA desde truck_status
│   │
│   ├── services/            # ← Servicios de infraestructura
│   │   ├── simulador.py     # APScheduler + lógica de avance
│   │   └── ws_manager.py    # Singleton WebSocket broadcast
│   │
│   ├── api/routes/          # ← Endpoints HTTP
│   │   └── eta_router.py
│   │
│   ├── db/
│   │   └── database.py      # Esquema SQLite + seed
│   │
│   └── core/
│       └── config.py        # Variables de entorno
│
└── requirements.txt

Capa de Dominio

Entidades (`domain/entities/ruta.py`)

@dataclass
class PuntoRuta:
    id: int
    ruta_id: str
    orden: int
    nombre: str
    lat: float
    lng: float
    tiempo_estimado_min: int

@dataclass
class Ruta:
    id: str
    nombre: str
    turno: str
    puntos: List[PuntoRuta]

@dataclass
class TruckStatus:
    route_id: str
    current_position_id: int
    last_update: datetime
    status: str  # EN_RUTA | AVERIADA | RETRASADA

Sin dependencias de frameworks — solo Python estándar.

Interfaz del Repositorio

class IRutaRepository(ABC):
    @abstractmethod
    async def obtener_ruta(self, ruta_id: str) -> Optional[Ruta]:
        pass
    
    @abstractmethod
    async def obtener_truck_status(self, ruta_id: str) -> Optional[TruckStatus]:
        pass

Capa de Datos

Repositorio SQLite (`data/repositories/ruta_repository.py`)

Lee de rutas, puntos_ruta, truck_status
Implementa IRutaRepository
Retorna entidades del dominio (no diccionarios crudos)

Ejemplo:

async def obtener_ruta(self, ruta_id: str) -> Optional[Ruta]:
    cursor.execute("SELECT * FROM rutas WHERE id = ?", (ruta_id,))
    row = cursor.fetchone()
    # ... mapea a entidad Ruta
    return Ruta(id=row[0], nombre=row[1], ...)

Capa de Casos de Uso

`obtener_eta.py`

Recibe address_id, calcula ETA combinando:

Posición actual del camión (truck_status.current_position_id)
Posición del domicilio del usuario (addresses.route_id)
Distancia entre puntos

Retorna:

{
  "eta_minutos": 22,
  "ventana": {"inicio": "7:20 pm", "fin": "7:35 pm"},
  "mensaje": "El camión está en camino..."
}

Servicios de Infraestructura

Simulador (`services/simulador.py`)

Responsabilidades:

Avanza truck_status.current_position_id cada tick (10s default)
Calcula ETA para cada domicilio en la ruta
Si ETA ≤ umbral (10 min) → broadcast WebSocket
Respeta notification_preferences del usuario
Usa notification_templates para formatear mensajes

Ciclo de vida:

# Arranque manual vía endpoint
POST /admin/route/RUTA-01/start
  → APScheduler.add_job(tick_simulador, interval=10)

# Cada tick
def tick_simulador():
    avanzar_camion()
    para cada domicilio en ruta:
        calcular_eta()
        si eta <= umbral:
            consultar_preferencias(user_id)
            si preferencia.notify_proximity:
                broadcast_ws(address_id, payload)

Eventos soportados:

ruta_iniciada → broadcast al arrancar
aproximandose → cuando ETA ≤ 10 min
falla_mecanica → POST /alerts/breakdown
ruta_tarde → POST /admin/route/{id}/delay

WebSocket Manager (`services/ws_manager.py`)

class WSManager:
    _instance = None  # Singleton
    connections: Dict[str, List[WebSocket]] = {}
    
    async def connect(self, address_id: str, ws: WebSocket):
        # Indexa por address_id como zona_key
        
    async def broadcast_zona(self, address_id: str, payload: dict):
        # Envía a todos los clientes conectados a ese address_id

Capa de API

Router (`api/routes/eta_router.py`)

@router.get("/eta/{address_id}")
async def get_eta(address_id: int):
    # Llama al use case
    result = await obtener_eta_use_case.execute(address_id)
    return result

@router.websocket("/ws/{address_id}")
async def websocket_endpoint(address_id: int, ws: WebSocket):
    await ws_manager.connect(address_id, ws)
    # Mantiene conexión abierta

Base de Datos

Tablas clave del Módulo B

-- Estado actual del camión
CREATE TABLE truck_status (
    route_id TEXT PRIMARY KEY,
    current_position_id INTEGER,  -- FK a puntos_ruta.id
    last_update TEXT,
    status TEXT  -- EN_RUTA | AVERIADA | RETRASADA
);

-- Definición de rutas
CREATE TABLE rutas (
    id TEXT PRIMARY KEY,     -- ej: "RUTA-01"
    nombre TEXT,
    turno TEXT               -- "matutino" | "nocturno"
);

-- Waypoints de cada ruta
CREATE TABLE puntos_ruta (
    id INTEGER PRIMARY KEY,
    ruta_id TEXT,
    orden INTEGER,           -- 1, 2, 3...
    nombre TEXT,
    lat REAL,
    lng REAL,
    tiempo_estimado_min INTEGER
);

-- Log de notificaciones enviadas
CREATE TABLE notificaciones (
    id INTEGER PRIMARY KEY,
    tipo TEXT,               -- "aproximandose" | "falla_mecanica"
    ruta_id TEXT,
    address_id INTEGER,
    mensaje TEXT,
    eta_minutos INTEGER,
    creada_en TEXT
);

Tablas compartidas con Persona A

-- Domicilios validados (Persona A los crea)
CREATE TABLE addresses (
    id INTEGER PRIMARY KEY,
    user_id INTEGER,
    alias TEXT,
    lat REAL,
    lng REAL,
    route_id TEXT,           -- Persona B lo usa para filtrar
    created_at TEXT
);

-- Preferencias de notificaciones (Persona A las gestiona)
CREATE TABLE notification_preferences (
    user_id INTEGER PRIMARY KEY,
    notify_proximity BOOLEAN,
    notify_breakdown BOOLEAN,
    notify_delay BOOLEAN,
    notify_route_start BOOLEAN
);

Flujo completo: desde arranque hasta notificación

1. POST /admin/route/RUTA-01/start
   └─> simulador.arrancar()
       └─> APScheduler inicia ticks cada 10s

2. Cada tick (10s)
   └─> truck_status.current_position_id += 1
       └─> para cada address en RUTA-01:
           └─> calcular ETA
               └─> si ETA <= 10 min:
                   └─> consultar notification_preferences
                       └─> si notify_proximity == True:
                           └─> formatear mensaje con template
                               └─> ws_manager.broadcast_zona(address_id, payload)

3. Flutter escucha en WS /ws/{address_id}
   └─> recibe JSON {"tipo": "aproximandose", "eta_minutos": 8}
       └─> muestra alerta push local

Variables de entorno (`.env`)

# Simulador
SIM_TICK_SECONDS=10              # Cada cuántos segundos avanza el camión
SIM_ETA_ALERT_MINUTES=10         # Umbral para notificar proximidad
SIM_SPEED_MULTIPLIER=1.0         # Acelerar simulación (2.0 = doble velocidad)

# Base de datos
DATABASE_PATH=./basura.db        # Ruta del archivo SQLite

# Servidor
HOST=0.0.0.0
PORT=8000
RELOAD=True

Testing local

# 1. Arrancar servidor
uvicorn app.main:app --reload

# 2. Verificar health
curl http://localhost:8000/health

# 3. Arrancar simulador
curl -X POST http://localhost:8000/admin/route/RUTA-01/start

# 4. Verificar estado del camión
curl http://localhost:8000/admin/route/RUTA-01/status

# 5. Consultar ETA de un domicilio
curl http://localhost:8000/eta/1

# 6. Forzar avería (para probar notificaciones)
curl -X POST http://localhost:8000/alerts/breakdown \
  -H "Content-Type: application/json" \
  -d '{"route_id": "RUTA-01"}'

Siguiente paso

Lee cómo Flutter consume estos endpoints:

Arquitectura del Backend (Módulo B)

Clean Architecture aplicada