onlinces/hackathon-acapulquitos-boys-c201275836cb46a5a486451f9caa83f5
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Añadir 01-backend_arquitectura.md

hack_23031391_8ff9d8
2026-05-23 02:57:55 +00:00
parent cd9b9226d9
commit 219a615cd9
1 changed files with 336 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

336
01-backend_arquitectura.md.-.md Normal file

@@ -0,0 +1,336 @@
# Arquitectura del Backend (Módulo B)
## Clean Architecture aplicada
```
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`)
```python
@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
```python
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:**
```python
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:
1. Posición actual del camión (`truck_status.current_position_id`)
2. Posición del domicilio del usuario (`addresses.route_id`)
3. Distancia entre puntos
**Retorna:**
```python
{
"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:**
```python
# 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`)
```python
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`)
```python
@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
```sql
-- 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
```sql
-- 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`)
```bash
# 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
```bash
# 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:
- [Contratos de API](../api/01-endpoints.md)
- [Integración WebSocket](../api/02-websocket.md)